Merge branch 'main' of https://git.getaix.tech/strepsiades/ecm_sync_system

docs/enum_refactor/init_project_data.md

@@ -2,7 +2,7 @@
 
 ## 1. 范围
 
-本文档只分析 backend 里的 `ProjectService._init_project_data()` 当前到底做了什么。
+本文档只分析 backend 里的 `ProjectService.init_project_data)` 当前到底做了什么。
 
 重点回答 4 个问题：
 
@@ -13,13 +13,13 @@
 
 ## 2. 触发时机
 
-`_init_project_data()` 在项目审批通过后执行。
+`init_project_data)` 在项目审批通过后执行。
 
 调用链是：
 
 - `ProjectService.audit_project()`
 - `ProjectService._audit_approval()`
-- `ProjectService._init_project_data()`
+- `ProjectService.init_project_data)`
 
 也就是说，这不是“创建项目主表时立即执行”的逻辑，而是“审批通过后批量灌初始化数据”的逻辑。
 
@@ -51,7 +51,7 @@
 
 - `configs = {config_key: decoded_json}`
 
-但在 `_init_project_data()` 自己内部，真正直接用到的 config 只有 1 个：
+但在 `init_project_data)` 自己内部，真正直接用到的 config 只有 1 个：
 
 - `contract_config`
 
@@ -140,7 +140,7 @@
 
 创建方式：
 
-- `_init_project_data()` 从 `configs["contract_config"]` 取出列表
+- `init_project_data)` 从 `configs["contract_config"]` 取出列表
 - `ContractService.init_from_project()` 根据每项的 `code`、`name`、`contract_type` 创建合同
 - 先删项目下旧合同，再批量插入新合同
 
@@ -169,7 +169,7 @@
 
 创建来源：
 
-- 不是 `_init_project_data()` 直接创建。
+- 不是 `init_project_data)` 直接创建。
 - 它是 `contract.init_from_project()` 内部继续调用 `material.init_from_contract()` 创建。
 
 当前来源拆分：
@@ -244,7 +244,7 @@
 
 创建来源：
 
-- `_init_project_data()` 自己没有直接给施工传 dict 或 config 数据
+- `init_project_data)` 自己没有直接给施工传 dict 或 config 数据
 - 只调用：`create_constructions_for_project(tenant_id, project_id)`
 
 但施工初始化内部会继续读取：
@@ -280,7 +280,7 @@
 创建来源：
 
 - 不依赖 dict
-- 也不依赖 `_init_project_data()` 读出来的 config
+- 也不依赖 `init_project_data)` 读出来的 config
 
 当前方式：
 
@@ -298,7 +298,7 @@
 
 ## 6. 这 6 组 dict 里，哪些真的在负责事情
 
-### 6.1 直接被 `_init_project_data()` 消费的 dict
+### 6.1 直接被 `init_project_data)` 消费的 dict
 
 - `preparation_{project_type}`
 - `production_{project_type}`
@@ -306,7 +306,7 @@
 
 这 3 组 dict 是真的在当前函数里直接承担初始化模板职责。
 
-### 6.2 被 `_init_project_data()` 取了，但当前函数里没有直接消费的 dict
+### 6.2 被 `init_project_data)` 取了，但当前函数里没有直接消费的 dict
 
 - `construction_contract_{project_type}`
 - `service_contract_{project_type}`
@@ -314,11 +314,11 @@
 
 其中：
 
-- `construction_contract_*`：当前只看到被 `_init_project_data()` 取出，但没有实际使用。
-- `service_contract_*`：当前也只看到被 `_init_project_data()` 取出，但没有实际使用。
+- `construction_contract_*`：当前只看到被 `init_project_data)` 取出，但没有实际使用。
+- `service_contract_*`：当前也只看到被 `init_project_data)` 取出，但没有实际使用。
 - `material_contract_*`：当前函数本身没用，但会在 `MaterialService.init_from_contract()` 里被重新读取一次，用于补单位。
 
-这说明当前 `_init_project_data()` 至少存在两类历史包袱：
+这说明当前 `init_project_data)` 至少存在两类历史包袱：
 
 - 取了但没用的 dict key
 - 同一套语义在上游合同 config 和下游 material dict 之间分叉
@@ -374,7 +374,7 @@
 
 ## 8. 额外现状问题
 
-### 8.1 `_init_project_data()` 读了整类 config，但只直接用了 `contract_config`
+### 8.1 `init_project_data)` 读了整类 config，但只直接用了 `contract_config`
 
 这会导致：
 
@@ -424,7 +424,7 @@
 
 ## 9. 结论
 
-可以把 `_init_project_data()` 当前的配置来源，拆成 4 类理解：
+可以把 `init_project_data)` 当前的配置来源，拆成 4 类理解：
 
 1. 纯 dict 初始化：`preparation`、`production`、`power`
 2. 纯 config 初始化：`contract`
@@ -439,14 +439,14 @@
 
 ## 10. 建议落地顺序
 
-1. 先把 `_init_project_data()` 里“取了但没直接用”的 `construction_contract_*`、`service_contract_*` 从函数里清出来，确认是否还能删掉。
+1. 先把 `init_project_data)` 里“取了但没直接用”的 `construction_contract_*`、`service_contract_*` 从函数里清出来，确认是否还能删掉。
 2. 再把 `preparation_{project_type}`、`production_{project_type}`、`power_{project_type}` 迁到 config。
 3. 再处理 `material_contract_*`，但前提是先统一 `contract_config` 和 material 初始化所需的单位/展示信息来源。
 4. 施工初始化不要和这批 dict 迁移混做，单独处理。
 
 ## 10.1 按本轮改造目标，初始化链路具体要改什么
 
-如果把这次目标收敛成“初始化统一走配置工具”，那么 `_init_project_data()` 不应该再只是分析对象，而是明确要往下面的形态改。
+如果把这次目标收敛成“初始化统一走配置工具”，那么 `init_project_data)` 不应该再只是分析对象，而是明确要往下面的形态改。
 
 ### A. 保留并整理的 config
 
@@ -472,7 +472,7 @@
 
 迁移后的目标形态是：
 
-- `_init_project_data()` 不再直接读这些 dict key
+- `init_project_data)` 不再直接读这些 dict key
 - 改由 `ConfigService.get_field_options()` 或等价工具读取字段候选项
 - service 仍只负责“删旧 + 批量重建”，不再关心候选项来自 dict 还是 config
 
@@ -493,9 +493,9 @@
 - 把 `unit`、`label`、`parts` 等字段并回正式 config 结构
 - 不再让单位这类业务字段挂在 dict `remark` 上
 
-### D. `_init_project_data()` 迁移后的读取原则
+### D. `init_project_data)` 迁移后的读取原则
 
-迁移后，`_init_project_data()` 本身应只做 2 件事：
+迁移后，`init_project_data)` 本身应只做 2 件事：
 
 - 找到当前项目类型下所需的候选项配置
 - 把候选项列表传给各领域初始化 service
@@ -513,11 +513,11 @@
 - `preparation`、`production`、`power` 初始化不再依赖 dict
 - `material` 初始化不再依赖 dict `remark` 补单位
 - 初始化候选项与前端下拉、create 校验使用同一套 config 真源
-- `_init_project_data()` 中不再保留只读取不用的 dict key
+- `init_project_data)` 中不再保留只读取不用的 dict key
 
 ## 11. 从接口创建视角反看这些字段是不是已经“主要 config 化”
 
-只看 `_init_project_data()`，容易得出“系统已经比较 config 驱动”的印象。
+只看 `init_project_data)`，容易得出“系统已经比较 config 驱动”的印象。
 
 但如果再抽样看接口创建路径，会发现当前真实情况是：
 

docs/enum_refactor/personnel.md

@@ -112,7 +112,7 @@ personnel 领域不是这轮“项目初始化 + 项目类型 field_config”第
 
 本地人员资质领域当前没有项目初始化批量灌数链路。
 
-因此本轮不把它纳入 `_init_project_data()` 第一批改造范围。
+因此本轮不把它纳入 `init_project_data)` 第一批改造范围。
 
 ### 5.2 Create 校验
 

docs/enum_refactor/plan.md

@@ -35,7 +35,7 @@
 
 ### 1. 优化项目初始化过程
 
-目标不是只分析 `_init_project_data()`，而是把初始化链路改成统一从配置工具取候选项：
+目标不是只分析 `init_project_data)`，而是把初始化链路改成统一从配置工具取候选项：
 
 - 整理项目初始化当前实际使用到的 config。
 - 把适合迁移的初始化候选项改为通过新的 config 工具获取后创建。
@@ -110,6 +110,12 @@
 - 创建：通过统一配置工具校验 code 类字段
 - 下拉：通过统一配置工具返回选项列表
 
+对本轮来说，这里的“统一出口”不是泛指，而是明确服务于 3 条业务链路：
+
+- 初始化：通过统一配置工具取候选项并创建数据
+- 创建：通过统一配置工具校验 code 类字段
+- 下拉：通过统一配置工具返回选项列表
+
 ## 3. 不在本次范围内的内容
 
 以下内容明确不在本次重构范围：
@@ -126,6 +132,11 @@
 - 不允许为了兼容旧逻辑继续新增 dict `remark` 之类的隐式字段承载方式。
 - 不允许为了减少改动，在新工具落地后继续保留“前端下拉走 config、初始化走 dict、create 校验走代码字典”的三头并存状态。
 
+这里补充两条边界：
+
+- 不允许为了兼容旧逻辑继续新增 dict `remark` 之类的隐式字段承载方式。
+- 不允许为了减少改动，在新工具落地后继续保留“前端下拉走 config、初始化走 dict、create 校验走代码字典”的三头并存状态。
+
 因此，本次允许字段级配置中的 `value / label` 与现有 dict 翻译短期重复。后续如有需要，可再通过脚本生成 dict 或做统一翻译收敛。
 
 ## 4. 核心设计
@@ -297,6 +308,17 @@ async def get_field_options_for_frontend(project_type: str, key: str, tenant_id:
     ...
 
 
+async def validate_field_values(project_type: str, items: list[tuple[str, str]], tenant_id: str | None = None) -> None:
+    ...
+```
+
+如果后续需要覆盖“批量校验多个字段”或“直接返回下拉结构”的场景，可以继续补：
+
+```python
+async def get_field_options_for_frontend(project_type: str, key: str, tenant_id: str | None = None) -> list[dict[str, Any]]:
+    ...
+
+
 async def validate_field_values(project_type: str, items: list[tuple[str, str]], tenant_id: str | None = None) -> None:
     ...
 ```
@@ -337,6 +359,12 @@ async def validate_field_values(project_type: str, items: list[tuple[str, str]],
 - 创建和编辑时：`validate_field_value()`
 - 下拉接口时：`get_field_options_for_frontend()` 或 `get_field_options()`
 
+另外，本轮会直接用到两种典型调用方式：
+
+- 初始化时：`get_field_options()` / `get_field_option_map()`
+- 创建和编辑时：`validate_field_value()`
+- 下拉接口时：`get_field_options_for_frontend()` 或 `get_field_options()`
+
 这里特别强调：
 
 - `validate` 是手动调用的
@@ -413,6 +441,12 @@ async def validate_field_values(project_type: str, items: list[tuple[str, str]],
 - 初始化改造前后，可以快速对比 config 差异
 - 下拉列表和创建校验依赖的 config，可以单独导出检查，不再手工翻长 CSV
 
+这块和本轮目标直接对应：
+
+- 字段级配置调整后，可以直接导出为 `config.csv`
+- 初始化改造前后，可以快速对比 config 差异
+- 下拉列表和创建校验依赖的 config，可以单独导出检查，不再手工翻长 CSV
+
 建议目录：
 
 - `scripts/initdata/json_config/`
@@ -507,6 +541,12 @@ async def validate_field_values(project_type: str, items: list[tuple[str, str]],
 - 创建：这个字段 create service 当前怎么校验，迁移后改成哪个 `validate_field_value()` 调用
 - 下拉：这个字段前端当前通过哪个接口取值，迁移后改成哪个 config key 统一返回
 
+对本轮来说，每个领域文档都至少要把下面 3 条链路写清楚：
+
+- 初始化：这个字段初始化时从哪里取候选项，迁移后改成哪个 config key
+- 创建：这个字段 create service 当前怎么校验，迁移后改成哪个 `validate_field_value()` 调用
+- 下拉：这个字段前端当前通过哪个接口取值，迁移后改成哪个 config key 统一返回
+
 ### 8.4 必须完整对齐集团侧 `enum_config`
 
 领域梳理时，不能只挑 depm 里已经处理过的字段，也不能只挑“看起来最容易改”的字段。
@@ -526,6 +566,7 @@ async def validate_field_values(project_type: str, items: list[tuple[str, str]],
 
 ## 9. 实施步骤
 
+### 第一阶段：公共能力
 ### 第一阶段：公共能力
 
 - 明确字段级配置固定 JSON 结构
@@ -538,10 +579,10 @@ async def validate_field_values(project_type: str, items: list[tuple[str, str]],
 
 ### 第二阶段：初始化改造
 
-- 逐项整理 `_init_project_data()` 当前真实依赖的 config / dict
+- 逐项整理 `init_project_data)` 当前真实依赖的 config / dict
 - 把 `preparation`、`production`、`power` 这类平铺型初始化模板迁到 config
 - 整理 `material` 初始化，把单位等附属属性从不适当的 dict 备注迁出
-- 清理 `_init_project_data()` 中取了但不该继续作为真源的 dict 依赖
+- 清理 `init_project_data)` 中取了但不该继续作为真源的 dict 依赖
 
 ### 第三阶段：创建校验改造
 
@@ -549,6 +590,7 @@ async def validate_field_values(project_type: str, items: list[tuple[str, str]],
 - 把当前分散在 service 内的硬编码白名单逐步迁到 config 工具
 - 保留真正应该继续存在的代码 enum，不把固定语义状态误迁到 config
 
+### 第四阶段：下拉列表改造
 ### 第四阶段：下拉列表改造
 
 - 梳理 code 类下拉接口
@@ -558,6 +600,16 @@ async def validate_field_values(project_type: str, items: list[tuple[str, str]],
 
 ### 第五阶段：清理和测试
 
+- 删除废弃 enum 和废弃代码字典
+- 删除不再适合作为真源的 dict 依赖
+- 补 service 层集成测试和初始化测试
+- 梳理 code 类下拉接口
+- 下拉统一改成通过 config 工具返回
+- 少量仍写死的下拉改为动态获取
+- 保证前端下拉、初始化候选项、create 校验三者同源
+
+### 第五阶段：清理和测试
+
 - 删除废弃 enum 和废弃代码字典
 - 删除不再适合作为真源的 dict 依赖
 - 补 service 层集成测试和初始化测试
@@ -572,6 +624,7 @@ async def validate_field_values(project_type: str, items: list[tuple[str, str]],
 - 一套 JSON / CSV / DB 三者之间的转换与维护规范
 - 一组领域级重构文档
 - 一组初始化 / 下拉 / create 的 service 层集成测试
+- 一组初始化 / 下拉 / create 的 service 层集成测试
 
 判断本次重构是否成功的标准不是“历史实现全部统一”，而是：
 
@@ -610,3 +663,35 @@ async def validate_field_values(project_type: str, items: list[tuple[str, str]],
 - 测试代码写在 `tests/` 下
 - 优先写 service 层集成测试，不在项目根目录放临时脚本
 - 如果需要初始化测试数据，应复用现有测试体系，不改 `schemas/`、不改 `filtered_datasource/`
+- 前端下拉、初始化、create 校验不再各自读不同真源
+
+## 11. 自动化测试要求
+
+这次测试不是只补 unit test，而是以 service 层集成测试为主。
+
+至少覆盖下面 3 组：
+
+### 11.1 初始化测试
+
+- 覆盖项目初始化入口
+- 校验初始化创建出的 preparation / production / contract / material / power / construction 数据是否来自目标 config
+- 校验错误配置时直接报错，不做静默降级
+
+### 11.2 下拉列表测试
+
+- 覆盖 contract / material / construction 等核心下拉接口或对应 service
+- 校验返回项是否来自预期 config key
+- 校验下拉返回和初始化候选项是否一致
+
+### 11.3 Create 校验测试
+
+- 直接走 service 层 create
+- 对合法 code 校验通过
+- 对非法 code 明确抛错
+- 校验不再依赖旧硬编码字典或 dict `remark`
+
+### 11.4 测试组织方式
+
+- 测试代码写在 `tests/` 下
+- 优先写 service 层集成测试，不在项目根目录放临时脚本
+- 如果需要初始化测试数据，应复用现有测试体系，不改 `schemas/`、不改 `filtered_datasource/`

docs/enum_refactor/preparation.md

@@ -100,13 +100,6 @@ preparation 属于本轮最适合先收口的一类：
 
 当前 preparation 初始化直接来自 `preparation_{project_type}` dict。
 
-同时本地还存在：
-
-- `PreparationCode` enum
-- `name_to_code`
-- `dict_to_preparation_code_name`
-- `list_to_preparation_code_name`
-
 本轮目标是：
 
 - 把 `preparation.code` 迁到 field config
@@ -129,17 +122,6 @@ preparation 当前没有单独的“节点编码下拉接口”，前端更多
 这轮文档上的要求是：
 
 - 如果后续需要独立 code 候选项下拉，应直接从 `preparation.code` field config 返回
-- 不再把 `list_to_preparation_code_name` 当成前端下拉真源
-
-### 5.4 清理项
-
-preparation 领域本轮改造完成后，应优先清理：
-
-- `preparation_{project_type}` dict 的初始化真源职责
-- `PreparationCode` 相关重复映射表
-- 列表排序里写死的 `sort_code`
-
-排序如果仍需要固定顺序，应迁到 field config 的附属字段中表达，例如 `weight` 或 `sort`。
 
 ## 6. 测试关注点
 

docs/enum_refactor/production.md

@@ -95,19 +95,12 @@ production 和 preparation 一样，属于本轮适合优先收口的平铺型
 
 当前 production 初始化直接来自 `production_{project_type}` dict。
 
-同时本地还存在：
-
-- `ProductionCode` enum
-- `dict_to_production_code_name`
-- `list_to_production_code_name`
-- 列表排序里的 `sort_code`
-
 本轮目标是：
 
 - 把 `production.code` 迁到 field config
 - 初始化改成通过 config 工具读取候选项后批量创建
 - 不再依赖 dict 作为初始化真源
-
+∏
 ### 5.2 Create 校验
 
 当前 production 领域也没有统一的 code 候选项校验入口。
@@ -124,17 +117,6 @@ production 当前也没有单独的 code 候选项下拉接口，更多是返回
 本轮要求是：
 
 - 如果需要单独 code 候选项下拉，应直接从 `production.code` field config 返回
-- 不再把 `list_to_production_code_name` 当成前端下拉真源
-
-### 5.4 清理项
-
-production 领域本轮改造完成后，应优先清理：
-
-- `production_{project_type}` dict 的初始化真源职责
-- `ProductionCode` 相关重复映射和展示列表
-- 列表排序中写死的 `sort_code`
-
-排序如果仍需要固定顺序，也应迁到配置附属字段中表达。
 
 ## 6. 测试关注点
 

docs/enum_refactor/project.md

@@ -29,7 +29,7 @@
 
 当前使用：
 
-- `ProjectService._init_project_data()` 用它作为初始化路由键，去取 `preparation_{project_type}`、`production_{project_type}`、`power_{project_type}` 等配置
+- `ProjectService.init_project_data)` 用它作为初始化路由键，去取 `preparation_{project_type}`、`production_{project_type}`、`power_{project_type}` 等配置
 - construction 模块的部分接口和工具也直接复用 `ProjectType`
 
 当前问题：

sync_state_machine/__init__.py

@@ -15,10 +15,10 @@ from .common.registry import DomainRegistry
 from .config import (
 	ApiDataSourceConfig,
 	BaseDataSourceConfig,
-	DbDataSourceConfig,
 	JsonlDataSourceConfig,
 	PipelineRunConfig,
 	build_config,
+	build_config_from_file,
 	build_default_config,
 	get_datasource_config_model,
 	get_datasource_factory,
@@ -40,7 +40,7 @@ except PackageNotFoundError:
 
 def ensure_domain_registry_loaded() -> None:
 	"""Ensure built-in domain registrations are imported."""
-	_domain  # keep module import for side effects
+	_ = _domain
 
 
 __all__ = [
@@ -50,9 +50,9 @@ __all__ = [
 	"PipelineRunConfig",
 	"ApiDataSourceConfig",
 	"BaseDataSourceConfig",
-	"DbDataSourceConfig",
 	"JsonlDataSourceConfig",
 	"build_config",
+	"build_config_from_file",
 	"build_default_config",
 	"get_datasource_config_model",
 	"get_datasource_factory",

sync_state_machine/config/__init__.py

@@ -1,4 +1,4 @@
-from .datasource_config import ApiDataSourceConfig, BaseDataSourceConfig, DbDataSourceConfig, JsonlDataSourceConfig, DataSourceConfig
+from .datasource_config import ApiDataSourceConfig, BaseDataSourceConfig, JsonlDataSourceConfig, DataSourceConfig
 from .persist_config import PersistConfig
 from .logging_config import LoggingConfig, resolve_log_file_path
 from ..datasource.type_registry import (
@@ -27,6 +27,7 @@ from .pipeline_config import PipelineRunConfig, DEFAULT_NODE_TYPES
 from .builder import (
     build_default_config,
     build_config,
+    build_config_from_file,
     apply_config_overrides,
     apply_env_overrides,
     config_snapshot,
@@ -35,7 +36,6 @@ from .builder import (
 __all__ = [
     "ApiDataSourceConfig",
     "BaseDataSourceConfig",
-    "DbDataSourceConfig",
     "JsonlDataSourceConfig",
     "PipelineRunConfig",
     "DataSourceConfig",
@@ -61,6 +61,7 @@ __all__ = [
     "DEFAULT_NODE_TYPES",
     "build_default_config",
     "build_config",
+    "build_config_from_file",
     "apply_config_overrides",
     "apply_env_overrides",
     "config_snapshot",

sync_state_machine/config/builder.py

@@ -187,3 +187,11 @@ def build_config(
         config = apply_config_overrides(config, overrides)
 
     return config
+
+
+def build_config_from_file(
+    *,
+    project_root: Path,
+    file_path: Union[str, Path],
+) -> PipelineRunConfig:
+    return build_config(project_root=project_root, overrides_file=Path(file_path))

sync_state_machine/config/datasource_config.py

@@ -22,11 +22,6 @@ class JsonlDataSourceConfig(BaseDataSourceConfig):
     read_only: bool = False
 
 
-class DbDataSourceConfig(BaseDataSourceConfig):
-
-    type: Literal["db"] = "db"
-
-
 class ApiDataSourceConfig(BaseDataSourceConfig):
 
     type: Literal["api"] = "api"
@@ -52,5 +47,4 @@ DataSourceConfig = Annotated[
 
 
 register_datasource_config_model("jsonl", JsonlDataSourceConfig)
-register_datasource_config_model("db", DbDataSourceConfig)
 register_datasource_config_model("api", ApiDataSourceConfig)

sync_state_machine/config/pipeline_config.py

@@ -38,7 +38,6 @@ class PipelineRunConfig(BaseModel):
     model_config = ConfigDict(extra="forbid", validate_assignment=True)
 
     node_types: List[str] = Field(default_factory=lambda: list(DEFAULT_NODE_TYPES))
-    target_project_ids: List[str] = Field(default_factory=list)
 
     local_datasource: DataSourceConfig
     remote_datasource: DataSourceConfig

sync_state_machine/config/run_presets.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import json
-import importlib
 from pathlib import Path
 from typing import Any, Dict, List, Tuple
 
@@ -29,43 +28,6 @@ def _replace_placeholders(value: Any, project_root: Path) -> Any:
     return value
 
 
-def _import_from_path(import_path: str) -> Any:
-    module_path, sep, attr_path = import_path.partition(":")
-    if not module_path:
-        raise ValueError(f"Invalid import path: {import_path}")
-
-    module = importlib.import_module(module_path)
-    if not sep:
-        return module
-
-    obj: Any = module
-    for attr_name in attr_path.split("."):
-        obj = getattr(obj, attr_name)
-    return obj
-
-
-def _apply_datasource_bootstraps(payload: Dict[str, Any]) -> Dict[str, Any]:
-    extensions = payload.pop("extensions", None)
-    if extensions is None:
-        return payload
-    if not isinstance(extensions, dict):
-        raise ValueError("extensions must be an object mapping")
-
-    bootstrap_paths = extensions.get("datasource_bootstraps") or []
-    if not isinstance(bootstrap_paths, list):
-        raise ValueError("extensions.datasource_bootstraps must be a list")
-
-    for item in bootstrap_paths:
-        if not isinstance(item, str) or not item.strip():
-            raise ValueError("extensions.datasource_bootstraps entries must be non-empty strings")
-        loaded = _import_from_path(item)
-        if not callable(loaded):
-            raise TypeError(f"datasource bootstrap must resolve to a callable: {item}")
-        loaded()
-
-    return payload
-
-
 def list_preset_names() -> List[str]:
     return sorted(PRESET_FILES.keys())
 
@@ -115,7 +77,8 @@ def load_overrides_from_file(file_path: Path, project_root: Path) -> Dict[str, A
     resolved = _replace_placeholders(payload, project_root)
     if not isinstance(resolved, dict):
         raise ValueError(f"Preset file must contain object mapping: {file_path}")
-    return _apply_datasource_bootstraps(resolved)
+    resolved.pop("extensions", None)
+    return resolved
 
 
 def load_named_preset_overrides(project_root: Path, preset_name: str) -> Tuple[Dict[str, Any], Path, bool]:

sync_state_machine/config/runtime_config.py

@@ -3,7 +3,7 @@
 Prefer importing from `sync_state_machine.config` directly.
 """
 
-from .datasource_config import ApiDataSourceConfig, DbDataSourceConfig, JsonlDataSourceConfig, DataSourceConfig
+from .datasource_config import ApiDataSourceConfig, JsonlDataSourceConfig, DataSourceConfig
 from .persist_config import PersistConfig
 from .logging_config import LoggingConfig
 from .strategy_config import (
@@ -17,6 +17,7 @@ from .pipeline_config import PipelineRunConfig, DEFAULT_NODE_TYPES
 from .builder import (
     build_default_config,
     build_config,
+    build_config_from_file,
     apply_config_overrides,
     apply_env_overrides,
     config_snapshot,
@@ -24,7 +25,6 @@ from .builder import (
 
 __all__ = [
     "ApiDataSourceConfig",
-    "DbDataSourceConfig",
     "JsonlDataSourceConfig",
     "DataSourceConfig",
     "PersistConfig",
@@ -38,6 +38,7 @@ __all__ = [
     "DEFAULT_NODE_TYPES",
     "build_default_config",
     "build_config",
+    "build_config_from_file",
     "apply_config_overrides",
     "apply_env_overrides",
     "config_snapshot",

sync_state_machine/datasource/__init__.py

@@ -23,13 +23,25 @@ from .handler import (
 from .task_result import TaskResult, TaskStatus
 
 # API DataSource
-from .api import ApiClient, ApiDataSource, BaseApiHandler
+from .api import ApiClient, ApiDataSource, BaseApiHandler, register as register_api_datasource
 
-# DB DataSource
+# DB base helpers
 from .db import DbDataSource, BaseDbHandler
 
 # JSONL DataSource  
-from .jsonl import JsonlDataSource, BaseJsonlHandler
+from .jsonl import JsonlDataSource, BaseJsonlHandler, register as register_jsonl_datasource
+
+
+_BUILTIN_DATASOURCE_TYPES_REGISTERED = False
+
+
+def ensure_builtin_datasource_types_registered(*, replace: bool = False) -> None:
+    global _BUILTIN_DATASOURCE_TYPES_REGISTERED
+    if _BUILTIN_DATASOURCE_TYPES_REGISTERED and not replace:
+        return
+    register_jsonl_datasource(replace=replace)
+    register_api_datasource(replace=replace)
+    _BUILTIN_DATASOURCE_TYPES_REGISTERED = True
 
 __all__ = [
     # 基础类
@@ -46,13 +58,14 @@ __all__ = [
     "TaskResult",
     "TaskStatus",
     "ValidationResult",
+    "ensure_builtin_datasource_types_registered",
     
     # API DataSource
     "ApiClient",
     "ApiDataSource",
     "BaseApiHandler",
 
-    # DB DataSource
+    # DB base helpers
     "DbDataSource",
     "BaseDbHandler",
     

sync_state_machine/datasource/api/__init__.py

@@ -1,15 +1,50 @@
+from __future__ import annotations
+
 """
 API DataSource 模块
 
 提供基于 HTTP API 的数据源实现。
 """
+from typing import TYPE_CHECKING
+
+from ..type_registry import register_datasource_type
 
 from .client import ApiClient
 from .datasource import ApiDataSource
 from .handler import BaseApiHandler
 
+if TYPE_CHECKING:
+    from ...config.datasource_config import ApiDataSourceConfig
+
+
+def build_api_datasource(config: ApiDataSourceConfig, endpoint_name: str, **_: object):
+    del endpoint_name
+    return ApiDataSource(
+        base_url=config.api_base_url,
+        uid=config.api_uid,
+        secret=config.api_secret,
+        debug=config.api_debug,
+        poll_max_retries=config.poll_max_retries,
+        poll_interval=config.poll_interval,
+        rate_limit_max_requests=config.api_rate_limit_max_requests,
+        rate_limit_window_seconds=config.api_rate_limit_window_seconds,
+    )
+
+
+def register(*, replace: bool = False) -> None:
+    from ...config.datasource_config import ApiDataSourceConfig
+
+    register_datasource_type(
+        "api",
+        config_model=ApiDataSourceConfig,
+        factory=build_api_datasource,
+        replace=replace,
+    )
+
 __all__ = [
     "ApiClient",
     "ApiDataSource",
     "BaseApiHandler",
+    "build_api_datasource",
+    "register",
 ]

sync_state_machine/datasource/datasource.py

@@ -88,6 +88,13 @@ class BaseDataSource(ABC):
         self._poll_interval = max(0, poll_interval)
         self._sm_runtime: Optional[StateMachineRuntime] = None
 
+    def set_target_project_ids(self, target_project_ids: List[str]) -> None:
+        self._target_project_ids = [str(item) for item in target_project_ids if str(item)]
+
+    @property
+    def target_project_ids(self) -> List[str]:
+        return list(self._target_project_ids)
+
     def _ensure_sm_runtime(self) -> StateMachineRuntime:
         if self._sm_runtime is None:
             cfg_path = Path(__file__).resolve().parents[1] / "config" / "node_state_machine.yaml"
@@ -957,6 +964,10 @@ class BaseDataSource(ABC):
         
         return summary
 
+    async def initialize(self) -> None:
+        """初始化数据源资源（子类可覆盖）"""
+        return None
+
     async def close(self) -> None:
         """关闭数据源资源（子类可覆盖以实现特定的清理逻辑）"""
-        pass  # 基类默认空实现
+        return None

sync_state_machine/datasource/db/__init__.py

@@ -1,4 +1,6 @@
-"""DB DataSource 模块。"""
+from __future__ import annotations
+
+"""DB DataSource 基类模块。"""
 
 from .datasource import DbDataSource
 from .handler import BaseDbHandler

sync_state_machine/datasource/jsonl/__init__.py

@@ -1,13 +1,53 @@
+from __future__ import annotations
+
 """
 JSONL DataSource 模块
 
 提供基于 JSONL 文件的数据源实现，用于测试和本地开发。
 """
 
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from ..type_registry import register_datasource_type
 from .datasource import JsonlDataSource
 from .handler import BaseJsonlHandler
+from .path_utils import prepare_remote_jsonl_dir
+
+if TYPE_CHECKING:
+    from ...config.datasource_config import JsonlDataSourceConfig
+
+
+def build_jsonl_datasource(
+    config: JsonlDataSourceConfig,
+    endpoint_name: str,
+    *,
+    project_root: Path | None = None,
+):
+    dir_path = Path(config.jsonl_dir)
+    if endpoint_name == "remote":
+        dir_path = prepare_remote_jsonl_dir(dir_path, project_root=project_root or Path.cwd())
+    elif not dir_path.exists() or not dir_path.is_dir():
+        raise FileNotFoundError(
+            f"{endpoint_name}_datasource.jsonl_dir does not exist or is not a directory: {dir_path}"
+        )
+    return JsonlDataSource(dir_path, read_only=config.read_only)
+
+
+def register(*, replace: bool = False) -> None:
+    from ...config.datasource_config import JsonlDataSourceConfig
+
+    register_datasource_type(
+        "jsonl",
+        config_model=JsonlDataSourceConfig,
+        factory=build_jsonl_datasource,
+        replace=replace,
+    )
 
 __all__ = [
     "JsonlDataSource",
     "BaseJsonlHandler",
+    "build_jsonl_datasource",
+    "prepare_remote_jsonl_dir",
+    "register",
 ]

sync_state_machine/datasource/jsonl/handler.py

@@ -50,24 +50,22 @@ class BaseJsonlHandler(BaseNodeHandler):
     ):
         super().__init__(node_type, node_class, schema)
         self.datasource = datasource
-        self._target_project_ids: Optional[Set[str]] = None
+        self._target_project_ids: List[str] = []
 
     def set_target_project_ids(self, target_project_ids: List[str]) -> None:
-        """注入项目过滤列表，供 load 阶段预过滤使用。"""
-        if target_project_ids:
-            self._target_project_ids = {str(pid) for pid in target_project_ids if str(pid)}
-        else:
-            self._target_project_ids = None
+        """注入项目过滤列表的本地副本。真正的过滤来源仍然是 datasource。"""
+        self._target_project_ids = [str(pid) for pid in target_project_ids if str(pid)]
 
     def _should_skip_item_by_project_filter(self, item: Dict[str, Any], item_id: str) -> bool:
-        if not self._target_project_ids:
+        target_project_ids = set(self.datasource.target_project_ids)
+        if not target_project_ids:
             return False
         if self.node_type == "project":
-            return item_id not in self._target_project_ids
+            return item_id not in target_project_ids
 
         project_id = item.get("project_id")
         if project_id:
-            return str(project_id) not in self._target_project_ids
+            return str(project_id) not in target_project_ids
         return False
     
     async def load(self) -> List['SyncNode']:

sync_state_machine/datasource/jsonl/path_utils.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+
+logger = logging.getLogger(__name__)
+
+
+def prepare_remote_jsonl_dir(remote_dir: Path, *, project_root: Path) -> Path:
+    project_root_path = project_root.resolve()
+    resolved_remote_dir = remote_dir.resolve(strict=False)
+
+    if resolved_remote_dir.exists():
+        if not resolved_remote_dir.is_dir():
+            raise FileNotFoundError(
+                f"remote_datasource.jsonl_dir exists but is not a directory: {resolved_remote_dir}"
+            )
+        return resolved_remote_dir
+
+    try:
+        resolved_remote_dir.relative_to(project_root_path)
+    except ValueError as exc:
+        raise FileNotFoundError(
+            "remote_datasource.jsonl_dir does not exist and auto-create is only allowed "
+            f"under project root: {resolved_remote_dir} (project_root={project_root_path})"
+        ) from exc
+
+    resolved_remote_dir.mkdir(parents=True, exist_ok=True)
+    logger.info("Created missing remote JSONL directory: %s", resolved_remote_dir)
+    return resolved_remote_dir

sync_state_machine/pipeline/__init__.py

@@ -7,7 +7,6 @@ from .full_sync_pipeline import FullSyncPipeline
 from ..config import (
     PipelineRunConfig,
     ApiDataSourceConfig,
-    DbDataSourceConfig,
     JsonlDataSourceConfig,
     DataSourceConfig,
     PersistConfig,
@@ -22,7 +21,9 @@ from .factory import (
     create_jsonl_to_api_pipeline,
     create_api_to_jsonl_pipeline,
     create_jsonl_to_jsonl_pipeline,
+    create_pipeline_from_file,
     create_pipeline_from_config,
+    run_pipeline_from_file,
     run_pipeline_from_config,
 )
 
@@ -32,11 +33,12 @@ __all__ = [
     "create_jsonl_to_api_pipeline",
     "create_jsonl_to_jsonl_pipeline",
     "create_api_to_jsonl_pipeline",
+    "create_pipeline_from_file",
     "create_pipeline_from_config",
+    "run_pipeline_from_file",
     "run_pipeline_from_config",
     "PipelineRunConfig",
     "ApiDataSourceConfig",
-    "DbDataSourceConfig",
     "JsonlDataSourceConfig",
     "DataSourceConfig",
     "PersistConfig",

sync_state_machine/pipeline/factory.py

@@ -23,19 +23,17 @@ from ..common.binding import BindingManager
 from ..common.collection import DataCollection
 from ..common.persistence import PersistenceBackend
 from ..common.registry import DomainRegistry
-from ..datasource.api.datasource import ApiDataSource
-from ..datasource.db.datasource import DbDataSource
-from ..datasource.jsonl.datasource import JsonlDataSource
-from ..datasource.type_registry import get_datasource_factory, register_datasource_factory
+from ..datasource import ensure_builtin_datasource_types_registered
+from ..datasource.type_registry import get_datasource_factory
 from ..config import (
     PipelineRunConfig,
     DataSourceConfig,
     ApiDataSourceConfig,
-    DbDataSourceConfig,
-    JsonlDataSourceConfig,
     PersistConfig,
+    JsonlDataSourceConfig,
     LoggingConfig,
     StrategyRuntimeConfig,
+    build_config_from_file,
     apply_config_overrides,
     OrphanAction,
     UpdateDirection,
@@ -55,27 +53,9 @@ def _project_root() -> Path:
 
 
 def _prepare_remote_jsonl_dir(remote_dir: Path, *, project_root: Optional[Path] = None) -> Path:
-    project_root_path = (project_root or _project_root()).resolve()
-    resolved_remote_dir = remote_dir.resolve(strict=False)
+    from ..datasource.jsonl.path_utils import prepare_remote_jsonl_dir
 
-    if resolved_remote_dir.exists():
-        if not resolved_remote_dir.is_dir():
-            raise FileNotFoundError(
-                f"remote_datasource.jsonl_dir exists but is not a directory: {resolved_remote_dir}"
-            )
-        return resolved_remote_dir
-
-    try:
-        resolved_remote_dir.relative_to(project_root_path)
-    except ValueError as exc:
-        raise FileNotFoundError(
-            "remote_datasource.jsonl_dir does not exist and auto-create is only allowed "
-            f"under project root: {resolved_remote_dir} (project_root={project_root_path})"
-        ) from exc
-
-    resolved_remote_dir.mkdir(parents=True, exist_ok=True)
-    logger.info("Created missing remote JSONL directory: %s", resolved_remote_dir)
-    return resolved_remote_dir
+    return prepare_remote_jsonl_dir(remote_dir, project_root=(project_root or _project_root()))
 
 
 def _ensure_pipeline_logger(
@@ -104,13 +84,8 @@ def _ensure_pipeline_logger(
 def _build_handler_config(
     node_type: str,
     ds_config: DataSourceConfig,
-    target_project_ids: List[str],
 ) -> Dict[str, Any]:
-    config = dict(ds_config.handler_configs.get(node_type, {}))
-
-    if node_type == "project" and target_project_ids:
-        config.setdefault("filter_project_id", target_project_ids)
-    return config
+    return dict(ds_config.handler_configs.get(node_type, {}))
 
 
 def _create_handler(
@@ -118,7 +93,6 @@ def _create_handler(
     node_type: str,
     ds_config: DataSourceConfig,
     datasource,
-    target_project_ids: List[str],
 ):
     datasource_type = ds_config.type
     handler_class = DomainRegistry.get_handler(node_type, datasource_type)
@@ -127,7 +101,7 @@ def _create_handler(
             f"Missing {datasource_type.upper()} handler for node_type: {node_type}"
         )
 
-    handler_config = _build_handler_config(node_type, ds_config, target_project_ids)
+    handler_config = _build_handler_config(node_type, ds_config)
     if datasource_type == "api":
         handler = handler_class(**handler_config)
     else:
@@ -138,22 +112,12 @@ def _create_handler(
     return handler
 
 
-def _resolve_datasource_target_project_ids(config: PipelineRunConfig) -> tuple[List[str], List[str]]:
-    fallback = [str(pid) for pid in config.target_project_ids if str(pid)]
-
-    local_cfg_ids = [str(pid) for pid in config.local_datasource.target_project_ids if str(pid)]
-    remote_cfg_ids = [str(pid) for pid in config.remote_datasource.target_project_ids if str(pid)]
-
-    local_target_ids = local_cfg_ids if local_cfg_ids else fallback
-    remote_target_ids = remote_cfg_ids if remote_cfg_ids else fallback
-    return local_target_ids, remote_target_ids
-
-
 async def _create_datasource(
     ds_config: DataSourceConfig,
     *,
     datasource_override=None,
     endpoint_name: str,
+    project_root: Path | None = None,
 ):
     if datasource_override is not None:
         return datasource_override
@@ -162,54 +126,24 @@ async def _create_datasource(
     if datasource_factory is None:
         raise RuntimeError(f"No datasource factory registered for type: {ds_config.type}")
 
-    datasource = datasource_factory(ds_config, endpoint_name)
+    factory_kwargs: dict[str, Any] = {}
+    signature = inspect.signature(datasource_factory)
+    if "project_root" in signature.parameters:
+        factory_kwargs["project_root"] = project_root or _project_root()
+
+    datasource = datasource_factory(ds_config, endpoint_name, **factory_kwargs)
     if inspect.isawaitable(datasource):
         datasource = await datasource
     return datasource
 
 
-def _build_jsonl_datasource(ds_config: JsonlDataSourceConfig, endpoint_name: str):
-    dir_path = Path(ds_config.jsonl_dir)
-    if endpoint_name == "remote":
-        dir_path = _prepare_remote_jsonl_dir(dir_path)
-    elif not dir_path.exists() or not dir_path.is_dir():
-        raise FileNotFoundError(
-            f"{endpoint_name}_datasource.jsonl_dir does not exist or is not a directory: {dir_path}"
-        )
-    return JsonlDataSource(dir_path, read_only=ds_config.read_only)
-
-
-def _build_db_datasource(ds_config: DbDataSourceConfig, endpoint_name: str):
-    del ds_config, endpoint_name
-    return DbDataSource()
-
-
-async def _build_api_datasource(ds_config: ApiDataSourceConfig, endpoint_name: str):
-    del endpoint_name
-    datasource = ApiDataSource(
-        base_url=ds_config.api_base_url,
-        uid=ds_config.api_uid,
-        secret=ds_config.api_secret,
-        debug=ds_config.api_debug,
-        poll_max_retries=ds_config.poll_max_retries,
-        poll_interval=ds_config.poll_interval,
-        rate_limit_max_requests=ds_config.api_rate_limit_max_requests,
-        rate_limit_window_seconds=ds_config.api_rate_limit_window_seconds,
-    )
+async def _initialize_datasource(datasource, *, endpoint_name: str) -> None:
+    try:
         await datasource.initialize()
-    return datasource
+    except Exception as exc:
+        raise RuntimeError(f"Failed to initialize {endpoint_name} datasource: {type(exc).__name__}: {exc}") from exc
 
-
-def _ensure_builtin_datasource_factories_registered() -> None:
-    if get_datasource_factory("jsonl") is None:
-        register_datasource_factory("jsonl", _build_jsonl_datasource)
-    if get_datasource_factory("db") is None:
-        register_datasource_factory("db", _build_db_datasource)
-    if get_datasource_factory("api") is None:
-        register_datasource_factory("api", _build_api_datasource)
-
-
-_ensure_builtin_datasource_factories_registered()
+ensure_builtin_datasource_types_registered()
 
 
 def _register_handlers_for_endpoint(
@@ -217,7 +151,6 @@ def _register_handlers_for_endpoint(
     datasource,
     ds_config: DataSourceConfig,
     node_types: List[str],
-    target_project_ids: List[str],
 ) -> None:
     for node_type in node_types:
         datasource.register_handler(
@@ -225,7 +158,6 @@ def _register_handlers_for_endpoint(
                 node_type=node_type,
                 ds_config=ds_config,
                 datasource=datasource,
-                target_project_ids=target_project_ids,
             )
         )
 
@@ -240,13 +172,11 @@ def _register_handlers(
         datasource=local_ds,
         ds_config=config.local_datasource,
         node_types=all_types,
-        target_project_ids=[str(pid) for pid in config.local_datasource.target_project_ids if str(pid)],
     )
     _register_handlers_for_endpoint(
         datasource=remote_ds,
         ds_config=config.remote_datasource,
         node_types=all_types,
-        target_project_ids=[str(pid) for pid in config.remote_datasource.target_project_ids if str(pid)],
     )
 
 
@@ -318,7 +248,6 @@ async def create_pipeline_from_config(
 
     all_types = list(config.node_types)
     strategy_map = _resolve_strategy_map(config, all_types)
-    local_target_project_ids, remote_target_project_ids = _resolve_datasource_target_project_ids(config)
 
     if config.persist.backend != "sqlite":
         raise NotImplementedError(f"Persistence backend not supported yet: {config.persist.backend}")
@@ -338,31 +267,20 @@ async def create_pipeline_from_config(
             config.local_datasource,
             datasource_override=local_datasource_override,
             endpoint_name="local",
+            project_root=_project_root(),
         )
+        await _initialize_datasource(local_ds, endpoint_name="local")
         remote_ds = await _create_datasource(
             config.remote_datasource,
             datasource_override=remote_datasource_override,
             endpoint_name="remote",
+            project_root=_project_root(),
         )
-        local_cfg_with_target = config.local_datasource.model_copy(
-            update={"target_project_ids": list(local_target_project_ids)},
-            deep=True,
-        )
-        remote_cfg_with_target = config.remote_datasource.model_copy(
-            update={"target_project_ids": list(remote_target_project_ids)},
-            deep=True,
-        )
-        handler_target_config = config.model_copy(
-            update={
-                "local_datasource": local_cfg_with_target,
-                "remote_datasource": remote_cfg_with_target,
-            },
-            deep=True,
-        )
+        await _initialize_datasource(remote_ds, endpoint_name="remote")
+        local_ds.set_target_project_ids([str(pid) for pid in config.local_datasource.target_project_ids if str(pid)])
+        remote_ds.set_target_project_ids([str(pid) for pid in config.remote_datasource.target_project_ids if str(pid)])
 
-        _register_handlers(handler_target_config, local_ds, remote_ds, all_types)
-        local_ds.set_target_project_ids(local_target_project_ids)
-        remote_ds.set_target_project_ids(remote_target_project_ids)
+        _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)
@@ -387,9 +305,6 @@ async def create_pipeline_from_config(
             remote_collection=remote_collection,
             binding_manager=binding_manager,
             enable_persistence=config.persist.enable,
-            target_project_ids=config.target_project_ids,
-            local_target_project_ids=local_target_project_ids,
-            remote_target_project_ids=remote_target_project_ids,
         )
 
         return pipeline
@@ -474,6 +389,40 @@ async def run_pipeline_from_config(
             await pipeline.close()
 
 
+async def create_pipeline_from_file(
+    *,
+    project_root: Path,
+    config_path: str | Path,
+    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,
+        local_datasource_override=local_datasource_override,
+        remote_datasource_override=remote_datasource_override,
+    )
+
+
+async def run_pipeline_from_file(
+    *,
+    project_root: Path,
+    config_path: str | Path,
+    title: str = "sync_state_machine pipeline",
+    print_summary: bool = True,
+    local_datasource_override=None,
+    remote_datasource_override=None,
+) -> Dict[str, Any]:
+    config = build_config_from_file(project_root=project_root, file_path=config_path)
+    return await run_pipeline_from_config(
+        config,
+        title=title,
+        print_summary=print_summary,
+        local_datasource_override=local_datasource_override,
+        remote_datasource_override=remote_datasource_override,
+    )
+
+
 async def create_jsonl_to_api_pipeline(
     *,
     # 路径配置
@@ -506,8 +455,12 @@ async def create_jsonl_to_api_pipeline(
 ) -> FullSyncPipeline:
     cfg = PipelineRunConfig(
         node_types=node_types,
+        local_datasource=JsonlDataSourceConfig(
+            type="jsonl",
+            jsonl_dir=local_jsonl_dir,
+            read_only=False,
             target_project_ids=target_project_ids or [],
-        local_datasource=JsonlDataSourceConfig(type="jsonl", jsonl_dir=local_jsonl_dir, read_only=False),
+        ),
         remote_datasource=ApiDataSourceConfig(
             type="api",
             api_base_url=api_base_url,
@@ -570,7 +523,6 @@ async def create_api_to_jsonl_pipeline(
 ) -> FullSyncPipeline:
     cfg = PipelineRunConfig(
         node_types=node_types,
-        target_project_ids=target_project_ids or [],
         local_datasource=ApiDataSourceConfig(
             type="api",
             api_base_url=api_base_url,
@@ -584,7 +536,12 @@ async def create_api_to_jsonl_pipeline(
             target_project_ids=target_project_ids or [],
             handler_configs=handler_configs or {},
         ),
-        remote_datasource=JsonlDataSourceConfig(type="jsonl", jsonl_dir=remote_jsonl_dir, read_only=False),
+        remote_datasource=JsonlDataSourceConfig(
+            type="jsonl",
+            jsonl_dir=remote_jsonl_dir,
+            read_only=False,
+            target_project_ids=target_project_ids or [],
+        ),
         strategies=[],
         persist=PersistConfig(
             db_path=persistence_db,
@@ -618,9 +575,18 @@ async def create_jsonl_to_jsonl_pipeline(
 ) -> FullSyncPipeline:
     cfg = PipelineRunConfig(
         node_types=node_types,
+        local_datasource=JsonlDataSourceConfig(
+            type="jsonl",
+            jsonl_dir=local_jsonl_dir,
+            read_only=False,
             target_project_ids=target_project_ids or [],
-        local_datasource=JsonlDataSourceConfig(type="jsonl", jsonl_dir=local_jsonl_dir, read_only=False),
-        remote_datasource=JsonlDataSourceConfig(type="jsonl", jsonl_dir=remote_jsonl_dir, read_only=False),
+        ),
+        remote_datasource=JsonlDataSourceConfig(
+            type="jsonl",
+            jsonl_dir=remote_jsonl_dir,
+            read_only=False,
+            target_project_ids=target_project_ids or [],
+        ),
         strategies=[],
         persist=PersistConfig(
             db_path=persistence_db,

sync_state_machine/pipeline/full_sync_pipeline.py

@@ -51,10 +51,6 @@ class FullSyncPipeline:
         load_order: Optional[List[str]] = None,
         sync_order: Optional[List[str]] = None,
         enable_persistence: bool = True,
-        # 项目过滤
-        target_project_ids: Optional[List[str]] = None,
-        local_target_project_ids: Optional[List[str]] = None,
-        remote_target_project_ids: Optional[List[str]] = None,
     ):
         # 验证必需参数（由工厂函数创建）
         if strategies is None:
@@ -70,17 +66,6 @@ class FullSyncPipeline:
         self.sync_order = sync_order or self.node_types
         self.persistence = persistence
         self.enable_persistence = enable_persistence
-        self.target_project_ids = target_project_ids or []
-        self.local_target_project_ids = (
-            list(local_target_project_ids)
-            if local_target_project_ids is not None
-            else list(self.target_project_ids)
-        )
-        self.remote_target_project_ids = (
-            list(remote_target_project_ids)
-            if remote_target_project_ids is not None
-            else list(self.target_project_ids)
-        )
         
         self.local_collection = local_collection
         self.remote_collection = remote_collection
@@ -187,12 +172,6 @@ class FullSyncPipeline:
         await self._load_from_datasource()
         self._logger.info("✅ Bootstrap: datasource fetched")
 
-        if self.local_target_project_ids or self.remote_target_project_ids:
-            local_deleted = await self.local_collection.filter_by_project_ids(self.local_target_project_ids)
-            remote_deleted = await self.remote_collection.filter_by_project_ids(self.remote_target_project_ids)
-            if local_deleted > 0 or remote_deleted > 0:
-                self._logger.info(f"🔍 Project filtering: deleted {local_deleted} local, {remote_deleted} remote nodes")
-
     async def phase_bind(self) -> None:
         strategy_map = {s.node_type: s for s in self.strategies}
         for node_type in self.sync_order:

tests/unit/test_datasource_target_project_ids_resolution.py

@@ -1,28 +1,23 @@
 from __future__ import annotations
 
-from pydantic import ValidationError
 import pytest
 
 from sync_state_machine.config import (
     PipelineRunConfig,
     JsonlDataSourceConfig,
     ApiDataSourceConfig,
-    DbDataSourceConfig,
     PersistConfig,
     LoggingConfig,
 )
-from sync_state_machine.pipeline.factory import _resolve_datasource_target_project_ids
 
 
 def _build_config(
     *,
-    top_level: list[str],
     local_ids: list[str] | None = None,
     remote_ids: list[str] | None = None,
 ) -> PipelineRunConfig:
     return PipelineRunConfig(
         node_types=["project"],
-        target_project_ids=top_level,
         local_datasource=JsonlDataSourceConfig(
             type="jsonl",
             jsonl_dir="./filtered_datasource/datasource/filtered",
@@ -39,49 +34,22 @@ def _build_config(
     )
 
 
-def test_datasource_target_ids_take_precedence_over_top_level() -> None:
-    cfg = _build_config(top_level=["p_top"], local_ids=["p_local"], remote_ids=["p_remote"])
+def test_pipeline_config_has_no_top_level_target_project_ids() -> None:
+    cfg = _build_config(local_ids=["p_local"], remote_ids=["p_remote"])
 
-    local_ids, remote_ids = _resolve_datasource_target_project_ids(cfg)
-
-    assert local_ids == ["p_local"]
-    assert remote_ids == ["p_remote"]
+    assert cfg.local_datasource.target_project_ids == ["p_local"]
+    assert cfg.remote_datasource.target_project_ids == ["p_remote"]
+    with pytest.raises(AttributeError):
+        _ = cfg.target_project_ids
 
 
-def test_datasource_target_ids_fallback_to_top_level_when_missing() -> None:
-    cfg = _build_config(top_level=["p_top"], local_ids=[], remote_ids=[])
-
-    local_ids, remote_ids = _resolve_datasource_target_project_ids(cfg)
-
-    assert local_ids == ["p_top"]
-    assert remote_ids == ["p_top"]
-
-
-def test_datasource_target_ids_support_mixed_override_and_fallback() -> None:
-    cfg = _build_config(top_level=["p_top"], local_ids=["p_local"], remote_ids=[])
-
-    local_ids, remote_ids = _resolve_datasource_target_project_ids(cfg)
-
-    assert local_ids == ["p_local"]
-    assert remote_ids == ["p_top"]
-
-
-def test_api_datasource_requires_target_project_ids() -> None:
-    with pytest.raises(ValidationError) as exc_info:
-        ApiDataSourceConfig(
+def test_api_datasource_allows_empty_target_project_ids() -> None:
+    cfg = ApiDataSourceConfig(
         type="api",
         api_base_url="https://example.com",
-            target_project_ids=[],  # Empty list should fail
     )
-    assert "List should have at least 1 item" in str(exc_info.value)
 
-    with pytest.raises(ValidationError) as exc_info:
-        ApiDataSourceConfig(
-            type="api",
-            api_base_url="https://example.com",
-            # Missing target_project_ids should fail
-        )
-    assert "Field required" in str(exc_info.value) or "missing" in str(exc_info.value).lower()
+    assert cfg.target_project_ids == []
 
 
 def test_api_datasource_rate_limit_config_fields() -> None:
@@ -102,11 +70,3 @@ def test_api_datasource_rate_limit_config_fields() -> None:
     )
     assert overridden.api_rate_limit_max_requests == 5
     assert overridden.api_rate_limit_window_seconds == 2.5
-
-
-def test_db_datasource_accepts_minimal_config() -> None:
-    cfg = DbDataSourceConfig(type="db")
-
-    assert cfg.type == "db"
-    assert cfg.target_project_ids == []
-    assert cfg.handler_configs == {}

tests/unit/test_datasource_type_registry.py

@@ -9,14 +9,15 @@ from sync_state_machine.common.registry import DomainRegistry
 from sync_state_machine.common.sync_node import SyncNode
 from sync_state_machine.config import (
     BaseDataSourceConfig,
-    DbDataSourceConfig,
     LoggingConfig,
     PersistConfig,
     PipelineRunConfig,
     build_config,
+    get_datasource_factory,
     register_datasource_type,
 )
 from sync_state_machine.datasource import BaseDataSource
+from sync_state_machine.datasource import ensure_builtin_datasource_types_registered
 from sync_state_machine.datasource.db import BaseDbHandler, DbDataSource
 from sync_state_machine.pipeline.factory import create_pipeline_from_config
 from sync_state_machine.sync_system.strategy import DefaultSyncStrategy
@@ -118,6 +119,13 @@ def _memory_factory(config: MemoryConfig, endpoint_name: str) -> MemoryDataSourc
     return MemoryDataSource(namespace=f"{endpoint_name}:{config.namespace}")
 
 
+def test_builtin_datasource_factories_are_registered() -> None:
+    ensure_builtin_datasource_types_registered(replace=True)
+
+    assert get_datasource_factory("jsonl") is not None
+    assert get_datasource_factory("api") is not None
+
+
 @pytest.mark.asyncio
 async def test_create_pipeline_from_config_supports_registered_custom_datasource_type(tmp_path: Path) -> None:
     register_datasource_type("memory", config_model=MemoryConfig, factory=_memory_factory, replace=True)
@@ -126,7 +134,7 @@ async def test_create_pipeline_from_config_supports_registered_custom_datasource
     config = PipelineRunConfig(
         node_types=["memory_demo"],
         local_datasource=MemoryConfig(type="memory", namespace="alpha"),
-        remote_datasource=DbDataSourceConfig(type="db"),
+        remote_datasource=MemoryConfig(type="memory", namespace="omega"),
         strategies=[],
         persist=PersistConfig(db_path=str(tmp_path / "memory-registry.db")),
         logging=LoggingConfig(initialize=False),
@@ -136,8 +144,10 @@ async def test_create_pipeline_from_config_supports_registered_custom_datasource
     try:
         assert isinstance(pipeline.local_datasource, MemoryDataSource)
         assert pipeline.local_datasource.namespace == "local:alpha"
+        assert isinstance(pipeline.remote_datasource, MemoryDataSource)
+        assert pipeline.remote_datasource.namespace == "remote:omega"
         assert isinstance(pipeline.local_datasource.get_handler("memory_demo"), MemoryHandler)
-        assert isinstance(pipeline.remote_datasource.get_handler("memory_demo"), MemoryRemoteDbHandler)
+        assert isinstance(pipeline.remote_datasource.get_handler("memory_demo"), MemoryHandler)
     finally:
         await pipeline.close()
 
@@ -176,14 +186,15 @@ def test_build_config_supports_same_file_datasource_bootstrap(tmp_path: Path, mo
     profile_path = tmp_path / "custom-memory.yaml"
     profile_path.write_text(
         "extensions:\n"
-        "  datasource_bootstraps:\n"
+        "  bootstraps:\n"
         "    - memory_bootstrap_demo:register\n"
         "node_types: []\n"
         "local_datasource:\n"
         "  type: memory_bootstrap\n"
         "  namespace: local-demo\n"
         "remote_datasource:\n"
-        "  type: db\n"
+        "  type: jsonl\n"
+        "  jsonl_dir: ${PROJECT_ROOT}/remote_datasource/datasource\n"
         "persist:\n"
         "  db_path: ${PROJECT_ROOT}/runtime/test-memory-bootstrap.db\n"
         "logging:\n"
@@ -195,4 +206,4 @@ def test_build_config_supports_same_file_datasource_bootstrap(tmp_path: Path, mo
 
     assert config.local_datasource.type == "memory_bootstrap"
     assert getattr(config.local_datasource, "namespace") == "local-demo"
-    assert config.remote_datasource.type == "db"
+    assert config.remote_datasource.type == "jsonl"

tests/unit/test_factory_db_handler_registration.py

@@ -6,13 +6,25 @@ from pydantic import BaseModel
 
 from sync_state_machine.common.registry import DomainRegistry
 from sync_state_machine.common.sync_node import SyncNode
-from sync_state_machine.config import DbDataSourceConfig, JsonlDataSourceConfig, LoggingConfig, PersistConfig, PipelineRunConfig
+from sync_state_machine.config import (
+    BaseDataSourceConfig,
+    JsonlDataSourceConfig,
+    LoggingConfig,
+    PersistConfig,
+    PipelineRunConfig,
+    register_datasource_type,
+)
 from sync_state_machine.datasource.db import BaseDbHandler, DbDataSource
 from sync_state_machine.datasource.jsonl import BaseJsonlHandler, JsonlDataSource
 from sync_state_machine.pipeline.factory import _register_handlers
 from sync_state_machine.sync_system.strategy import DefaultSyncStrategy
 
 
+class TestRegisteredDbConfig(BaseDataSourceConfig):
+    __test__ = False
+    type: str = "test_db"
+
+
 class TestDbSchema(BaseModel):
     __test__ = False
     id: str
@@ -56,7 +68,18 @@ class TestJsonlHandler(BaseJsonlHandler):
         super().__init__(datasource, "test_db_handler", TestDbNode, TestDbSchema)
 
 
+def _register_test_db_datasource_type() -> None:
+    register_datasource_type(
+        "test_db",
+        config_model=TestRegisteredDbConfig,
+        factory=lambda config, endpoint_name: DbDataSource(),
+        replace=True,
+    )
+
+
 def test_register_handlers_uses_db_handler_for_db_datasource(tmp_path: Path) -> None:
+    _register_test_db_datasource_type()
+
     if not DomainRegistry.is_registered("test_db_handler"):
         DomainRegistry.register(
             node_type="test_db_handler",
@@ -65,11 +88,11 @@ def test_register_handlers_uses_db_handler_for_db_datasource(tmp_path: Path) ->
             strategy_class=TestDbStrategy,
             jsonl_handler_class=TestJsonlHandler,
         )
-    DomainRegistry.register_db_handler("test_db_handler", TestDbHandler)
+    DomainRegistry.register_handler("test_db_handler", "test_db", TestDbHandler)
 
     config = PipelineRunConfig(
         node_types=["test_db_handler"],
-        local_datasource=DbDataSourceConfig(type="db"),
+        local_datasource=TestRegisteredDbConfig(type="test_db"),
         remote_datasource=JsonlDataSourceConfig(
             type="jsonl",
             jsonl_dir=str(tmp_path),

tests/unit/test_factory_endpoint_initialization.py

@@ -7,12 +7,23 @@ from pydantic import BaseModel
 
 from sync_state_machine.common.registry import DomainRegistry
 from sync_state_machine.common.sync_node import SyncNode
-from sync_state_machine.config import DbDataSourceConfig, LoggingConfig, PersistConfig, PipelineRunConfig
+from sync_state_machine.config import (
+    BaseDataSourceConfig,
+    LoggingConfig,
+    PersistConfig,
+    PipelineRunConfig,
+    register_datasource_type,
+)
 from sync_state_machine.datasource.db import BaseDbHandler, DbDataSource
 from sync_state_machine.pipeline.factory import create_pipeline_from_config
 from sync_state_machine.sync_system.strategy import DefaultSyncStrategy
 
 
+class EndpointInitDbConfig(BaseDataSourceConfig):
+    __test__ = False
+    type: str = "endpoint_init_db"
+
+
 class EndpointInitSchema(BaseModel):
     __test__ = False
     id: str
@@ -57,6 +68,7 @@ def _ensure_registered() -> None:
         reg.node_class = EndpointInitNode
         reg.strategy_class = EndpointInitStrategy
         reg.db_handler_class = EndpointInitDbHandler
+        reg.handler_classes["endpoint_init_db"] = EndpointInitDbHandler
         return
 
     DomainRegistry.register(
@@ -66,6 +78,16 @@ def _ensure_registered() -> None:
         strategy_class=EndpointInitStrategy,
         db_handler_class=EndpointInitDbHandler,
     )
+    DomainRegistry.register_handler(node_type, "endpoint_init_db", EndpointInitDbHandler)
+
+
+def _register_endpoint_init_datasource_type() -> None:
+    register_datasource_type(
+        "endpoint_init_db",
+        config_model=EndpointInitDbConfig,
+        factory=lambda config, endpoint_name: DbDataSource(),
+        replace=True,
+    )
 
 
 @pytest.mark.asyncio
@@ -74,16 +96,17 @@ async def test_factory_initializes_same_datasource_type_with_distinct_handler_co
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     _ensure_registered()
+    _register_endpoint_init_datasource_type()
 
     config = PipelineRunConfig(
         node_types=["factory_endpoint_demo"],
-        local_datasource=DbDataSourceConfig(
-            type="db",
+        local_datasource=EndpointInitDbConfig(
+            type="endpoint_init_db",
             handler_configs={"factory_endpoint_demo": {"side": "local", "batch_size": 10}},
             target_project_ids=["project_local"],
         ),
-        remote_datasource=DbDataSourceConfig(
-            type="db",
+        remote_datasource=EndpointInitDbConfig(
+            type="endpoint_init_db",
             handler_configs={"factory_endpoint_demo": {"side": "remote", "batch_size": 20}},
             target_project_ids=["project_remote"],
         ),

tests/unit/test_run_presets_yaml_loader.py

@@ -4,6 +4,7 @@ from pathlib import Path
 
 import pytest
 
+from sync_state_machine.config import build_config_from_file
 from sync_state_machine.config.run_presets import load_overrides_from_file
 
 
@@ -29,3 +30,97 @@ local_datasource:
 
     assert overrides["logging"]["suppress_node_logs"] is True
     assert str(tmp_path) in overrides["local_datasource"]["jsonl_dir"]
+
+
+def test_load_overrides_executes_generic_extensions_bootstraps(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    bootstrap_module = tmp_path / "profile_bootstrap_demo.py"
+    bootstrap_module.write_text(
+        "BOOTSTRAP_CALLED = False\n"
+        "\n"
+        "def register():\n"
+        "    global BOOTSTRAP_CALLED\n"
+        "    BOOTSTRAP_CALLED = True\n",
+        encoding="utf-8",
+    )
+    monkeypatch.syspath_prepend(str(tmp_path))
+
+    yaml_file = tmp_path / "profile_with_bootstrap.yaml"
+    yaml_file.write_text(
+        """
+extensions:
+  bootstraps:
+    - profile_bootstrap_demo:register
+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
+""".strip(),
+        encoding="utf-8",
+    )
+
+    load_overrides_from_file(yaml_file, project_root=tmp_path)
+
+    import profile_bootstrap_demo
+
+    assert profile_bootstrap_demo.BOOTSTRAP_CALLED is True
+
+
+def test_build_config_from_file_uses_registered_bootstraps(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    bootstrap_module = tmp_path / "config_from_file_bootstrap.py"
+    bootstrap_module.write_text(
+        "from pydantic import ConfigDict\n"
+        "from sync_state_machine.config import BaseDataSourceConfig, register_datasource_type\n"
+        "from sync_state_machine.datasource import BaseDataSource\n"
+        "\n"
+        "class FileBackedConfig(BaseDataSourceConfig):\n"
+        "    model_config = ConfigDict(extra=\"forbid\", validate_assignment=True)\n"
+        "    type: str = \"file_backed\"\n"
+        "    namespace: str = \"demo\"\n"
+        "\n"
+        "class FileBackedDatasource(BaseDataSource):\n"
+        "    pass\n"
+        "\n"
+        "def build_datasource(config, endpoint_name):\n"
+        "    return FileBackedDatasource()\n"
+        "\n"
+        "def register():\n"
+        "    register_datasource_type(\n"
+        "        \"file_backed\",\n"
+        "        config_model=FileBackedConfig,\n"
+        "        factory=build_datasource,\n"
+        "        replace=True,\n"
+        "    )\n",
+        encoding="utf-8",
+    )
+    monkeypatch.syspath_prepend(str(tmp_path))
+
+    profile = tmp_path / "profile.yaml"
+    profile.write_text(
+        """
+extensions:
+  bootstraps:
+    - config_from_file_bootstrap:register
+local_datasource:
+  type: file_backed
+  namespace: local-demo
+remote_datasource:
+  type: jsonl
+  jsonl_dir: ${PROJECT_ROOT}/remote_datasource/datasource
+persist:
+  db_path: ${PROJECT_ROOT}/runtime/test.db
+logging:
+  initialize: false
+""".strip(),
+        encoding="utf-8",
+    )
+
+    config = build_config_from_file(project_root=tmp_path, file_path=profile)
+
+    assert config.local_datasource.type == "file_backed"
+    assert getattr(config.local_datasource, "namespace") == "local-demo"