feat(tplus): 新增 T+ 只读同步服务与同步管理

backend/app/services/tplus/README.md

@@ -0,0 +1,21 @@
+# T+ 只读同步
+
+> AI 摘要：读取外部 SQL Server 的原始快照，并管理持久化任务。关键文件为 task_service、worker、sync_service、relationship_audit 和 query_service；管理员 API 为 `/api/v1/system/tplus`。
+
+## 职责与边界
+
+- `task_service` 创建/重试任务，控制请求幂等和事务，保留原批次与异常；HTTP 不执行源库长查询。
+- `worker.run_once` 取得 MySQL 执行锁后恢复中断任务、领取一个任务；读取和关联核验共用锁。
+- `sync_service` 主键分页读取白名单实体，每页保存原始快照，相同源 ID 和行哈希去重。
+- `relationship_audit` 执行 `db/queries/tplus/audit/queries.sql`，只返回系统字典和数量；按收款方向与单据类型核对关联。
+- `query_service` 分页元数据，按需加载白名单报告；不返回原始 payload、驱动异常或源配置。
+
+## 主要入口
+
+管理员 API 提供概况、批次/异常分页、详情、创建与失败重试。CLI 为 `python -m app.cli tplus-worker --once/--watch`；原 `tplus-stage --entity personnel` 可继续直接读取。
+
+连接配置、启动命令和异常处理参见项目 `docs/T+本地恢复与同步开发手册.md`。后端与 worker 必须指向同一 MySQL 业务库和 T+ 账套，查询文件随部署交付。
+
+## 验收限制
+
+当前只写 MySQL 暂存与管理表，不写工资业务表、不推进时间水位。关联、成本、日期和负向事件的业务规则待客户确认；SQL Server 2022 的实测不能替代客户 2008 R2（10.50.x）的兼容验收。

backend/app/services/tplus/__init__.py

@@ -0,0 +1,6 @@
+"""T+ 只读读取、字段映射和同步 staging。"""
+
+from app.services.tplus.reader import TPlusReader
+from app.services.tplus.sync_service import TPlusSyncService
+
+__all__ = ["TPlusReader", "TPlusSyncService"]

backend/app/services/tplus/mapping.py

@@ -0,0 +1,136 @@
+"""T+ 字段映射元数据。
+
+这里记录候选字段和确认状态，不把字典中的候选直接当成业务契约。
+高风险口径在客户确认前保持 ``PENDING``，同步器只允许写入原始 staging。
+"""
+
+from dataclasses import dataclass
+from enum import StrEnum
+
+
+class MappingStatus(StrEnum):
+    """字段映射的证据状态。"""
+
+    CONFIRMED = "CONFIRMED"
+    CANDIDATE = "CANDIDATE"
+    MISSING = "MISSING"
+    PENDING = "PENDING"
+
+
+@dataclass(frozen=True, slots=True)
+class TPlusMapping:
+    """一个实体的源表、候选字段和目标表说明。"""
+
+    entity_type: str
+    source_table: str
+    source_fields: tuple[str, ...]
+    target_table: str
+    status: MappingStatus
+    unresolved: tuple[str, ...] = ()
+    version: str = "0.2"
+    schema_evidence: str = "ceshi20261026.bak，2026-10-27 恢复副本结构核验"
+
+
+TPLUS_MAPPINGS: tuple[TPlusMapping, ...] = (
+    TPlusMapping(
+        "personnel",
+        "AA_Person",
+        ("id", "code", "name", "disabled", "mobilePhoneNo", "officePhoneNo", "iddepartment"),
+        "personnel",
+        MappingStatus.CANDIDATE,
+        ("工号是否等于 code", "经理隶属关系", "入职日期", "试用期"),
+    ),
+    TPlusMapping(
+        "products",
+        "AA_Inventory",
+        ("id", "code", "name", "specification", "idunit", "idinventoryclass"),
+        "products",
+        MappingStatus.CANDIDATE,
+        ("库存数量语义", "参考价来源", "系统产品分类"),
+    ),
+    TPlusMapping(
+        "customers",
+        "AA_Partner",
+        ("id", "code", "name", "Contact", "representative", "disabled", "updated"),
+        "customers",
+        MappingStatus.CANDIDATE,
+        ("客户唯一键", "改名/合并规则", "系统客户分类"),
+    ),
+    TPlusMapping(
+        "businesses",
+        "SA_SaleOrder",
+        ("id", "code", "voucherdate", "idcustomer", "idclerk", "idbusinesstype", "updated"),
+        "businesses",
+        MappingStatus.PENDING,
+        ("成本口径", "业务负责人覆盖规则", "含税口径"),
+    ),
+    TPlusMapping(
+        "shipments",
+        "SA_SaleDelivery",
+        (
+            "id",
+            "code",
+            "voucherdate",
+            "SendDate",
+            "deliveryDate",
+            "auditeddate",
+            "SourceVoucherCode",
+            "updated",
+        ),
+        "shipments",
+        MappingStatus.PENDING,
+        ("发货日期", "订单关联路径", "退货/红冲规则", "运费来源"),
+    ),
+    TPlusMapping(
+        "receipts",
+        "ARAP_ReceivePayment",
+        (
+            "id",
+            "code",
+            "voucherDate",
+            "auditedDate",
+            "createdTime",
+            "SaleOrderCode",
+            "SaleOrderID",
+            "ts",
+        ),
+        "receipts",
+        MappingStatus.PENDING,
+        ("到账日期", "收款方式", "核销状态", "回款与发货关联"),
+    ),
+)
+
+DETAIL_MAPPINGS: tuple[TPlusMapping, ...] = (
+    TPlusMapping(
+        "business_items",
+        "SA_SaleOrder_b",
+        ("id", "idSaleOrderDTO", "LatestCost"),
+        "business_items",
+        MappingStatus.PENDING,
+        ("成本口径", "含税口径"),
+    ),
+    TPlusMapping(
+        "shipment_items",
+        "SA_SaleDelivery_b",
+        ("ID", "idSaleDeliveryDTO", "saleOrderId"),
+        "shipment_items",
+        MappingStatus.PENDING,
+        ("发货归属", "红冲和退货"),
+    ),
+    TPlusMapping(
+        "receipt_items",
+        "ARAP_ReceivePayment_b",
+        ("id", "idArapReceivePaymentDTO", "saleorderid"),
+        "receipts",
+        MappingStatus.PENDING,
+        ("回款拆分", "核销语义", "增量水位"),
+    ),
+)
+
+
+def get_mapping(entity_type: str) -> TPlusMapping:
+    """按实体类型取得映射；未知实体直接拒绝。"""
+    for mapping in (*TPLUS_MAPPINGS, *DETAIL_MAPPINGS):
+        if mapping.entity_type == entity_type:
+            return mapping
+    raise KeyError(f"未知 T+ 实体类型：{entity_type}")

backend/app/services/tplus/query_service.py

@@ -0,0 +1,122 @@
+"""同步管理只读查询；只公开受控元数据，不公开源数据和原异常。"""
+
+from app.core.exceptions import NotFoundError
+from app.core.response import PageResult
+from app.core.tplus_db import resolve_tplus_settings
+from app.repositories import tplus_repository as repository
+from app.schemas.tplus import (
+    TPlusBatchDetailOut,
+    TPlusBatchOut,
+    TPlusErrorOut,
+    TPlusOverviewOut,
+)
+from app.services.tplus import task_service
+from app.services.tplus.mapping import DETAIL_MAPPINGS, TPLUS_MAPPINGS
+from app.services.tplus.relationship_audit import public_report
+
+ERROR_MESSAGES = {
+    "READ_ERROR": "源数据读取失败，请检查只读连接和字段映射",
+    "STAGE_ERROR": "原始数据保存失败，请核对字段类型后重试",
+    "INTERRUPTED": "执行进程中断，可重新读取，已保存数据会自动去重",
+    "EXECUTION_ERROR": "任务执行失败，请检查只读连接和字段映射",
+    "TYPE_UNRESOLVED": "关联核验所需的单据类型缺失或不唯一",
+}
+_ENTITY_TYPES = {item.entity_type for item in (*TPLUS_MAPPINGS, *DETAIL_MAPPINGS)}
+
+
+def batch_out(batch, active_parents: set[str]) -> TPlusBatchOut:
+    """只使用元数据字段组装输出，历史批次范围保持未知。"""
+    entities = batch.entity_types_json
+    scope_valid = (
+        isinstance(entities, list)
+        and bool(entities)
+        and all(isinstance(item, str) and item in _ENTITY_TYPES for item in entities)
+    )
+    return TPlusBatchOut(
+        batch_no=batch.batch_no,
+        task_type="AUDIT" if batch.sync_mode == "AUDIT" else "READ",
+        status=batch.status,
+        entity_types=entities if scope_valid else None,
+        source_record_count=batch.source_record_count,
+        staged_record_count=batch.staged_record_count,
+        skipped_record_count=batch.skipped_record_count,
+        failed_record_count=batch.failed_record_count,
+        parent_batch_no=batch.parent_batch_no,
+        create_time=batch.create_time,
+        started_at=batch.started_at,
+        finished_at=batch.finished_at,
+        can_retry=batch.status in ("FAILED", "PARTIAL")
+        and (batch.sync_mode == "AUDIT" or (batch.sync_mode == "FULL" and scope_valid))
+        and batch.batch_no not in active_parents,
+        error_summary="任务有异常，请查看异常明细"
+        if batch.status in ("FAILED", "PARTIAL")
+        else None,
+    )
+
+
+def get_batch(session, number: str) -> TPlusBatchDetailOut:
+    """按需加载报告，未找到时统一返回 404。"""
+    batch = repository.get_batch(session, number, with_result=True)
+    if batch is None:
+        raise NotFoundError("同步批次不存在")
+    out = batch_out(batch, repository.active_retry_parents(session, [number]))
+    return TPlusBatchDetailOut(
+        **out.model_dump(), audit_result=public_report(batch.audit_result_json)
+    )
+
+
+def list_batches(session, query):
+    """分页批次并批量查询重试状态。"""
+    rows, total = repository.list_batches(session, query)
+    active = repository.active_retry_parents(session, [row.batch_no for row in rows])
+    return PageResult[TPlusBatchOut](
+        records=[batch_out(row, active) for row in rows],
+        total=total,
+        page_num=query.page_num,
+        page_size=query.page_size,
+    )
+
+
+def _safe_id(value) -> str | None:
+    if isinstance(value, str) and value.isascii() and value.isdigit() and len(value) <= 10:
+        return value if int(value) <= 2147483647 else None
+    return None
+
+
+def list_errors(session, number, query):
+    """错误按编码生成固定说明，避免驱动异常泄漏连接参数。"""
+    if repository.get_batch(session, number) is None:
+        raise NotFoundError("同步批次不存在")
+    rows, total = repository.list_errors(session, number, query)
+    records = [
+        TPlusErrorOut(
+            id=row.id,
+            entity_type=row.entity_type if row.entity_type in _ENTITY_TYPES else "system",
+            source_id=_safe_id(row.source_id),
+            error_code=row.error_code if row.error_code in ERROR_MESSAGES else "UNKNOWN",
+            error_message=ERROR_MESSAGES.get(row.error_code, "任务未完成，请核对配置后重试"),
+            retry_count=row.retry_count,
+            create_time=row.create_time,
+        )
+        for row in rows
+    ]
+    return PageResult[TPlusErrorOut](
+        records=records, total=total, page_num=query.page_num, page_size=query.page_size
+    )
+
+
+def overview(session) -> TPlusOverviewOut:
+    """只校验配置，不在页面查询期间建立 T+ 连接。"""
+    try:
+        resolve_tplus_settings(task_service.get_settings())
+        configured, message = True, "只读配置已就绪，实际连接结果以任务执行为准"
+    except RuntimeError:
+        configured, message = False, "T+ 只读配置未就绪，请联系管理员配置"
+    latest = repository.latest_audit(session)
+    return TPlusOverviewOut(
+        is_configured=configured,
+        configuration_message=message,
+        status_counts=repository.status_counts(session),
+        total_staged_records=repository.total_staged_records(session),
+        latest_audit=public_report(latest.audit_result_json) if latest else None,
+    )

backend/app/services/tplus/reader.py

@@ -0,0 +1,95 @@
+"""T+ 六类源表的只读读取器。"""
+
+from collections.abc import Iterator
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import BigInteger, Integer, bindparam, text
+from sqlalchemy.orm import Session
+
+from app.services.tplus.mapping import TPLUS_MAPPINGS, get_mapping
+
+_ENTITY_QUERY_FILES = {
+    "personnel": "personnel.sql",
+    "products": "products.sql",
+    "customers": "customers.sql",
+    "businesses": "businesses.sql",
+    "shipments": "shipments.sql",
+    "receipts": "receipts.sql",
+    "business_items": "business_items.sql",
+    "shipment_items": "shipment_items.sql",
+    "receipt_items": "receipt_items.sql",
+}
+
+
+class TPlusReader:
+    """执行白名单 SQL 文件，禁止调用方传入任意表名或 SQL。"""
+
+    def __init__(self, session: Session, query_dir: Path | None = None) -> None:
+        self.session = session
+        self.query_dir = query_dir or self._default_query_dir()
+
+    @staticmethod
+    def _default_query_dir() -> Path:
+        return Path(__file__).resolve().parents[4] / "db" / "queries" / "tplus"
+
+    def iter_entity(
+        self,
+        entity_type: str,
+        *,
+        watermark: datetime | None = None,
+        batch_size: int = 500,
+    ) -> Iterator[dict[str, Any]]:
+        """按实体流式读取，水位只通过绑定参数传入。"""
+        if entity_type not in _ENTITY_QUERY_FILES:
+            raise KeyError(f"不支持的 T+ 实体类型：{entity_type}")
+        if batch_size < 1 or batch_size > 5000:
+            raise ValueError("batch_size 必须在 1 到 5000 之间")
+        if watermark is not None:
+            raise ValueError("增量水位语义尚未确认，当前只支持全量读取和哈希去重")
+        get_mapping(entity_type)
+        sql_path = self.query_dir / _ENTITY_QUERY_FILES[entity_type]
+        if not sql_path.is_file():
+            raise FileNotFoundError(f"缺少 T+ 查询文件：{sql_path}")
+
+        statement = text(sql_path.read_text(encoding="utf-8")).bindparams(
+            bindparam("batch_size", type_=Integer()), bindparam("after_id", type_=BigInteger())
+        )
+        after_id = -2147483649  # 所有源表主键均已核验为 SQL Server int，包含其最小值。
+        while True:
+            result = self.session.execute(
+                statement, {"batch_size": batch_size, "after_id": after_id}
+            )
+            try:
+                rows = [dict(row) for row in result.mappings().all()]
+            finally:
+                result.close()
+            if not rows:
+                return
+            next_id = int(rows[-1]["id"])
+            if next_id <= after_id:
+                raise ValueError("T+ 源主键未递增，停止读取以避免重复循环")
+            yield from rows
+            after_id = next_id
+
+    def fetch_entity(
+        self,
+        entity_type: str,
+        *,
+        watermark: datetime | None = None,
+        max_rows: int = 5000,
+    ) -> list[dict[str, Any]]:
+        """读取有限数量的实体行，供核验命令和单元测试使用。"""
+        if max_rows < 1:
+            raise ValueError("max_rows 必须大于 0")
+        rows: list[dict[str, Any]] = []
+        for row in self.iter_entity(entity_type, watermark=watermark):
+            rows.append(row)
+            if len(rows) >= max_rows:
+                break
+        return rows
+
+    def inventory(self) -> tuple[str, ...]:
+        """返回当前适配器支持的实体类型。"""
+        return tuple(mapping.entity_type for mapping in TPLUS_MAPPINGS)

backend/app/services/tplus/relationship_audit.py

@@ -0,0 +1,135 @@
+"""读取实际系统字典并统计带单据类型的收款关联，不推断计提口径。"""
+
+import re
+from datetime import datetime
+from pathlib import Path
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.core.logging import scrub
+from app.services.tplus.reader import TPlusReader
+
+_COLUMNS = {
+    "source_version": ("source_version",),
+    "voucher_types": ("id", "code", "name"),
+    "state_enums": ("id", "code", "name", "enum_name"),
+    "receipt_kinds": ("voucher_type_code", "voucher_type_name", "receive_flag", "row_count"),
+    "receipt_line_kinds": ("voucher_type_id", "voucher_type_name", "receive_flag", "row_count"),
+    "sales_links": (
+        "row_count",
+        "delivery_matches",
+        "detail_matches",
+        "direct_order_matches",
+        "detail_order_matches",
+        "unambiguous_order_matches",
+        "invalid_order_ids",
+        "conflicting_order_ids",
+    ),
+    "order_coverage": ("row_count", "without_order", "single_order", "multiple_orders"),
+}
+PENDING_CONFIRMATIONS = (
+    ("cost_basis", "绩效成本取值", "订单最新成本或出库成本的类型、时点和含税口径需业务确认"),
+    ("receipt_date", "到账日", "单据日、审核日和明细 arrivalDate 的业务含义需确认"),
+    ("ship_date", "发货日", "回款天数应使用哪个发货日期需确认"),
+    ("allocation", "多单分摊", "一笔收款涉及多张销货单或订单时，不能自动把整笔金额记给一单"),
+    ("negative_events", "退货与红冲", "负数、冲销及已结算业务的调整方式需确认"),
+)
+
+
+def public_report(raw: dict | None) -> dict | None:
+    """历史报告也按当前字段白名单输出，禁止透传未知嵌套数据。"""
+    if not isinstance(raw, dict):
+        return None
+    result = {
+        key: TPlusRelationshipAudit._safe_value(raw.get(key))
+        for key in (
+            "schema_version",
+            "captured_at",
+            "source_version",
+            "is_target_engine",
+            "relationship_status",
+        )
+    }
+    result["compatibility_status"] = "PENDING"
+    for section, columns in _COLUMNS.items():
+        if section in ("source_version", "voucher_types"):
+            continue
+        value = raw.get(section)
+        if isinstance(value, list):
+            result[section] = [
+                {key: TPlusRelationshipAudit._safe_value(row.get(key)) for key in columns}
+                for row in value[:200]
+                if isinstance(row, dict)
+            ]
+        elif isinstance(value, dict):
+            result[section] = {
+                key: TPlusRelationshipAudit._safe_value(value.get(key)) for key in columns
+            }
+        else:
+            result[section] = None
+    result["pending_confirmations"] = [
+        {"key": k, "title": t, "reason": r} for k, t, r in PENDING_CONFIRMATIONS
+    ]
+    return result
+
+
+class TPlusRelationshipAudit:
+    """执行固定核验查询；结果只包含系统字典和数量，不包含客户记录。"""
+
+    def __init__(self, session: Session, query_dir: Path | None = None) -> None:
+        self.session = session
+        root = query_dir or TPlusReader._default_query_dir()
+        source = (root / "audit" / "queries.sql").read_text(encoding="utf-8")
+        sections = re.split(r"(?m)^-- @query (\w+)\s*$", source)
+        self.queries = dict(zip(sections[1::2], sections[2::2], strict=True))
+        if set(self.queries) != set(_COLUMNS):
+            raise ValueError("关联核验查询清单不完整")
+
+    def fetch(self, name: str, **params) -> list[dict]:
+        """只接受登记的查询名与绑定参数，过滤非合同字段。"""
+        if name not in self.queries:
+            raise ValueError("不支持的核验查询")
+        result = self.session.execute(text(self.queries[name]), params)
+        try:
+            return [
+                {key: self._safe_value(row.get(key)) for key in _COLUMNS[name]}
+                for row in result.mappings().all()
+            ]
+        finally:
+            result.close()
+
+    @staticmethod
+    def _safe_value(value):
+        if value is None or isinstance(value, (int, bool)):
+            return value
+        return scrub(re.sub(r"[\x00-\x1f]", " ", str(value)))[:200]
+
+    def run(self) -> dict:
+        """生成事实报告；类型缺失或有歧义时不执行该关联链路。"""
+        version = self.fetch("source_version")[0]["source_version"]
+        types = self.fetch("voucher_types")
+        report = {
+            "schema_version": 1,
+            "captured_at": datetime.now().isoformat(),
+            "source_version": version,
+            "is_target_engine": str(version).startswith("10.50."),
+            "compatibility_status": "PENDING",
+            "state_enums": self.fetch("state_enums"),
+            "receipt_kinds": self.fetch("receipt_kinds"),
+            "receipt_line_kinds": self.fetch("receipt_line_kinds"),
+            "sales_links": None,
+            "order_coverage": None,
+            "relationship_status": "UNRESOLVED",
+            "relationship_basis": "主表按 isReceiveFlag=1（是否收款）筛选；明细按 SM_VoucherType 的 SA04 类型关联。主表 idvouchertype 是来源类型，不能据此筛本单类型。",
+            "pending_confirmations": [
+                {"key": k, "title": t, "reason": r} for k, t, r in PENDING_CONFIRMATIONS
+            ],
+        }
+        sale = [r["id"] for r in types if r["code"] == "SA04"]
+        if len(sale) == 1:
+            params = {"sale_delivery_type_id": sale[0], "receive_flag": 1}
+            report["sales_links"] = self.fetch("sales_links", **params)[0]
+            report["order_coverage"] = self.fetch("order_coverage", **params)[0]
+            report["relationship_status"] = "OBSERVED"
+        return report

backend/app/services/tplus/staging.py

@@ -0,0 +1,123 @@
+"""T+ 原始行 staging 服务。"""
+
+import hashlib
+import json
+from datetime import date, datetime
+from decimal import Decimal
+from typing import Any
+
+from sqlalchemy import insert, select, tuple_
+from sqlalchemy.orm import Session
+
+from app.models import TPlusSyncRecord
+
+
+def _json_default(value: Any) -> str:
+    """将日期、Decimal 等源字段转换为稳定 JSON 文本。"""
+    if isinstance(value, (date, datetime)):
+        return value.isoformat()
+    if isinstance(value, Decimal):
+        return str(value)
+    if isinstance(value, (bytes, bytearray, memoryview)):
+        return "0x" + bytes(value).hex()
+    raise TypeError("T+ 源行包含无法序列化的数据类型")
+
+
+def stage_row(
+    session: Session,
+    *,
+    batch_no: str,
+    entity_type: str,
+    source_table: str,
+    row: dict[str, Any],
+) -> tuple[TPlusSyncRecord, bool]:
+    """幂等写入单条原始行；返回记录和是否新插入。"""
+    values = prepare_row(batch_no, entity_type, source_table, row)
+    existing = session.execute(
+        select(TPlusSyncRecord).where(
+            TPlusSyncRecord.entity_type == entity_type,
+            TPlusSyncRecord.source_id == values["source_id"],
+            TPlusSyncRecord.row_hash == values["row_hash"],
+        )
+    ).scalar_one_or_none()
+    if existing is not None:
+        return existing, False
+
+    record = TPlusSyncRecord(**values)
+    session.add(record)
+    session.flush()
+    return record, True
+
+
+def prepare_row(batch_no: str, entity_type: str, source_table: str, row: dict) -> dict:
+    """规范化原始值，生成与 JSON 列序列化一致的哈希。"""
+    source_id = str(row["id"]) if row.get("id") is not None else ""
+    if not source_id:
+        raise ValueError("T+ 源行缺少 id，无法建立幂等键")
+    source_no = row.get("code")
+    payload = json.dumps(
+        row, ensure_ascii=False, sort_keys=True, default=_json_default, allow_nan=False
+    )
+    return {
+        "batch_no": batch_no,
+        "entity_type": entity_type,
+        "source_table": source_table,
+        "source_id": source_id,
+        "source_no": str(source_no).strip() if source_no is not None else None,
+        "source_line_no": str(row.get("line_no")) if row.get("line_no") is not None else None,
+        "source_updated_at": _source_datetime(row),
+        "row_hash": hashlib.sha256(payload.encode("utf-8")).hexdigest(),
+        "payload_json": json.loads(payload),
+    }
+
+
+def stage_rows(session: Session, batch_no: str, mapping, rows: list[dict]) -> tuple[int, list]:
+    """批量去重和插入；数据库拒绝整批时逐条隔离，保留有效行。"""
+    records, errors = {}, []
+    for row in rows:
+        try:
+            values = prepare_row(batch_no, mapping.entity_type, mapping.source_table, row)
+            records[(values["source_id"], values["row_hash"])] = values
+        except (ValueError, TypeError) as exc:
+            errors.append((row, type(exc).__name__))
+    if not records:
+        return 0, errors
+    existing = set(
+        session.execute(
+            select(TPlusSyncRecord.source_id, TPlusSyncRecord.row_hash).where(
+                TPlusSyncRecord.entity_type == mapping.entity_type,
+                tuple_(TPlusSyncRecord.source_id, TPlusSyncRecord.row_hash).in_(list(records)),
+            )
+        ).all()
+    )
+    pending = [v for k, v in records.items() if k not in existing]
+    if not pending:
+        return 0, errors
+    try:
+        with session.begin_nested():
+            session.execute(insert(TPlusSyncRecord), pending)
+        return len(pending), errors
+    except Exception:
+        # 唯一约束竞争或坏行触发回滚后，再逐条处理，不能丢掉同批有效行。
+        inserted = 0
+        for row in rows:
+            try:
+                with session.begin_nested():
+                    _, is_new = stage_row(
+                        session,
+                        batch_no=batch_no,
+                        entity_type=mapping.entity_type,
+                        source_table=mapping.source_table,
+                        row=row,
+                    )
+                inserted += int(is_new)
+            except Exception as exc:
+                if not any(old is row for old, _ in errors):
+                    errors.append((row, type(exc).__name__))
+        return inserted, errors
+
+
+def _source_datetime(row: dict[str, Any]) -> datetime | None:
+    """只接收真实时间值；SQL Server rowversion 不能当作日期。"""
+    value = row.get("source_updated_at") or row.get("updated")
+    return value if isinstance(value, datetime) else None

backend/app/services/tplus/sync_service.py

@@ -0,0 +1,131 @@
+"""T+ 只读快照同步：批量 staging、幂等去重和失败隔离。"""
+
+from collections.abc import Iterable
+from datetime import datetime
+from itertools import batched
+from pathlib import Path
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.core.config import get_settings
+from app.core.tplus_lock import sync_lock
+from app.models import TPlusSyncBatch, TPlusSyncError
+from app.services.tplus.mapping import TPlusMapping, get_mapping
+from app.services.tplus.reader import TPlusReader
+from app.services.tplus.staging import stage_rows
+from app.services.tplus.task_service import fail_batch, new_batch_no, recover_running
+
+
+class TPlusSyncService:
+    """按分页提交原始快照，业务表始终不参与本服务。"""
+
+    def __init__(self, source_session: Session, target_session: Session) -> None:
+        self.source_session = source_session
+        self.target_session = target_session
+        configured = get_settings().t_plus_query_dir
+        self.reader = TPlusReader(source_session, Path(configured) if configured else None)
+
+    def run(
+        self,
+        *,
+        entity_types: Iterable[str],
+        watermark: datetime | None = None,
+        batch_no: str | None = None,
+        batch_size: int = 500,
+    ) -> TPlusSyncBatch:
+        """执行可重复的全量读取，提交每页进度；字段语义未定时拒绝时间增量。"""
+        entities = tuple(dict.fromkeys(entity_types))
+        if not entities or not 1 <= batch_size <= 5000:
+            raise ValueError("至少指定一个实体，batch_size 必须在 1 到 5000 之间")
+        if watermark is not None:
+            raise ValueError("增量水位语义尚未确认，当前只支持全量读取和哈希去重")
+        for entity in entities:
+            get_mapping(entity)
+        with sync_lock(self.target_session):
+            recover_running(self.target_session)
+            batch = self._start_direct(entities, batch_no)
+            try:
+                return self.execute_claimed(batch, batch_size)
+            except BaseException:
+                self.target_session.rollback()
+                fail_batch(self.target_session, batch, "INTERRUPTED")
+                raise
+
+    def _start_direct(self, entities, batch_no):
+        batch = TPlusSyncBatch(
+            batch_no=batch_no or new_batch_no(),
+            sync_mode="FULL",
+            status="RUNNING",
+            source_database=get_settings().t_plus_database or None,
+            started_at=datetime.now(),
+            source_record_count=0,
+            staged_record_count=0,
+            failed_record_count=0,
+            skipped_record_count=0,
+            entity_types_json=list(entities),
+        )
+        self.target_session.add(batch)
+        self.target_session.commit()
+        return batch
+
+    def execute_claimed(self, batch: TPlusSyncBatch, batch_size: int = 500):
+        """处理已持锁且标记 RUNNING 的任务；仅由直接入口或 worker 调用。"""
+        if batch.status != "RUNNING" or not batch.entity_types_json:
+            raise ValueError("任务尚未领取或读取范围不完整")
+        mappings = [get_mapping(entity) for entity in batch.entity_types_json]
+        for mapping in mappings:
+            try:
+                self._read_entity(batch, mapping, batch_size)
+            except Exception as exc:
+                self.target_session.rollback()
+                self._record_error(batch, mapping, {}, "READ_ERROR", type(exc).__name__)
+                self.target_session.commit()
+        batch.status = "SUCCEEDED"
+        if batch.failed_record_count:
+            failed_rows = (
+                self.target_session.scalar(
+                    select(func.count())
+                    .select_from(TPlusSyncError)
+                    .where(
+                        TPlusSyncError.batch_no == batch.batch_no,
+                        TPlusSyncError.error_code == "STAGE_ERROR",
+                    )
+                )
+                or 0
+            )
+            batch.status = "PARTIAL" if batch.source_record_count > failed_rows else "FAILED"
+            batch.error_message = "部分数据未完成，请查询同步错误记录并重跑全量快照"
+        batch.finished_at = datetime.now()
+        batch.watermark_to = None
+        self.target_session.commit()
+        return batch
+
+    def _read_entity(self, batch: TPlusSyncBatch, mapping: TPlusMapping, size: int) -> None:
+        """批量写入一类数据；每页提交后即使进程中断也可幂等重跑。"""
+        source = self.reader.iter_entity(mapping.entity_type, batch_size=size)
+        for chunk in batched(source, size):
+            inserted, errors = stage_rows(self.target_session, batch.batch_no, mapping, list(chunk))
+            batch.source_record_count += len(chunk)
+            batch.staged_record_count += inserted
+            batch.skipped_record_count += len(chunk) - inserted - len(errors)
+            for row, reason in errors:
+                self._record_error(batch, mapping, row, "STAGE_ERROR", reason)
+            self.target_session.commit()
+
+    def _record_error(
+        self, batch: TPlusSyncBatch, mapping: TPlusMapping, row: dict, code: str, reason: str
+    ) -> None:
+        """只记录错误类别和源 ID，避免数据库异常携带个人信息或凭据。"""
+        self.target_session.add(
+            TPlusSyncError(
+                batch_no=batch.batch_no,
+                entity_type=mapping.entity_type,
+                source_table=mapping.source_table,
+                source_id=str(row["id"]) if row.get("id") is not None else None,
+                source_line_no=str(row["line_no"]) if row.get("line_no") is not None else None,
+                error_code=code,
+                error_message=f"源数据处理失败（{reason}），请核对连接和字段",
+            )
+        )
+        batch.failed_record_count += 1

backend/app/services/tplus/task_service.py

@@ -0,0 +1,184 @@
+"""持久化任务创建和重试，不在 HTTP 事务中读取源库。"""
+
+import secrets
+from contextlib import nullcontext
+from dataclasses import dataclass
+from datetime import datetime
+
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.core.config import get_settings
+from app.core.exceptions import BadRequestError, NotFoundError
+from app.core.tplus_db import resolve_tplus_settings
+from app.models import TPlusSyncBatch, TPlusSyncError, User
+from app.repositories import tplus_repository as repository
+from app.schemas.tplus import TPlusCreateIn, TPlusRetryIn
+from app.services import audit_service
+from app.services.audit_service import AuditRecord, RequestMeta
+
+
+@dataclass(frozen=True)
+class TaskActor:
+    """HTTP 操作者与请求元信息，CLI 调用不传。"""
+
+    user: User
+    meta: RequestMeta | None = None
+
+
+def new_batch_no() -> str:
+    """生成最长 24 字符的可追溯批次号。"""
+    return f"TP{datetime.now():%Y%m%d%H%M%S}{secrets.token_hex(4)}"
+
+
+def _audit(session, actor, action, data):
+    if actor is None:
+        return nullcontext()
+    return audit_service.audit(
+        session,
+        operator=actor.user,
+        meta=actor.meta,
+        record=AuditRecord("T+数据同步", action, "T+ 数据任务" + action, data),
+    )
+
+
+def _source_database() -> str:
+    try:
+        return resolve_tplus_settings(get_settings()).t_plus_database
+    except RuntimeError as exc:
+        raise BadRequestError(str(exc)) from None
+
+
+def _same_intent(batch, mode, entities, parent):
+    if (
+        batch.sync_mode != mode
+        or sorted(batch.entity_types_json or []) != sorted(entities)
+        or batch.parent_batch_no != parent
+    ):
+        raise BadRequestError("幂等键已用于不同的任务请求")
+    return batch
+
+
+def _insert(session, values):
+    try:
+        with session.begin_nested():
+            batch = TPlusSyncBatch(**values)
+            session.add(batch)
+            session.flush()
+        return batch, True
+    except IntegrityError:
+        batch = repository.get_request(session, values["request_key"], for_update=True)
+        if batch is None:
+            raise
+        return _same_intent(
+            batch, values["sync_mode"], values["entity_types_json"], values["parent_batch_no"]
+        ), False
+
+
+def _values(payload, database, actor, parent=None):
+    return {
+        "batch_no": new_batch_no(),
+        "request_key": str(payload.request_key),
+        "status": "QUEUED",
+        "sync_mode": "AUDIT" if payload.task_type == "AUDIT" else "FULL",
+        "entity_types_json": sorted(payload.entity_types),
+        "source_database": database,
+        "requested_by": actor.user.id if actor else None,
+        "parent_batch_no": parent,
+        "started_at": None,
+        "source_record_count": 0,
+        "staged_record_count": 0,
+        "failed_record_count": 0,
+        "skipped_record_count": 0,
+    }
+
+
+def create_task(session: Session, payload: TPlusCreateIn, actor: TaskActor | None = None):
+    """原子创建任务；重放已存在的意图不依赖此刻的源连接状态。"""
+    with _audit(session, actor, "创建读取任务", payload.model_dump(mode="json")):
+        existing = repository.get_request(session, str(payload.request_key))
+        mode = "AUDIT" if payload.task_type == "AUDIT" else "FULL"
+        if existing:
+            batch = _same_intent(existing, mode, payload.entity_types, None)
+        else:
+            batch, _ = _insert(session, _values(payload, _source_database(), actor))
+        session.commit()
+    return batch
+
+
+def retry_task(
+    session: Session, parent_no: str, payload: TPlusRetryIn, actor: TaskActor | None = None
+):
+    """锁父批次后创建一个新任务；不覆盖原批次及异常历史。"""
+    with _audit(
+        session,
+        actor,
+        "重试读取任务",
+        {"parent_batch_no": parent_no, "request_key": str(payload.request_key)},
+    ):
+        existing = repository.get_request(session, str(payload.request_key))
+        if existing:
+            if existing.parent_batch_no != parent_no:
+                raise BadRequestError("幂等键已用于不同的重试请求")
+            session.commit()
+            batch = existing
+        else:
+            parent = repository.get_batch(session, parent_no, for_update=True)
+            if parent is None:
+                raise NotFoundError("同步批次不存在")
+            request = _retry_request(parent, payload)
+            active = repository.active_retry(session, parent_no)
+            if active:
+                if active.request_key != str(payload.request_key):
+                    raise BadRequestError("该批次已有未完成的重试任务，请刷新列表查看")
+                batch = active
+            else:
+                batch, inserted = _insert(
+                    session, _values(request, _source_database(), actor, parent_no)
+                )
+                if inserted:
+                    repository.increment_retry_errors(session, parent_no)
+            session.commit()
+    return batch
+
+
+def _retry_request(parent, payload):
+    if parent.status not in ("FAILED", "PARTIAL"):
+        raise BadRequestError("只能重试失败或部分失败的批次")
+    if parent.sync_mode not in ("FULL", "AUDIT"):
+        raise BadRequestError("该同步方式尚未启用，请重新选择读取任务")
+    if parent.sync_mode != "AUDIT" and parent.entity_types_json is None:
+        raise BadRequestError("历史批次未记录范围，请重新选择读取类型")
+    return TPlusCreateIn(
+        task_type="AUDIT" if parent.sync_mode == "AUDIT" else "READ",
+        entity_types=parent.entity_types_json or [],
+        request_key=payload.request_key,
+    )
+
+
+def fail_batch(session: Session, batch: TPlusSyncBatch, code: str) -> None:
+    """只保存固定错误摘要；即使原异常含连接密码也不直接落库。"""
+    messages = {
+        "INTERRUPTED": "执行进程中断，请重新读取；已暂存数据会自动去重",
+        "EXECUTION_ERROR": "任务执行失败，请检查只读连接和字段映射后重试",
+        "TYPE_UNRESOLVED": "核验所需单据类型缺失或不唯一，请核对源字典",
+    }
+    message = messages.get(code, "任务未完成，请核对运行配置")
+    batch.status, batch.finished_at, batch.error_message = "FAILED", datetime.now(), message
+    batch.failed_record_count = (batch.failed_record_count or 0) + 1
+    session.add(
+        TPlusSyncError(
+            batch_no=batch.batch_no,
+            entity_type="system",
+            source_table="system",
+            error_code=code,
+            error_message=message,
+        )
+    )
+    session.commit()
+
+
+def recover_running(session: Session) -> None:
+    """仅在持有执行锁时调用；锁证明之前的执行者已经退出。"""
+    for batch in repository.running_batches(session):
+        fail_batch(session, batch, "INTERRUPTED")

backend/app/services/tplus/worker.py

@@ -0,0 +1,61 @@
+"""持久化任务执行者；获取连接级锁后才恢复或领取任务。"""
+
+from datetime import datetime
+from pathlib import Path
+
+from app.core.tplus_db import get_tplus_session_factory, resolve_tplus_settings
+from app.core.tplus_lock import TPlusBusyError, sync_lock
+from app.repositories import tplus_repository as repository
+from app.services.tplus import task_service
+from app.services.tplus.relationship_audit import TPlusRelationshipAudit
+from app.services.tplus.sync_service import TPlusSyncService
+
+
+def run_once(session, source_factory=None):
+    """返回本轮处理的批次号；忙锁或队列为空时不假称已执行。"""
+    try:
+        with sync_lock(session):
+            task_service.recover_running(session)
+            batch = repository.next_queued(session)
+            if batch is None:
+                session.rollback()
+                return None
+            batch.status, batch.started_at = "RUNNING", datetime.now()
+            session.commit()
+            _execute(session, batch, source_factory)
+            return batch.batch_no
+    except TPlusBusyError:
+        session.rollback()
+        return None
+
+
+def _execute(session, batch, source_factory):
+    try:
+        settings = resolve_tplus_settings(task_service.get_settings())
+        if batch.source_database and batch.source_database != settings.t_plus_database:
+            raise RuntimeError("任务排队后的源账套配置已改变")
+        factory = source_factory or get_tplus_session_factory()
+        with factory() as source:
+            if batch.sync_mode == "AUDIT":
+                root = Path(settings.t_plus_query_dir) if settings.t_plus_query_dir else None
+                report = TPlusRelationshipAudit(source, root).run()
+                batch.audit_result_json = report
+                batch.source_record_count = sum(row["row_count"] for row in report["receipt_kinds"])
+                batch.finished_at, batch.status = datetime.now(), "SUCCEEDED"
+                if report["relationship_status"] != "OBSERVED":
+                    task_service.fail_batch(session, batch, "TYPE_UNRESOLVED")
+                else:
+                    session.commit()
+            else:
+                TPlusSyncService(source, session).execute_claimed(batch)
+    except BaseException as exc:
+        session.rollback()
+        task_service.fail_batch(
+            session,
+            batch,
+            "INTERRUPTED"
+            if isinstance(exc, (KeyboardInterrupt, SystemExit))
+            else "EXECUTION_ERROR",
+        )
+        if isinstance(exc, (KeyboardInterrupt, SystemExit)):
+            raise