diff --git a/scripts/tplus/profile_source.py b/scripts/tplus/profile_source.py index 1699e5e..e9aa707 100644 --- a/scripts/tplus/profile_source.py +++ b/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")) - cs = ("DRIVER={" + c["driver"] + "};SERVER=" + c["host"] + "," + str(c["port"]) - + ";DATABASE=" + c["database"] + ";UID=" + c["user"] - + ";PWD={" + c["password"].replace("}", "}}") + "};") + resolved = resolve_tplus_settings( + get_settings().model_copy( + update={"t_plus_enabled": True, "t_plus_connection_file": str(args.connection_file)} + ) + ) + engine = build_source_engine(resolved) result = {} - with pyodbc.connect(cs, timeout=10) as connection: - connection.timeout = 120 - cursor = connection.cursor() - for name, sql in QUERIES.items(): - cursor.execute(sql) - 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) + try: + connection = engine.raw_connection() + try: + connection.driver_connection.timeout = resolved.t_plus_query_timeout + cursor = connection.cursor() + for name, sql in QUERIES.items(): + cursor.execute(sql) + 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 diff --git a/scripts/tplus/verify_staging.py b/scripts/tplus/verify_staging.py index 16a2f62..bfbf47c 100644 --- a/scripts/tplus/verify_staging.py +++ b/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")) - settings = get_settings() - settings.t_plus_enabled = True - for key in ("host", "port", "database", "user", "password"): - setattr(settings, "t_plus_" + key, config[key]) - settings.t_plus_odbc_driver = config["driver"] - source_factory, target_factory = get_tplus_session_factory(), get_session_factory() + resolved = resolve_tplus_settings( + get_settings().model_copy( + update={"t_plus_enabled": True, "t_plus_connection_file": str(args.connection_file)} + ) + ) + engine = build_source_engine(resolved) + 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)