diff --git a/.gitignore b/.gitignore index 5c29604..3c437e2 100644 --- a/.gitignore +++ b/.gitignore @@ -95,4 +95,5 @@ run_profiles/*.yml # Sync System Runtime Data _runtime/ test_results/ -*.code-workspace \ No newline at end of file +*.code-workspace +runtime/ \ No newline at end of file diff --git a/README.md b/README.md index 04fab82..18fa1ed 100644 --- a/README.md +++ b/README.md @@ -58,15 +58,21 @@ ## 5. 使用方法 +运行配置遵循一个约定： +- 跟踪在仓库里的 `run_profiles/*.yaml` 与 `run_profiles/preset/*.default.yaml` 只写必要的覆盖项。 +- pipeline 启动时会打印并写入一份 **Resolved Full Config**，把省略的默认值全部补齐，便于排障和审阅。 +- 字段含义与全量示例见：`docs/runtime_config_guide.md` + ### JSONL 同步 -- 基线配置（优先）：`ecm-sync-run --config_path=run_profiles/preset/jsonl_to_jsonl.default.yaml` -- 兼容旧入口：`python run.py --config_path=run_profiles/preset/jsonl_to_jsonl.default.yaml` -- 使用自定义配置：先从 `run_profiles/preset/jsonl_to_jsonl.default.yaml` 复制到 `run_profiles/jsonl_to_jsonl.local.yaml`，按需修改后运行：`ecm-sync-run --config_path=run_profiles/jsonl_to_jsonl.local.yaml` +- 已安装命令行入口：`ecm-sync-run --config_path=run_profiles/preset/jsonl_to_jsonl.default.yaml` +- 仓库内直接运行：`python run.py --config_path=run_profiles/preset/jsonl_to_jsonl.default.yaml` +- 多项目 JSONL 示例：`python run.py --config_path=run_profiles/preset/jsonl_to_jsonl.multi_project.default.yaml` +- 自定义配置：从 `run_profiles/preset/` 复制一份到 `run_profiles/*.local.yaml`，只改有差异的字段即可。 ### API 同步 -- 基线配置：`ecm-sync-run --config_path=run_profiles/preset/jsonl_to_api.default.yaml` -- 兼容旧入口：`python run.py --config_path=run_profiles/preset/jsonl_to_api.default.yaml` -- 使用自定义配置：先从 `run_profiles/preset/jsonl_to_api.default.yaml` 复制到 `run_profiles/jsonl_to_api.local.yaml`，按需修改后运行：`ecm-sync-run --config_path=run_profiles/jsonl_to_api.local.yaml` +- 已安装命令行入口：`ecm-sync-run --config_path=run_profiles/preset/jsonl_to_api.default.yaml` +- 仓库内直接运行：`python run.py --config_path=run_profiles/preset/jsonl_to_api.default.yaml` +- 如果是单项目联调，优先参考仓库内现成示例：`run_profiles/jsonl_to_api.yaml` ### 库模式使用 - 本地可编辑安装：`pip install -e .` @@ -75,6 +81,7 @@ ### 配置与校验 - 模板配置：`run_profiles/preset/*.default.yaml` - 建议先跑 default，再在 `run_profiles/*.local.yaml` 做覆盖（该目录 YAML 已默认忽略） +- 跟踪中的模板与示例配置都采用“只写必要字段”的风格；启动日志里的 `Resolved Full Config` 才是最终生效配置。 - 状态机配置校验：`python tools/validate_config.py --config sync_state_machine/config/node_state_machine.yaml` - 状态机图生成：`python tools/render_graph.py --format mermaid --out docs/state_machine.mmd` @@ -88,6 +95,7 @@ - UI 调试：`python -m ui.main --host 127.0.0.1 --port 8765` - 运维与排障：`docs/operations.md` - 远程测试环境：`tools/remote_env.py` +- 脚本化回归与数据集构建：`scripts/README.md` ## 文档导航 diff --git a/docs/runtime_config_guide.md b/docs/runtime_config_guide.md index 5271a4e..d642a19 100644 --- a/docs/runtime_config_guide.md +++ b/docs/runtime_config_guide.md @@ -2,6 +2,91 @@ 本文档从业务视角说明配置字段含义、对阶段行为的影响，以及推荐的覆盖方式。 +## 0. 配置约定 + +当前运行配置分成两层： + +1. 输入配置 + - 你写在 `run_profiles/*.yaml` 或 `run_profiles/preset/*.default.yaml` 里的内容。 + - 只建议写必要覆盖项，不要把默认值全部展开。 +2. Resolved Full Config + - pipeline 启动时打印并写入日志的全量配置。 + - 其中已经补齐 datasource 默认值、策略默认值和日志/持久化默认值。 + +也就是说： +- 模板应当尽量短。 +- 排障时看启动日志里的 **Resolved Full Config**。 + +### 0.1 一个最小输入示例 + +```yaml +local_datasource: + type: jsonl + jsonl_dir: ${PROJECT_ROOT}/filtered_datasource/datasource/filtered +remote_datasource: + type: jsonl + jsonl_dir: ${PROJECT_ROOT}/remote_datasource/datasource +persist: + db_path: ${PROJECT_ROOT}/test_results/sync_state_machine/demo.db +strategies: + project: + config: + auto_bind: true + auto_bind_fields: + - name +``` + +### 0.2 对应的全量 resolved 结果骨架 + +```yaml +node_types: + - company + - supplier + - user + - project + ... +local_datasource: + type: jsonl + jsonl_dir: /abs/path/filtered_datasource/datasource/filtered + read_only: false + handler_configs: {} +remote_datasource: + type: jsonl + jsonl_dir: /abs/path/remote_datasource/datasource + read_only: false + handler_configs: {} +strategies: + - node_type: project + skip_sync: false + config: + auto_bind: true + auto_bind_fields: + - name + pre_bind_data_id: [] + depend_fields: + company_id: company + local_orphan_action: create_remote + remote_orphan_action: none + update_direction: push + field_direction_key: null + field_direction_overrides: {} + post_check_ignore_fields: [] + compare_ignore_fields: [] +persist: + backend: sqlite + db_path: /abs/path/test_results/sync_state_machine/demo.db + backend_options: {} + enable: true + wipe_on_start: false +logging: + initialize: true + file_path: null + file_dir: null + file_name_pattern: null + level: 20 + suppress_node_logs: true +``` + ## 1. 推荐使用方式（先 default，再覆盖） 1. 先使用基线配置运行： @@ -48,7 +133,7 @@ ## 4. 策略字段（Strategy） -每个 node_type 都建议显式写出以下字段，便于审阅与排障。 +每个 node_type 的 **Resolved Full Config** 都会包含以下字段，但输入配置不需要全部显式写出。 ### 4.1 `update_direction` @@ -97,6 +182,32 @@ - `remote_datasource.handler_configs.supplier.load_mode: persisted_only` - `persisted_only` 表示只使用已持久化数据，不主动从外部拉新数据（具体以对应 handler 实现为准）。 +### 4.7 `field_direction_key` / `field_direction_overrides` + +- 这两个字段只在少数 node_type 上需要，典型例子是 `project_detail_data`。 +- `field_direction_key` + - 指定从节点 data 里取哪个字段作为“方向查表键”。 + - 例如 `key`，表示用 `data["key"]` 决定这条记录是 push 还是 pull。 +- `field_direction_overrides` + - 指定“字段值 -> 更新方向”的映射。 + - 没命中的值继续使用默认 `update_direction`。 + +示例： + +```yaml +project_detail_data: + config: + field_direction_key: key + field_direction_overrides: + ensure_production: pull + climb_production: pull + challenge_production: pull +``` + +含义： +- `key=ensure_production/climb_production/challenge_production` 时，从远端拉回本地。 +- 其余 key 继续按默认 `update_direction=push` 处理。 + ## 5. 阶段行为与配置关系 - bind 阶段 @@ -108,8 +219,9 @@ - update 阶段 - `update_direction` 决定是否和向哪侧更新。 - 写入后 reload（Pipeline 内置） - - 当某个 node_type 的 create 或 update 发生实际写入成功后，pipeline 会自动对该 node_type 重新执行两侧 `load`。 - - 目的：让后续 update 判断基于最新远端/本地数据，支持“create 后再补一次 update”的链路。 + - 对 API / 需要重新拉取权威状态的数据源，create 或 update 成功后仍会按 node_type reload。 + - 对纯 JSONL -> JSONL pipeline，reload 会跳过，直接复用当前内存状态。 + - 目的：避免对离线 JSONL 数据做无意义重复扫描，同时保留 API 数据源的权威状态刷新。 - 同步后一致性检查（Pipeline 内置） - 每个 node_type 在本轮同步后会基于绑定对执行数据一致性校验（比较时忽略 `id/_id` 字段）。 - 若发现差异，会在日志中打印差异样本，并在汇总表新增 `Consistent` 列展示 `Y` / `N(差异数)`。 diff --git a/pytest.ini b/pytest.ini index 3b758c4..a09de4a 100644 --- a/pytest.ini +++ b/pytest.ini @@ -10,4 +10,3 @@ filterwarnings = ignore:'asyncio.set_event_loop_policy' is deprecated and slated for removal in Python 3.16:DeprecationWarning:pytest_asyncio\.plugin # 也可以设置异步测试的默认 loop scope asyncio_default_fixture_loop_scope = function -addopts = --ignore=scripts/test_api_sync.py --ignore=scripts/test_api_sync_top10.py --ignore=scripts/test_api_write_real_sample.py diff --git a/run.py b/run.py index b75bb9f..1694bc1 100644 --- a/run.py +++ b/run.py @@ -11,7 +11,7 @@ if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from sync_state_machine.config import load_overrides_from_file -from sync_state_machine.pipeline import build_config, run_pipeline_from_config +from sync_state_machine.pipeline import run_profile_from_file def _resolve_config_path(raw_path: str) -> Path: @@ -34,14 +34,10 @@ async def _main() -> int: if not cfg_path.exists() or not cfg_path.is_file(): raise FileNotFoundError(f"Config file not found: {cfg_path}") - overrides = load_overrides_from_file(cfg_path, PROJECT_ROOT) - config = build_config(project_root=PROJECT_ROOT, overrides=overrides) - - mode = f"{config.local_datasource.type}->{config.remote_datasource.type}" print(f"🧩 Loaded config file: {cfg_path}") - await run_pipeline_from_config( - config, - title=f"sync_state_machine pipeline ({mode})", + await run_profile_from_file( + project_root=PROJECT_ROOT, + config_path=cfg_path, print_summary=True, ) return 0 diff --git a/run_profiles/README.md b/run_profiles/README.md index 813b744..be6786f 100644 --- a/run_profiles/README.md +++ b/run_profiles/README.md @@ -3,19 +3,35 @@ - 路径：项目根目录 `run_profiles/` - 命名：建议使用 `*.local.yaml`（该目录下 YAML 默认会被 git 忽略） +当前约定： +- 仓库里跟踪的配置文件只写必要字段，不重复抄默认值。 +- pipeline 启动时会打印 **Resolved Full Config**，把省略字段补齐。 +- 如果你要看完整字段和字段含义，不看模板，直接看 `docs/runtime_config_guide.md`。 + 模板来源： - `run_profiles/preset/jsonl_to_jsonl.default.yaml` - `run_profiles/preset/jsonl_to_api.default.yaml` +- `run_profiles/preset/jsonl_to_jsonl.multi_project.default.yaml` 推荐流程： 1. 从模板复制到本目录，例如： - - `run_profiles/jsonl_to_jsonl.local.yaml` - - `run_profiles/jsonl_to_api.local.yaml` + - `run_profiles/jsonl_to_jsonl.local.yaml` + - `run_profiles/jsonl_to_api.local.yaml` 2. 按需修改后运行： - - `ecm-sync-run --config_path=run_profiles/jsonl_to_jsonl.local.yaml` + - `ecm-sync-run --config_path=run_profiles/jsonl_to_jsonl.local.yaml` 兼容旧入口： - `python run.py --config_path=run_profiles/jsonl_to_jsonl.local.yaml` +多项目示例： +- `run_profiles/jsonl_to_jsonl.multi_project.yaml` +- `run_profiles/jsonl_to_jsonl.multi_project.global.yaml` +- `run_profiles/jsonl_to_jsonl.multi_project.project.yaml` + 支持占位符： - `${PROJECT_ROOT}`：运行时替换为项目根目录绝对路径。 + +现成示例： +- `run_profiles/jsonl_to_api.yaml`：单项目 JSONL -> API 联调示例 +- `run_profiles/jsonl_to_api.test.yaml`：auto-test 使用的 JSONL -> API 示例 +- `run_profiles/jsonl_to_jsonl.multi_project.yaml`：多项目 JSONL -> JSONL 编排入口 diff --git a/run_profiles/preset/README.md b/run_profiles/preset/README.md index 15611af..b2dcb9c 100644 --- a/run_profiles/preset/README.md +++ b/run_profiles/preset/README.md @@ -2,20 +2,35 @@ 该目录存放版本管理的运行配置模板（default）。 +模板约定： +- 这里只保留必要覆盖项。 +- 省略的字段会在加载时从默认 datasource 配置和各 domain 的 `default_config` 自动补齐。 +- 真正生效的全量配置，以启动日志里的 **Resolved Full Config** 为准。 + 当前模板： - `jsonl_to_jsonl.default.yaml` - `jsonl_to_api.default.yaml` +- `jsonl_to_jsonl.multi_project.default.yaml` +- `jsonl_to_jsonl.multi_project.global.default.yaml` +- `jsonl_to_jsonl.multi_project.project.default.yaml` 使用方式： 1. 复制模板到项目根目录 `run_profiles/`，并重命名为你自己的文件（如 `*.local.yaml`）。 -2. 修改后通过入口运行： +2. 只改和当前环境/业务差异相关的字段。 +3. 修改后通过入口运行： - `ecm-sync-run --config_path=run_profiles/jsonl_to_jsonl.local.yaml` 兼容旧入口： - `python run.py --config_path=run_profiles/jsonl_to_jsonl.local.yaml` +多项目配置说明： +- `jsonl_to_jsonl.multi_project.default.yaml` 是多 pipeline 编排入口。 +- `jsonl_to_jsonl.multi_project.global.default.yaml` 定义全局公共 domain 的 pipeline。 +- `jsonl_to_jsonl.multi_project.project.default.yaml` 定义单项目 pipeline 默认配置。 +- `project_bootstrap_node_types` 用于显式声明项目级 pipeline 需要从 global scope 继承到项目 scope 的公共节点类型，例如 `company`、`supplier`。 + 说明： - `run_profiles/*.yaml` / `run_profiles/*.yml` 默认被 `.gitignore` 忽略，用于本地私有配置。 - 模板内可使用 `${PROJECT_ROOT}` 占位符。 -- 模板已显式展开 datasource 与 strategies 关键字段，便于集中审查。 +- 完整字段定义与字段含义见 `docs/runtime_config_guide.md`。 - API 数据源支持全局限流配置：`api_rate_limit_max_requests` 与 `api_rate_limit_window_seconds`。 \ No newline at end of file diff --git a/run_profiles/preset/jsonl_to_api.default.yaml b/run_profiles/preset/jsonl_to_api.default.yaml index 45dd7a7..5f68a03 100644 --- a/run_profiles/preset/jsonl_to_api.default.yaml +++ b/run_profiles/preset/jsonl_to_api.default.yaml @@ -1,32 +1,33 @@ node_types: -- company -- supplier -- project -- contract -- contract_change -- construction_task -- construction_task_detail -- material -- material_detail -- contract_settlement -- contract_settlement_detail -- personnel -- personnel_detail -- preparation -- production -- lar -- lar_change -- project_detail_data -- units + - company + - supplier + - project + - contract + - contract_change + - construction_task + - construction_task_detail + - material + - material_detail + - contract_settlement + - contract_settlement_detail + - personnel + - personnel_detail + - preparation + - production + - lar + - lar_change + - project_detail_data + - units + local_datasource: type: jsonl jsonl_dir: ${PROJECT_ROOT}/filtered_datasource/datasource/filtered - read_only: false handler_configs: project: data_id_filter: - "0f9c3e22-f4bb-4803-825a-4f238eeb06d5" - "f3e02e84-e81c-4aa6-88d4-f5aa108c8227" + remote_datasource: type: api api_base_url: http://localhost:8000/api/third/v2 @@ -36,257 +37,34 @@ remote_datasource: api_rate_limit_max_requests: 30 api_rate_limit_window_seconds: 10.0 poll_max_retries: 1 - poll_interval: 0.5 handler_configs: project: data_id_filter: - # 请替换为实际的项目ID，至少需要一个 - "0f9c3e22-f4bb-4803-825a-4f238eeb06d5" - "f3e02e84-e81c-4aa6-88d4-f5aa108c8227" - supplier: {} + persist: - backend: sqlite db_path: ${PROJECT_ROOT}/test_results/sync_state_machine/simple_pipeline_sm.db - enable: true - wipe_on_start: false + logging: - initialize: true file_dir: ${PROJECT_ROOT}/test_results/sync_state_machine file_name_pattern: simple_pipeline_sm_{timestamp}.log - level: 20 + strategies: company: - skip_sync: false + preset: pull_only config: - auto_bind: true auto_bind_fields: - - credit_code - depend_fields: {} - local_orphan_action: none - remote_orphan_action: create_local - update_direction: pull + - credit_code + supplier: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - credit_code - depend_fields: {} - local_orphan_action: none - remote_orphan_action: create_local - update_direction: pull + preset: pull_only + project: - skip_sync: false config: - auto_bind: false - auto_bind_fields: [] - pre_bind_data_id: [] - depend_fields: - company_id: company local_orphan_action: none - remote_orphan_action: none - update_direction: push - contract: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - project_id - - company_id - - code - depend_fields: - project_id: project - company_id: supplier - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - contract_change: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - contract_id - - change_time - - title - depend_fields: - project_id: project - contract_id: contract - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - construction_task: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - contract_id - - task_type - depend_fields: - contract_id: contract - project_id: project - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - construction_task_detail: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - parent_id - - date - depend_fields: - parent_id: construction_task - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - material: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - contract_id - - material_type - depend_fields: - contract_id: contract - project_id: project - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - material_detail: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - parent_id - - date - depend_fields: - parent_id: material - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - contract_settlement: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - contract_id - depend_fields: - contract_id: contract - project_id: project - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - contract_settlement_detail: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - parent_id - - date - depend_fields: - parent_id: contract_settlement - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - personnel: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - project_id - - code - depend_fields: - project_id: project - contract_id: contract - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - personnel_detail: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - parent_id - - date - depend_fields: - parent_id: personnel - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - preparation: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - project_id - - code - depend_fields: - project_id: project - local_orphan_action: none - remote_orphan_action: none - update_direction: push + production: - skip_sync: false config: - auto_bind: true - auto_bind_fields: - - project_id - - code - depend_fields: - project_id: project - local_orphan_action: none - remote_orphan_action: none - update_direction: push - lar: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - project_id - depend_fields: - project_id: project - local_orphan_action: none - remote_orphan_action: none - update_direction: push - lar_change: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - project_id - - change_time - - title - depend_fields: - project_id: project - local_orphan_action: none - remote_orphan_action: none - update_direction: push - project_detail_data: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - project_id - - key - depend_fields: - project_id: project - local_orphan_action: none - remote_orphan_action: none - update_direction: push # 未在 field_direction_overrides 中列出的 key 默认 push - field_direction_key: key # 以节点 data.key 字段值作为方向查表键 - field_direction_overrides: - ensure_production: pull # 远程 → 本地 - climb_production: pull # 远程 → 本地 - challenge_production: pull # 远程 → 本地 - units: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - project_id - - serial_number - depend_fields: - project_id: project - local_orphan_action: none - remote_orphan_action: none - update_direction: push + post_check_ignore_fields: + - is_exist diff --git a/run_profiles/preset/jsonl_to_jsonl.default.yaml b/run_profiles/preset/jsonl_to_jsonl.default.yaml index 74e1925..3062940 100644 --- a/run_profiles/preset/jsonl_to_jsonl.default.yaml +++ b/run_profiles/preset/jsonl_to_jsonl.default.yaml @@ -22,261 +22,46 @@ node_types: local_datasource: type: jsonl jsonl_dir: ${PROJECT_ROOT}/filtered_datasource/datasource/filtered - read_only: false - handler_configs: {} remote_datasource: type: jsonl jsonl_dir: ${PROJECT_ROOT}/remote_datasource/datasource - read_only: false - handler_configs: {} persist: - backend: sqlite db_path: ${PROJECT_ROOT}/test_results/sync_state_machine/full_sync_pipeline_jsonl_sm.db - enable: true - wipe_on_start: false logging: - initialize: true file_dir: ${PROJECT_ROOT}/test_results/sync_state_machine file_name_pattern: jsonl_pipeline_sm_{timestamp}.log - level: 20 strategies: company: - skip_sync: false config: - auto_bind: true auto_bind_fields: - - credit_code - depend_fields: {} + - credit_code local_orphan_action: create_remote remote_orphan_action: none update_direction: push user: - skip_sync: false config: - auto_bind: true - auto_bind_fields: - - username - depend_fields: {} local_orphan_action: create_remote remote_orphan_action: none update_direction: push supplier: - skip_sync: false config: - auto_bind: true - auto_bind_fields: - - credit_code - depend_fields: {} local_orphan_action: create_remote remote_orphan_action: none update_direction: push project: - skip_sync: false config: auto_bind: true auto_bind_fields: - - name - depend_fields: - company_id: company - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - contract: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - project_id - - company_id - - code - depend_fields: - project_id: project - company_id: supplier - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - contract_change: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - contract_id - - change_time - - title - depend_fields: - project_id: project - contract_id: contract - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - construction_task: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - contract_id - - task_type - depend_fields: - contract_id: contract - project_id: project - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - construction_task_detail: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - parent_id - - date - depend_fields: - parent_id: construction_task - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - material: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - contract_id - - material_type - depend_fields: - contract_id: contract - project_id: project - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - material_detail: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - parent_id - - date - depend_fields: - parent_id: material - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - contract_settlement: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - contract_id - depend_fields: - contract_id: contract - project_id: project - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - contract_settlement_detail: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - parent_id - - date - depend_fields: - parent_id: contract_settlement - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - personnel: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - project_id - - code - depend_fields: - project_id: project - contract_id: contract - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - personnel_detail: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - parent_id - - date - depend_fields: - parent_id: personnel - local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push - preparation: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - project_id - - code - depend_fields: - project_id: project - local_orphan_action: none - remote_orphan_action: none - update_direction: push - production: - skip_sync: false - config: - auto_bind: true - auto_bind_fields: - - project_id - - code - depend_fields: - project_id: project - local_orphan_action: none - remote_orphan_action: none - update_direction: push + - name lar: - skip_sync: false config: - auto_bind: true - auto_bind_fields: - - project_id - depend_fields: - project_id: project local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push lar_change: - skip_sync: false config: - auto_bind: true - auto_bind_fields: - - project_id - - change_time - - title - depend_fields: - project_id: project local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push project_detail_data: - skip_sync: false config: - auto_bind: true - auto_bind_fields: - - project_id - - key - depend_fields: - project_id: project local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push units: - skip_sync: false config: - auto_bind: true - auto_bind_fields: - - project_id - - serial_number - depend_fields: - project_id: project local_orphan_action: create_remote - remote_orphan_action: none - update_direction: push diff --git a/run_profiles/preset/jsonl_to_jsonl.multi_project.default.yaml b/run_profiles/preset/jsonl_to_jsonl.multi_project.default.yaml new file mode 100644 index 0000000..86ac637 --- /dev/null +++ b/run_profiles/preset/jsonl_to_jsonl.multi_project.default.yaml @@ -0,0 +1,12 @@ +global_config: run_profiles/preset/jsonl_to_jsonl.multi_project.global.default.yaml +default_project_config: run_profiles/preset/jsonl_to_jsonl.multi_project.project.default.yaml + +project_bootstrap_node_types: + - company + - supplier + +projects: + - local_project_id: f3e02e84-e81c-4aa6-88d4-f5aa108c8227 + remote_project_id: f3e02e84-e81c-4aa6-88d4-f5aa108c8227 + - local_project_id: 0f9c3e22-f4bb-4803-825a-4f238eeb06d5 + remote_project_id: 0f9c3e22-f4bb-4803-825a-4f238eeb06d5 diff --git a/run_profiles/preset/jsonl_to_jsonl.multi_project.global.default.yaml b/run_profiles/preset/jsonl_to_jsonl.multi_project.global.default.yaml new file mode 100644 index 0000000..ead35b6 --- /dev/null +++ b/run_profiles/preset/jsonl_to_jsonl.multi_project.global.default.yaml @@ -0,0 +1,29 @@ +node_types: + - company + - supplier +local_datasource: + type: jsonl + jsonl_dir: ${PROJECT_ROOT}/working_local_datasource/20260302-020003 +remote_datasource: + type: jsonl + jsonl_dir: ${PROJECT_ROOT}/runtime/tmp/multi_project_jsonl_remote +persist: + db_path: ${PROJECT_ROOT}/test_results/sync_state_machine/multi_project_jsonl.db + wipe_on_start: true +logging: + file_dir: ${PROJECT_ROOT}/test_results/sync_state_machine + file_name_pattern: multi_project_global_{timestamp}.log +strategies: + company: + config: + auto_bind_fields: + - code + - name + local_orphan_action: create_remote + remote_orphan_action: none + update_direction: push + supplier: + config: + local_orphan_action: create_remote + remote_orphan_action: none + update_direction: push diff --git a/run_profiles/preset/jsonl_to_jsonl.multi_project.project.default.yaml b/run_profiles/preset/jsonl_to_jsonl.multi_project.project.default.yaml new file mode 100644 index 0000000..7596505 --- /dev/null +++ b/run_profiles/preset/jsonl_to_jsonl.multi_project.project.default.yaml @@ -0,0 +1,47 @@ +node_types: + - project + - contract + - contract_change + - construction_task + - construction_task_detail + - material + - material_detail + - contract_settlement + - contract_settlement_detail + - personnel + - personnel_detail + - preparation + - production + - lar + - lar_change + - project_detail_data + - units +local_datasource: + type: jsonl + jsonl_dir: ${PROJECT_ROOT}/working_local_datasource/20260302-020003 +remote_datasource: + type: jsonl + jsonl_dir: ${PROJECT_ROOT}/runtime/tmp/multi_project_jsonl_remote +persist: + db_path: ${PROJECT_ROOT}/test_results/sync_state_machine/multi_project_jsonl.db +logging: + file_dir: ${PROJECT_ROOT}/test_results/sync_state_machine + file_name_pattern: multi_project_project_{timestamp}.log +strategies: + project: + config: + auto_bind: true + auto_bind_fields: + - name + lar: + config: + local_orphan_action: create_remote + lar_change: + config: + local_orphan_action: create_remote + project_detail_data: + config: + local_orphan_action: create_remote + units: + config: + local_orphan_action: create_remote diff --git a/scripts/README.md b/scripts/README.md index 28eb2b6..a21db3c 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -1,44 +1,41 @@ # 脚本目录 -## 文件说明 +本目录当前主要放两类脚本： +- 自动化测试编排脚本 +- 本地数据集构建/分析脚本 -### test_api_sync.py -主要的API数据同步测试脚本。 +旧的 `scripts/test_api_sync.py` 与 `API_SYNC_GUIDE.md` 已不存在，不再作为当前入口。 -**功能：** -- 从远程API获取所有GET/LIST接口数据 -- 自动分页获取全量数据 -- 保存为JSONL格式到 `remote_datasource` 目录 -- 支持单项目模式或全量同步 +## 当前常用入口 + +### 自动化测试 +- `scripts/auto_test_init.py` + - 初始化 auto-test 运行环境，负责清理状态、恢复测试数据、拉起后端。 +- `scripts/auto_test_clear.py` + - 清理 auto-test 运行环境。 +- `scripts/run_auto_test.py` + - 运行指定 domain 的多轮自动回归并输出报告。 +- `scripts/run_auto_test_all_domains.py` + - 聚合运行多个 domain 的自动回归。 +- `scripts/run_large_scale_auto_test.py` + - 大规模自动回归入口。 + +### 数据集与分析 +- `scripts/build_filtered_datasource_for_projects.py` + - 从完整 datasource 中抽取指定项目，生成可复现的精简 JSONL 数据集。 +- `scripts/analyze_state_signatures.py` + - 分析状态签名与差异。 + +## 推荐命令 + +这些命令目前可直接使用： -**运行方式：** ```bash -python scripts/test_api_sync.py +python scripts/build_filtered_datasource_for_projects.py --help +python scripts/run_auto_test.py --help ``` -**数据类型覆盖：** -- 基础数据: company, user, supplier, project -- 项目数据: contract, construction_task, material, contract_settlement, personnel -- 项目扩展: preparation, production, lar, units (从all_info获取) -- 明细数据: construction_task_detail, material_detail, contract_settlement_detail, personnel_detail +## 进一步说明 -**配置说明：** -在文件中修改以下配置： -```python -BASE_URL = "http://localhost:8000/api/third/v2" -UID = "your-uid" -SECRET_KEY = "your-secret-key" -TARGET_PROJECT_ID = "xxx-xxx-xxx" # 单项目模式，设为None则全量同步 -``` - -## 输出结果 - -数据保存在 `remote_datasource/datasource/` 目录下： -- `company_1.jsonl` - 公司数据 -- `user_1.jsonl` - 用户数据 -- `project_1.jsonl` - 项目数据 -- ... 其他数据类型 - -## 更多文档 - -详细文档请查看项目根目录的 [API_SYNC_GUIDE.md](../API_SYNC_GUIDE.md) +- 自动化测试总览：`docs/testing_guide.md` +- 大规模自动回归方法：`docs/auto_test_spec/agent_auto_test_method.md` diff --git a/sync_state_machine/__init__.py b/sync_state_machine/__init__.py index fb7fdc8..e943b10 100644 --- a/sync_state_machine/__init__.py +++ b/sync_state_machine/__init__.py @@ -25,12 +25,15 @@ from .config import ( list_registered_datasource_types, load_named_preset_overrides, load_overrides_from_file, + MultiProjectRunConfig, + build_multi_project_config_from_file, + is_multi_project_payload, register_datasource_config_model, register_datasource_factory, register_datasource_type, ) from .datasource import ApiClient, ApiDataSource, BaseApiHandler, BaseJsonlHandler, JsonlDataSource -from .pipeline import FullSyncPipeline, create_pipeline_from_config, run_pipeline_from_config +from .pipeline import FullSyncPipeline, MultiProjectPipeline, create_pipeline_from_config, run_pipeline_from_config, run_profile_from_file try: __version__ = version("ecm-sync-system") @@ -59,6 +62,9 @@ __all__ = [ "list_registered_datasource_types", "load_overrides_from_file", "load_named_preset_overrides", + "MultiProjectRunConfig", + "build_multi_project_config_from_file", + "is_multi_project_payload", "register_datasource_config_model", "register_datasource_factory", "register_datasource_type", @@ -68,6 +74,8 @@ __all__ = [ "BaseApiHandler", "BaseJsonlHandler", "FullSyncPipeline", + "MultiProjectPipeline", "create_pipeline_from_config", "run_pipeline_from_config", + "run_profile_from_file", ] diff --git a/sync_state_machine/cli.py b/sync_state_machine/cli.py index 3a20303..75170b6 100644 --- a/sync_state_machine/cli.py +++ b/sync_state_machine/cli.py @@ -5,8 +5,7 @@ import asyncio import logging from pathlib import Path -from .config import build_config, load_overrides_from_file -from .pipeline import run_pipeline_from_config +from .pipeline import run_profile_from_file def package_project_root() -> Path: @@ -35,14 +34,10 @@ async def _main() -> int: if not cfg_path.exists() or not cfg_path.is_file(): raise FileNotFoundError(f"Config file not found: {cfg_path}") - overrides = load_overrides_from_file(cfg_path, project_root) - config = build_config(project_root=project_root, overrides=overrides) - - mode = f"{config.local_datasource.type}->{config.remote_datasource.type}" print(f"🧩 Loaded config file: {cfg_path}") - await run_pipeline_from_config( - config, - title=f"sync_state_machine pipeline ({mode})", + await run_profile_from_file( + project_root=project_root, + config_path=cfg_path, print_summary=True, ) return 0 diff --git a/sync_state_machine/common/binding.py b/sync_state_machine/common/binding.py index d8bc65d..33d355e 100644 --- a/sync_state_machine/common/binding.py +++ b/sync_state_machine/common/binding.py @@ -12,9 +12,10 @@ class BindingManager: BindingManager 负责维护不同 DataCollection 间节点的绑定关系。 所有查询均在内存中进行，持久化仅作为备份。 """ - def __init__(self, persistence: PersistenceBackend, auto_persist: bool = True): + def __init__(self, persistence: PersistenceBackend, auto_persist: bool = True, scope: str = "global"): self.persistence = persistence self.auto_persist = auto_persist + self.scope = scope # { node_type: { local_id: remote_id } } self._bindings: Dict[str, Dict[str, Optional[str]]] = {} self._deleted: Dict[str, set[str]] = {} @@ -55,7 +56,7 @@ class BindingManager: logger.debug(log_line) if self.auto_persist: - await self.persistence.save_binding(node_type, local_id, {"remote_id": remote_id}) + await self.persistence.save_binding(self.scope, node_type, local_id, {"remote_id": remote_id}) async def unbind(self, node_type: str, local_id: str, manual: bool = False): """ @@ -80,7 +81,7 @@ class BindingManager: ) if self.auto_persist: - await self.persistence.delete_binding(node_type, local_id) + await self.persistence.delete_binding(self.scope, node_type, local_id) else: self._get_deleted(node_type).add(local_id) @@ -119,25 +120,71 @@ class BindingManager: async def load_from_persistence(self, node_type: str): """从持久化层加载特定类型的绑定关系到内存""" - raw_bindings = await self.persistence.load_bindings(node_type) + raw_bindings = await self.persistence.load_bindings(self.scope, node_type) bindings = self._get_type_bindings(node_type) + bindings.clear() self._get_deleted(node_type).clear() for rb in raw_bindings: payload = rb.get("payload") or {} bindings[rb["local_id"]] = payload.get("remote_id") - async def persist(self) -> None: + def import_bindings(self, node_type: str, bindings: Dict[str, Optional[str]]) -> None: + """导入既有绑定到当前 manager，仅写内存，不触发持久化。""" + target = self._get_type_bindings(node_type) + target.clear() + target.update(dict(bindings)) + self._get_deleted(node_type).clear() + + async def prune_missing_nodes( + self, + *, + local_collection, + remote_collection, + node_types: Optional[List[str]] = None, + ) -> int: + target_types = list(node_types) if node_types is not None else list(self._bindings.keys()) + deleted_count = 0 + + for node_type in target_types: + bindings = self._get_type_bindings(node_type) + if not bindings: + continue + + valid_local_ids = {node.node_id for node in local_collection.filter(node_type=node_type)} + valid_remote_ids = {node.node_id for node in remote_collection.filter(node_type=node_type)} + + stale_local_ids = [ + local_id + for local_id, remote_id in bindings.items() + if local_id not in valid_local_ids or (remote_id is not None and remote_id not in valid_remote_ids) + ] + + for local_id in stale_local_ids: + del bindings[local_id] + deleted_count += 1 + if self.auto_persist: + await self.persistence.delete_binding(self.scope, node_type, local_id) + else: + self._get_deleted(node_type).add(local_id) + + return deleted_count + + async def persist(self, *, exclude_node_types: Optional[List[str]] = None) -> None: """将所有绑定关系写回持久化层""" if not self._bindings and not self._deleted: logger.info("🔍 [BindingManager.persist] 跳过：无绑定变更") return + excluded = set(node_type for node_type in (exclude_node_types or []) if node_type) + type_count = 0 saved_total = 0 deleted_total = 0 touched_types: List[str] = [] for node_type, bindings in self._bindings.items(): + if node_type in excluded: + continue deleted = list(self._get_deleted(node_type)) binding_count = len(bindings) deleted_count = len(deleted) @@ -149,13 +196,13 @@ class BindingManager: touched_types.append(node_type) if deleted: - await self.persistence.delete_bindings_bulk(node_type, deleted) + await self.persistence.delete_bindings_bulk(self.scope, node_type, deleted) self._get_deleted(node_type).clear() if bindings: - await self.persistence.save_bindings_bulk(node_type, bindings) + await self.persistence.save_bindings_bulk(self.scope, node_type, bindings) touched_text = ",".join(touched_types) if touched_types else "none" logger.info( - f"🔍 [BindingManager.persist] 完成: types={type_count}, saved={saved_total}, deleted={deleted_total}, touched=[{touched_text}]" + f"🔍 [BindingManager.persist] 完成: scope={self.scope}, types={type_count}, saved={saved_total}, deleted={deleted_total}, touched=[{touched_text}]" ) diff --git a/sync_state_machine/common/collection.py b/sync_state_machine/common/collection.py index f9ba777..0f0f426 100644 --- a/sync_state_machine/common/collection.py +++ b/sync_state_machine/common/collection.py @@ -1,4 +1,5 @@ import uuid +import copy from enum import Enum from typing import Dict, List, Optional, Any, Set, Callable, Literal from .sync_node import SyncNode @@ -22,11 +23,13 @@ class DataCollection: def __init__( self, collection_id: str, + scope: str = "global", persistence: Optional[PersistenceBackend] = None, auto_persist: bool = False, sm_runtime: Optional[Any] = None, ): self.collection_id = collection_id + self.scope = scope self.persistence = persistence self.auto_persist = auto_persist self._sm_runtime = sm_runtime @@ -67,7 +70,7 @@ class DataCollection: self._nodes[node.node_id] = node self._deleted_node_ids.discard(node.node_id) if self.persistence and self.auto_persist: - await self.persistence.save_node(self.collection_id, node) + await self.persistence.save_node(self.collection_id, self.scope, node) async def update(self, node: SyncNode): old_node = self._nodes.get(node.node_id) @@ -85,7 +88,7 @@ class DataCollection: self._nodes[node.node_id] = node self._deleted_node_ids.discard(node.node_id) if self.persistence and self.auto_persist: - await self.persistence.save_node(self.collection_id, node) + await self.persistence.save_node(self.collection_id, self.scope, node) async def delete(self, node_id: str): old_node = self._nodes.get(node_id) @@ -96,7 +99,7 @@ class DataCollection: del self._nodes[node_id] if self.persistence: if self.auto_persist: - await self.persistence.delete_node(self.collection_id, node_id) + await self.persistence.delete_node(self.collection_id, self.scope, node_id) else: self._deleted_node_ids.add(node_id) @@ -370,7 +373,7 @@ class DataCollection: self._data_id_to_node_id = {} self._deleted_node_ids.clear() - node_dicts = await self.persistence.load_nodes(self.collection_id) + node_dicts = await self.persistence.load_nodes(self.collection_id, self.scope) from .types import SyncAction, SyncStatus, BindingStatus from .registry import DomainRegistry @@ -491,7 +494,7 @@ class DataCollection: node_types = {node.node_type for node in self._nodes.values()} for node_type in node_types: - override_map = await self.persistence.load_bind_data_id_overrides(node_type) + override_map = await self.persistence.load_bind_data_id_overrides(node_type, self.scope) for node in self.filter(node_type=node_type): bind_data_id = override_map.get(node.node_id) if bind_data_id: @@ -499,20 +502,169 @@ class DataCollection: else: node.context.pop("bind_data_id", None) - async def persist(self) -> None: + async def load_from_persistence_excluding(self, node_types: Optional[List[str]] = None): + """从持久化恢复数据，但跳过指定 node_type。""" + if not self.persistence: + return + + self._nodes = {} + self._data_id_to_node_id = {} + self._deleted_node_ids.clear() + + node_dicts = await self.persistence.load_nodes( + self.collection_id, + self.scope, + exclude_node_types=node_types, + ) + + from .types import SyncAction, SyncStatus, BindingStatus + from .registry import DomainRegistry + + for node_dict in node_dicts: + node_type = node_dict["node_type"] + node_class = DomainRegistry.get_node_class(node_type) + assert issubclass(node_class, SyncNode) + + parse_errors: list[str] = [] + + raw_action = node_dict.get("action", SyncAction.NONE) + if isinstance(raw_action, str): + try: + action_value = SyncAction(raw_action) + except ValueError: + parse_errors.append(f"invalid action={raw_action}") + action_value = SyncAction.NONE + elif isinstance(raw_action, SyncAction): + action_value = raw_action + else: + if raw_action is None: + action_value = SyncAction.NONE + else: + parse_errors.append(f"invalid action type={type(raw_action).__name__}") + action_value = SyncAction.NONE + + raw_status = node_dict.get("status", SyncStatus.PENDING) + if isinstance(raw_status, str): + try: + status_value = SyncStatus(raw_status) + except ValueError: + parse_errors.append(f"invalid status={raw_status}") + status_value = SyncStatus.FAILED + elif isinstance(raw_status, SyncStatus): + status_value = raw_status + else: + if raw_status is None: + status_value = SyncStatus.PENDING + else: + parse_errors.append(f"invalid status type={type(raw_status).__name__}") + status_value = SyncStatus.FAILED + + raw_binding_status = node_dict.get("binding_status", BindingStatus.UNCHECKED) + if isinstance(raw_binding_status, str): + try: + binding_status_value = BindingStatus(raw_binding_status) + except ValueError: + parse_errors.append(f"invalid binding_status={raw_binding_status}") + binding_status_value = BindingStatus.ABNORMAL + elif isinstance(raw_binding_status, BindingStatus): + binding_status_value = raw_binding_status + else: + if raw_binding_status is None: + binding_status_value = BindingStatus.UNCHECKED + else: + parse_errors.append( + f"invalid binding_status type={type(raw_binding_status).__name__}" + ) + binding_status_value = BindingStatus.ABNORMAL + + raw_context = node_dict.get("context", {}) + context_value = raw_context if isinstance(raw_context, dict) else {} + error_value = node_dict.get("error") + + if parse_errors: + context_value = dict(context_value) + context_value["bootstrap_blocked"] = True + context_value["bootstrap_block_reason"] = "invalid_persisted_enum" + context_value["bootstrap_block_errors"] = parse_errors + binding_status_value = BindingStatus.ABNORMAL + action_value = SyncAction.NONE + status_value = SyncStatus.FAILED + details = "; ".join(parse_errors) + error_value = f"LOAD_INVALID_ENUM: {details}" + + node = node_class( + node_id=node_dict["node_id"], + data_id=node_dict.get("data_id", ""), + depend_ids=node_dict.get("depend_ids", []), + data=node_dict.get("data"), + origin_data=node_dict.get("origin_data"), + action=action_value, + status=status_value, + binding_status=binding_status_value, + error=error_value, + sync_log=None, + context=context_value, + ) + node.context["_loaded_from_persistence"] = True + self._nodes[node.node_id] = node + + if node.data_id and node.data_id != "": + existing_node_id = self._data_id_to_node_id.get(node.data_id) + if existing_node_id is not None and existing_node_id != node.node_id: + raise ValueError( + f"Duplicate data_id detected while loading: {node.data_id} already bound to node_id={existing_node_id}" + ) + self._data_id_to_node_id[node.data_id] = node.node_id + + loaded_node_types = {node.node_type for node in self._nodes.values()} + for node_type in loaded_node_types: + override_map = await self.persistence.load_bind_data_id_overrides(node_type, self.scope) + for node in self.filter(node_type=node_type): + bind_data_id = override_map.get(node.node_id) + if bind_data_id: + node.context["bind_data_id"] = bind_data_id + else: + node.context.pop("bind_data_id", None) + + async def import_nodes(self, nodes: List[SyncNode]) -> None: + """导入现有节点到当前 collection，仅写内存，不触发持久化。""" + for source_node in nodes: + node = copy.deepcopy(source_node) + if node.node_id in self._nodes: + continue + + if node.data_id and node.data_id != "": + existing_node_id = self._data_id_to_node_id.get(node.data_id) + if existing_node_id is not None and existing_node_id != node.node_id: + raise ValueError( + f"Duplicate data_id detected: {node.data_id} already bound to node_id={existing_node_id}" + ) + self._data_id_to_node_id[node.data_id] = node.node_id + + self._nodes[node.node_id] = node + self._deleted_node_ids.discard(node.node_id) + + async def persist(self, *, exclude_node_types: Optional[List[str]] = None) -> None: """将当前 collection 全量持久化到后端""" if not self.persistence: return # 先落库删除（用于 auto_persist=False 的场景） if self._deleted_node_ids: - await self.persistence.delete_nodes_bulk(self.collection_id, list(self._deleted_node_ids)) + await self.persistence.delete_nodes_bulk(self.collection_id, self.scope, list(self._deleted_node_ids)) self._deleted_node_ids.clear() if not self._nodes: return - await self.persistence.save_nodes_bulk(self.collection_id, list(self._nodes.values())) + excluded = set(node_type for node_type in (exclude_node_types or []) if node_type) + nodes_to_persist = [ + node for node in self._nodes.values() if node.node_type not in excluded + ] + if not nodes_to_persist: + return + + await self.persistence.save_nodes_bulk(self.collection_id, self.scope, nodes_to_persist) async def filter_by_project_ids(self, data_id_filter: List[str]) -> int: """按目标项目 data_id 集合过滤当前 collection，返回删除数量。""" diff --git a/sync_state_machine/common/persistence.py b/sync_state_machine/common/persistence.py index 1b3b9f2..61df8e1 100644 --- a/sync_state_machine/common/persistence.py +++ b/sync_state_machine/common/persistence.py @@ -114,6 +114,7 @@ class PersistenceBackend: await self.conn.execute(""" CREATE TABLE IF NOT EXISTS nodes ( collection_id TEXT NOT NULL, + scope TEXT NOT NULL DEFAULT 'global', node_id TEXT NOT NULL, node_type TEXT NOT NULL, data_id TEXT, @@ -126,7 +127,7 @@ class PersistenceBackend: error TEXT, sync_log TEXT, context TEXT, - PRIMARY KEY (collection_id, node_id) + PRIMARY KEY (collection_id, scope, node_id) ) """) await self._ensure_nodes_schema() @@ -140,8 +141,49 @@ class PersistenceBackend: cursor = await self.conn.execute("PRAGMA table_info(nodes)") rows = await cursor.fetchall() columns = [r[1] for r in rows] if rows else [] + if "scope" not in columns: + await self.conn.execute(""" + CREATE TABLE IF NOT EXISTS nodes_new ( + collection_id TEXT NOT NULL, + scope TEXT NOT NULL DEFAULT 'global', + node_id TEXT NOT NULL, + node_type TEXT NOT NULL, + data_id TEXT, + bind_data_id TEXT, + data TEXT, + origin_data TEXT, + action TEXT, + status TEXT, + binding_status TEXT, + error TEXT, + sync_log TEXT, + context TEXT, + PRIMARY KEY (collection_id, scope, node_id) + ) + """) + await self.conn.execute(""" + INSERT INTO nodes_new ( + collection_id, scope, node_id, node_type, data_id, bind_data_id, + data, origin_data, action, status, binding_status, error, sync_log, context + ) + SELECT + collection_id, 'global', node_id, node_type, data_id, bind_data_id, + data, origin_data, action, status, binding_status, error, sync_log, context + FROM nodes + """) + await self.conn.execute("DROP TABLE nodes") + await self.conn.execute("ALTER TABLE nodes_new RENAME TO nodes") + cursor = await self.conn.execute("PRAGMA table_info(nodes)") + rows = await cursor.fetchall() + columns = [r[1] for r in rows] if rows else [] if "bind_data_id" not in columns: await self.conn.execute("ALTER TABLE nodes ADD COLUMN bind_data_id TEXT") + await self.conn.execute( + "CREATE INDEX IF NOT EXISTS idx_nodes_collection_scope_type ON nodes (collection_id, scope, node_type)" + ) + await self.conn.execute( + "CREATE INDEX IF NOT EXISTS idx_nodes_collection_scope_data_id ON nodes (collection_id, scope, data_id)" + ) async def _ensure_bindings_schema(self) -> None: """确保 bindings 表为 (node_type, local_id, remote_id) 结构。 @@ -155,46 +197,60 @@ class PersistenceBackend: if not columns: await self.conn.execute(""" CREATE TABLE IF NOT EXISTS bindings ( + scope TEXT NOT NULL DEFAULT 'global', node_type TEXT NOT NULL, local_id TEXT NOT NULL, remote_id TEXT, last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - PRIMARY KEY (node_type, local_id) + PRIMARY KEY (scope, node_type, local_id) ) """) return - if "payload" in columns and "remote_id" not in columns: - # 迁移旧结构 payload -> remote_id + if "scope" not in columns or ("payload" in columns and "remote_id" not in columns): + # 迁移旧结构到 scope + remote_id 新结构 await self.conn.execute(""" CREATE TABLE IF NOT EXISTS bindings_new ( + scope TEXT NOT NULL DEFAULT 'global', node_type TEXT NOT NULL, local_id TEXT NOT NULL, remote_id TEXT, last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - PRIMARY KEY (node_type, local_id) + PRIMARY KEY (scope, node_type, local_id) ) """) - async with self.conn.execute("SELECT node_type, local_id, payload FROM bindings") as cur: - old_rows = await cur.fetchall() - for row in old_rows: - try: - payload_dict = json.loads(row[2]) if row[2] else {} - except json.JSONDecodeError: - payload_dict = {} - remote_id = payload_dict.get("remote_id") - await self.conn.execute( - "INSERT OR REPLACE INTO bindings_new (node_type, local_id, remote_id) VALUES (?, ?, ?)", - (row[0], row[1], remote_id), - ) + if "payload" in columns and "remote_id" not in columns: + async with self.conn.execute("SELECT node_type, local_id, payload FROM bindings") as cur: + old_rows = await cur.fetchall() + for row in old_rows: + try: + payload_dict = json.loads(row[2]) if row[2] else {} + except json.JSONDecodeError: + payload_dict = {} + remote_id = payload_dict.get("remote_id") + await self.conn.execute( + "INSERT OR REPLACE INTO bindings_new (scope, node_type, local_id, remote_id) VALUES (?, ?, ?, ?)", + ("global", row[0], row[1], remote_id), + ) + else: + async with self.conn.execute("SELECT node_type, local_id, remote_id FROM bindings") as cur: + old_rows = await cur.fetchall() + for row in old_rows: + await self.conn.execute( + "INSERT OR REPLACE INTO bindings_new (scope, node_type, local_id, remote_id) VALUES (?, ?, ?, ?)", + ("global", row[0], row[1], row[2]), + ) await self.conn.execute("DROP TABLE bindings") await self.conn.execute("ALTER TABLE bindings_new RENAME TO bindings") + await self.conn.execute( + "CREATE INDEX IF NOT EXISTS idx_bindings_scope_type_remote ON bindings (scope, node_type, remote_id)" + ) # --- DataCollection 支持 --- - async def save_node(self, collection_id: str, node: "SyncNode"): + async def save_node(self, collection_id: str, scope: str, node: "SyncNode"): """保存或更新单个节点（depend_ids不持久化，运行时从data实时计算）""" bind_data_id = None if isinstance(node.context, dict): @@ -203,12 +259,13 @@ class PersistenceBackend: bind_data_id = str(value) await self.conn.execute(""" INSERT OR REPLACE INTO nodes ( - collection_id, node_id, node_type, data_id, bind_data_id, + collection_id, scope, node_id, node_type, data_id, bind_data_id, data, origin_data, action, status, binding_status, error, sync_log, context ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( collection_id, + scope, node.node_id, node.node_type, node.data_id, @@ -224,7 +281,7 @@ class PersistenceBackend: )) await self.conn.commit() - async def save_nodes_bulk(self, collection_id: str, nodes: List["SyncNode"]): + async def save_nodes_bulk(self, collection_id: str, scope: str, nodes: List["SyncNode"]): """批量保存或更新节点 (单次提交，depend_ids不持久化)""" if not nodes: return @@ -237,6 +294,7 @@ class PersistenceBackend: bind_data_id = str(value) rows.append(( collection_id, + scope, node.node_id, node.node_type, node.data_id, @@ -253,32 +311,49 @@ class PersistenceBackend: await self.conn.executemany(""" INSERT OR REPLACE INTO nodes ( - collection_id, node_id, node_type, data_id, bind_data_id, + collection_id, scope, node_id, node_type, data_id, bind_data_id, data, origin_data, action, status, binding_status, error, sync_log, context ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, rows) await self.conn.commit() - async def delete_node(self, collection_id: str, node_id: str): + async def delete_node(self, collection_id: str, scope: str, node_id: str): """删除单个节点""" - await self.conn.execute("DELETE FROM nodes WHERE collection_id = ? AND node_id = ?", (collection_id, node_id)) + await self.conn.execute( + "DELETE FROM nodes WHERE collection_id = ? AND scope = ? AND node_id = ?", + (collection_id, scope, node_id), + ) await self.conn.commit() - async def delete_nodes_bulk(self, collection_id: str, node_ids: List[str]): + async def delete_nodes_bulk(self, collection_id: str, scope: str, node_ids: List[str]): """批量删除节点 (单次提交)""" if not node_ids: return - rows = [(collection_id, node_id) for node_id in node_ids] + rows = [(collection_id, scope, node_id) for node_id in node_ids] await self.conn.executemany( - "DELETE FROM nodes WHERE collection_id = ? AND node_id = ?", + "DELETE FROM nodes WHERE collection_id = ? AND scope = ? AND node_id = ?", rows, ) await self.conn.commit() - async def load_nodes(self, collection_id: str) -> List[Dict[str, Any]]: + async def load_nodes( + self, + collection_id: str, + scope: str, + *, + exclude_node_types: Optional[List[str]] = None, + ) -> List[Dict[str, Any]]: """从特定 collection 中加载所有节点（depend_ids初始化为空，稍后从data实时计算）""" - async with self.conn.execute("SELECT * FROM nodes WHERE collection_id = ?", (collection_id,)) as cursor: + sql = "SELECT * FROM nodes WHERE collection_id = ? AND scope = ?" + params: List[Any] = [collection_id, scope] + filtered_types = [node_type for node_type in (exclude_node_types or []) if node_type] + if filtered_types: + placeholders = ", ".join("?" for _ in filtered_types) + sql += f" AND node_type NOT IN ({placeholders})" + params.extend(filtered_types) + + async with self.conn.execute(sql, tuple(params)) as cursor: rows = await cursor.fetchall() results = [] for row in rows: @@ -303,52 +378,52 @@ class PersistenceBackend: # --- BindingManager 支持 --- - async def save_binding(self, node_type: str, local_id: str, payload_dict: Dict[str, Any]): + async def save_binding(self, scope: str, node_type: str, local_id: str, payload_dict: Dict[str, Any]): """保存或更新绑定关系及其元数据""" remote_id = payload_dict.get("remote_id") if payload_dict else None await self.conn.execute(""" - INSERT OR REPLACE INTO bindings (node_type, local_id, remote_id) - VALUES (?, ?, ?) - """, (node_type, local_id, remote_id)) + INSERT OR REPLACE INTO bindings (scope, node_type, local_id, remote_id) + VALUES (?, ?, ?, ?) + """, (scope, node_type, local_id, remote_id)) await self.conn.commit() - async def save_bindings_bulk(self, node_type: str, bindings: Dict[str, Optional[str]]): + async def save_bindings_bulk(self, scope: str, node_type: str, bindings: Dict[str, Optional[str]]): """批量保存或更新绑定关系 (单次提交)""" if not bindings: return rows = [] for local_id, remote_id in bindings.items(): - rows.append((node_type, local_id, remote_id)) + rows.append((scope, node_type, local_id, remote_id)) await self.conn.executemany( - "INSERT OR REPLACE INTO bindings (node_type, local_id, remote_id) VALUES (?, ?, ?)", + "INSERT OR REPLACE INTO bindings (scope, node_type, local_id, remote_id) VALUES (?, ?, ?, ?)", rows, ) await self.conn.commit() - async def delete_binding(self, node_type: str, local_id: str): + async def delete_binding(self, scope: str, node_type: str, local_id: str): """删除绑定关系""" await self.conn.execute( - "DELETE FROM bindings WHERE node_type = ? AND local_id = ?", - (node_type, local_id) + "DELETE FROM bindings WHERE scope = ? AND node_type = ? AND local_id = ?", + (scope, node_type, local_id) ) await self.conn.commit() - async def delete_bindings_bulk(self, node_type: str, local_ids: List[str]): + async def delete_bindings_bulk(self, scope: str, node_type: str, local_ids: List[str]): """批量删除绑定关系 (单次提交)""" if not local_ids: return - rows = [(node_type, local_id) for local_id in local_ids] + rows = [(scope, node_type, local_id) for local_id in local_ids] await self.conn.executemany( - "DELETE FROM bindings WHERE node_type = ? AND local_id = ?", + "DELETE FROM bindings WHERE scope = ? AND node_type = ? AND local_id = ?", rows, ) await self.conn.commit() - async def load_bindings(self, node_type: str) -> List[Dict[str, Any]]: + async def load_bindings(self, scope: str, node_type: str) -> List[Dict[str, Any]]: """回写该类型下的所有绑定关系到内存""" async with self.conn.execute( - "SELECT local_id, remote_id FROM bindings WHERE node_type = ?", - (node_type,) + "SELECT local_id, remote_id FROM bindings WHERE scope = ? AND node_type = ?", + (scope, node_type) ) as cursor: rows = await cursor.fetchall() return [ @@ -356,7 +431,7 @@ class PersistenceBackend: for r in rows ] - async def load_bind_data_id_overrides(self, node_type: str) -> Dict[str, Optional[str]]: + async def load_bind_data_id_overrides(self, node_type: str, scope: str) -> Dict[str, Optional[str]]: """基于 bindings 表计算 node_id -> bind_data_id 覆盖映射。 规则： @@ -369,7 +444,9 @@ class PersistenceBackend: LEFT JOIN nodes rn ON rn.node_id = b.remote_id AND rn.node_type = b.node_type + AND rn.scope = b.scope WHERE b.node_type = ? + AND b.scope = ? UNION ALL @@ -378,10 +455,12 @@ class PersistenceBackend: LEFT JOIN nodes ln ON ln.node_id = b.local_id AND ln.node_type = b.node_type + AND ln.scope = b.scope WHERE b.node_type = ? + AND b.scope = ? AND b.remote_id IS NOT NULL """ - async with self.conn.execute(sql, (node_type, node_type)) as cursor: + async with self.conn.execute(sql, (node_type, scope, node_type, scope)) as cursor: rows = await cursor.fetchall() result: Dict[str, Optional[str]] = {} for row in rows: @@ -391,6 +470,74 @@ class PersistenceBackend: result[str(src_node_id)] = row["bind_data_id"] return result + async def delete_scope_node_types( + self, + *, + scope: str, + node_types: List[str], + ) -> None: + filtered_types = [node_type for node_type in node_types if node_type] + if not filtered_types: + return + + placeholders = ", ".join("?" for _ in filtered_types) + await self.conn.execute( + f"DELETE FROM nodes WHERE scope = ? AND node_type IN ({placeholders})", + (scope, *filtered_types), + ) + await self.conn.execute( + f"DELETE FROM bindings WHERE scope = ? AND node_type IN ({placeholders})", + (scope, *filtered_types), + ) + await self.conn.commit() + + async def copy_nodes_between_scopes( + self, + *, + collection_id: str, + source_scope: str, + target_scope: str, + node_types: List[str], + ) -> None: + if source_scope == target_scope or not node_types: + return + placeholders = ", ".join("?" for _ in node_types) + sql = f""" + INSERT OR REPLACE INTO nodes ( + collection_id, scope, node_id, node_type, data_id, bind_data_id, + data, origin_data, action, status, binding_status, error, sync_log, context + ) + SELECT + collection_id, ?, node_id, node_type, data_id, bind_data_id, + data, origin_data, action, status, binding_status, error, sync_log, context + FROM nodes + WHERE collection_id = ? + AND scope = ? + AND node_type IN ({placeholders}) + """ + await self.conn.execute(sql, (target_scope, collection_id, source_scope, *node_types)) + await self.conn.commit() + + async def copy_bindings_between_scopes( + self, + *, + source_scope: str, + target_scope: str, + node_types: List[str], + ) -> None: + if source_scope == target_scope or not node_types: + return + placeholders = ", ".join("?" for _ in node_types) + sql = f""" + INSERT OR REPLACE INTO bindings (scope, node_type, local_id, remote_id) + SELECT ?, node_type, local_id, remote_id + FROM bindings + WHERE scope = ? + AND node_type IN ({placeholders}) + """ + await self.conn.execute(sql, (target_scope, source_scope, *node_types)) + await self.conn.commit() + async def dump_to_runtime(self, filename: str): """将当前状态转储到 _runtime 文件夹 (使用 VACUUM INTO)""" runtime_path = os.path.join("_runtime", filename) diff --git a/sync_state_machine/config/__init__.py b/sync_state_machine/config/__init__.py index 56d8b06..4939566 100644 --- a/sync_state_machine/config/__init__.py +++ b/sync_state_machine/config/__init__.py @@ -24,6 +24,12 @@ from .strategy_config import ( ConfigPresets, ) from .pipeline_config import PipelineRunConfig, DEFAULT_NODE_TYPES +from .multi_project_config import ( + MultiProjectItemConfig, + MultiProjectRunConfig, + build_multi_project_config_from_file, + is_multi_project_payload, +) from .builder import ( build_default_config, build_config, @@ -58,6 +64,10 @@ __all__ = [ "StrategyRuntimeConfig", "ConfigPresets", "DEFAULT_NODE_TYPES", + "MultiProjectItemConfig", + "MultiProjectRunConfig", + "build_multi_project_config_from_file", + "is_multi_project_payload", "build_default_config", "build_config", "build_config_from_file", diff --git a/sync_state_machine/config/builder.py b/sync_state_machine/config/builder.py index 3ddf1f9..d02eb8d 100644 --- a/sync_state_machine/config/builder.py +++ b/sync_state_machine/config/builder.py @@ -63,13 +63,22 @@ def _apply_strategy_overrides( runtime_cfg = strategy_map[node_type] override = dict(raw_override or {}) + deprecated_keys = [key for key in ("force_sync", "load_mode") if key in override] + if deprecated_keys: + joined = ", ".join(deprecated_keys) + raise ValueError( + f"Strategy override for {node_type!r} contains unsupported legacy keys: {joined}. " + "Use skip_sync under strategies. or datasource.handler_configs instead." + ) + preset = override.pop("preset", None) if preset: runtime_cfg.config = ConfigPresets.get_preset(str(preset)) - override.pop("skip_sync", None) - override.pop("force_sync", None) - override.pop("load_mode", None) + skip_sync = override.pop("skip_sync", None) + + if skip_sync is not None: + runtime_cfg.skip_sync = bool(skip_sync) if "config" in override and isinstance(override["config"], dict): runtime_cfg.config = runtime_cfg.config.model_validate( diff --git a/sync_state_machine/config/multi_project_config.py b/sync_state_machine/config/multi_project_config.py new file mode 100644 index 0000000..7584e62 --- /dev/null +++ b/sync_state_machine/config/multi_project_config.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, ConfigDict + +from .run_presets import load_overrides_from_file + + +class MultiProjectItemConfig(BaseModel): + model_config = ConfigDict(extra="forbid", validate_assignment=True) + + local_project_id: str + remote_project_id: str + config_path: Optional[str] = None + + @property + def scope(self) -> str: + return f"project_id:{self.local_project_id}" + + +class MultiProjectRunConfig(BaseModel): + model_config = ConfigDict(extra="forbid", validate_assignment=True) + + global_config: str + default_project_config: str + project_bootstrap_node_types: List[str] = [] + projects: List[MultiProjectItemConfig] + + +def is_multi_project_payload(payload: Dict[str, Any]) -> bool: + return all(key in payload for key in ("global_config", "default_project_config", "projects")) + + +def build_multi_project_config_from_file(*, project_root: Path, file_path: str | Path) -> MultiProjectRunConfig: + path = Path(file_path) + overrides = load_overrides_from_file(path, project_root) + return MultiProjectRunConfig.model_validate(overrides) \ No newline at end of file diff --git a/sync_state_machine/config/runtime_config.py b/sync_state_machine/config/runtime_config.py index 99fc37d..301012f 100644 --- a/sync_state_machine/config/runtime_config.py +++ b/sync_state_machine/config/runtime_config.py @@ -14,6 +14,12 @@ from .strategy_config import ( ConfigPresets, ) from .pipeline_config import PipelineRunConfig, DEFAULT_NODE_TYPES +from .multi_project_config import ( + MultiProjectItemConfig, + MultiProjectRunConfig, + build_multi_project_config_from_file, + is_multi_project_payload, +) from .builder import ( build_default_config, build_config, @@ -35,6 +41,10 @@ __all__ = [ "ConfigPresets", "PipelineRunConfig", "DEFAULT_NODE_TYPES", + "MultiProjectItemConfig", + "MultiProjectRunConfig", + "build_multi_project_config_from_file", + "is_multi_project_payload", "build_default_config", "build_config", "build_config_from_file", diff --git a/sync_state_machine/config/strategy_config.py b/sync_state_machine/config/strategy_config.py index e1e79fc..cad8314 100644 --- a/sync_state_machine/config/strategy_config.py +++ b/sync_state_machine/config/strategy_config.py @@ -47,10 +47,9 @@ class StrategyConfig(BaseModel): field_direction_key: Optional[str] = None field_direction_overrides: Dict[str, UpdateDirection] = Field(default_factory=dict) - # post-check 一致性比较时，忽略列表型字段内各元素的特定子键。 - # 格式: {字段名: [子键, ...]}，如 {"plan_node": ["is_audit"]}。 - # 用于排除后端硬写死、永远与本地不一致的字段，避免误报。 - post_check_ignore_list_item_fields: Dict[str, List[str]] = Field(default_factory=dict) + # post-check 一致性比较时忽略的顶层字段。 + # 只影响最终一致性评估，不影响 create/update 的差异检测。 + post_check_ignore_fields: List[str] = Field(default_factory=list) # schema diff / validate 时忽略的顶层字段。 compare_ignore_fields: List[str] = Field(default_factory=list) @@ -62,6 +61,7 @@ class StrategyRuntimeConfig(BaseModel): model_config = ConfigDict(validate_assignment=True, extra="forbid") node_type: str + skip_sync: bool = False config: StrategyConfig = Field(default_factory=StrategyConfig) diff --git a/sync_state_machine/datasource/handler.py b/sync_state_machine/datasource/handler.py index c5ee35e..65b96f3 100644 --- a/sync_state_machine/datasource/handler.py +++ b/sync_state_machine/datasource/handler.py @@ -361,11 +361,15 @@ class BaseNodeHandler(ABC, Generic[T]): self.handler_config = dict(config) def get_data_id_filter(self) -> List[str]: - value = self.handler_config.get("data_id_filter", []) - if isinstance(value, str): - return [value] if value else [] - if isinstance(value, list): - return [str(item) for item in value if str(item)] + if "data_id_filter" in self.handler_config: + value = self.handler_config.get("data_id_filter", []) + if isinstance(value, str): + return [value] if value else [] + if isinstance(value, list): + return [str(item) for item in value if str(item)] + return [] + if self.node_type != "project" and self._collection is not None: + return [node.data_id for node in self._collection.filter(node_type="project") if node.data_id] return [] def should_skip_by_data_id_filter(self, item: Dict[str, Any], item_id: str) -> bool: diff --git a/sync_state_machine/datasource/jsonl/datasource.py b/sync_state_machine/datasource/jsonl/datasource.py index 6e3fc8f..6429a32 100644 --- a/sync_state_machine/datasource/jsonl/datasource.py +++ b/sync_state_machine/datasource/jsonl/datasource.py @@ -10,8 +10,10 @@ JsonlDataSource - JSONL 文件数据源 from __future__ import annotations +import json +import re from pathlib import Path -from typing import Dict, Any, Set +from typing import Dict, Any, Set, Callable, List, Optional from ..datasource import BaseDataSource @@ -46,6 +48,7 @@ class JsonlDataSource(BaseDataSource): # 用于 create/update/delete 的内存缓存 self._data: Dict[str, Dict[str, Dict[str, Any]]] = {} # {node_type: {id: data}} self._dirty_types: Set[str] = set() + self._node_items_cache: Dict[str, Dict[str, Any]] = {} def _on_node_loaded(self, handler, node) -> None: node.append_log( @@ -61,6 +64,98 @@ class JsonlDataSource(BaseDataSource): node.append_log( f"JSONL链路[poll] task_id={task_id} status={result.status.value}" ) + + def invalidate_node_cache(self, node_type: str) -> None: + self._node_items_cache.pop(node_type, None) + + def _matching_files(self, node_type: str) -> List[Path]: + if not self.dir_path.exists(): + return [] + pattern = re.compile(rf"^{re.escape(node_type)}_\d+\.jsonl$") + return sorted( + file_path + for file_path in self.dir_path.iterdir() + if pattern.match(file_path.name) + ) + + def _build_cache_entry( + self, + *, + node_type: str, + extract_id: Callable[[Dict[str, Any]], str], + ) -> Dict[str, Any]: + files = self._matching_files(node_type) + signature = tuple( + (file_path.name, file_path.stat().st_mtime_ns, file_path.stat().st_size) + for file_path in files + ) + cached = self._node_items_cache.get(node_type) + if cached and cached.get("signature") == signature: + return cached + + items: List[Dict[str, Any]] = [] + data_id_index: Dict[str, List[Dict[str, Any]]] = {} + project_index: Dict[str, List[Dict[str, Any]]] = {} + no_project_items: List[Dict[str, Any]] = [] + + for file_path in files: + with open(file_path, "r", encoding="utf-8") as f: + for line_num, line in enumerate(f, 1): + line = line.strip() + if not line: + continue + + try: + item = json.loads(line) + except json.JSONDecodeError: + print(f"Warning: Invalid JSON in {file_path.name}:{line_num}") + continue + + items.append(item) + + item_id = extract_id(item) + if item_id: + data_id_index.setdefault(str(item_id), []).append(item) + + project_id = item.get("project_id") + if project_id: + project_index.setdefault(str(project_id), []).append(item) + else: + no_project_items.append(item) + + entry = { + "signature": signature, + "items": items, + "data_id_index": data_id_index, + "project_index": project_index, + "no_project_items": no_project_items, + } + self._node_items_cache[node_type] = entry + return entry + + def get_items_for_load( + self, + *, + node_type: str, + extract_id: Callable[[Dict[str, Any]], str], + data_id_filter: Optional[List[str]] = None, + filter_on_project_id: bool = False, + ) -> List[Dict[str, Any]]: + entry = self._build_cache_entry(node_type=node_type, extract_id=extract_id) + filters = [str(item) for item in (data_id_filter or []) if str(item)] + if not filters: + return list(entry["items"]) + + if not filter_on_project_id: + matched: List[Dict[str, Any]] = [] + for data_id in filters: + matched.extend(entry["data_id_index"].get(data_id, [])) + return matched + + matched = list(entry["no_project_items"]) + for project_id in filters: + matched.extend(entry["project_index"].get(project_id, [])) + return matched # load_all 和 sync_all 继承自 BaseDataSource # 保存逻辑在 BaseJsonlHandler 中实现 diff --git a/sync_state_machine/datasource/jsonl/handler.py b/sync_state_machine/datasource/jsonl/handler.py index b0305d1..fb91601 100644 --- a/sync_state_machine/datasource/jsonl/handler.py +++ b/sync_state_machine/datasource/jsonl/handler.py @@ -12,7 +12,6 @@ BaseJsonlHandler - JSONL Handler 基类 from __future__ import annotations import json -import re from typing import Dict, List, Any, TYPE_CHECKING, Optional, Set from uuid import uuid4 @@ -64,40 +63,26 @@ class BaseJsonlHandler(BaseNodeHandler): if not self.datasource.dir_path.exists(): return nodes - - # 查找所有匹配的文件：{node_type}_N.jsonl - pattern = re.compile(rf"^{re.escape(self.node_type)}_\d+\.jsonl$") - - for file_path in self.datasource.dir_path.iterdir(): - if not pattern.match(file_path.name): - continue - - # 读取文件 - with open(file_path, "r", encoding="utf-8") as f: - for line_num, line in enumerate(f, 1): - line = line.strip() - if not line: - continue - - try: - item = json.loads(line) - item_id = self.extract_id(item) - if self.should_skip_by_data_id_filter(item, item_id): - continue - - # 创建节点 - node = self.create_node(item) - nodes.append(node) - - # 同时加载到缓存 - if item_id: - bucket = self.datasource._data.setdefault(self.node_type, {}) - bucket[item_id] = item - - except json.JSONDecodeError as e: - print(f"Warning: Invalid JSON in {file_path.name}:{line_num}: {e}") - # 继续处理下一行 + data_id_filter = self.get_data_id_filter() + items = self.datasource.get_items_for_load( + node_type=self.node_type, + extract_id=self.extract_id, + data_id_filter=data_id_filter, + filter_on_project_id=self.node_type != "project", + ) + + for item in items: + item_id = self.extract_id(item) + if self.should_skip_by_data_id_filter(item, item_id): + continue + + node = self.create_node(item) + nodes.append(node) + + if item_id: + bucket = self.datasource._data.setdefault(self.node_type, {}) + bucket[item_id] = item return nodes @@ -276,6 +261,7 @@ class BaseJsonlHandler(BaseNodeHandler): # 清除该类型的 dirty 标记 self.datasource._dirty_types.discard(self.node_type) + self.datasource.invalidate_node_cache(self.node_type) except Exception as e: print(f"Error saving {self.node_type}: {str(e)}") diff --git a/sync_state_machine/domain/construction/sync_strategy.py b/sync_state_machine/domain/construction/sync_strategy.py index 6c25cee..21217ec 100644 --- a/sync_state_machine/domain/construction/sync_strategy.py +++ b/sync_state_machine/domain/construction/sync_strategy.py @@ -18,3 +18,13 @@ class ConstructionTaskSyncStrategy(DefaultSyncStrategy[ConstructionResponse]): remote_orphan_action=OrphanAction.NONE, update_direction=UpdateDirection.PUSH, ) + + def preprocess_compare_payload(self, payload: dict) -> dict: + plan_nodes = payload.get("plan_node") + if not isinstance(plan_nodes, list): + return payload + payload["plan_node"] = [ + {key: value for key, value in item.items() if key != "is_audit"} if isinstance(item, dict) else item + for item in plan_nodes + ] + return payload diff --git a/sync_state_machine/domain/contract_settlement/sync_strategy.py b/sync_state_machine/domain/contract_settlement/sync_strategy.py index 009387b..1d5da17 100644 --- a/sync_state_machine/domain/contract_settlement/sync_strategy.py +++ b/sync_state_machine/domain/contract_settlement/sync_strategy.py @@ -17,4 +17,14 @@ class ContractSettlementSyncStrategy(DefaultSyncStrategy[ContractSettlementRespo local_orphan_action=OrphanAction.CREATE_REMOTE, remote_orphan_action=OrphanAction.NONE, update_direction=UpdateDirection.PUSH, - ) \ No newline at end of file + ) + + def preprocess_compare_payload(self, payload: dict) -> dict: + plan_nodes = payload.get("plan_node") + if not isinstance(plan_nodes, list): + return payload + payload["plan_node"] = [ + {key: value for key, value in item.items() if key != "is_audit"} if isinstance(item, dict) else item + for item in plan_nodes + ] + return payload \ No newline at end of file diff --git a/sync_state_machine/domain/material/sync_strategy.py b/sync_state_machine/domain/material/sync_strategy.py index 43b6ad9..e5c7967 100644 --- a/sync_state_machine/domain/material/sync_strategy.py +++ b/sync_state_machine/domain/material/sync_strategy.py @@ -17,4 +17,14 @@ class MaterialSyncStrategy(DefaultSyncStrategy[MaterialResponse]): local_orphan_action=OrphanAction.CREATE_REMOTE, remote_orphan_action=OrphanAction.NONE, update_direction=UpdateDirection.PUSH, - ) \ No newline at end of file + ) + + def preprocess_compare_payload(self, payload: dict) -> dict: + plan_nodes = payload.get("plan_node") + if not isinstance(plan_nodes, list): + return payload + payload["plan_node"] = [ + {key: value for key, value in item.items() if key != "is_audit"} if isinstance(item, dict) else item + for item in plan_nodes + ] + return payload \ No newline at end of file diff --git a/sync_state_machine/pipeline/__init__.py b/sync_state_machine/pipeline/__init__.py index ca078f4..6e19dfe 100644 --- a/sync_state_machine/pipeline/__init__.py +++ b/sync_state_machine/pipeline/__init__.py @@ -4,6 +4,7 @@ Pipeline Layer - Pipeline orchestration and workflow management from .copy_data_pipeline import CopyDataPipeline from .full_sync_pipeline import FullSyncPipeline +from .multi_project_pipeline import MultiProjectPipeline from ..config import ( PipelineRunConfig, ApiDataSourceConfig, @@ -24,11 +25,13 @@ from .factory import ( create_pipeline_from_config, run_pipeline_from_file, run_pipeline_from_config, + run_profile_from_file, ) __all__ = [ "CopyDataPipeline", "FullSyncPipeline", + "MultiProjectPipeline", "create_jsonl_to_api_pipeline", "create_jsonl_to_jsonl_pipeline", "create_api_to_jsonl_pipeline", @@ -36,6 +39,7 @@ __all__ = [ "create_pipeline_from_config", "run_pipeline_from_file", "run_pipeline_from_config", + "run_profile_from_file", "PipelineRunConfig", "ApiDataSourceConfig", "JsonlDataSourceConfig", diff --git a/sync_state_machine/pipeline/factory.py b/sync_state_machine/pipeline/factory.py index 57cb82a..8c5d595 100644 --- a/sync_state_machine/pipeline/factory.py +++ b/sync_state_machine/pipeline/factory.py @@ -27,18 +27,23 @@ from ..datasource import ensure_builtin_datasource_types_registered from ..datasource.type_registry import get_datasource_factory from ..config import ( PipelineRunConfig, + MultiProjectRunConfig, DataSourceConfig, ApiDataSourceConfig, PersistConfig, JsonlDataSourceConfig, LoggingConfig, + build_config, StrategyRuntimeConfig, build_config_from_file, + build_multi_project_config_from_file, apply_config_overrides, OrphanAction, UpdateDirection, resolve_log_file_path, config_snapshot, + is_multi_project_payload, + load_overrides_from_file, ) from ..sync_system.strategy import BaseSyncStrategy from .run_logger import init_pipeline_logger @@ -180,6 +185,17 @@ def _register_handlers( ) +def _build_collections_and_bindings( + *, + persistence: PersistenceBackend, + scope: str, +) -> tuple[DataCollection, DataCollection, BindingManager]: + local_collection = DataCollection("local", scope=scope, persistence=persistence, auto_persist=False) + remote_collection = DataCollection("remote", scope=scope, persistence=persistence, auto_persist=False) + binding_manager = BindingManager(persistence, auto_persist=False, scope=scope) + return local_collection, remote_collection, binding_manager + + def _resolve_strategy_map(config: PipelineRunConfig, node_types: List[str]) -> Dict[str, StrategyRuntimeConfig]: if config.strategies: return {item.node_type: item for item in config.strategies} @@ -196,6 +212,28 @@ def _resolve_strategy_map(config: PipelineRunConfig, node_types: List[str]) -> D return resolved +def _build_pipeline_run_config( + *, + node_types: List[str], + local_datasource: DataSourceConfig, + remote_datasource: DataSourceConfig, + persist: PersistConfig, + logging_config: LoggingConfig, + strategy_overrides: Optional[Dict[str, Dict[str, Any]]] = None, +) -> PipelineRunConfig: + cfg = PipelineRunConfig( + node_types=node_types, + local_datasource=local_datasource, + remote_datasource=remote_datasource, + strategies=[], + persist=persist, + logging=logging_config, + ) + if strategy_overrides: + cfg = apply_config_overrides(cfg, {"strategies": strategy_overrides}) + return cfg + + def _build_strategies( config: PipelineRunConfig, node_types: List[str], @@ -225,6 +263,7 @@ def _build_strategies( runtime_cfg = strategy_map.get(node_type) if runtime_cfg is not None: strategy.set_config(runtime_cfg.config.model_copy(deep=True)) + strategy.skip_sync = bool(runtime_cfg.skip_sync) strategies.append(strategy) return strategies @@ -233,6 +272,10 @@ def _build_strategies( async def create_pipeline_from_config( config: PipelineRunConfig, *, + scope: str = "global", + pipeline_name: str | None = None, + bootstrap_binding_node_types: Optional[List[str]] = None, + ephemeral_node_types: Optional[List[str]] = None, local_datasource_override=None, remote_datasource_override=None, ) -> FullSyncPipeline: @@ -279,9 +322,10 @@ async def create_pipeline_from_config( await _initialize_datasource(remote_ds, endpoint_name="remote") _register_handlers(config, local_ds, remote_ds, all_types) - local_collection = DataCollection("local", persistence, auto_persist=False) - remote_collection = DataCollection("remote", persistence, auto_persist=False) - binding_manager = BindingManager(persistence, auto_persist=False) + local_collection, remote_collection, binding_manager = _build_collections_and_bindings( + persistence=persistence, + scope=scope, + ) strategies = _build_strategies( config, @@ -301,6 +345,10 @@ async def create_pipeline_from_config( local_collection=local_collection, remote_collection=remote_collection, binding_manager=binding_manager, + scope=scope, + pipeline_name=pipeline_name, + bootstrap_binding_node_types=bootstrap_binding_node_types, + ephemeral_node_types=ephemeral_node_types, enable_persistence=config.persist.enable, ) @@ -325,6 +373,8 @@ async def run_pipeline_from_config( *, title: str = "sync_state_machine pipeline", print_summary: bool = True, + scope: str = "global", + bootstrap_binding_node_types: Optional[List[str]] = None, local_datasource_override=None, remote_datasource_override=None, ) -> Dict[str, Any]: @@ -341,7 +391,7 @@ async def run_pipeline_from_config( resolved_cfg = config_snapshot(config) resolved_cfg_text = json.dumps(resolved_cfg, ensure_ascii=False, indent=2) logger.info( - "Resolved Config Summary: node_types=%d local=%s remote=%s", + "Resolved Full Config Summary: node_types=%d local=%s remote=%s", len(config.node_types), config.local_datasource.type, config.remote_datasource.type, @@ -349,7 +399,7 @@ async def run_pipeline_from_config( if config.logging.file_path: try: with open(config.logging.file_path, "a", encoding="utf-8") as fp: - fp.write("Resolved Config:\n") + fp.write("Resolved Full Config:\n") fp.write(resolved_cfg_text) fp.write("\n") except Exception as exc: @@ -362,15 +412,18 @@ async def run_pipeline_from_config( print(f"🚀 Creating {title}") print(f"📝 Log file: {config.logging.file_path or '-'}") print( - f"🔧 Resolved Config Summary: node_types={len(config.node_types)} " + f"🔧 Resolved Full Config Summary: node_types={len(config.node_types)} " f"local={config.local_datasource.type} " f"remote={config.remote_datasource.type}" ) - print(f"🔧 Resolved Config:\n{resolved_cfg_text}") + print(f"🔧 Resolved Full Config:\n{resolved_cfg_text}") print("=" * 80) pipeline = await create_pipeline_from_config( config, + scope=scope, + pipeline_name=title, + bootstrap_binding_node_types=bootstrap_binding_node_types, local_datasource_override=local_datasource_override, remote_datasource_override=remote_datasource_override, ) @@ -390,12 +443,20 @@ async def create_pipeline_from_file( *, project_root: Path, config_path: str | Path, + scope: str = "global", + pipeline_name: str | None = None, + bootstrap_binding_node_types: Optional[List[str]] = None, + ephemeral_node_types: Optional[List[str]] = None, local_datasource_override=None, remote_datasource_override=None, ) -> FullSyncPipeline: config = build_config_from_file(project_root=project_root, file_path=config_path) return await create_pipeline_from_config( config, + scope=scope, + pipeline_name=pipeline_name, + bootstrap_binding_node_types=bootstrap_binding_node_types, + ephemeral_node_types=ephemeral_node_types, local_datasource_override=local_datasource_override, remote_datasource_override=remote_datasource_override, ) @@ -407,6 +468,8 @@ async def run_pipeline_from_file( config_path: str | Path, title: str = "sync_state_machine pipeline", print_summary: bool = True, + scope: str = "global", + bootstrap_binding_node_types: Optional[List[str]] = None, local_datasource_override=None, remote_datasource_override=None, ) -> Dict[str, Any]: @@ -415,11 +478,41 @@ async def run_pipeline_from_file( config, title=title, print_summary=print_summary, + scope=scope, + bootstrap_binding_node_types=bootstrap_binding_node_types, local_datasource_override=local_datasource_override, remote_datasource_override=remote_datasource_override, ) +async def run_profile_from_file( + *, + project_root: Path, + config_path: str | Path, + print_summary: bool = True, +) -> Dict[str, Any]: + cfg_path = Path(config_path) + raw = load_overrides_from_file(cfg_path, project_root) + if is_multi_project_payload(raw): + from .multi_project_pipeline import MultiProjectPipeline + + multi_cfg = MultiProjectRunConfig.model_validate(raw) + pipeline = MultiProjectPipeline(project_root=project_root, config_path=cfg_path, config=multi_cfg) + if print_summary: + print("\n" + "=" * 80) + print(f"🚀 Creating multi-project pipeline from: {cfg_path}") + print("=" * 80) + return await pipeline.run() + + config = build_config(project_root=project_root, overrides=raw) + mode = f"{config.local_datasource.type}->{config.remote_datasource.type}" + return await run_pipeline_from_config( + config, + title=f"sync_state_machine pipeline ({mode})", + print_summary=print_summary, + ) + + async def create_jsonl_to_api_pipeline( *, # 路径配置 @@ -438,7 +531,7 @@ async def create_jsonl_to_api_pipeline( # API 配置 api_debug: bool = False, poll_max_retries: int = 10, - poll_interval: float = 0.5, + poll_interval: float = 0.5, api_rate_limit_max_requests: int = 30, api_rate_limit_window_seconds: float = 10.0, local_handler_configs: Optional[Dict[str, Dict[str, Any]]] = None, @@ -448,7 +541,7 @@ async def create_jsonl_to_api_pipeline( logger_file_path: Optional[str] = None, logger_level: int = logging.INFO, ) -> FullSyncPipeline: - cfg = PipelineRunConfig( + cfg = _build_pipeline_run_config( node_types=node_types, local_datasource=JsonlDataSourceConfig( type="jsonl", @@ -468,20 +561,18 @@ async def create_jsonl_to_api_pipeline( api_rate_limit_window_seconds=api_rate_limit_window_seconds, handler_configs=remote_handler_configs or {}, ), - strategies=[], persist=PersistConfig( db_path=persistence_db, enable=enable_persistence, wipe_on_start=wipe_persistence_on_start, ), - logging=LoggingConfig( + logging_config=LoggingConfig( initialize=initialize_logger, file_path=logger_file_path, level=logger_level, ), + strategy_overrides=strategy_overrides, ) - if strategy_overrides: - cfg = apply_config_overrides(cfg, {"strategies": strategy_overrides}) return await create_pipeline_from_config(cfg) @@ -503,7 +594,7 @@ async def create_api_to_jsonl_pipeline( # API 配置 api_debug: bool = False, poll_max_retries: int = 10, - poll_interval: float = 0.5, + poll_interval: float = 0.5, api_rate_limit_max_requests: int = 30, api_rate_limit_window_seconds: float = 10.0, local_handler_configs: Optional[Dict[str, Dict[str, Any]]] = None, @@ -513,7 +604,7 @@ async def create_api_to_jsonl_pipeline( logger_file_path: Optional[str] = None, logger_level: int = logging.INFO, ) -> FullSyncPipeline: - cfg = PipelineRunConfig( + cfg = _build_pipeline_run_config( node_types=node_types, local_datasource=ApiDataSourceConfig( type="api", @@ -533,20 +624,18 @@ async def create_api_to_jsonl_pipeline( read_only=False, handler_configs=remote_handler_configs or {}, ), - strategies=[], persist=PersistConfig( db_path=persistence_db, enable=enable_persistence, wipe_on_start=wipe_persistence_on_start, ), - logging=LoggingConfig( + logging_config=LoggingConfig( initialize=initialize_logger, file_path=logger_file_path, level=logger_level, ), + strategy_overrides=strategy_overrides, ) - if strategy_overrides: - cfg = apply_config_overrides(cfg, {"strategies": strategy_overrides}) return await create_pipeline_from_config(cfg) @@ -565,7 +654,7 @@ async def create_jsonl_to_jsonl_pipeline( logger_file_path: Optional[str] = None, logger_level: int = logging.INFO, ) -> FullSyncPipeline: - cfg = PipelineRunConfig( + cfg = _build_pipeline_run_config( node_types=node_types, local_datasource=JsonlDataSourceConfig( type="jsonl", @@ -579,18 +668,16 @@ async def create_jsonl_to_jsonl_pipeline( read_only=False, handler_configs=remote_handler_configs or {}, ), - strategies=[], persist=PersistConfig( db_path=persistence_db, enable=enable_persistence, wipe_on_start=wipe_persistence_on_start, ), - logging=LoggingConfig( + logging_config=LoggingConfig( initialize=initialize_logger, file_path=logger_file_path, level=logger_level, ), + strategy_overrides=strategy_overrides, ) - if strategy_overrides: - cfg = apply_config_overrides(cfg, {"strategies": strategy_overrides}) return await create_pipeline_from_config(cfg) diff --git a/sync_state_machine/pipeline/full_sync_pipeline.py b/sync_state_machine/pipeline/full_sync_pipeline.py index 1d81e68..3725bdc 100644 --- a/sync_state_machine/pipeline/full_sync_pipeline.py +++ b/sync_state_machine/pipeline/full_sync_pipeline.py @@ -12,6 +12,7 @@ from pathlib import Path if TYPE_CHECKING: from ..datasource.datasource import BaseDataSource +from ..datasource.jsonl import JsonlDataSource from ..common.collection import DataCollection from ..common.binding import BindingManager from ..common.persistence import PersistenceBackend @@ -47,6 +48,10 @@ class FullSyncPipeline: local_collection: DataCollection, remote_collection: DataCollection, binding_manager: BindingManager, + scope: str = "global", + pipeline_name: str | None = None, + bootstrap_binding_node_types: Optional[List[str]] = None, + ephemeral_node_types: Optional[List[str]] = None, node_types: Optional[List[str]] = None, load_order: Optional[List[str]] = None, sync_order: Optional[List[str]] = None, @@ -61,7 +66,11 @@ class FullSyncPipeline: self.local_datasource = local_datasource self.remote_datasource = remote_datasource self.strategies = strategies + self.scope = scope + self.pipeline_name = pipeline_name or f"pipeline[{scope}]" self.node_types = node_types or [s.node_type for s in self.strategies] + self.bootstrap_binding_node_types = list(dict.fromkeys(bootstrap_binding_node_types or [])) + self.ephemeral_node_types = list(dict.fromkeys(ephemeral_node_types or [])) self.load_order = load_order or self.node_types self.sync_order = sync_order or self.node_types self.persistence = persistence @@ -91,6 +100,9 @@ class FullSyncPipeline: self._logger = logger self._closed = False self._loaded_node_types: set[str] = set() + self._preloaded_local_nodes: List[Any] = [] + self._preloaded_remote_nodes: List[Any] = [] + self._preloaded_bindings: Dict[str, Dict[str, Optional[str]]] = {} def _resolve_pipeline_runtime(self) -> StateMachineRuntime: if self.strategies: @@ -104,17 +116,17 @@ class FullSyncPipeline: async def run(self) -> Dict[str, Any]: try: await self.phase_bootstrap() - self._logger.info("✅ Phase bootstrap completed") + self._logger.info("✅ Phase bootstrap completed (%s)", self.pipeline_name) await self.phase_sync_by_node_type() - self._logger.info("✅ Phase sync-by-node-type completed") + self._logger.info("✅ Phase sync-by-node-type completed (%s)", self.pipeline_name) post_ok = await self.phase_post_check() self.stats["post_check_passed"] = post_ok - self._logger.info("✅ Phase post-check completed") + self._logger.info("✅ Phase post-check completed (%s)", self.pipeline_name) await self.phase_persistence() - self._logger.info("✅ Phase persistence completed") + self._logger.info("✅ Phase persistence completed (%s)", self.pipeline_name) return self.stats finally: await self.close() @@ -173,9 +185,14 @@ class FullSyncPipeline: await self._load_persistence_state() self._logger.info("✅ Bootstrap: persistence loaded") + await self._apply_preloaded_state() + self._logger.info("✅ Bootstrap: shared preload applied") + self._prepare_datasources() self._logger.info("✅ Bootstrap: datasource prepared") + await self._apply_project_scope_cleanup() + async def phase_bind(self) -> None: self._prepare_datasources() strategy_map = {s.node_type: s for s in self.strategies} @@ -257,10 +274,17 @@ class FullSyncPipeline: if not self.enable_persistence: return + excluded_node_types = self.ephemeral_node_types or None + # 加载持久化数据 - await self.local_collection.load_from_persistence() - await self.remote_collection.load_from_persistence() - for node_type in self.node_types: + await self.local_collection.load_from_persistence_excluding(excluded_node_types) + await self.remote_collection.load_from_persistence_excluding(excluded_node_types) + binding_node_types = [ + node_type + for node_type in dict.fromkeys([*self.node_types, *self.bootstrap_binding_node_types]) + if node_type not in set(self.ephemeral_node_types) + ] + for node_type in binding_node_types: await self.binding_manager.load_from_persistence(node_type) self._logger.info("🔄 Persistence loaded (state preserved; pending E01 normalization)") @@ -278,11 +302,67 @@ class FullSyncPipeline: self._logger.info(f"🧹 Reset cleanup: {total_cleaned} CREATE-failed zombies cleaned") self._logger.info("🔄 Post-load reset cleanup completed") + async def _apply_preloaded_state(self) -> None: + if self._preloaded_local_nodes: + await self.local_collection.import_nodes(self._preloaded_local_nodes) + if self._preloaded_remote_nodes: + await self.remote_collection.import_nodes(self._preloaded_remote_nodes) + for node_type, bindings in self._preloaded_bindings.items(): + self.binding_manager.import_bindings(node_type, bindings) + + def preload_shared_state( + self, + *, + local_nodes: List[Any], + remote_nodes: List[Any], + bindings_by_type: Dict[str, Dict[str, Optional[str]]], + ) -> None: + self._preloaded_local_nodes = list(local_nodes) + self._preloaded_remote_nodes = list(remote_nodes) + self._preloaded_bindings = { + node_type: dict(bindings) + for node_type, bindings in bindings_by_type.items() + } + def _prepare_datasources(self) -> None: """为后续按 node_type 加载准备 collection 绑定。""" self.local_datasource.set_collection(self.local_collection) self.remote_datasource.set_collection(self.remote_collection) + def _get_project_scope_filter_ids(self, datasource: "BaseDataSource") -> List[str]: + try: + handler = datasource.get_handler("project") + get_data_id_filter = getattr(handler, "get_data_id_filter", None) + if callable(get_data_id_filter): + raw_filter = get_data_id_filter() + if isinstance(raw_filter, list): + return list(dict.fromkeys(raw_filter)) + except Exception: + pass + return [] + + async def _apply_project_scope_cleanup(self) -> None: + local_project_ids = self._get_project_scope_filter_ids(self.local_datasource) + remote_project_ids = self._get_project_scope_filter_ids(self.remote_datasource) + if not local_project_ids and not remote_project_ids: + return + + deleted_local = await self.local_collection.filter_by_project_ids(local_project_ids) + deleted_remote = await self.remote_collection.filter_by_project_ids(remote_project_ids) + deleted_bindings = await self.binding_manager.prune_missing_nodes( + local_collection=self.local_collection, + remote_collection=self.remote_collection, + node_types=list(dict.fromkeys([*self.node_types, *self.bootstrap_binding_node_types])), + ) + + if deleted_local or deleted_remote or deleted_bindings: + self._logger.info( + "🧹 Project-scope cleanup: local_deleted=%d remote_deleted=%d bindings_deleted=%d", + deleted_local, + deleted_remote, + deleted_bindings, + ) + async def _load_node_type(self, node_type: str, *, reason: str) -> None: self._logger.info(f"🔄 Load node_type={node_type}, reason={reason}") await self.local_datasource.load_all(order=[node_type]) @@ -300,7 +380,13 @@ class FullSyncPipeline: action_summary = summary.get(action_key, {}) if isinstance(summary, dict) else {} return int(action_summary.get("success", 0)) if isinstance(action_summary, dict) else 0 + def _should_skip_reload(self) -> bool: + return isinstance(self.local_datasource, JsonlDataSource) and isinstance(self.remote_datasource, JsonlDataSource) + async def _reload_node_type(self, node_type: str, *, reason: str) -> None: + if self._should_skip_reload(): + self._logger.info("⏭️ Reload skipped for JSONL pipeline: node_type=%s reason=%s", node_type, reason) + return await self._load_node_type(node_type, reason=reason) async def _evaluate_consistency_for_node_type(self, node_type: str) -> Dict[str, Any]: @@ -309,28 +395,18 @@ class FullSyncPipeline: compare_to_remote = True if strategy and strategy.config.update_direction == UpdateDirection.PULL: compare_to_remote = False - ignore_list_item_fields = dict(strategy.config.post_check_ignore_list_item_fields) if strategy else {} - - # 从远端 handler 的 update_schemas 提取只读字段过滤白名单 - post_check_fields: List[str] | None = None - try: - remote_handler = self.remote_datasource.get_handler(node_type) - fields = remote_handler.get_update_fields() - if fields: - post_check_fields = fields - except Exception: - pass + ignore_fields = list(strategy.config.post_check_ignore_fields) if strategy else [] result = await evaluate_consistency_for_node_type( node_type=node_type, + strategy=strategy, local_collection=self.local_collection, remote_collection=self.remote_collection, binding_manager=self.binding_manager, logger=self._logger, depend_fields=depend_fields, compare_to_remote=compare_to_remote, - ignore_list_item_fields=ignore_list_item_fields, - post_check_fields=post_check_fields, + ignore_fields=ignore_fields, ) self._consistency_by_type[node_type] = result return result @@ -430,9 +506,12 @@ class FullSyncPipeline: 所有类型均已同步完毕后，统一做一次全量 reload（触发后端派生字段计算）， 再对所有类型做一致性评估，确保 post-check 结果反映最终状态。 """ - self._logger.info("🔄 Post-check: reloading all node types for final consistency evaluation...") - for node_type in self.node_types: - await self._reload_node_type(node_type, reason="post-check final reload") + if self._should_skip_reload(): + self._logger.info("⏭️ Post-check reload skipped for JSONL pipeline") + else: + self._logger.info("🔄 Post-check: reloading all node types for final consistency evaluation...") + for node_type in self.node_types: + await self._reload_node_type(node_type, reason="post-check final reload") for node_type in self.node_types: await self._evaluate_consistency_for_node_type(node_type) @@ -454,11 +533,11 @@ class FullSyncPipeline: return self._logger.info("🔍 Persisting local collection...") - await self.local_collection.persist() + await self.local_collection.persist(exclude_node_types=self.ephemeral_node_types) self._logger.info("🔍 Persisting remote collection...") - await self.remote_collection.persist() + await self.remote_collection.persist(exclude_node_types=self.ephemeral_node_types) self._logger.info("🔍 Persisting binding manager...") - await self.binding_manager.persist() + await self.binding_manager.persist(exclude_node_types=self.ephemeral_node_types) self._logger.info("✅ All data persisted successfully") # ========== Summary & Reporting ========== diff --git a/sync_state_machine/pipeline/multi_project_pipeline.py b/sync_state_machine/pipeline/multi_project_pipeline.py new file mode 100644 index 0000000..4848bd2 --- /dev/null +++ b/sync_state_machine/pipeline/multi_project_pipeline.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +import logging +import copy +from pathlib import Path +from typing import Dict, Optional, List, TypedDict, Any + +from ..config import ( + MultiProjectItemConfig, + MultiProjectRunConfig, + PipelineRunConfig, + apply_config_overrides, + build_config_from_file, +) +from .factory import create_pipeline_from_config + + +logger = logging.getLogger(__name__) + + +class _SharedStateSnapshot(TypedDict): + local_nodes: List[Any] + remote_nodes: List[Any] + bindings_by_type: Dict[str, Dict[str, Optional[str]]] + + +def _resolve_embedded_config_path(raw_path: str, *, project_root: Path, profile_path: Path) -> Path: + candidate = Path(raw_path) + if candidate.is_absolute(): + return candidate + + profile_relative = (profile_path.parent / candidate).resolve() + if profile_relative.exists(): + return profile_relative + return (project_root / candidate).resolve() + + +def _project_scope_overrides(item: MultiProjectItemConfig, *, node_types: list[str]) -> Dict[str, object]: + del node_types + local_handler_configs = { + "project": {"data_id_filter": [item.local_project_id]}, + } + remote_handler_configs = { + "project": {"data_id_filter": [item.remote_project_id]}, + } + return { + "local_datasource": { + "handler_configs": local_handler_configs, + }, + "remote_datasource": { + "handler_configs": remote_handler_configs, + }, + } + + +def _resolve_project_bootstrap_node_types( + *, + config: MultiProjectRunConfig, + shared_node_types: list[str], +) -> list[str]: + configured = [node_type for node_type in config.project_bootstrap_node_types if node_type] + if configured: + return list(dict.fromkeys(configured)) + return list(dict.fromkeys(shared_node_types)) + + +class MultiProjectPipeline: + def __init__(self, *, project_root: Path, config_path: Path, config: MultiProjectRunConfig): + self.project_root = project_root + self.config_path = config_path + self.config = config + + def _load_pipeline_config(self, raw_path: str) -> PipelineRunConfig: + resolved = _resolve_embedded_config_path(raw_path, project_root=self.project_root, profile_path=self.config_path) + return build_config_from_file(project_root=self.project_root, file_path=resolved) + + def _prepare_project_config( + self, + *, + shared_persist: Dict[str, object], + item: MultiProjectItemConfig, + base_path: Optional[str], + ) -> PipelineRunConfig: + raw_path = base_path or self.config.default_project_config + cfg = self._load_pipeline_config(raw_path) + cfg = apply_config_overrides(cfg, {"persist": {**shared_persist, "wipe_on_start": False}}) + return apply_config_overrides(cfg, _project_scope_overrides(item, node_types=list(cfg.node_types))) + + async def _snapshot_shared_state(self, *, pipeline, node_types: List[str]) -> _SharedStateSnapshot: + local_nodes = [ + copy.deepcopy(node) + for node_type in node_types + for node in pipeline.local_collection.filter(node_type=node_type) + ] + remote_nodes = [ + copy.deepcopy(node) + for node_type in node_types + for node in pipeline.remote_collection.filter(node_type=node_type) + ] + bindings_by_type: Dict[str, Dict[str, Optional[str]]] = {} + for node_type in node_types: + bindings = await pipeline.binding_manager.get_all_bindings(node_type) + bindings_by_type[node_type] = {local_id: remote_id for local_id, remote_id in bindings} + return { + "local_nodes": local_nodes, + "remote_nodes": remote_nodes, + "bindings_by_type": bindings_by_type, + } + + async def run(self) -> Dict[str, Dict[str, object]]: + results: Dict[str, Dict[str, object]] = {} + + global_config = self._load_pipeline_config(self.config.global_config) + shared_persist = global_config.persist.model_dump(mode="json") + shared_node_types = list(global_config.node_types) + project_bootstrap_node_types = _resolve_project_bootstrap_node_types( + config=self.config, + shared_node_types=shared_node_types, + ) + logger.info("🚀 Multi-project pipeline start: projects=%d", len(self.config.projects)) + global_pipeline = await create_pipeline_from_config( + global_config, + scope="global", + pipeline_name="multi-project pipeline [global]", + ) + try: + results["global"] = await global_pipeline.run() + shared_state = await self._snapshot_shared_state( + pipeline=global_pipeline, + node_types=project_bootstrap_node_types, + ) + finally: + await global_pipeline.close() + + for item in self.config.projects: + project_config = self._prepare_project_config( + shared_persist=shared_persist, + item=item, + base_path=item.config_path, + ) + scope = item.scope + title = f"multi-project pipeline [{scope}]" + logger.info( + "🚀 Project pipeline start: scope=%s local_project_id=%s remote_project_id=%s", + scope, + item.local_project_id, + item.remote_project_id, + ) + project_pipeline = await create_pipeline_from_config( + project_config, + scope=scope, + pipeline_name=title, + bootstrap_binding_node_types=project_bootstrap_node_types, + ephemeral_node_types=project_bootstrap_node_types, + ) + try: + project_pipeline.preload_shared_state( + local_nodes=shared_state["local_nodes"], + remote_nodes=shared_state["remote_nodes"], + bindings_by_type=shared_state["bindings_by_type"], + ) + results[scope] = await project_pipeline.run() + finally: + await project_pipeline.close() + + return results \ No newline at end of file diff --git a/sync_state_machine/pipeline/summary_report.py b/sync_state_machine/pipeline/summary_report.py index af4700e..6745eb6 100644 --- a/sync_state_machine/pipeline/summary_report.py +++ b/sync_state_machine/pipeline/summary_report.py @@ -52,18 +52,17 @@ async def print_pipeline_summary(*, logger, node_types, binding_manager, local_d logger.info(" pass") continue - logger.info(" field mismatch/total diff_schemas samples") - logger.info(" ------------------------------------------------------------------------------") + logger.info(" field mismatch/total samples") + logger.info(" -----------------------------------------------------------") for field_report in differing_fields: field_name = str(field_report.get("field_name", "")) field_mismatch = int(field_report.get("mismatch_count", 0)) field_total = int(field_report.get("total_count", 0)) - schema_refs = ",".join(str(item) for item in (field_report.get("schema_refs", []) or [])) or "-" samples = field_report.get("samples", []) or [] sample_text = _format_sample(samples[0]) if samples else "-" logger.info( f" {_clip(field_name, 30):<30} {field_mismatch:>4}/{field_total:<4} " - f"{_clip(schema_refs, 28):<28} {sample_text}" + f"{sample_text}" ) def print_collection_summary(title: str, datasource, collection) -> None: diff --git a/sync_state_machine/sync_system/strategy.py b/sync_state_machine/sync_system/strategy.py index c053158..53c7006 100644 --- a/sync_state_machine/sync_system/strategy.py +++ b/sync_state_machine/sync_system/strategy.py @@ -20,6 +20,7 @@ from ..engine import ( StateMachineRuntime, ) from ..validation import SchemaDiffValidator +from .strategy_ops.compare_ops import normalized_data_for_compare T = TypeVar("T", bound=BaseModel) logger = logging.getLogger(__name__) @@ -186,12 +187,28 @@ class BaseSyncStrategy(Generic[T]): self._validate_config() logger.info(f"[{self.node_type}] Applied preset: {preset_name}") + def preprocess_compare_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]: + return payload + + def normalize_compare_payload( + self, + data: Optional[Dict[str, Any]], + data_id_map: Optional[Dict[str, str]] = None, + *, + ignore_fields: Optional[set[str]] = None, + ) -> Dict[str, Any]: + prepared = self.preprocess_compare_payload(dict(data or {})) + return normalized_data_for_compare( + prepared, + data_id_map, + ignore_fields=ignore_fields, + ) + def get_schema_diff_validator(self) -> SchemaDiffValidator: if self._schema_diff_validator is None: self._schema_diff_validator = SchemaDiffValidator( schema_name=self.schema.__name__, ignore_fields=self.config.compare_ignore_fields, - ignore_list_item_fields=self.config.post_check_ignore_list_item_fields, ) return self._schema_diff_validator diff --git a/sync_state_machine/sync_system/strategy_ops/compare_ops.py b/sync_state_machine/sync_system/strategy_ops/compare_ops.py index 72277f6..3d58768 100644 --- a/sync_state_machine/sync_system/strategy_ops/compare_ops.py +++ b/sync_state_machine/sync_system/strategy_ops/compare_ops.py @@ -2,7 +2,7 @@ from __future__ import annotations import copy import inspect -from typing import Any, Dict, Mapping, Optional +from typing import Any, Dict, Optional async def build_data_id_normalization_map( @@ -39,42 +39,18 @@ def is_id_like_field(field_name: Optional[str]) -> bool: return lowered.endswith("_id") or lowered.endswith("_ids") -def _strip_ignored_list_item_fields( - field_name: Optional[str], - value: Any, - ignore_list_item_fields: Mapping[str, list[str]], -) -> Any: - if not isinstance(value, list): - return value - - ignored_keys = set(ignore_list_item_fields.get(field_name or "", [])) - if not ignored_keys: - return value - - normalized_items = [] - for item in value: - if isinstance(item, dict): - normalized_items.append({key: nested for key, nested in item.items() if key not in ignored_keys}) - else: - normalized_items.append(item) - return normalized_items - - def normalize_compare_value( field_name: Optional[str], value: Any, data_id_map: Dict[str, str], - ignore_list_item_fields: Optional[Mapping[str, list[str]]] = None, ) -> Any: - ignored_list_fields = ignore_list_item_fields or {} if isinstance(value, dict): return { - key: normalize_compare_value(key, nested, data_id_map, ignored_list_fields) + key: normalize_compare_value(key, nested, data_id_map) for key, nested in value.items() } if isinstance(value, list): - value = _strip_ignored_list_item_fields(field_name, value, ignored_list_fields) - return [normalize_compare_value(field_name, item, data_id_map, ignored_list_fields) for item in value] + return [normalize_compare_value(field_name, item, data_id_map) for item in value] if value == "": value = None @@ -88,7 +64,6 @@ def normalized_data_for_compare( data: Optional[Dict[str, Any]], data_id_map: Optional[Dict[str, str]] = None, ignore_fields: Optional[set[str]] = None, - ignore_list_item_fields: Optional[Mapping[str, list[str]]] = None, ) -> Dict[str, Any]: payload = copy.deepcopy(data or {}) payload.pop("id", None) @@ -96,7 +71,7 @@ def normalized_data_for_compare( ignored = ignore_fields or set() mapping = data_id_map or {} return { - key: normalize_compare_value(key, value, mapping, ignore_list_item_fields) + key: normalize_compare_value(key, value, mapping) for key, value in payload.items() if key not in ignored } diff --git a/sync_state_machine/sync_system/strategy_ops/post_check_ops.py b/sync_state_machine/sync_system/strategy_ops/post_check_ops.py index d806a84..ba49987 100644 --- a/sync_state_machine/sync_system/strategy_ops/post_check_ops.py +++ b/sync_state_machine/sync_system/strategy_ops/post_check_ops.py @@ -4,34 +4,7 @@ from typing import Any, Dict, List from ...sync_system.resolve_ids import IDResolver from ...validation import SchemaDiffValidator -from .compare_ops import ( - build_data_id_normalization_map, - normalized_data_for_compare, -) - - -def _strip_list_item_fields(payload: Dict, ignore_map: Dict[str, List[str]]) -> Dict: - """从 payload 中列表型字段的每个元素里，剔除指定子键。 - - 例如 ignore_map={"plan_node": ["is_audit"]} 会把 payload["plan_node"][*]["is_audit"] 全部去掉。 - payload 本身不会被修改（浅拷贝）。 - """ - if not ignore_map: - return payload - result = dict(payload) - for field_name, sub_keys in ignore_map.items(): - if field_name not in result: - continue - items = result[field_name] - if not isinstance(items, list): - continue - stripped = [] - for item in items: - if isinstance(item, dict): - item = {k: v for k, v in item.items() if k not in sub_keys} - stripped.append(item) - result[field_name] = stripped - return result +from .compare_ops import build_data_id_normalization_map def _compact_repr(value: Any, *, max_len: int = 80) -> str: @@ -54,20 +27,23 @@ def _append_post_check_error(node, message: str) -> None: async def evaluate_consistency_for_node_type( *, node_type: str, + strategy, local_collection, remote_collection, binding_manager, logger, depend_fields: Dict[str, str] | None = None, compare_to_remote: bool = True, - ignore_list_item_fields: Dict[str, List[str]] | None = None, - post_check_fields: List[str] | None = None, + ignore_fields: List[str] | None = None, ) -> Dict[str, Any]: records = await binding_manager.get_all_records(node_type) checked_pairs = 0 depend_fields = dict(depend_fields or {}) id_resolver = IDResolver(local_collection, remote_collection, binding_manager) - validator = SchemaDiffValidator(schema_name=node_type) + validator = SchemaDiffValidator( + schema_name=node_type, + ignore_fields=ignore_fields, + ) data_id_map = await build_data_id_normalization_map( node_type=node_type, local_collection=local_collection, @@ -117,14 +93,8 @@ async def evaluate_consistency_for_node_type( if resolved_remote is not None: remote_data = resolved_remote - local_payload = normalized_data_for_compare(local_data, data_id_map) - remote_payload = normalized_data_for_compare(remote_data, data_id_map) - if post_check_fields: - local_payload = {k: v for k, v in local_payload.items() if k in post_check_fields} - remote_payload = {k: v for k, v in remote_payload.items() if k in post_check_fields} - if ignore_list_item_fields: - local_payload = _strip_list_item_fields(local_payload, ignore_list_item_fields) - remote_payload = _strip_list_item_fields(remote_payload, ignore_list_item_fields) + local_payload = strategy.normalize_compare_payload(local_data, data_id_map) + remote_payload = strategy.normalize_compare_payload(remote_data, data_id_map) diff_map = validator.compare_payloads( local_payload, remote_payload, diff --git a/sync_state_machine/sync_system/strategy_ops/update_ops.py b/sync_state_machine/sync_system/strategy_ops/update_ops.py index 67a9101..fddcedb 100644 --- a/sync_state_machine/sync_system/strategy_ops/update_ops.py +++ b/sync_state_machine/sync_system/strategy_ops/update_ops.py @@ -55,17 +55,15 @@ def default_needs_update( return False ignore_fields = set(strategy.config.compare_ignore_fields) - normalized_source = normalized_data_for_compare( + normalized_source = strategy.normalize_compare_payload( source_data, data_id_map, ignore_fields=ignore_fields, - ignore_list_item_fields=strategy.config.post_check_ignore_list_item_fields, ) - normalized_target = normalized_data_for_compare( + normalized_target = strategy.normalize_compare_payload( target_data, data_id_map, ignore_fields=ignore_fields, - ignore_list_item_fields=strategy.config.post_check_ignore_list_item_fields, ) diff_map = strategy.get_schema_diff_validator().compare_payloads( normalized_source, @@ -87,28 +85,26 @@ def resolve_needs_update_check(strategy) -> Callable: def _collect_diff_descriptions( + strategy, source_data, target_data, *, data_id_map, ignore_fields: set[str], - ignore_list_item_fields: dict[str, list[str]], max_items: int = 8, ) -> List[str]: if source_data is None or target_data is None: return [] - normalized_source = normalized_data_for_compare( + normalized_source = strategy.normalize_compare_payload( source_data, data_id_map, ignore_fields=ignore_fields, - ignore_list_item_fields=ignore_list_item_fields, ) - normalized_target = normalized_data_for_compare( + normalized_target = strategy.normalize_compare_payload( target_data, data_id_map, ignore_fields=ignore_fields, - ignore_list_item_fields=ignore_list_item_fields, ) diff_map = collect_payload_diffs( normalized_source, @@ -206,11 +202,11 @@ async def add_update_action( has_diff = bool(should_update) if has_diff: diff_descriptions = _collect_diff_descriptions( + strategy, data, target_node.get_data(), data_id_map=data_id_map or {}, ignore_fields=set(strategy.config.compare_ignore_fields), - ignore_list_item_fields=dict(strategy.config.post_check_ignore_list_item_fields), ) if steps_changed: diff --git a/sync_state_machine/validation/schema_diff.py b/sync_state_machine/validation/schema_diff.py index 816a893..2d61ab6 100644 --- a/sync_state_machine/validation/schema_diff.py +++ b/sync_state_machine/validation/schema_diff.py @@ -1,7 +1,7 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import Any, Iterable, Mapping +from typing import Any, Iterable from ..sync_system.strategy_ops.compare_ops import normalized_data_for_compare @@ -28,12 +28,10 @@ class SchemaDiffValidator: *, schema_name: str, ignore_fields: Iterable[str] | None = None, - ignore_list_item_fields: Mapping[str, list[str]] | None = None, sample_limit: int = 3, ) -> None: self.schema_name = schema_name self.ignore_fields = {str(field) for field in (ignore_fields or []) if str(field)} - self.ignore_list_item_fields = dict(ignore_list_item_fields or {}) self.sample_limit = max(1, sample_limit) self.reset() @@ -56,12 +54,10 @@ class SchemaDiffValidator: local_normalized = normalized_data_for_compare( dict(local_payload or {}), ignore_fields=self.ignore_fields, - ignore_list_item_fields=self.ignore_list_item_fields, ) remote_normalized = normalized_data_for_compare( dict(remote_payload or {}), ignore_fields=self.ignore_fields, - ignore_list_item_fields=self.ignore_list_item_fields, ) field_names = sorted(set(local_normalized) | set(remote_normalized)) diff --git a/tests/README.md b/tests/README.md index fb4a430..550ddc7 100644 --- a/tests/README.md +++ b/tests/README.md @@ -12,6 +12,7 @@ - 工具链专项：`pytest tests/integration/test_toolchain_config_and_graph.py -q` - UI 专项：`pytest tests/integration/test_ui_api_smoke.py -q` - 测试总览与远程环境工具：`docs/testing_guide.md` +- 最近常用的定向回归：`pytest tests/unit/test_update_ops.py tests/unit/test_post_check_dependency_id_normalization.py -q` ## 设计原则 - 不绕过 schema/状态机校验；配置错误必须显式失败。 @@ -26,5 +27,5 @@ - 覆盖核心业务链路（bind/create/update 与 full pipeline）。 ## 大规模自动回归 -- 按轮迭代的大规模自动测试方法见 [auto_test_spec/agent_auto_test_method.md](auto_test_spec/agent_auto_test_method.md)。 +- 按轮迭代的大规模自动测试方法见 [../docs/auto_test_spec/agent_auto_test_method.md](../docs/auto_test_spec/agent_auto_test_method.md)。 - 执行时优先采用“单 domain 复现 -> 修复/分类 -> 全量聚合回归 -> 记录总计变化”的节奏，避免一次携带过多上下文。 diff --git a/tests/integration/test_multi_project_pipeline_jsonl_jsonl.py b/tests/integration/test_multi_project_pipeline_jsonl_jsonl.py new file mode 100644 index 0000000..c7829ba --- /dev/null +++ b/tests/integration/test_multi_project_pipeline_jsonl_jsonl.py @@ -0,0 +1,157 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +from sync_state_machine.common.persistence import PersistenceBackend +from sync_state_machine.pipeline import run_profile_from_file +from sync_state_machine.pipeline.full_sync_pipeline import FullSyncPipeline + + +def _write_yaml(path: Path, content: str) -> None: + path.write_text(content.strip() + "\n", encoding="utf-8") + + +@pytest.mark.asyncio +async def test_multi_project_pipeline_runs_jsonl_to_jsonl_with_shared_persistence( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + pytest.importorskip("yaml") + + async def _skip_post_check(self) -> bool: + return True + + monkeypatch.setattr(FullSyncPipeline, "phase_post_check", _skip_post_check) + + project_root = Path(__file__).resolve().parents[2] + local_dir = project_root / "working_local_datasource" / "20260302-020003" + remote_dir = tmp_path / "remote_jsonl" + db_path = tmp_path / "multi_project.db" + logs_dir = tmp_path / "logs" + remote_dir.mkdir(parents=True, exist_ok=True) + logs_dir.mkdir(parents=True, exist_ok=True) + + global_cfg = tmp_path / "global.yaml" + project_cfg = tmp_path / "project.yaml" + multi_cfg = tmp_path / "multi.yaml" + + _write_yaml( + global_cfg, + f""" +node_types: + - project +local_datasource: + type: jsonl + jsonl_dir: {local_dir.as_posix()} + read_only: false +remote_datasource: + type: jsonl + jsonl_dir: {remote_dir.as_posix()} + read_only: false +persist: + backend: sqlite + db_path: {db_path.as_posix()} + enable: true + wipe_on_start: true +logging: + initialize: true + file_dir: {logs_dir.as_posix()} + file_name_pattern: multi_global_{{timestamp}}.log + level: 20 +strategies: + project: + skip_sync: false + config: + auto_bind: true + auto_bind_fields: + - name + depend_fields: {{}} + local_orphan_action: create_remote + remote_orphan_action: none + update_direction: push + """, + ) + + _write_yaml( + project_cfg, + f""" +node_types: + - lar +local_datasource: + type: jsonl + jsonl_dir: {local_dir.as_posix()} + read_only: false +remote_datasource: + type: jsonl + jsonl_dir: {remote_dir.as_posix()} + read_only: false +persist: + backend: sqlite + db_path: {db_path.as_posix()} + enable: true + wipe_on_start: false +logging: + initialize: true + file_dir: {logs_dir.as_posix()} + file_name_pattern: multi_project_{{timestamp}}.log + level: 20 +strategies: + lar: + skip_sync: false + config: + auto_bind: true + auto_bind_fields: + - project_id + depend_fields: + project_id: project + local_orphan_action: create_remote + remote_orphan_action: none + update_direction: push + """, + ) + + _write_yaml( + multi_cfg, + f""" +global_config: {global_cfg.as_posix()} +default_project_config: {project_cfg.as_posix()} +projects: + - local_project_id: f3e02e84-e81c-4aa6-88d4-f5aa108c8227 + remote_project_id: f3e02e84-e81c-4aa6-88d4-f5aa108c8227 + - local_project_id: 0f9c3e22-f4bb-4803-825a-4f238eeb06d5 + remote_project_id: 0f9c3e22-f4bb-4803-825a-4f238eeb06d5 + """, + ) + + results = await run_profile_from_file(project_root=project_root, config_path=multi_cfg, print_summary=False) + + assert "global" in results + assert "project_id:f3e02e84-e81c-4aa6-88d4-f5aa108c8227" in results + assert "project_id:0f9c3e22-f4bb-4803-825a-4f238eeb06d5" in results + + _, scope_rows = PersistenceBackend.query_records( + backend="sqlite", + db_path=str(db_path), + sql="SELECT DISTINCT scope FROM nodes ORDER BY scope", + ) + scopes = [row["scope"] for row in scope_rows] + assert scopes == [ + "global", + "project_id:0f9c3e22-f4bb-4803-825a-4f238eeb06d5", + "project_id:f3e02e84-e81c-4aa6-88d4-f5aa108c8227", + ] + + _, copied_rows = PersistenceBackend.query_records( + backend="sqlite", + db_path=str(db_path), + sql=( + "SELECT scope, collection_id, node_type, COUNT(1) AS c " + "FROM nodes " + "WHERE scope <> 'global' AND node_type = 'project' " + "GROUP BY scope, collection_id, node_type " + "ORDER BY scope, collection_id, node_type" + ), + ) + assert copied_rows == [] \ 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 444c042..c79f156 100644 --- a/tests/unit/test_full_sync_pipeline_downstream_reload.py +++ b/tests/unit/test_full_sync_pipeline_downstream_reload.py @@ -1,11 +1,13 @@ from __future__ import annotations from types import SimpleNamespace +from pathlib import Path import pytest from sync_state_machine.common.binding import BindingManager from sync_state_machine.common.collection import DataCollection +from sync_state_machine.datasource.jsonl import JsonlDataSource from sync_state_machine.pipeline.full_sync_pipeline import FullSyncPipeline @@ -93,4 +95,52 @@ async def test_phase_sync_loads_each_node_type_before_strategy(monkeypatch: pyte ("project", "create wrote data"), ("preparation", "pre-strategy refresh"), ("contract", "pre-strategy refresh"), + ] + + +@pytest.mark.asyncio +async def test_phase_sync_skips_reload_for_jsonl_datasources( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + project_strategy = _StubStrategy("project") + + pipeline = FullSyncPipeline( + local_datasource=JsonlDataSource(tmp_path / "local", read_only=True), + remote_datasource=JsonlDataSource(tmp_path / "remote", read_only=True), + strategies=[project_strategy], + persistence=_FakePersistence(), + local_collection=DataCollection("local"), + remote_collection=DataCollection("remote"), + binding_manager=BindingManager(_FakePersistence(), auto_persist=False), + node_types=["project"], + load_order=["project"], + sync_order=["project"], + enable_persistence=False, + ) + + reload_calls: list[tuple[str, str]] = [] + + async def fake_commit_creates(node_type: str) -> bool: + return node_type == "project" + + async def fake_commit_updates(node_type: str) -> bool: + del node_type + return False + + async def fake_load_node_type(node_type: str, *, reason: str) -> None: + reload_calls.append((node_type, reason)) + + async def fake_normalize(node_type: str) -> None: + del node_type + + monkeypatch.setattr(pipeline, "commit_creates", fake_commit_creates) + monkeypatch.setattr(pipeline, "commit_updates", fake_commit_updates) + monkeypatch.setattr(pipeline, "_load_node_type", fake_load_node_type) + monkeypatch.setattr(pipeline, "normalize_create_success_to_update_entry", fake_normalize) + + await pipeline.phase_sync_by_node_type() + + assert reload_calls == [ + ("project", "pre-strategy refresh"), ] \ No newline at end of file diff --git a/tests/unit/test_multi_project_pipeline_config.py b/tests/unit/test_multi_project_pipeline_config.py new file mode 100644 index 0000000..73767d6 --- /dev/null +++ b/tests/unit/test_multi_project_pipeline_config.py @@ -0,0 +1,123 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml +from pydantic import BaseModel + +from sync_state_machine.config.multi_project_config import build_multi_project_config_from_file +from sync_state_machine.common.binding import BindingManager +from sync_state_machine.common.collection import DataCollection +from sync_state_machine.common.persistence import PersistenceBackend +from sync_state_machine.common.sync_node import SyncNode +from sync_state_machine.pipeline.multi_project_pipeline import ( + _project_scope_overrides, + _resolve_project_bootstrap_node_types, +) + + +def test_build_multi_project_config_reads_explicit_project_bootstrap_node_types(tmp_path: Path) -> None: + config_path = tmp_path / "multi.yaml" + config_path.write_text( + yaml.safe_dump( + { + "global_config": "run_profiles/global.yaml", + "default_project_config": "run_profiles/project.yaml", + "project_bootstrap_node_types": ["company", "supplier"], + "projects": [ + { + "local_project_id": "local-project", + "remote_project_id": "remote-project", + } + ], + }, + allow_unicode=True, + sort_keys=False, + ), + encoding="utf-8", + ) + + config = build_multi_project_config_from_file(project_root=tmp_path, file_path=config_path) + + assert config.project_bootstrap_node_types == ["company", "supplier"] + + +def test_project_scope_overrides_only_inject_project_handler_filter() -> None: + item = type("Item", (), {"local_project_id": "local-project", "remote_project_id": "remote-project"})() + + overrides = _project_scope_overrides(item, node_types=["project", "contract", "user"]) + + assert overrides == { + "local_datasource": { + "handler_configs": { + "project": {"data_id_filter": ["local-project"]}, + } + }, + "remote_datasource": { + "handler_configs": { + "project": {"data_id_filter": ["remote-project"]}, + } + }, + } + + +def test_resolve_project_bootstrap_node_types_prefers_explicit_config() -> None: + config = type( + "Config", + (), + {"project_bootstrap_node_types": ["company", "supplier", "company"]}, + )() + + resolved = _resolve_project_bootstrap_node_types( + config=config, + shared_node_types=["company", "supplier", "project"], + ) + + assert resolved == ["company", "supplier"] + + +class _ProjectSchema(BaseModel): + id: str + + +class _ProjectNode(SyncNode[dict]): + node_type = "project" + schema = _ProjectSchema + + +@pytest.mark.asyncio +async def test_binding_manager_prunes_stale_bindings_after_project_scope_cleanup() -> None: + persistence = PersistenceBackend(":memory:") + await persistence.initialize() + try: + local_collection = DataCollection("local", persistence=persistence, auto_persist=False) + remote_collection = DataCollection("remote", persistence=persistence, auto_persist=False) + binding_manager = BindingManager(persistence, auto_persist=False, scope="project_id:p1") + + local_node = _ProjectNode(node_id="local-project-1", data={"id": "p1"}, data_id="p1") + remote_node = _ProjectNode(node_id="remote-project-1", data={"id": "p1"}, data_id="p1") + stale_local_node = _ProjectNode(node_id="local-project-2", data={"id": "p2"}, data_id="p2") + stale_remote_node = _ProjectNode(node_id="remote-project-2", data={"id": "p2"}, data_id="p2") + + await local_collection.add(local_node) + await local_collection.add(stale_local_node) + await remote_collection.add(remote_node) + await remote_collection.add(stale_remote_node) + await binding_manager.bind("project", "local-project-1", "remote-project-1") + await binding_manager.bind("project", "local-project-2", "remote-project-2") + + await local_collection.filter_by_project_ids(["p1"]) + await remote_collection.filter_by_project_ids(["p1"]) + + deleted = await binding_manager.prune_missing_nodes( + local_collection=local_collection, + remote_collection=remote_collection, + node_types=["project"], + ) + + assert deleted == 1 + assert await binding_manager.get_remote_id("project", "local-project-1") == "remote-project-1" + assert await binding_manager.get_remote_id("project", "local-project-2") is None + finally: + await persistence.close() \ No newline at end of file diff --git a/tests/unit/test_post_check_dependency_id_normalization.py b/tests/unit/test_post_check_dependency_id_normalization.py index 3dbaed4..4d0a393 100644 --- a/tests/unit/test_post_check_dependency_id_normalization.py +++ b/tests/unit/test_post_check_dependency_id_normalization.py @@ -10,7 +10,9 @@ from sync_state_machine.common.collection import DataCollection from sync_state_machine.common.persistence import PersistenceBackend from sync_state_machine.common.types import BindingStatus, SyncAction, SyncStatus from sync_state_machine.domain.construction.sync_node import ConstructionTaskSyncNode +from sync_state_machine.domain.construction.sync_strategy import ConstructionTaskSyncStrategy from sync_state_machine.domain.contract.sync_node import ContractSyncNode +from sync_state_machine.domain.contract.sync_strategy import ContractSyncStrategy from sync_state_machine.sync_system.strategy_ops.post_check_ops import evaluate_consistency_for_node_type @@ -138,8 +140,16 @@ async def test_post_check_uses_same_id_resolver_as_create_update_for_depend_ids( await remote_collection.add(remote_task) await binding_manager.bind("construction_task", local_task.node_id, remote_task.node_id) + strategy = ConstructionTaskSyncStrategy( + "construction_task", + local_collection, + remote_collection, + binding_manager, + ) + result = await evaluate_consistency_for_node_type( node_type="construction_task", + strategy=strategy, local_collection=local_collection, remote_collection=remote_collection, binding_manager=binding_manager, @@ -152,3 +162,92 @@ async def test_post_check_uses_same_id_resolver_as_create_update_for_depend_ids( assert result["mismatch_count"] == 0 await persistence.close() + + +@pytest.mark.asyncio +async def test_post_check_ignores_only_explicit_configured_fields(tmp_path: Path) -> None: + db_path = tmp_path / "state.db" + persistence = PersistenceBackend(str(db_path)) + await persistence.initialize() + + local_collection = DataCollection("local", persistence=persistence, auto_persist=False) + remote_collection = DataCollection("remote", persistence=persistence, auto_persist=False) + binding_manager = BindingManager(persistence, auto_persist=False) + + local_contract = ContractSyncNode( + node_id="local-contract-node", + data_id="local-contract-1", + data={ + "id": "local-contract-1", + "project_id": "", + "name": "合同A-LOCAL", + "short_name": "合同A", + "code": "C-LOCAL", + "contract_type": "施工", + "currency": "CNY", + "manager_name": "张三", + "manager_phone": "13800000000", + "sign_company": "示例公司", + "company_id": "company-1", + "sign_date": "2025-01-01", + "credit_code": None, + "total_price": 1.0, + "sub_name": None, + }, + binding_status=BindingStatus.NORMAL, + action=SyncAction.NONE, + status=SyncStatus.SUCCESS, + ) + remote_contract = ContractSyncNode( + node_id="remote-contract-node", + data_id="remote-contract-1", + data={ + "id": "remote-contract-1", + "project_id": "", + "name": "合同A-REMOTE", + "short_name": "合同A", + "code": "C-REMOTE", + "contract_type": "施工", + "currency": "CNY", + "manager_name": "张三", + "manager_phone": "13800000000", + "sign_company": "示例公司", + "company_id": "company-1", + "sign_date": "2025-01-01", + "credit_code": None, + "total_price": 1.0, + "sub_name": None, + }, + binding_status=BindingStatus.NORMAL, + action=SyncAction.NONE, + status=SyncStatus.SUCCESS, + ) + + await local_collection.add(local_contract) + await remote_collection.add(remote_contract) + await binding_manager.bind("contract", local_contract.node_id, remote_contract.node_id) + + strategy = ContractSyncStrategy( + "contract", + local_collection, + remote_collection, + binding_manager, + ) + + result = await evaluate_consistency_for_node_type( + node_type="contract", + strategy=strategy, + local_collection=local_collection, + remote_collection=remote_collection, + binding_manager=binding_manager, + logger=logging.getLogger("sync_state_machine"), + compare_to_remote=True, + ignore_fields=["code"], + ) + + field_names = {field["field_name"] for field in result["fields"]} + assert "code" not in field_names + assert "name" in field_names + assert result["ignore_fields"] == ["code"] + + await persistence.close() diff --git a/tests/unit/test_run_presets_yaml_loader.py b/tests/unit/test_run_presets_yaml_loader.py index 4a407c1..f22a07c 100644 --- a/tests/unit/test_run_presets_yaml_loader.py +++ b/tests/unit/test_run_presets_yaml_loader.py @@ -6,7 +6,7 @@ import pytest from pydantic import ConfigDict -from sync_state_machine.config import BaseDataSourceConfig, build_config_from_file, register_datasource_type +from sync_state_machine.config import BaseDataSourceConfig, build_config_from_file, config_snapshot, register_datasource_type from sync_state_machine.config.run_presets import load_overrides_from_file from sync_state_machine.datasource import BaseDataSource @@ -123,3 +123,84 @@ logging: assert config.local_datasource.type == "file_backed" assert getattr(config.local_datasource, "namespace") == "local-demo" + + +def test_build_config_from_sparse_profile_resolves_full_strategy_defaults(tmp_path: Path) -> None: + profile = tmp_path / "sparse_profile.yaml" + profile.write_text( + """ +node_types: + - project + - production +local_datasource: + type: jsonl + jsonl_dir: ${PROJECT_ROOT}/filtered_datasource/datasource/filtered +remote_datasource: + type: jsonl + jsonl_dir: ${PROJECT_ROOT}/remote_datasource/datasource +persist: + db_path: ${PROJECT_ROOT}/runtime/test.db +logging: + initialize: false +strategies: + project: + config: + local_orphan_action: none + post_check_ignore_fields: + - code + production: + config: + post_check_ignore_fields: + - is_exist +""".strip(), + encoding="utf-8", + ) + + config = build_config_from_file(project_root=tmp_path, file_path=profile) + strategies = {item.node_type: item for item in config.strategies} + + project = strategies["project"].config + assert project.auto_bind is False + assert project.auto_bind_fields == [] + assert project.depend_fields == {"company_id": "company"} + assert project.local_orphan_action == "none" + assert project.remote_orphan_action == "none" + assert project.update_direction == "push" + assert project.post_check_ignore_fields == ["code"] + assert project.compare_ignore_fields == [] + + production = strategies["production"].config + assert production.auto_bind_fields == ["project_id", "code"] + assert production.depend_fields == {"project_id": "project"} + assert production.post_check_ignore_fields == ["is_exist"] + + snapshot = config_snapshot(config) + assert snapshot["strategies"][0]["config"]["compare_ignore_fields"] == [] + assert snapshot["strategies"][0]["config"]["field_direction_overrides"] == {} + + +def test_build_config_from_file_rejects_legacy_strategy_keys(tmp_path: Path) -> None: + profile = tmp_path / "legacy_profile.yaml" + profile.write_text( + """ +node_types: + - supplier +local_datasource: + type: jsonl + jsonl_dir: ${PROJECT_ROOT}/filtered_datasource/datasource/filtered +remote_datasource: + type: jsonl + jsonl_dir: ${PROJECT_ROOT}/remote_datasource/datasource +persist: + db_path: ${PROJECT_ROOT}/runtime/test.db +logging: + initialize: false +strategies: + supplier: + load_mode: persisted_only +""".strip(), + encoding="utf-8", + ) + + with pytest.raises(ValueError, match="unsupported legacy keys: load_mode"): + build_config_from_file(project_root=tmp_path, file_path=profile) diff --git a/tests/unit/test_scope_persistence_partitioning.py b/tests/unit/test_scope_persistence_partitioning.py new file mode 100644 index 0000000..a97fae9 --- /dev/null +++ b/tests/unit/test_scope_persistence_partitioning.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +from sync_state_machine.common.binding import BindingManager +from sync_state_machine.common.collection import DataCollection +from sync_state_machine.common.persistence import PersistenceBackend +from tests.fixtures.mock_domain import build_project_node, register_test_domain + + +@pytest.mark.asyncio +async def test_scope_partitioning_and_copy_between_scopes(tmp_path: Path) -> None: + register_test_domain() + + db_path = tmp_path / "scope_partitioning.db" + persistence = PersistenceBackend(str(db_path)) + await persistence.initialize() + + global_collection = DataCollection("local", scope="global", persistence=persistence, auto_persist=False) + global_node = build_project_node("node-global-1", "project-global-1", name="Global Project", code="GP-1") + await global_collection.add(global_node) + await global_collection.persist() + + global_bindings = BindingManager(persistence, auto_persist=False, scope="global") + await global_bindings.bind("test_project", global_node.node_id, "remote-node-1") + await global_bindings.persist() + + target_scope = "project_id:abc" + await persistence.copy_nodes_between_scopes( + collection_id="local", + source_scope="global", + target_scope=target_scope, + node_types=["test_project"], + ) + await persistence.copy_bindings_between_scopes( + source_scope="global", + target_scope=target_scope, + node_types=["test_project"], + ) + + scoped_collection = DataCollection("local", scope=target_scope, persistence=persistence, auto_persist=False) + await scoped_collection.load_from_persistence() + scoped_nodes = scoped_collection.filter(node_type="test_project") + assert len(scoped_nodes) == 1 + assert scoped_nodes[0].data_id == "project-global-1" + + scoped_bindings = BindingManager(persistence, auto_persist=False, scope=target_scope) + await scoped_bindings.load_from_persistence("test_project") + assert await scoped_bindings.get_remote_id("test_project", "node-global-1") == "remote-node-1" + + reloaded_global = DataCollection("local", scope="global", persistence=persistence, auto_persist=False) + await reloaded_global.load_from_persistence() + assert len(reloaded_global.filter(node_type="test_project")) == 1 + + await persistence.close() \ No newline at end of file diff --git a/tests/unit/test_summary_report.py b/tests/unit/test_summary_report.py new file mode 100644 index 0000000..d175e6e --- /dev/null +++ b/tests/unit/test_summary_report.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +import logging +from io import StringIO + +import pytest + +from sync_state_machine.pipeline.summary_report import print_pipeline_summary + + +class _FakeBindingManager: + async def get_all_records(self, node_type: str) -> list[dict]: + return [] + + +class _FakeDataSource: + def get_action_summary(self) -> dict: + return {} + + +class _FakeCollection: + def filter(self, *, node_type: str) -> list: + return [] + + +@pytest.mark.asyncio +async def test_pipeline_summary_omits_diff_schemas_column() -> None: + stream = StringIO() + logger = logging.getLogger("tests.summary_report") + logger.handlers.clear() + logger.setLevel(logging.INFO) + logger.propagate = False + + handler = logging.StreamHandler(stream) + handler.setFormatter(logging.Formatter("%(message)s")) + logger.addHandler(handler) + + try: + await print_pipeline_summary( + logger=logger, + node_types=["project"], + binding_manager=_FakeBindingManager(), + local_datasource=_FakeDataSource(), + remote_datasource=_FakeDataSource(), + local_collection=_FakeCollection(), + remote_collection=_FakeCollection(), + consistency_by_type={ + "project": { + "checked_pairs": 1, + "mismatch_count": 1, + "fields": [ + { + "field_name": "actual_production_date", + "mismatch_count": 1, + "total_count": 1, + "schema_refs": ["ProjectResponseBase", "ProjectBaseInfoUpdate"], + "samples": [{"local": None, "remote": "2025-11-14"}], + } + ], + } + }, + stats={ + "loaded": 1, + "synced": 0, + "failed": 0, + "post_check_passed": True, + }, + ) + finally: + logger.removeHandler(handler) + + output = stream.getvalue() + assert "diff_schemas" not in output + assert "ProjectResponseBase" not in output + assert "ProjectBaseInfoUpdate" not in output + assert "field mismatch/total samples" in output + assert "actual_production_date" in output + assert "None/'2025-11-14'" in output \ No newline at end of file diff --git a/tests/unit/test_target_project_preload_filter.py b/tests/unit/test_target_project_preload_filter.py index 4952c5f..03ff756 100644 --- a/tests/unit/test_target_project_preload_filter.py +++ b/tests/unit/test_target_project_preload_filter.py @@ -77,3 +77,38 @@ async def test_jsonl_non_project_load_filters_rows_by_project_id(tmp_path): loaded = datasource._collection.filter(node_type="contract") # type: ignore[union-attr] assert {node.data_id for node in loaded} == {"c1", "c3"} + + +@pytest.mark.asyncio +async def test_jsonl_non_project_load_inherits_project_filter_from_loaded_projects(tmp_path): + project_file = tmp_path / "project_1.jsonl" + project_file.write_text( + "\n".join([ + json.dumps({"id": "p1"}, ensure_ascii=False), + ]) + + "\n", + encoding="utf-8", + ) + contract_file = tmp_path / "contract_1.jsonl" + contract_file.write_text( + "\n".join([ + json.dumps({"id": "c1", "project_id": "p1"}, ensure_ascii=False), + json.dumps({"id": "c2", "project_id": "p2"}, ensure_ascii=False), + json.dumps({"id": "c3"}, ensure_ascii=False), + ]) + + "\n", + encoding="utf-8", + ) + + datasource = JsonlDataSource(tmp_path) + project_handler = BaseJsonlHandler(datasource, "project", _ProjectNode, _ProjectSchema) + project_handler.set_handler_config({"data_id_filter": ["p1"]}) + contract_handler = BaseJsonlHandler(datasource, "contract", _BizNode, _BizSchema) + datasource.register_handler(project_handler) + datasource.register_handler(contract_handler) + datasource.set_collection(DataCollection("local")) + + await datasource.load_all(order=["project", "contract"]) + + loaded = datasource._collection.filter(node_type="contract") # type: ignore[union-attr] + assert {node.data_id for node in loaded} == {"c1", "c3"} diff --git a/tests/unit/test_update_ops.py b/tests/unit/test_update_ops.py index 2ac3d4e..4f4628a 100644 --- a/tests/unit/test_update_ops.py +++ b/tests/unit/test_update_ops.py @@ -2,6 +2,7 @@ import pytest from unittest.mock import AsyncMock, patch, MagicMock from sync_state_machine.sync_system.config import UpdateDirection from sync_state_machine.sync_system.strategy_ops.update_ops import run_update +from sync_state_machine.domain.construction.sync_strategy import ConstructionTaskSyncStrategy @pytest.mark.asyncio async def test_run_update_with_direction_overrides(): @@ -65,3 +66,23 @@ async def test_run_update_with_direction_overrides(): node_empty = FakeNode(None) assert push_filter(node_empty) is True assert pull_filter(node_empty) is False + + +def test_construction_task_compare_payload_ignores_plan_node_is_audit(): + strategy = ConstructionTaskSyncStrategy( + "construction_task", + MagicMock(), + MagicMock(), + MagicMock(), + ) + + local_payload = { + "plan_node": [{"name": "A", "is_audit": True}], + "task_type": "demo", + } + remote_payload = { + "plan_node": [{"name": "A", "is_audit": False}], + "task_type": "demo", + } + + assert strategy.normalize_compare_payload(local_payload) == strategy.normalize_compare_payload(remote_payload)