From bdc7bd5069eb1235f75b6b4992a09f8732a0947e Mon Sep 17 00:00:00 2001 From: strepsiades Date: Mon, 30 Mar 2026 14:51:01 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A7=84=E8=8C=83=E4=BA=86update=20data?= =?UTF-8?q?=E7=9A=84=E6=9E=84=E5=BB=BA=E6=96=B9=E5=BC=8F=EF=BC=8C=E7=BB=9F?= =?UTF-8?q?=E4=B8=80=E4=BD=BF=E7=94=A8`build=5Fupdate=5Frequest=5Fdata`?= =?UTF-8?q?=E6=96=B9=E6=B3=95=E6=9D=A5=E5=A4=84=E7=90=86=E6=9B=B4=E6=96=B0?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9A=84=E6=9E=84=E5=BB=BA=E5=92=8C=E8=BF=87?= =?UTF-8?q?=E6=BB=A4=E9=80=BB=E8=BE=91=E3=80=82=E8=BF=99=E4=B8=AA=E6=96=B9?= =?UTF-8?q?=E6=B3=95=E4=BC=9A=E6=A0=B9=E6=8D=AE=E6=8F=90=E4=BE=9B=E7=9A=84?= =?UTF-8?q?=E5=8E=9F=E5=A7=8B=E6=95=B0=E6=8D=AE=E3=80=81=E6=9B=B4=E6=96=B0?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=92=8C=E5=AF=B9=E5=BA=94=E7=9A=84schema?= =?UTF-8?q?=E6=9D=A5=E7=94=9F=E6=88=90=E6=9C=80=E7=BB=88=E7=9A=84=E6=9B=B4?= =?UTF-8?q?=E6=96=B0=E8=AF=B7=E6=B1=82=E6=95=B0=E6=8D=AE=EF=BC=8C=E5=90=8C?= =?UTF-8?q?=E6=97=B6=E4=BC=9A=E8=87=AA=E5=8A=A8=E6=8E=92=E9=99=A4=E6=8E=89?= =?UTF-8?q?=E6=9C=AA=E5=8F=98=E5=8C=96=E7=9A=84=E5=AD=97=E6=AE=B5=E5=92=8C?= =?UTF-8?q?=E7=A9=BA=E5=80=BC=E5=AD=97=E6=AE=B5=EF=BC=8C=E4=BB=8E=E8=80=8C?= =?UTF-8?q?=E7=A1=AE=E4=BF=9D=E5=8F=AA=E5=8F=91=E9=80=81=E5=BF=85=E8=A6=81?= =?UTF-8?q?=E7=9A=84=E6=9B=B4=E6=96=B0=E4=BF=A1=E6=81=AF=E5=88=B0=E5=90=8E?= =?UTF-8?q?=E7=AB=AF=E6=8E=A5=E5=8F=A3=E3=80=82=E8=BF=99=E7=A7=8D=E6=96=B9?= =?UTF-8?q?=E5=BC=8F=E4=B8=8D=E4=BB=85=E7=AE=80=E5=8C=96=E4=BA=86=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E9=80=BB=E8=BE=91=EF=BC=8C=E8=BF=98=E6=8F=90=E9=AB=98?= =?UTF-8?q?=E4=BA=86=E4=BB=A3=E7=A0=81=E7=9A=84=E5=8F=AF=E7=BB=B4=E6=8A=A4?= =?UTF-8?q?=E6=80=A7=E5=92=8C=E4=B8=80=E8=87=B4=E6=80=A7=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/datasource_guide.md | 4 +- sync_state_machine/config/builder.py | 4 + sync_state_machine/config/strategy_config.py | 5 + sync_state_machine/datasource/api/handler.py | 156 +++++++++--------- .../domain/construction/api_handler.py | 4 +- .../construction_task_detail/api_handler.py | 6 +- .../domain/contract/api_handler.py | 2 +- .../domain/contract_change/api_handler.py | 2 +- .../domain/contract_settlement/api_handler.py | 4 +- .../contract_settlement_detail/api_handler.py | 6 +- sync_state_machine/domain/lar/api_handler.py | 2 +- .../domain/lar_change/api_handler.py | 2 +- .../domain/material/api_handler.py | 4 +- .../domain/material_detail/api_handler.py | 6 +- .../domain/personnel/api_handler.py | 2 +- .../domain/personnel_detail/api_handler.py | 6 +- .../domain/preparation/api_handler.py | 2 +- .../domain/production/api_handler.py | 2 +- .../domain/project/api_handler.py | 6 +- .../domain/units/api_handler.py | 2 +- sync_state_machine/pipeline/factory.py | 1 + .../pipeline/full_sync_pipeline.py | 14 ++ sync_state_machine/sync_system/strategy.py | 5 + .../sync_system/strategy_ops/bind_ops.py | 119 ++++++++++--- .../test_strategy_business_points_migrated.py | 24 +++ ..._base_api_handler_update_schema_helpers.py | 94 +++++++++++ ...st_full_sync_pipeline_downstream_reload.py | 54 +++++- tests/unit/test_strategy_runtime_config.py | 7 +- 28 files changed, 416 insertions(+), 129 deletions(-) create mode 100644 tests/unit/test_base_api_handler_update_schema_helpers.py diff --git a/docs/datasource_guide.md b/docs/datasource_guide.md index 16078a7..22dabc1 100644 --- a/docs/datasource_guide.md +++ b/docs/datasource_guide.md @@ -131,8 +131,8 @@ API 数据源中,校验分为三层: - 例如 `ContractCreate.model_validate(...)`、`ContractUpdate.model_validate(...)` - load 响应也会在解析后转为领域 schema 2. 通用更新过滤与差异识别 - - `_filter_update_data`:仅在 schema 相关字段有变化时提交更新 - - `_get_schema_diff`:用于差异字段识别与调试输出 + - `build_update_request_data`:update 入口,仅在 schema 相关字段有变化时提交更新 + - `get_update_schema_diff`:update 入口,用于差异字段识别与调试输出 3. 轮询接口输入校验 - `PushIdsSchema` 校验 push_ids 格式后再请求 `/push/result` diff --git a/sync_state_machine/config/builder.py b/sync_state_machine/config/builder.py index d02eb8d..7a220f1 100644 --- a/sync_state_machine/config/builder.py +++ b/sync_state_machine/config/builder.py @@ -76,10 +76,14 @@ def _apply_strategy_overrides( runtime_cfg.config = ConfigPresets.get_preset(str(preset)) skip_sync = override.pop("skip_sync", None) + skip_post_check = override.pop("skip_post_check", None) if skip_sync is not None: runtime_cfg.skip_sync = bool(skip_sync) + if skip_post_check is not None: + runtime_cfg.skip_post_check = bool(skip_post_check) + if "config" in override and isinstance(override["config"], dict): runtime_cfg.config = runtime_cfg.config.model_validate( _deep_merge(runtime_cfg.config.model_dump(), override.pop("config")) diff --git a/sync_state_machine/config/strategy_config.py b/sync_state_machine/config/strategy_config.py index fed4fa1..35cf5d3 100644 --- a/sync_state_machine/config/strategy_config.py +++ b/sync_state_machine/config/strategy_config.py @@ -59,6 +59,10 @@ class StrategyConfig(BaseModel): # schema diff / validate 时忽略的顶层字段。 compare_ignore_fields: List[str] = Field(default_factory=list) + # 当 auto-bind 命中“多对一”候选时,允许稳定选择一条进行绑定,剩余候选按 orphan/create 流程继续处理。 + # 默认关闭,避免在一般场景下引入隐式匹配。 + allow_many_to_one_auto_bind: bool = False + class StrategyRuntimeConfig(BaseModel): """Pipeline 层策略运行配置(按 node_type)。""" @@ -67,6 +71,7 @@ class StrategyRuntimeConfig(BaseModel): node_type: str skip_sync: bool = False + skip_post_check: bool = False config: StrategyConfig = Field(default_factory=StrategyConfig) diff --git a/sync_state_machine/datasource/api/handler.py b/sync_state_machine/datasource/api/handler.py index 027fae9..cbdd394 100644 --- a/sync_state_machine/datasource/api/handler.py +++ b/sync_state_machine/datasource/api/handler.py @@ -17,7 +17,7 @@ import asyncio from abc import abstractmethod from copy import deepcopy from typing import ClassVar, Dict, List, Any, Optional, TYPE_CHECKING, TypeVar -from pydantic import BaseModel +from pydantic import BaseModel, ValidationError from ...common.type_safety import get_pydantic_model_fields from ..handler import BaseNodeHandler @@ -51,9 +51,6 @@ class BaseApiHandler(BaseNodeHandler[T]): # post-check 将只比较这些 schema 覆盖的字段,过滤后端只读的派生字段。 self.update_schemas: List[type] = [] - def set_collection(self, collection: 'DataCollection') -> None: - self._collection = collection - def get_update_fields(self, *, exclude: frozenset = frozenset({'id'})) -> List[str]: """从 update_schemas 中提取所有可写字段名,排除标识键(默认排除 id)。""" fields: set = set() @@ -167,10 +164,6 @@ class BaseApiHandler(BaseNodeHandler[T]): payload["steps"] = self._get_request_steps(node) if node is not None else [] return payload - def _build_full_update_data(self, new_data: Dict[str, Any], schema: Any) -> Dict[str, Any]: - validated = schema.model_validate(new_data) - return validated.model_dump(exclude_none=True) - def _mark_approval_snapshot_synced(self, node: 'SyncNode') -> None: snapshot = self._get_active_approval_snapshot(node) if snapshot: @@ -255,8 +248,53 @@ class BaseApiHandler(BaseNodeHandler[T]): return None except Exception as e: return str(e) + + def _get_update_schema_payload( + self, + schema: Any, + data: Dict[str, Any], + *, + require_valid: bool, + ) -> Dict[str, Any] | None: + """Update 内部实现:提取指定 schema 下的 payload。 + + 这里明确只服务“update schema”,不负责 response model 一致性校验。 + 这个方法只给 BaseApiHandler 的 update 入口方法复用,不作为 handler 入口。 + + 处理顺序: + 1. 先尝试用 update schema 校验,成功则直接使用校验后的 model_dump。 + 2. 如果校验失败且 require_valid=False,则退化为只抽取 schema 定义的字段。 + + 之所以允许退化,是因为 update 比较时两边快照常常不完整: + 某些字段在旧快照缺失,但新数据已经有值。此时应该继续比较 schema + 相关字段,而不是把它误判成“无差异”。 + + 空字符串和 None 会被统一过滤掉,因为当前 update handler 并不把它们 + 当成“清空字段”的显式语义,而是当作“不参与本次更新”。 + """ + try: + validated = schema.model_validate(data) + payload = validated.model_dump(exclude_unset=True) + except ValidationError: + if require_valid: + return None + + schema_fields = get_pydantic_model_fields(schema) + if not schema_fields: + return {} + payload = { + field_name: data.get(field_name) + for field_name in schema_fields.keys() + if field_name in data + } + + return { + field_name: value + for field_name, value in payload.items() + if value not in (None, "") + } - def _get_schema_diff( + def get_update_schema_diff( self, schema, data: Dict[str, Any], @@ -264,8 +302,7 @@ class BaseApiHandler(BaseNodeHandler[T]): node_id: Optional[str] = None, force_full_schema: bool = False, ) -> Dict[str, Any] | None: - """ - 获取数据在指定 schema 下的差异字段 + """Update 入口:获取指定 update schema 下的差异字段。 Args: schema: Pydantic schema 类 @@ -276,46 +313,37 @@ class BaseApiHandler(BaseNodeHandler[T]): Returns: Dict[str, Any] | None: 有差异返回差异字段字典,无差异或字段不全返回 None """ - from pydantic import ValidationError - - try: - new_model = schema.model_validate(data) - origin_model = schema.model_validate(origin_data) - - new_dict = new_model.model_dump(exclude_unset=True) - origin_dict = origin_model.model_dump(exclude_unset=True) - - if new_dict == origin_dict: - if force_full_schema: - return new_dict - return None - - # 返回有差异的字段 - diff_fields = {k: v for k, v in new_dict.items() if origin_dict.get(k) != v} - - if diff_fields: - # 打印差异详情 - node_info = f"[{node_id}]" if node_id else "" - print(f" [DIFF] {self.node_type}{node_info} schema={schema.__name__}") - for k, v in diff_fields.items(): - orig_v = origin_dict.get(k) - print(f" • {k}: {orig_v!r} → {v!r}") - - return diff_fields if diff_fields else None - - except ValidationError: - # schema 字段不全,视为无差异 + # 当前待发送的数据必须能构成合法 update payload;允许旧快照不完整。 + new_dict = self._get_update_schema_payload(schema, data, require_valid=True) + origin_dict = self._get_update_schema_payload(schema, origin_data, require_valid=False) + + if new_dict is None or origin_dict is None: return None + + if new_dict == origin_dict: + if force_full_schema: + return new_dict + return None + + diff_fields = {k: v for k, v in new_dict.items() if origin_dict.get(k) != v} + + if diff_fields: + node_info = f"[{node_id}]" if node_id else "" + print(f" [DIFF] {self.node_type}{node_info} schema={schema.__name__}") + for k, v in diff_fields.items(): + orig_v = origin_dict.get(k) + print(f" • {k}: {orig_v!r} → {v!r}") + + return diff_fields if diff_fields else None - def _filter_update_data( + def build_update_request_data( self, new_data: Dict[str, Any], original_data: Dict[str, Any], schema: Optional[Any] = None, force_full_schema: bool = False, ) -> Dict[str, Any]: - """ - 过滤出需要更新的字段 + """Update 入口:构造最终要发送给更新接口的请求数据。 规则: 1. 提取 schema 定义的所有字段(包括必填) @@ -342,29 +370,10 @@ class BaseApiHandler(BaseNodeHandler[T]): update_data[key] = new_value return update_data - # 1. 提取 schema 定义的所有字段 - schema_fields = set(schema_fields.keys()) - schema_data = {} - has_change = False - - for key in schema_fields: - new_value = new_data.get(key) - - # 排除空值 - if new_value == "" or new_value is None: - continue - - schema_data[key] = new_value - - # 检查是否变化 - original_value = original_data.get(key) - if new_value != original_value: - # 记录具体差异以便排查 - # print(f" [DEBUG] Field '{key}' changed: '{original_value}' -> '{new_value}'") - has_change = True - - # 2. 只有存在变化时才返回数据 - if has_change or force_full_schema: + schema_data = self._get_update_schema_payload(schema, new_data, require_valid=True) or {} + origin_schema_data = self._get_update_schema_payload(schema, original_data, require_valid=False) or {} + + if force_full_schema or schema_data != origin_schema_data: return schema_data return {} @@ -374,16 +383,12 @@ class BaseApiHandler(BaseNodeHandler[T]): new_data: Dict[str, Any], origin_data: Dict[str, Any], ) -> List[str]: - schema_fields = get_pydantic_model_fields(schema) - if not schema_fields: - return [] - changed: List[str] = [] - for field_name in schema_fields.keys(): - if new_data.get(field_name) != origin_data.get(field_name): - changed.append(field_name) - return changed + new_payload = self._get_update_schema_payload(schema, new_data, require_valid=True) or {} + origin_payload = self._get_update_schema_payload(schema, origin_data, require_valid=False) or {} + field_names = sorted(set(new_payload) | set(origin_payload)) + return [field_name for field_name in field_names if new_payload.get(field_name) != origin_payload.get(field_name)] - def _build_update_schema_status( + def build_update_schema_status( self, *, schema_name: str, @@ -393,6 +398,7 @@ class BaseApiHandler(BaseNodeHandler[T]): will_update: bool, update_error: Optional[str] = None, ) -> str: + """Update 入口:生成 update schema 的状态说明文本。""" new_error = self._validate_schema(new_data, schema) origin_error = self._validate_schema(origin_data, schema) changed_fields = self._schema_changed_fields(schema, new_data, origin_data) diff --git a/sync_state_machine/domain/construction/api_handler.py b/sync_state_machine/domain/construction/api_handler.py index b5987cc..f9c99a0 100644 --- a/sync_state_machine/domain/construction/api_handler.py +++ b/sync_state_machine/domain/construction/api_handler.py @@ -190,7 +190,7 @@ class ConstructionTaskApiHandler(BaseApiHandler): base_update_error = "show_name changed but task_type != QT" schema_status_lines.append( - self._build_update_schema_status( + self.build_update_schema_status( schema_name="ConstructionUpdate", schema=ConstructionUpdate, new_data=data, @@ -238,7 +238,7 @@ class ConstructionTaskApiHandler(BaseApiHandler): plan_update_error = str(e) schema_status_lines.append( - self._build_update_schema_status( + self.build_update_schema_status( schema_name="ConstructionPlanUpdate", schema=ConstructionPlanUpdate, new_data=data, diff --git a/sync_state_machine/domain/construction_task_detail/api_handler.py b/sync_state_machine/domain/construction_task_detail/api_handler.py index d08e053..48facf4 100644 --- a/sync_state_machine/domain/construction_task_detail/api_handler.py +++ b/sync_state_machine/domain/construction_task_detail/api_handler.py @@ -124,9 +124,9 @@ class ConstructionTaskDetailApiHandler(BaseApiHandler): origin_data = node.get_origin_data() or {} - # 使用 _get_schema_diff 检查差异并打印日志 + # 使用 get_update_schema_diff 作为 update 入口检查差异并打印日志 force_steps_update = self._steps_changed(node) - diff_fields = self._get_schema_diff( + diff_fields = self.get_update_schema_diff( ConstructionDetailUpdate, data, origin_data, @@ -145,7 +145,7 @@ class ConstructionTaskDetailApiHandler(BaseApiHandler): continue # 需要完整的 schema 数据(包括必填字段) - update_data = self._filter_update_data( + update_data = self.build_update_request_data( data, origin_data, ConstructionDetailUpdate, diff --git a/sync_state_machine/domain/contract/api_handler.py b/sync_state_machine/domain/contract/api_handler.py index d98ca60..466b6fa 100644 --- a/sync_state_machine/domain/contract/api_handler.py +++ b/sync_state_machine/domain/contract/api_handler.py @@ -256,7 +256,7 @@ class ContractApiHandler(BaseApiHandler): # 过滤出需要更新的字段(排除空值、未变化的字段) origin_data = node.get_origin_data() or {} force_steps_update = self._steps_changed(node) - update_data = self._filter_update_data( + update_data = self.build_update_request_data( new_data=data, original_data=origin_data, schema=ContractUpdate, diff --git a/sync_state_machine/domain/contract_change/api_handler.py b/sync_state_machine/domain/contract_change/api_handler.py index 4ea75e5..000c279 100644 --- a/sync_state_machine/domain/contract_change/api_handler.py +++ b/sync_state_machine/domain/contract_change/api_handler.py @@ -93,7 +93,7 @@ class ContractChangeApiHandler(BaseApiHandler): origin_data = node.get_origin_data() or {} force_steps_update = self._steps_changed(node) - update_data = self._filter_update_data( + update_data = self.build_update_request_data( node_data, origin_data, PostContractChange, diff --git a/sync_state_machine/domain/contract_settlement/api_handler.py b/sync_state_machine/domain/contract_settlement/api_handler.py index 84833ab..00912fe 100644 --- a/sync_state_machine/domain/contract_settlement/api_handler.py +++ b/sync_state_machine/domain/contract_settlement/api_handler.py @@ -140,7 +140,7 @@ class ContractSettlementApiHandler(BaseApiHandler): # 1. 基础信息更新 (ContractSettlementUpdate) force_steps_update = self._steps_changed(node) - base_diff = self._get_schema_diff( + base_diff = self.get_update_schema_diff( ContractSettlementUpdate, data, origin_data, @@ -167,7 +167,7 @@ class ContractSettlementApiHandler(BaseApiHandler): error = f"API error: {e}" # 2. 计划信息更新 (ContractSettlementPlanUpdate) - plan_diff = self._get_schema_diff(ContractSettlementPlanUpdate, data, origin_data, node.node_id) + plan_diff = self.get_update_schema_diff(ContractSettlementPlanUpdate, data, origin_data, node.node_id) if not error and plan_diff: try: plan_model = ContractSettlementPlanUpdate.model_validate(data) diff --git a/sync_state_machine/domain/contract_settlement_detail/api_handler.py b/sync_state_machine/domain/contract_settlement_detail/api_handler.py index 3e54141..72ccbcf 100644 --- a/sync_state_machine/domain/contract_settlement_detail/api_handler.py +++ b/sync_state_machine/domain/contract_settlement_detail/api_handler.py @@ -122,9 +122,9 @@ class ContractSettlementDetailApiHandler(BaseApiHandler): origin_data = node.get_origin_data() or {} - # 使用 _get_schema_diff 检查差异并打印日志 + # 使用 get_update_schema_diff 作为 update 入口检查差异并打印日志 force_steps_update = self._steps_changed(node) - diff_fields = self._get_schema_diff( + diff_fields = self.get_update_schema_diff( ContractSettlementDetailUpdate, data, origin_data, @@ -143,7 +143,7 @@ class ContractSettlementDetailApiHandler(BaseApiHandler): continue # 需要完整的 schema 数据(包括必填字段) - update_data = self._filter_update_data( + update_data = self.build_update_request_data( data, origin_data, ContractSettlementDetailUpdate, diff --git a/sync_state_machine/domain/lar/api_handler.py b/sync_state_machine/domain/lar/api_handler.py index 32d80f3..027320e 100644 --- a/sync_state_machine/domain/lar/api_handler.py +++ b/sync_state_machine/domain/lar/api_handler.py @@ -118,7 +118,7 @@ class LarApiHandler(BaseApiHandler): for schema, api_func in update_configs: # 获取过滤后的更新数据 - update_data = self._filter_update_data( + update_data = self.build_update_request_data( node_data, origin_data, schema, diff --git a/sync_state_machine/domain/lar_change/api_handler.py b/sync_state_machine/domain/lar_change/api_handler.py index 7f1437d..ff4d175 100644 --- a/sync_state_machine/domain/lar_change/api_handler.py +++ b/sync_state_machine/domain/lar_change/api_handler.py @@ -104,7 +104,7 @@ class LarChangeApiHandler(BaseApiHandler): continue origin_data = node.get_origin_data() or {} force_steps_update = self._steps_changed(node) - update_data = self._filter_update_data( + update_data = self.build_update_request_data( node_data, origin_data, LarChangeUpdateSchema, diff --git a/sync_state_machine/domain/material/api_handler.py b/sync_state_machine/domain/material/api_handler.py index 132608a..a8d6a0e 100644 --- a/sync_state_machine/domain/material/api_handler.py +++ b/sync_state_machine/domain/material/api_handler.py @@ -106,7 +106,7 @@ class MaterialApiHandler(BaseApiHandler): # 1. 基础信息更新 (MaterialUpdate) force_steps_update = self._steps_changed(node) - base_diff = self._get_schema_diff( + base_diff = self.get_update_schema_diff( MaterialUpdate, data, origin_data, @@ -131,7 +131,7 @@ class MaterialApiHandler(BaseApiHandler): error = f"{ERROR_VALIDATION_FAILED}: {e}" # 2. 计划信息更新 (MaterialPlanUpdate) - plan_diff = self._get_schema_diff(MaterialPlanUpdate, data, origin_data, node.node_id) + plan_diff = self.get_update_schema_diff(MaterialPlanUpdate, data, origin_data, node.node_id) if not error and plan_diff: try: plan_model = MaterialPlanUpdate.model_validate(data) diff --git a/sync_state_machine/domain/material_detail/api_handler.py b/sync_state_machine/domain/material_detail/api_handler.py index 3575e23..157f381 100644 --- a/sync_state_machine/domain/material_detail/api_handler.py +++ b/sync_state_machine/domain/material_detail/api_handler.py @@ -144,9 +144,9 @@ class MaterialDetailApiHandler(BaseApiHandler): origin_data = node.get_origin_data() or {} - # 使用 _get_schema_diff 检查差异并打印日志 + # 使用 get_update_schema_diff 作为 update 入口检查差异并打印日志 force_steps_update = self._steps_changed(node) - diff_fields = self._get_schema_diff( + diff_fields = self.get_update_schema_diff( MaterialDetailUpdate, data, origin_data, @@ -165,7 +165,7 @@ class MaterialDetailApiHandler(BaseApiHandler): continue # 需要完整的 schema 数据(包括必填字段) - update_data = self._filter_update_data( + update_data = self.build_update_request_data( data, origin_data, MaterialDetailUpdate, diff --git a/sync_state_machine/domain/personnel/api_handler.py b/sync_state_machine/domain/personnel/api_handler.py index f95ea5e..790f607 100644 --- a/sync_state_machine/domain/personnel/api_handler.py +++ b/sync_state_machine/domain/personnel/api_handler.py @@ -121,7 +121,7 @@ class PersonnelApiHandler(BaseApiHandler): # 1. 基础信息更新 (PowerUpdate) force_steps_update = self._steps_changed(node) - base_diff = self._get_schema_diff( + base_diff = self.get_update_schema_diff( PowerUpdate, data, origin_data, diff --git a/sync_state_machine/domain/personnel_detail/api_handler.py b/sync_state_machine/domain/personnel_detail/api_handler.py index 6666b01..06fc6cc 100644 --- a/sync_state_machine/domain/personnel_detail/api_handler.py +++ b/sync_state_machine/domain/personnel_detail/api_handler.py @@ -111,9 +111,9 @@ class PersonnelDetailApiHandler(BaseApiHandler): origin_data = node.get_origin_data() or {} - # 使用 _get_schema_diff 检查差异并打印日志 + # 使用 get_update_schema_diff 作为 update 入口检查差异并打印日志 force_steps_update = self._steps_changed(node) - diff_fields = self._get_schema_diff( + diff_fields = self.get_update_schema_diff( PowerDetailUpdate, data, origin_data, @@ -132,7 +132,7 @@ class PersonnelDetailApiHandler(BaseApiHandler): continue # 需要完整的 schema 数据(包括必填字段) - update_data = self._filter_update_data( + update_data = self.build_update_request_data( data, origin_data, PowerDetailUpdate, diff --git a/sync_state_machine/domain/preparation/api_handler.py b/sync_state_machine/domain/preparation/api_handler.py index 0cd0bea..13e7001 100644 --- a/sync_state_machine/domain/preparation/api_handler.py +++ b/sync_state_machine/domain/preparation/api_handler.py @@ -94,7 +94,7 @@ class PreparationApiHandler(BaseApiHandler): origin_data = node.get_origin_data() or {} force_steps_update = self._steps_changed(node) - update_data = self._filter_update_data( + update_data = self.build_update_request_data( node_data, origin_data, PostPreparation, diff --git a/sync_state_machine/domain/production/api_handler.py b/sync_state_machine/domain/production/api_handler.py index 1024839..8e03307 100644 --- a/sync_state_machine/domain/production/api_handler.py +++ b/sync_state_machine/domain/production/api_handler.py @@ -99,7 +99,7 @@ class ProductionApiHandler(BaseApiHandler): # 获取过滤后的更新数据 force_steps_update = self._steps_changed(node) - update_data = self._filter_update_data( + update_data = self.build_update_request_data( node_data, origin_data, PostProduction, diff --git a/sync_state_machine/domain/project/api_handler.py b/sync_state_machine/domain/project/api_handler.py index 67d138a..141bc3e 100644 --- a/sync_state_machine/domain/project/api_handler.py +++ b/sync_state_machine/domain/project/api_handler.py @@ -234,7 +234,7 @@ class ProjectApiHandler(BaseApiHandler): # 遍历所有更新 schema,检查差异并更新 for schema, api_func in update_configs: - update_data = self._get_schema_diff( + update_data = self.get_update_schema_diff( schema, data, original_data, @@ -244,9 +244,9 @@ class ProjectApiHandler(BaseApiHandler): if not update_data: continue - # 执行更新(差异已由 _get_schema_diff 打印) + # 执行更新(差异已由 get_update_schema_diff 打印) try: - full_update_data = self._filter_update_data( + full_update_data = self.build_update_request_data( data, original_data, schema, diff --git a/sync_state_machine/domain/units/api_handler.py b/sync_state_machine/domain/units/api_handler.py index bf2434a..0b63609 100644 --- a/sync_state_machine/domain/units/api_handler.py +++ b/sync_state_machine/domain/units/api_handler.py @@ -94,7 +94,7 @@ class UnitsApiHandler(BaseApiHandler): origin_data = node.get_origin_data() or {} force_steps_update = self._steps_changed(node) - update_data = self._filter_update_data( + update_data = self.build_update_request_data( node_data, origin_data, PostGeneratorUnits, diff --git a/sync_state_machine/pipeline/factory.py b/sync_state_machine/pipeline/factory.py index 01dde15..85b27b8 100644 --- a/sync_state_machine/pipeline/factory.py +++ b/sync_state_machine/pipeline/factory.py @@ -264,6 +264,7 @@ def _build_strategies( if runtime_cfg is not None: strategy.set_config(runtime_cfg.config.model_copy(deep=True)) strategy.skip_sync = bool(runtime_cfg.skip_sync) + strategy.skip_post_check = bool(runtime_cfg.skip_post_check) strategies.append(strategy) return strategies diff --git a/sync_state_machine/pipeline/full_sync_pipeline.py b/sync_state_machine/pipeline/full_sync_pipeline.py index e1f17ca..a25bffa 100644 --- a/sync_state_machine/pipeline/full_sync_pipeline.py +++ b/sync_state_machine/pipeline/full_sync_pipeline.py @@ -509,8 +509,22 @@ class FullSyncPipeline: else: self._logger.info("🔄 Post-check: reloading all node types for final consistency evaluation...") for node_type in self.node_types: + strategy = next((item for item in self.strategies if item.node_type == node_type), None) + if strategy is not None and strategy.skip_post_check: + self._logger.info("⏭️ Post-check reload skipped: node_type=%s", node_type) + continue await self._reload_node_type(node_type, reason="post-check final reload") for node_type in self.node_types: + strategy = next((item for item in self.strategies if item.node_type == node_type), None) + if strategy is not None and strategy.skip_post_check: + self._logger.info("⏭️ Post-check skipped: node_type=%s", node_type) + self._consistency_by_type[node_type] = { + "ok": True, + "checked_pairs": 0, + "mismatch_count": 0, + "skipped": True, + } + continue await self._evaluate_consistency_for_node_type(node_type) self.stats["consistency"] = { diff --git a/sync_state_machine/sync_system/strategy.py b/sync_state_machine/sync_system/strategy.py index 2537ca1..7140498 100644 --- a/sync_state_machine/sync_system/strategy.py +++ b/sync_state_machine/sync_system/strategy.py @@ -45,6 +45,7 @@ class BaseSyncStrategy(Generic[T]): skip_sync: bool - 如果为 True,则跳过该类型的 bind/create/update/sync 阶段 数据仍会被加载,但不会执行任何同步操作 适用于只需要加载数据供其他类型引用的场景(如 supplier) + skip_post_check: bool - 如果为 True,则跳过该类型最终 post-check 的 reload 与一致性校验 """ # 类变量:默认配置(子类可覆盖) default_config: StrategyConfig = StrategyConfig() @@ -52,6 +53,8 @@ class BaseSyncStrategy(Generic[T]): schema: Type[T] = None # type: ignore # 类变量:是否默认跳过同步(子类可覆盖) default_skip_sync: bool = False + # 类变量:是否默认跳过 post-check(子类可覆盖) + default_skip_post_check: bool = False def __init__( self, @@ -71,6 +74,8 @@ class BaseSyncStrategy(Generic[T]): self.config = self.default_config.model_copy(deep=True) # 是否跳过同步(可运行时修改) self.skip_sync = self.default_skip_sync + # 是否跳过 post-check(可运行时修改) + self.skip_post_check = self.default_skip_post_check # 统一由 Pipeline 注入 runtime;未注入时在执行阶段显式报错 self.sm_runtime: Optional[StateMachineRuntime] = None self._schema_diff_validator: Optional[SchemaDiffValidator] = None diff --git a/sync_state_machine/sync_system/strategy_ops/bind_ops.py b/sync_state_machine/sync_system/strategy_ops/bind_ops.py index 75a87fc..f0d2b47 100644 --- a/sync_state_machine/sync_system/strategy_ops/bind_ops.py +++ b/sync_state_machine/sync_system/strategy_ops/bind_ops.py @@ -16,6 +16,12 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +def _stable_pick_bind_pair(local_candidates: List["SyncNode"], remote_candidates: List["SyncNode"]): + local_node = sorted(local_candidates, key=lambda item: ((item.data_id or ""), item.node_id))[0] + remote_node = sorted(remote_candidates, key=lambda item: ((item.data_id or ""), item.node_id))[0] + return local_node, remote_node + + async def run_bind(strategy, **kwargs): logger.info(f"[{strategy.node_type}] Starting Binding Pipeline...") @@ -519,29 +525,78 @@ async def phase_auto_binding(strategy, local_nodes: List["SyncNode"], remote_nod } ) - if (len(l_remain) > 1 and len(r_remain) >= 1) or (len(l_remain) == 1 and len(r_remain) > 1): - for n in l_remain: - if n.binding_status == BindingStatus.MISSING: - pending_auto_bind_updates.append( - { - "node": n, - "outcome": "AMBIGUOUS", - "create_enabled": None, - "bind_remote_node_id": None, - } - ) + many_local_one_remote = len(l_remain) > 1 and len(r_remain) == 1 + one_local_many_remote = len(l_remain) == 1 and len(r_remain) > 1 + + if strategy.config.allow_many_to_one_auto_bind and (many_local_one_remote or one_local_many_remote): + chosen_local, chosen_remote = _stable_pick_bind_pair(l_remain, r_remain) + + if chosen_local.binding_status == BindingStatus.MISSING: + pending_auto_bind_updates.append( + { + "node": chosen_local, + "outcome": "ONE_TO_ONE", + "create_enabled": None, + "bind_remote_node_id": chosen_remote.node_id, + } + ) + if chosen_remote.binding_status == BindingStatus.MISSING: + pending_auto_bind_updates.append( + { + "node": chosen_remote, + "outcome": "ONE_TO_ONE", + "create_enabled": None, + "bind_remote_node_id": None, + } + ) + + for n in l_remain: + if n.node_id == chosen_local.node_id or n.binding_status != BindingStatus.MISSING: + continue + pending_auto_bind_updates.append( + { + "node": n, + "outcome": "ZERO", + "create_enabled": local_create_enabled, + "bind_remote_node_id": None, + } + ) - if (len(r_remain) > 1 and len(l_remain) >= 1) or (len(r_remain) == 1 and len(l_remain) > 1): for n in r_remain: - if n.binding_status == BindingStatus.MISSING: - pending_auto_bind_updates.append( - { - "node": n, - "outcome": "AMBIGUOUS", - "create_enabled": None, - "bind_remote_node_id": None, - } - ) + if n.node_id == chosen_remote.node_id or n.binding_status != BindingStatus.MISSING: + continue + pending_auto_bind_updates.append( + { + "node": n, + "outcome": "ZERO", + "create_enabled": remote_create_enabled, + "bind_remote_node_id": None, + } + ) + else: + if (len(l_remain) > 1 and len(r_remain) >= 1) or (len(l_remain) == 1 and len(r_remain) > 1): + for n in l_remain: + if n.binding_status == BindingStatus.MISSING: + pending_auto_bind_updates.append( + { + "node": n, + "outcome": "AMBIGUOUS", + "create_enabled": None, + "bind_remote_node_id": None, + } + ) + + if (len(r_remain) > 1 and len(l_remain) >= 1) or (len(r_remain) == 1 and len(l_remain) > 1): + for n in r_remain: + if n.binding_status == BindingStatus.MISSING: + pending_auto_bind_updates.append( + { + "node": n, + "outcome": "AMBIGUOUS", + "create_enabled": None, + "bind_remote_node_id": None, + } + ) for item in pending_auto_bind_updates: node = item["node"] @@ -572,3 +627,25 @@ async def phase_auto_binding(strategy, local_nodes: List["SyncNode"], remote_nod reason = f"{reason} | detail={detail}" node.error = reason logger.debug(f"[{strategy.node_type}] 自动绑定阻断: node={node.node_id}, reason={reason}") + + outcome_counts: Dict[str, int] = {} + create_enabled_count = 0 + bound_pairs = 0 + for item in pending_auto_bind_updates: + outcome = str(item.get("outcome") or "NONE") + outcome_counts[outcome] = outcome_counts.get(outcome, 0) + 1 + if item.get("create_enabled"): + create_enabled_count += 1 + if item.get("bind_remote_node_id"): + bound_pairs += 1 + + if pending_auto_bind_updates: + ordered = ", ".join(f"{key}={outcome_counts[key]}" for key in sorted(outcome_counts)) + logger.info( + "[%s] Auto-bind summary: candidates=%d, bind_pairs=%d, create_enabled=%d, outcomes={%s}", + strategy.node_type, + len(pending_auto_bind_updates), + bound_pairs, + create_enabled_count, + ordered, + ) diff --git a/tests/integration/test_strategy_business_points_migrated.py b/tests/integration/test_strategy_business_points_migrated.py index b01c3c8..a53abda 100644 --- a/tests/integration/test_strategy_business_points_migrated.py +++ b/tests/integration/test_strategy_business_points_migrated.py @@ -193,6 +193,30 @@ async def test_auto_binding_skip_on_ambiguous_migrated(setup_env): assert await bm.get_local_id("test_project", "r1") is None +@pytest.mark.asyncio +async def test_auto_binding_many_to_one_can_bind_stable_pick_and_leave_rest_for_create(setup_env): + local, remote, bm, _ = setup_env + + strategy = TestProjectStrategy("test_project", local, remote, bm) + strategy.update_config( + auto_bind=True, + auto_bind_fields=["code"], + depend_fields={}, + local_orphan_action=OrphanAction.CREATE_REMOTE, + allow_many_to_one_auto_bind=True, + ) + + await local.add(build_project_node("l2", "P2", name="B", code="AMB-2")) + await local.add(build_project_node("l1", "P1", name="A", code="AMB-2")) + await remote.add(build_project_node("r1", "RP1", name="A", code="AMB-2")) + + await strategy.bind() + + assert await bm.get_remote_id("test_project", "l1") == "r1" + assert await bm.get_remote_id("test_project", "l2") is None + assert await bm.get_local_id("test_project", "r1") == "l1" + + @pytest.mark.asyncio async def test_pre_bind_data_id_works_when_auto_bind_disabled(setup_env): local, remote, bm, _ = setup_env diff --git a/tests/unit/test_base_api_handler_update_schema_helpers.py b/tests/unit/test_base_api_handler_update_schema_helpers.py new file mode 100644 index 0000000..43db175 --- /dev/null +++ b/tests/unit/test_base_api_handler_update_schema_helpers.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel + +from sync_state_machine.datasource.api.handler import BaseApiHandler +from sync_state_machine.datasource.task_result import TaskResult + + +class _UpdateSchema(BaseModel): + id: str + required_value: float + optional_text: str | None = None + + +class _DummyApiHandler(BaseApiHandler[BaseModel]): + _node_type = "dummy" + _node_class = object # type: ignore[assignment] + _schema = BaseModel + + async def create_all(self, nodes): + return [TaskResult.success(node_id="dummy")] + + async def update_all(self, nodes): + return [TaskResult.success(node_id="dummy")] + + async def delete_all(self, nodes): + return [TaskResult.success(node_id="dummy")] + + async def load(self): + return [] + + +def test_get_update_schema_diff_falls_back_to_schema_fields_when_origin_is_incomplete() -> None: + handler = _DummyApiHandler() + + diff = handler.get_update_schema_diff( + _UpdateSchema, + {"id": "node-1", "required_value": 4.0, "optional_text": "ok"}, + {"id": "node-1"}, + node_id="node-1", + ) + + assert diff == {"required_value": 4.0, "optional_text": "ok"} + + +def test_build_update_request_data_uses_same_schema_projection_as_diff() -> None: + handler = _DummyApiHandler() + + payload = handler.build_update_request_data( + {"id": "node-1", "required_value": 4.0, "optional_text": "ok", "ignored": "x"}, + {"id": "node-1"}, + schema=_UpdateSchema, + ) + + assert payload == {"id": "node-1", "required_value": 4.0, "optional_text": "ok"} + + +def test_schema_changed_fields_matches_update_payload_comparison() -> None: + handler = _DummyApiHandler() + + changed_fields = handler._schema_changed_fields( + _UpdateSchema, + {"id": "node-1", "required_value": 4.0, "optional_text": "ok"}, + {"id": "node-1", "required_value": 4.0}, + ) + + assert changed_fields == ["optional_text"] + + +def test_update_schema_payload_filters_empty_values() -> None: + handler = _DummyApiHandler() + + payload = handler._get_update_schema_payload( + _UpdateSchema, + {"id": "node-1", "required_value": 4.0, "optional_text": ""}, + require_valid=False, + ) + + assert payload == {"id": "node-1", "required_value": 4.0} + + +def test_get_update_schema_diff_skips_schema_when_new_data_is_still_incomplete() -> None: + handler = _DummyApiHandler() + + diff = handler.get_update_schema_diff( + _UpdateSchema, + {"id": "node-1"}, + {"id": "node-1", "required_value": 2.0}, + node_id="node-1", + ) + + assert diff is None \ No newline at end of file diff --git a/tests/unit/test_full_sync_pipeline_downstream_reload.py b/tests/unit/test_full_sync_pipeline_downstream_reload.py index abb8d37..732b234 100644 --- a/tests/unit/test_full_sync_pipeline_downstream_reload.py +++ b/tests/unit/test_full_sync_pipeline_downstream_reload.py @@ -31,6 +31,7 @@ class _StubStrategy: def __init__(self, node_type: str): self.node_type = node_type self.skip_sync = False + self.skip_post_check = False self.config = SimpleNamespace( depend_fields={}, update_direction=None, @@ -193,4 +194,55 @@ async def test_evaluate_consistency_merges_compare_and_post_check_ignore_fields( full_sync_pipeline_module.evaluate_consistency_for_node_type = original assert captured["ignore_fields"] == ["updated_at", "code", "audit_status"] - assert result["ignore_fields"] == ["updated_at", "code", "audit_status"] \ No newline at end of file + assert result["ignore_fields"] == ["updated_at", "code", "audit_status"] + + +@pytest.mark.asyncio +async def test_phase_post_check_skips_configured_node_types(monkeypatch: pytest.MonkeyPatch) -> None: + supplier_strategy = _StubStrategy("supplier") + supplier_strategy.skip_post_check = True + project_strategy = _StubStrategy("project") + + pipeline = FullSyncPipeline( + local_datasource=_FakeDataSource(), + remote_datasource=_FakeDataSource(), + strategies=[supplier_strategy, project_strategy], + persistence=_FakePersistence(), + local_collection=DataCollection("local"), + remote_collection=DataCollection("remote"), + binding_manager=BindingManager(_FakePersistence(), auto_persist=False), + node_types=["supplier", "project"], + load_order=["supplier", "project"], + sync_order=["supplier", "project"], + enable_persistence=False, + ) + + reload_calls: list[str] = [] + evaluated: list[str] = [] + + async def fake_reload(node_type: str, *, reason: str) -> None: + del reason + reload_calls.append(node_type) + + async def fake_evaluate(node_type: str): + evaluated.append(node_type) + return {"ok": True, "checked_pairs": 1, "mismatch_count": 0} + + async def fake_print_summary() -> None: + return None + + monkeypatch.setattr(pipeline, "_reload_node_type", fake_reload) + monkeypatch.setattr(pipeline, "_evaluate_consistency_for_node_type", fake_evaluate) + monkeypatch.setattr(pipeline, "print_summary", fake_print_summary) + + result = await pipeline.phase_post_check() + + assert result is True + assert reload_calls == ["project"] + assert evaluated == ["project"] + assert pipeline._consistency_by_type["supplier"] == { + "ok": True, + "checked_pairs": 0, + "mismatch_count": 0, + "skipped": True, + } \ No newline at end of file diff --git a/tests/unit/test_strategy_runtime_config.py b/tests/unit/test_strategy_runtime_config.py index a5420fc..16e7f1c 100644 --- a/tests/unit/test_strategy_runtime_config.py +++ b/tests/unit/test_strategy_runtime_config.py @@ -8,6 +8,7 @@ def test_strategy_runtime_config_skip_sync_defaults_false() -> None: runtime_config = StrategyRuntimeConfig(node_type="project") assert runtime_config.skip_sync is False + assert runtime_config.skip_post_check is False def test_apply_strategy_overrides_allows_skip_sync_override() -> None: @@ -16,6 +17,7 @@ def test_apply_strategy_overrides_allows_skip_sync_override() -> None: strategy_overrides={ "project": { "skip_sync": True, + "skip_post_check": True, } }, ) @@ -23,6 +25,7 @@ def test_apply_strategy_overrides_allows_skip_sync_override() -> None: assert len(resolved) == 1 assert resolved[0].node_type == "project" assert resolved[0].skip_sync is True + assert resolved[0].skip_post_check is True def test_strategy_config_supports_both_post_check_ignore_fields() -> None: @@ -30,8 +33,10 @@ def test_strategy_config_supports_both_post_check_ignore_fields() -> None: { "post_check_ignore_fields": ["code"], "post_check_ignore_list_item_fields": {"plan_node": ["is_audit"]}, + "allow_many_to_one_auto_bind": True, } ) assert config.post_check_ignore_fields == ["code"] - assert config.post_check_ignore_list_item_fields == {"plan_node": ["is_audit"]} \ No newline at end of file + assert config.post_check_ignore_list_item_fields == {"plan_node": ["is_audit"]} + assert config.allow_many_to_one_auto_bind is True \ No newline at end of file