refactor(脚本): 复用应用连接校验与 URL 转义

verify_staging 原先 setattr 改写全局 Settings 单例，绕过
_ConnectionFile 严格校验与 sa 账号拒绝；profile_source 手工
拼接 ODBC 连接串且仅转义密码，database/user 含分号时可注入
连接属性。两者统一改为 model_copy 副本 + resolve_tplus_settings
完整校验 + SQLAlchemy URL 自建引擎，与 check_compatibility
同一路径，顺带消除 pyodbc connect 上下文的提交语义隐患。

真实源库连接的功能验证待本机 T+ 环境执行。

Co-Authored-By: Claude Code <noreply@anthropic.com>

scripts/tplus/profile_source.py

@@ -2,9 +2,15 @@
 
 import argparse
 import json
+import sys
 from pathlib import Path
 
-import pyodbc
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "backend"))
+
+from sqlalchemy import Engine, create_engine
+
+from app.core.config import Settings, get_settings
+from app.core.tplus_db import resolve_tplus_settings
 
 QUERIES = {
     "personnel_dates": "SELECT COUNT_BIG(*) AS total, SUM(CASE WHEN updated IS NULL THEN 1 ELSE 0 END) AS null_updated FROM AA_Person",
@@ -20,25 +26,43 @@ QUERIES = {
 }
 
 
+def build_source_engine(resolved: Settings) -> Engine:
+    """复用应用的 URL 转义和 TLS 配置自建引擎，不修改全局设置。"""
+    return create_engine(
+        resolved.t_plus_database_url,
+        connect_args={"timeout": resolved.t_plus_connect_timeout},
+        hide_parameters=True,
+    )
+
+
 def main() -> int:
     """执行固定白名单统计，并保存无凭据、无明细的 JSON。"""
     parser = argparse.ArgumentParser(description="T+ 只读数据质量统计")
     parser.add_argument("--connection-file", required=True, type=Path)
     parser.add_argument("--report", required=True, type=Path)
     args = parser.parse_args()
-    c = json.loads(args.connection_file.read_text(encoding="utf-8-sig"))
+    resolved = resolve_tplus_settings(
-    cs = ("DRIVER={" + c["driver"] + "};SERVER=" + c["host"] + "," + str(c["port"])
+        get_settings().model_copy(
-          + ";DATABASE=" + c["database"] + ";UID=" + c["user"]
+            update={"t_plus_enabled": True, "t_plus_connection_file": str(args.connection_file)}
-          + ";PWD={" + c["password"].replace("}", "}}") + "};")
+        )
+    )
+    engine = build_source_engine(resolved)
     result = {}
-    with pyodbc.connect(cs, timeout=10) as connection:
+    try:
-        connection.timeout = 120
+        connection = engine.raw_connection()
-        cursor = connection.cursor()
+        try:
-        for name, sql in QUERIES.items():
+            connection.driver_connection.timeout = resolved.t_plus_query_timeout
-            cursor.execute(sql)
+            cursor = connection.cursor()
-            names = [column[0] for column in cursor.description]
+            for name, sql in QUERIES.items():
-            result[name] = [dict(zip(names, row, strict=True)) for row in cursor.fetchall()]
+                cursor.execute(sql)
-            print(name, json.dumps(result[name]), flush=True)
+                names = [column[0] for column in cursor.description]
+                result[name] = [dict(zip(names, row, strict=True)) for row in cursor.fetchall()]
+                print(name, json.dumps(result[name]), flush=True)
+            cursor.close()
+        finally:
+            connection.close()
+    finally:
+        engine.dispose()
     args.report.parent.mkdir(parents=True, exist_ok=True)
     args.report.write_text(json.dumps(result, indent=2), encoding="utf-8")
     return 0

scripts/tplus/verify_staging.py

@@ -7,16 +7,38 @@ from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "backend"))
 
-from sqlalchemy import func, select
+from sqlalchemy import Engine, create_engine, event, func, select
+from sqlalchemy.orm import sessionmaker
 
-from app.core.config import get_settings
+from app.core.config import Settings, get_settings
 from app.core.db import dispose_engine, get_session_factory
-from app.core.tplus_db import dispose_tplus_engine, get_tplus_session_factory
+from app.core.tplus_db import resolve_tplus_settings
 from app.models import TPlusSyncRecord
 from app.services.tplus.mapping import DETAIL_MAPPINGS, TPLUS_MAPPINGS
 from app.services.tplus.sync_service import TPlusSyncService
 
 
+def build_source_engine(resolved: Settings) -> Engine:
+    """复用应用的 URL 转义和 TLS 配置自建引擎，不修改全局设置。
+
+    不能走 get_tplus_session_factory()：它读取全局单例配置，看不到本脚本
+    显式指定的连接文件；因此这里自建引擎，用完由调用方 dispose。
+    """
+    engine = create_engine(
+        resolved.t_plus_database_url,
+        pool_pre_ping=True,
+        pool_recycle=1800,
+        connect_args={"timeout": resolved.t_plus_connect_timeout},
+        hide_parameters=True,
+    )
+
+    def configure_timeout(connection, _record):
+        connection.timeout = resolved.t_plus_query_timeout
+
+    event.listen(engine, "connect", configure_timeout)
+    return engine
+
+
 def main() -> int:
     """执行一次或重复快照，并将计数结果保存到指定报告。"""
     parser = argparse.ArgumentParser(description="T+ 本地只读快照与幂等对账")
@@ -24,17 +46,20 @@ def main() -> int:
     parser.add_argument("--report", type=Path, required=True)
     parser.add_argument("--repeat", action="store_true")
     args = parser.parse_args()
-    config = json.loads(args.connection_file.read_text(encoding="utf-8-sig"))
+    resolved = resolve_tplus_settings(
-    settings = get_settings()
+        get_settings().model_copy(
-    settings.t_plus_enabled = True
+            update={"t_plus_enabled": True, "t_plus_connection_file": str(args.connection_file)}
-    for key in ("host", "port", "database", "user", "password"):
+        )
-        setattr(settings, "t_plus_" + key, config[key])
+    )
-    settings.t_plus_odbc_driver = config["driver"]
+    engine = build_source_engine(resolved)
-    source_factory, target_factory = get_tplus_session_factory(), get_session_factory()
+    source_factory, target_factory = (
+        sessionmaker(engine, expire_on_commit=False),
+        get_session_factory(),
+    )
     reports, failed = [], False
     try:
         with source_factory() as source, target_factory() as target:
-            for run in range(1, 3 if args.repeat else 2):
+            for run in range(2 if args.repeat else 1):
                 for mapping in (*TPLUS_MAPPINGS, *DETAIL_MAPPINGS):
                     batch = TPlusSyncService(source, target).run(entity_types=[mapping.entity_type])
                     count = target.scalar(
@@ -55,14 +80,14 @@ def main() -> int:
                     print(json.dumps(row), flush=True)
                     reports.append(row)
                     failed |= batch.status != "SUCCEEDED" or (
-                        run == 2 and batch.staged_record_count != 0
+                        run == 1 and batch.staged_record_count != 0
                     )
                     args.report.parent.mkdir(parents=True, exist_ok=True)
                     args.report.write_text(
                         json.dumps(reports, ensure_ascii=False, indent=2), encoding="utf-8"
                     )
     finally:
-        dispose_tplus_engine()
+        engine.dispose()
         dispose_engine()
     return int(failed)