影刀RPA店群自动化架构实战：Python协同多店铺类型差异化管理与动态流程适配
同样是上货，品牌旗舰店要走审核流程，个人店可以直接发布。
店群矩阵自动化突破运营极限！
用同一套自动化脚本覆盖所有店铺类型，就像用一把钥匙开所有的锁——早晚会撬坏几把。
店群运营到一定规模后，店铺类型会自然分化。 拼多多有品牌店、专卖店、个人店、海外购店铺；TEMU有本地仓店铺和跨境店铺；TikTok Shop有本土店和跨境店。每种店铺类型的后台界面、操作流程、平台规则都不尽相同。
早期我们用一套通用流程打天下，结果就是品牌店因为跳过了必填的资质字段而上架失败，个人店却在等待一个永远弹不出来的审核对话框直到超时。 
后来我们设计了一套店铺类型感知的动态流程适配系统，让同一个业务目标（如上架商品）能根据店铺的实际类型和配置，自动选择最合适的执行路径。
temu店群自动化报活动案例
一、店铺画像：不只是ID和名称
要让系统理解店铺的差异，首先需要为每个店铺建立丰富的“画像”。
画像不仅包含基础信息，还包含影响自动化操作的结构化特征。
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional
class ShopKind(Enum):
BRAND = "brand" # 品牌旗舰店
FRANCHISE = "franchise" # 专卖店
INDIVIDUAL = "individual" # 个人店
OVERSEAS = "overseas" # 海外购
CROSS_BORDER = "cross_border" # 跨境店铺
class VerificationLevel(Enum):
HIGH = "high" # 需要严格资质审核
MEDIUM = "medium" # 部分字段需要验证
LOW = "low" # 几乎无需额外验证
@dataclass
class ShopProfile:
shop_id: str
platform: str
kind: ShopKind
verification_level: VerificationLevel
required_fields: List[str] = field(default_factory=list)
optional_fields: List[str] = field(default_factory=list)
upload_strategy: str = "direct" # direct / draft / review
max_images_per_product: int = 10
supported_payment_methods: List[str] = field(default_factory=list)
has_warehouse: bool = False
auto_approve_price_changes: bool = False
custom_features: Dict[str, bool] = field(default_factory=dict)
known_quirks: List[str] = field(default_factory=list)
```
`known_quirks` 这个字段记录了我们在长期运营中发现的店铺特异性行为。
比如某品牌店的商品编辑页面会多一个“品牌授权编号”必填字段，而平台官方文档里根本没有提到。
**很多团队最开始都会忽略这些细节，等到流程在个别店铺反复失败时才回头补。**
–––
## 二、流程变体与动态路由
同一类业务操作（如上架商品），针对不同店铺类型可能有不同的流程变体。
```yaml
flow_family: product_upload
variants:
– name: product_upload_brand
– applicable_to:
– kind: [brand, franchise]
– verification_level: [high, medium]
– modules:
– – pdd/login
– – pdd/navigate_to_product_create
– – pdd/fill_brand_authorization # 品牌授权信息
– – pdd/upload_qualification_certs # 上传资质证书
– – pdd/fill_product_info
– – pdd/upload_images
– – pdd/submit_for_review # 提交审核而非直接发布
– name: product_upload_individual
– applicable_to:
– kind: [individual]
– modules:
– – pdd/login
– – pdd/navigate_to_product_create
– – pdd/fill_product_info
– – pdd/upload_images
– – pdd/submit_direct # 直接上架，无需审核
– name: product_upload_default
– applicable_to: {} # 兜底流程
– modules:
– – pdd/login
– – pdd/navigate_to_product_create
– – pdd/fill_product_info
– – pdd/upload_images
– – pdd/submit_direct
– ```
Python端的路由引擎在任务创建时，根据店铺画像自动匹配最合适的流程变体。
```python
class FlowRouter:
def __init__(self, flow_registry, shop_profile_service):
self.registry = flow_registry
self.profiles = shop_profile_service
async def resolve_flow(self, shop_id: str, flow_family: str) –> str:
profile = await self.profiles.get_profile(shop_id)
variants = self.registry.get_variants(flow_family)
best_match = None
best_score = –1
for variant in variants:
if self._matches(profile, variant.get("applicable_to", {})):
score = self._specificity(variant.get("applicable_to", {}))
if score > best_score:
best_score = score
best_match = variant
if best_match:
return best_match["name"]
# 使用兜底流程
default = self.registry.get_default(flow_family)
if default:
return default["name"]
raise FlowNotFoundError(f"No flow found for {flow_family} on shop {shop_id}")
def _matches(self, profile: ShopProfile, conditions: dict) –> bool:
if conditions.get("kind") and profile.kind.value not in conditions["kind"]:
return False
if conditions.get("verification_level") and profile.verification_level.value not in conditions["verification_level"]:
return False
if conditions.get("custom_features"):
for feature, required in conditions["custom_features"].items():
if profile.custom_features.get(feature) != required:
return False
return True
def _specificity(self, conditions: dict) –> int:
return len(conditions)
```
这样，当系统为品牌店创建上货任务时，自动选用包含资质审核步骤的流程变体；为个人店铺自动选用直接发布的流程变体。
运营无需手动为每个店铺选择不同流程。
–––
## 三、动态步骤注入与跳过
有些差异不在整个流程层面，而是某个具体步骤的有无。
比如“上传品牌授权书”这个步骤，只在品牌店需要，个人店应完全跳过。
我们在指令序列中增加了条件步骤的支持。
```json
{
"steps": [
{"action": "navigate", "url": "/product/create"},
{
"action": "conditional",
"condition": "{{ shop.verification_level == 'high' }}",
"steps": [
{"action": "upload_file", "locator": "#auth-doc", "value_from": "product.auth_doc"},
{"action": "type_text", "locator": "#brand-code", "value_from": "shop.brand_code"}
]
},
{"action": "type_text", "locator": "#title-input", "value_from": "product.title"}
]
}
```
Python执行引擎在遇到 `conditional` 步骤时，先评估条件表达式，如果为真则展开子步骤，否则跳过。
```python
class ConditionalStepExecutor:
def __init__(self, expression_engine):
self.engine = expression_engine
async def execute(self, step: dict, context: dict):
condition = step.get("condition", "true")
result = self.engine.evaluate(condition, context)
if result:
for sub_step in step.get("steps", []):
await self.execute_sub_step(sub_step, context)
```
条件表达式支持简单的逻辑运算，并从上下文中获取店铺画像字段进行判断。
这让流程具备了“自适应”能力，而无需为每种店铺类型维护完全独立的流程文件。
–––
## 四、店铺特性知识库的持续更新
平台在不断更新规则，店铺类型也在变化。
有些店铺在运营过程中获得了品牌授权，从个人店升级为品牌店；有些店铺被平台加入了新的审核要求。
我们建立了一个店铺特性变更检测与知识库更新机制：
1. 每当任务出现“字段缺失”、“审核失败”等特定错误时，自动分析错误详情，提取可能的新增必填字段
2. 2. 当多个同类型店铺重复出现相同错误模式时，自动更新该类型店铺的画像模板
3. 3. 运营也可以通过管理后台手动更新某个店铺的 `known_quirks`
```python
class ShopQuirkDetector:
def __init__(self, db, profile_service):
self.db = db
self.profiles = profile_service
async def analyze_failure_patterns(self):
# 查询最近24小时内特定类型错误的聚集
rows = await self.db.fetch("""
SELECT shop_id, error_type, error_detail, COUNT(*) as cnt
FROM task_failures
WHERE created_at > NOW() – INTERVAL '24 hours'
AND error_type IN ('missing_field', 'verification_failed', 'unexpected_element')
GROUP BY shop_id, error_type, error_detail
HAVING COUNT(*) >= 3
""")
for row in rows:
shop = await self.profiles.get_profile(row['shop_id'])
quirk = self._infer_quirk(row['error_detail'])
if quirk and quirk not in shop.known_quirks:
shop.known_quirks.append(quirk)
await self.profiles.update_profile(shop)
logger.info(f"Added quirk '{quirk}' to shop {row['shop_id']}")
def _infer_quirk(self, error_detail: str) –> Optional[str]:
# 从错误详情中推断店铺特性
if "brand_authorization_number" in error_detail:
return "requires_brand_auth_number"
if "warehouse_region" in error_detail:
return "requires_warehouse_region"
return None
```
这套机制让我们从“被动应对”转为“主动学习”，系统越跑越了解每家店铺的特性。
–––
## 五、跨平台类型映射
不同平台的店铺分类体系不同，但业务含义有对应关系。
我们维护了一套跨平台店铺类型映射表，让规则引擎和路由系统能跨平台理解店铺特征。
```python
CROSS_PLATFORM_TYPE_MAP = {
("pdd", "brand"): {"temu": "brand_store", "tiktok": "official_shop"},
("pdd", "individual"): {"temu": "individual_seller", "tiktok": "individual_shop"},
("temu", "local_warehouse"): {"pdd": "brand", "tiktok": "local_shop"},
("temu", "cross_border"): {"tiktok": "cross_border_shop"},
}
```
当某个店铺在拼多多上是品牌店，系统在规划其TEMU上货流程时，能自动使用对应的品牌店铺流程变体。
–––
## 六、店铺配置模板的继承与覆盖
店铺画像中的很多字段来自“类型模板”，而不是逐店配置。
我们实现了模板继承机制：个人店铺默认继承“个人店模板”的所有字段，但可以覆盖特定字段。
```yaml
template: individual_base
overrides:
max_images_per_product: 15 # 这家店铺购买了额外的图片额度
known_quirks:
– requires_extra_category_field # 这家店因历史原因多了一个类目字段
– ```
渲染后的完整画像由模板 + 覆盖合并而成，确保了类型共性的一致性和个体差异的灵活性。
–––
## 七、监控与度量
– 各店铺类型的流程匹配正确率
– – 条件分支命中率（哪些条件最常触发）
– – 店铺特性自动发现数量
– – 因店铺特性不匹配导致的任务失败率
– – 新店铺自动归类准确率
当某类型的店铺失败率突然升高时，可能是平台对该类型店铺做了策略调整，需要更新模板和流程变体。
–––
## 八、踩坑记录
**模板粒度不够导致“例外”泛滥。**
早期我们只分了“品牌店”和“非品牌店”两种模板，结果非品牌店的个体差异巨大，流程频繁出错。
后来逐步细化出了专卖店、个人店、跨境店等子类型，并在模板中增加了可选字段的显式标注。
**条件表达式过于复杂。**
起初条件表达式支持嵌套逻辑和函数调用，运营配置的门槛太高，经常写错。
我们简化了表达式语法，只保留基本的比较和AND/OR组合，并提供了可视化的条件编辑器。
**画像过时。**
店铺在平台上升级了类型（如从个人店升级为品牌店），但系统中的画像没同步，导致流程持续走错分支。
我们增加了定期的店铺信息同步任务，从平台API抓取最新的店铺信息并与画像比对，发现不一致及时告警。
–––
## 九、写在最后
店群不是一堆完全相同的店铺的集合，而是一个包含多种类型、不同特性、持续变化的有机体。
用一套僵化的脚本去覆盖所有差异，短期省钱，长期填坑。
通过店铺画像、流程路由、条件步骤和知识库的持续更新，我们让自动化系统具备了识别和适应差异的能力。
> 自动化的真正智能，不在于能做多少事，而在于能辨别不同的情境，并做出正确的选择。
–––
*作者：林焱*
