AiDOPWarehouse/doc/plan/sql/uat-data/run_uat_s8_exception_fact_seed.py

#!/usr/bin/env python3
"""Seed recent tenant/factory-scoped S8 exception lifecycle facts."""

from __future__ import annotations

import argparse
import json
from datetime import datetime, timedelta
from typing import Any

from apply_sql_file import connect


DEFAULT_TENANT = 838257186181189
DEFAULT_FACTORY = 1
PREFIX = "UAT-S8-A-"


def seed(tenant_id: int, factory_id: int, dry_run: bool) -> dict[str, Any]:
    result: dict[str, Any] = {
        "tenantId": tenant_id,
        "factoryId": factory_id,
        "dryRun": dry_run,
        "exceptions": 14,
        "closed": 10,
    }
    conn = connect()
    try:
        conn.begin()
        with conn.cursor() as cur:
            if not dry_run:
                cur.execute(
                    "DELETE FROM ado_s8_exception WHERE tenant_id=%s AND exception_code LIKE %s",
                    (tenant_id, f"{PREFIX}%"),
                )
                now = datetime.now().replace(microsecond=0)
                for index in range(14):
                    seq = index + 1
                    source_module = f"S{index % 7 + 1}"
                    created_at = now - timedelta(days=13 - index, hours=6)
                    is_closed = index < 10
                    closed_at = created_at + timedelta(hours=2 + index % 8) if is_closed else None
                    status = "CLOSED" if is_closed else "ASSIGNED"
                    severity = ("LOW", "MEDIUM", "HIGH", "CRITICAL")[index % 4]
                    code = f"{PREFIX}{seq:03d}"
                    cur.execute(
                        """
                        INSERT INTO ado_s8_exception
                          (tenant_id,factory_id,exception_code,title,description,scene_code,source_type,
                           status,severity,priority_score,priority_level,
                           occurrence_dept_id,responsible_dept_id,responsible_group_id,
                           assignee_id,reporter_id,sla_deadline,timeout_flag,
                           created_at,updated_at,assigned_at,closed_at,is_deleted,
                           created_by,updated_by,process_node_code,related_object_code,module_code,
                           exception_type_code,source_payload,dedup_key,last_detected_at,
                           consecutive_hit_count,consecutive_miss_count,stage_code,rule_mechanism)
                        VALUES
                          (%s,%s,%s,%s,%s,'UAT_OPERATIONS','UAT_GENERATOR',
                           %s,%s,%s,%s,
                           1,1,1,1,1,%s,0,
                           %s,%s,%s,%s,0,
                           'UAT_GENERATOR','UAT_GENERATOR','UAT_STAGE',%s,%s,
                           'UAT_EXCEPTION',%s,%s,%s,
                           1,0,%s,'RULE')
                        """,
                        (
                            tenant_id,
                            factory_id,
                            code,
                            f"UAT 异常事件 {seq:03d}",
                            "用于验证 S8 异常数量与处理时效正式汇总链",
                            status,
                            severity,
                            60 + index,
                            "P1" if severity in ("HIGH", "CRITICAL") else "P2",
                            created_at + timedelta(hours=12),
                            created_at,
                            closed_at or created_at + timedelta(hours=1),
                            created_at + timedelta(minutes=15),
                            closed_at,
                            f"UAT-OBJECT-{seq:03d}",
                            source_module,
                            json.dumps({"source": "uat", "seq": seq}, ensure_ascii=False),
                            code,
                            closed_at or created_at + timedelta(hours=1),
                            source_module,
                        ),
                    )
        if dry_run:
            conn.rollback()
        else:
            conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()
    return result


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--tenant", type=int, default=DEFAULT_TENANT)
    parser.add_argument("--factory", type=int, default=DEFAULT_FACTORY)
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()
    print(json.dumps(seed(args.tenant, args.factory, args.dry_run), ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()