alien_py_cloud/scripts/reconcile_contract_status.py

"""
合同签署状态对账/回刷脚本（一次性运维工具）

背景：历史回调逻辑把"平台方自动盖章(SIGN_MISSON_COMPLETE / signResult=2)"误判为整份合同已签署，
导致部分 contract_document 被错误标记为 status=1，并写入了早 8 小时(UTC)的 signing_time。
本脚本以 e签宝 流程真实状态 signFlowStatus 为准，回刷 contract_document 与 contract_bundle 状态。

判定规则（signFlowStatus: 0草稿 1签署中 2完成 3撤销 4终止 5过期 6删除 7拒签）：
  - == 2 完成：保持/置为已签署(status=1)，并用 signFlowFinishTime 重算东八区 signing_time / effective_time / expiry_time，补全 download_url
  - != 2     ：置为未签署(status=0)，清空 signing_time / effective_time / expiry_time

安全说明：
  - 默认 dry-run，只打印将要做的变更，不写库；加 --apply 才真正提交。
  - 通过 APP_ENV 选择环境（与服务一致）：生产为 APP_ENV=produ。

用法示例（PowerShell）：
  $env:APP_ENV="produ"; python scripts/reconcile_contract_status.py                 # 预览（dry-run）
  $env:APP_ENV="produ"; python scripts/reconcile_contract_status.py --apply         # 实际回刷
  $env:APP_ENV="produ"; python scripts/reconcile_contract_status.py --status 1      # 只检查当前已签署的
  $env:APP_ENV="produ"; python scripts/reconcile_contract_status.py --sign-flow-id xxxx  # 只处理某一条
"""
import argparse
import datetime
import json
import os
import sys
import time

# 确保可以从项目根目录导入业务模块
ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if ROOT_DIR not in sys.path:
    sys.path.insert(0, ROOT_DIR)

from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session

from alien_gateway.config import settings
from alien_contract.db.models.bundle import ContractBundle
from alien_contract.db.models.document import ContractDocument
from alien_contract.infrastructure.esign.main import query_sign_flow_detail, file_download_url
from alien_contract.services.contract_server import _ms_to_naive_cn, _extract_download_url

ESIGN_FLOW_STATUS_COMPLETED = 2
FLOW_STATUS_DESC = {
    0: "草稿",
    1: "签署中",
    2: "完成",
    3: "撤销",
    4: "终止",
    5: "过期",
    6: "删除",
    7: "拒签",
}


def _query_flow(sign_flow_id: str):
    """返回 (signFlowStatus, signFlowFinishTime毫秒)；失败返回 (None, None)"""
    try:
        resp = query_sign_flow_detail(sign_flow_id)
        data = json.loads(resp)
    except Exception as exc:  # noqa: BLE001
        print(f"    [WARN] 查询流程失败 sign_flow_id={sign_flow_id} err={exc}")
        return None, None
    body = data.get("data") if isinstance(data, dict) else None
    if not isinstance(body, dict):
        print(f"    [WARN] 流程查询无 data sign_flow_id={sign_flow_id} resp={data}")
        return None, None
    return body.get("signFlowStatus"), body.get("signFlowFinishTime")


def _fetch_download_url(sign_flow_id: str):
    try:
        resp = file_download_url(sign_flow_id)
        url, _ = _extract_download_url(resp)
        return url
    except Exception as exc:  # noqa: BLE001
        print(f"    [WARN] 获取下载链接失败 sign_flow_id={sign_flow_id} err={exc}")
        return None


def _derive_times(finish_ms):
    signing_dt = _ms_to_naive_cn(finish_ms)
    effective_dt = expiry_dt = None
    if signing_dt:
        effective_dt = (signing_dt + datetime.timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
        expiry_dt = effective_dt + datetime.timedelta(days=365)
    return signing_dt, effective_dt, expiry_dt


def _recalc_bundle_status(session: Session, bundle_id: int) -> str:
    rows = session.execute(
        select(ContractDocument.status).where(
            ContractDocument.bundle_id == bundle_id,
            ContractDocument.delete_flag == 0,
        )
    ).all()
    statuses = [r[0] for r in rows]
    if not statuses:
        return "未签署"
    if all(s == 1 for s in statuses):
        return "已签署"
    if any(s == 1 for s in statuses):
        return "审核中"
    return "未签署"


def main():
    parser = argparse.ArgumentParser(description="合同签署状态对账/回刷")
    parser.add_argument("--apply", action="store_true", help="实际写库（默认 dry-run 仅预览）")
    parser.add_argument("--status", choices=["0", "1", "all"], default="all", help="只检查指定 status 的文档，默认 all")
    parser.add_argument("--sign-flow-id", default=None, help="只处理指定 sign_flow_id")
    parser.add_argument("--sleep", type=float, default=0.1, help="每条查询之间的间隔秒数，默认0.1")
    args = parser.parse_args()

    dry_run = not args.apply

    print("=" * 80)
    print(f"环境 APP_ENV = {settings.APP_ENV}")
    print(f"目标数据库   = {settings.DB_HOST}:{settings.DB_PORT}/{settings.DB_NAME}")
    print(f"模式         = {'DRY-RUN（仅预览，不写库）' if dry_run else '!!! APPLY（实际写库）!!!'}")
    print("=" * 80)

    engine = create_engine(settings.SQLALCHEMY_DATABASE_URI, pool_pre_ping=True)

    conditions = [ContractDocument.delete_flag == 0]
    if args.sign_flow_id:
        conditions.append(ContractDocument.sign_flow_id == args.sign_flow_id)
    elif args.status != "all":
        conditions.append(ContractDocument.status == int(args.status))

    counters = {"total": 0, "to_signed": 0, "to_unsigned": 0, "time_fixed": 0, "unchanged": 0, "skipped": 0}
    affected_bundle_ids: set[int] = set()

    with Session(engine) as session:
        docs = session.execute(select(ContractDocument).where(*conditions).order_by(ContractDocument.id)).scalars().all()
        print(f"待检查文档数：{len(docs)}\n")

        for doc in docs:
            counters["total"] += 1
            flow_status, finish_ms = _query_flow(doc.sign_flow_id)
            time.sleep(args.sleep)
            status_desc = FLOW_STATUS_DESC.get(flow_status, "未知")
            head = (
                f"[doc#{doc.id} bundle#{doc.bundle_id}] {doc.contract_name} "
                f"sign_flow_id={doc.sign_flow_id} 当前status={doc.status} -> e签宝={flow_status}({status_desc})"
            )

            if flow_status is None:
                counters["skipped"] += 1
                print(f"{head}  => 跳过（查询失败）")
                continue

            if flow_status == ESIGN_FLOW_STATUS_COMPLETED:
                signing_dt, effective_dt, expiry_dt = _derive_times(finish_ms)
                new_download = doc.download_url or ""
                if not new_download:
                    fetched = _fetch_download_url(doc.sign_flow_id)
                    if fetched:
                        new_download = fetched

                needs_status = doc.status != 1
                needs_time = (
                    doc.signing_time != signing_dt
                    or doc.effective_time != effective_dt
                    or doc.expiry_time != expiry_dt
                )
                needs_download = bool(new_download) and (doc.download_url or "") != new_download

                if not (needs_status or needs_time or needs_download):
                    counters["unchanged"] += 1
                    print(f"{head}  => 无需变更（已正确为已签署）")
                    continue

                if needs_status:
                    counters["to_signed"] += 1
                if needs_time and not needs_status:
                    counters["time_fixed"] += 1

                print(
                    f"{head}  => 置为已签署: status {doc.status}->1, "
                    f"signing_time {doc.signing_time!r}->{signing_dt!r}, "
                    f"effective {doc.effective_time!r}->{effective_dt!r}, "
                    f"expiry {doc.expiry_time!r}->{expiry_dt!r}"
                    + (", 补全download_url" if needs_download else "")
                )
                # 改动只作用于当前事务（dry-run 结束会 rollback），便于后续 bundle 状态准确重算
                doc.status = 1
                doc.signing_time = signing_dt
                doc.effective_time = effective_dt
                doc.expiry_time = expiry_dt
                if needs_download:
                    doc.download_url = new_download
                affected_bundle_ids.add(doc.bundle_id)
            else:
                if doc.status == 0 and doc.signing_time is None and doc.effective_time is None and doc.expiry_time is None:
                    counters["unchanged"] += 1
                    print(f"{head}  => 无需变更（已正确为未签署）")
                    continue
                counters["to_unsigned"] += 1
                print(
                    f"{head}  => 回退为未签署: status {doc.status}->0, 清空 signing/effective/expiry "
                    f"(原 signing_time={doc.signing_time!r})"
                )
                doc.status = 0
                doc.signing_time = None
                doc.effective_time = None
                doc.expiry_time = None
                affected_bundle_ids.add(doc.bundle_id)

        # 重算受影响合同包整体状态
        if affected_bundle_ids:
            print("\n--- 合同包状态重算 ---")
        for bundle_id in sorted(affected_bundle_ids):
            bundle = session.get(ContractBundle, bundle_id)
            if not bundle:
                continue
            new_status = _recalc_bundle_status(session, bundle_id)
            if bundle.status != new_status:
                print(f"[bundle#{bundle_id}] {bundle.subject_name} status {bundle.status} -> {new_status}")
                bundle.status = new_status
            else:
                print(f"[bundle#{bundle_id}] {bundle.subject_name} status 不变({bundle.status})")

        if dry_run:
            session.rollback()
        else:
            session.commit()

    print("\n" + "=" * 80)
    print("对账汇总：")
    print(f"  检查文档总数        : {counters['total']}")
    print(f"  置为已签署(含纠正)  : {counters['to_signed']}")
    print(f"  仅修正签署时间      : {counters['time_fixed']}")
    print(f"  回退为未签署        : {counters['to_unsigned']}")
    print(f"  无需变更            : {counters['unchanged']}")
    print(f"  跳过(查询失败)      : {counters['skipped']}")
    print(f"  受影响合同包        : {len(affected_bundle_ids)}")
    print(f"  模式                : {'DRY-RUN（未写库）' if dry_run else 'APPLY（已提交）'}")
    print("=" * 80)
    if dry_run:
        print("提示：以上为预览。确认无误后加 --apply 实际执行。")


if __name__ == "__main__":
    main()