PayrollManagementSystem/scripts/tplus/check_compatibility.py

"""只读核验备份源同版本 SQL Server；2022 PoC 不能获得兼容通过结果。"""

import argparse
import json
import sys
from datetime import datetime
from decimal import Decimal
from pathlib import Path

import pyodbc
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "backend"))

from app.core.config import Settings, get_settings
from app.core.tplus_db import resolve_tplus_settings
from app.services.tplus.mapping import DETAIL_MAPPINGS, TPLUS_MAPPINGS
from app.services.tplus.reader import TPlusReader


def compatibility_settings(path: Path) -> Settings:
    """复用应用的连接文件校验、URL 转义和 TLS 配置，不修改全局设置。"""
    settings = get_settings().model_copy(
        update={"t_plus_enabled": True, "t_plus_connection_file": str(path)}
    )
    return resolve_tplus_settings(settings)


def _version_matches(actual: str, expected: str) -> bool:
    def parts(value):
        values = value.split(".")
        return tuple(values + ["0"] * (4 - len(values)))

    return parts(actual) == parts(expected)


def _inspect_connection(connection, expected_version: str) -> dict:
    cursor = connection.cursor()
    try:
        version = str(
            cursor.execute(
                "SELECT CAST(SERVERPROPERTY('ProductVersion') AS varchar(80))"
            ).fetchone()[0]
        )
        values = cursor.execute(
            "SELECT N'中文核验', CAST('12345.6789' AS decimal(19,4)), "
            "CAST('2008-01-02T03:04:05' AS datetime)"
        ).fetchone()
        permissions = cursor.execute(
            "SELECT HAS_PERMS_BY_NAME(DB_NAME(),'DATABASE','INSERT'), "
            "HAS_PERMS_BY_NAME(DB_NAME(),'DATABASE','UPDATE'), "
            "HAS_PERMS_BY_NAME(DB_NAME(),'DATABASE','DELETE')"
        ).fetchone()
    finally:
        cursor.close()
    checks = {
        "sqlserver_2008r2": version.startswith("10.50."),
        "backup_source_version": _version_matches(version, expected_version),
        "unicode_roundtrip": values[0] == "中文核验",
        "decimal_exact": values[1] == Decimal("12345.6789"),
        "datetime_exact": values[2] == datetime(2008, 1, 2, 3, 4, 5),
        "no_dml_permissions": all(x == 0 for x in permissions),
    }
    return {
        "server_version": version,
        "expected_version": expected_version,
        "driver_name": connection.getinfo(pyodbc.SQL_DRIVER_NAME),
        "driver_version": connection.getinfo(pyodbc.SQL_DRIVER_VER),
        "checks": checks,
        "compatible": all(checks.values()),
    }


def _inspect_adapter(engine, report: dict) -> None:
    checks = {}
    with Session(engine) as session:
        reader = TPlusReader(session)
        for mapping in (*TPLUS_MAPPINGS, *DETAIL_MAPPINGS):
            try:
                reader.fetch_entity(mapping.entity_type, max_rows=1)
                checks[mapping.entity_type] = True
            except Exception as exc:
                session.rollback()
                checks[mapping.entity_type] = False
                report.setdefault("errors", {})[mapping.entity_type] = type(exc).__name__
    report["adapter_queries"] = checks
    report["compatible"] = report["compatible"] and all(checks.values())


def main() -> int:
    """检查版本、中文、金额、日期和写权限，报告不包含凭据或客户记录。"""
    parser = argparse.ArgumentParser(description="SQL Server 2008 R2 兼容性验收")
    parser.add_argument("--connection-file", required=True, type=Path)
    parser.add_argument("--report", required=True, type=Path)
    parser.add_argument("--expected-version", default="10.50.1600", help="备份源版本基准")
    args = parser.parse_args()
    engine = None
    try:
        settings = compatibility_settings(args.connection_file)
        engine = create_engine(
            settings.t_plus_database_url,
            hide_parameters=True,
            connect_args={"timeout": settings.t_plus_connect_timeout},
        )
        connection = engine.raw_connection()
        try:
            connection.driver_connection.timeout = settings.t_plus_query_timeout
            report = _inspect_connection(connection, args.expected_version)
        finally:
            connection.close()
        _inspect_adapter(engine, report)
    except Exception as exc:
        # 驱动异常可能包含完整连接串，报告仅保留类型。
        report = {"compatible": False, "error_type": type(exc).__name__}
    finally:
        if engine is not None:
            engine.dispose()
    args.report.parent.mkdir(parents=True, exist_ok=True)
    args.report.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(report, ensure_ascii=False))
    return 0 if report["compatible"] else 2


if __name__ == "__main__":
    raise SystemExit(main())