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

#!/usr/bin/env python3
"""Create effective UAT target configurations for every enabled KPI."""

from __future__ import annotations

import argparse
import json
from datetime import datetime
from decimal import Decimal
from typing import Any

from apply_sql_file import connect


DEFAULT_TENANT = 838257186181189
DEFAULT_FACTORY = 1
BATCH_ID = "UAT-KPI-TARGET-20260821"


def semantic_target(metric: dict[str, Any]) -> Decimal:
    name = str(metric.get("MetricName") or "")
    unit = str(metric.get("Unit") or "").strip().upper()
    module = str(metric.get("ModuleCode") or "")
    direction = str(metric.get("Direction") or "")

    if unit == "PPM":
        return Decimal("5000")
    if unit == "%" or "率" in name or "OEE" in name.upper() or "效率" in name:
        return Decimal("95")
    if unit in {"H", "小时"}:
        return Decimal("24")
    if unit in {"天", "DAY", "DAYS"}:
        if "周转" in name:
            return Decimal("30")
        return {
            "S1": Decimal("3"),
            "S3": Decimal("30"),
            "S5": Decimal("14"),
            "S6": Decimal("20"),
            "S7": Decimal("8"),
            "S9": Decimal("3"),
        }.get(module, Decimal("7"))
    if unit in {"个/人", "行/人", "单/人", "批次/人"} or "人效" in name:
        return {
            "S1": Decimal("100"),
            "S3": Decimal("231.1"),
            "S5": Decimal("7"),
            "S6": Decimal("2.1"),
            "S7": Decimal("10"),
        }.get(module, Decimal("10"))
    if unit in {"个", "人", "单", "批", "次"}:
        return Decimal("0") if direction == "lower_is_better" else Decimal("1")
    return Decimal("1")


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,
        "inserted": 0,
        "existing": 0,
        "fromHistory": 0,
        "fromAncestor": 0,
        "fromSemanticBaseline": 0,
    }
    conn = connect()
    try:
        conn.begin()
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT Id,MetricCode,ModuleCode,MetricLevel,ParentId,MetricName,Unit,Direction,TenantId
                FROM ado_smart_ops_kpi_master
                WHERE IsEnabled=1 AND TenantId IN (0,%s)
                ORDER BY MetricCode,TenantId DESC
                """,
                (tenant_id,),
            )
            metrics_by_code: dict[str, dict[str, Any]] = {}
            metrics_by_id: dict[int, dict[str, Any]] = {}
            for row in cur.fetchall():
                metrics_by_code.setdefault(row["MetricCode"], row)
                metrics_by_id[int(row["Id"])] = row

            cur.execute(
                """
                SELECT metric_code,target_value
                FROM ado_smart_ops_kpi_target_config
                WHERE tenant_id=%s AND factory_id=%s AND status=1
                  AND effective_from<=NOW() AND (effective_to IS NULL OR effective_to>=NOW())
                """,
                (tenant_id, factory_id),
            )
            active_targets = {row["metric_code"]: Decimal(row["target_value"]) for row in cur.fetchall()}

            cur.execute(
                """
                SELECT metric_code,target_value,biz_date FROM ado_s9_kpi_value_l1_day
                WHERE tenant_id=%s AND factory_id=%s AND is_deleted=0 AND target_value IS NOT NULL
                UNION ALL
                SELECT metric_code,target_value,biz_date FROM ado_s9_kpi_value_l2_day
                WHERE tenant_id=%s AND factory_id=%s AND is_deleted=0 AND target_value IS NOT NULL
                UNION ALL
                SELECT metric_code,target_value,biz_date FROM ado_s9_kpi_value_l3_day
                WHERE tenant_id=%s AND factory_id=%s AND is_deleted=0 AND target_value IS NOT NULL
                UNION ALL
                SELECT metric_code,target_value,biz_date FROM ado_s9_kpi_value_l4_day
                WHERE tenant_id=%s AND factory_id=%s AND is_deleted=0 AND target_value IS NOT NULL
                ORDER BY metric_code,biz_date DESC
                """,
                (tenant_id, factory_id) * 4,
            )
            history_targets: dict[str, Decimal] = {}
            for row in cur.fetchall():
                history_targets.setdefault(row["metric_code"], Decimal(row["target_value"]))

            resolved = dict(active_targets)
            effective_from = datetime(datetime.now().year, 1, 1)
            now = datetime.now().replace(microsecond=0)
            for code, metric in metrics_by_code.items():
                if code in active_targets:
                    result["existing"] += 1
                    continue

                target = history_targets.get(code)
                source = "history"
                if target is None:
                    parent_id = metric.get("ParentId")
                    while parent_id:
                        parent = metrics_by_id.get(int(parent_id))
                        if not parent:
                            break
                        parent_code = parent["MetricCode"]
                        if str(parent.get("Unit") or "") == str(metric.get("Unit") or ""):
                            target = resolved.get(parent_code) or history_targets.get(parent_code)
                            if target is not None:
                                source = "ancestor"
                                break
                        parent_id = parent.get("ParentId")
                if target is None:
                    target = semantic_target(metric)
                    source = "semantic"

                resolved[code] = target
                result[{
                    "history": "fromHistory",
                    "ancestor": "fromAncestor",
                    "semantic": "fromSemanticBaseline",
                }[source]] += 1
                if not dry_run:
                    cur.execute(
                        """
                        INSERT INTO ado_smart_ops_kpi_target_config
                          (tenant_id,factory_id,metric_code,target_value,effective_from,effective_to,
                           source_type,source_batch_id,status,remark,create_time,update_time)
                        VALUES
                          (%s,%s,%s,%s,%s,NULL,'UAT_BASELINE',%s,1,
                           'UAT验收目标基线：优先沿用历史/同单位上级目标，否则按指标单位与方向设置',%s,%s)
                        """,
                        (
                            tenant_id,
                            factory_id,
                            code,
                            target,
                            effective_from,
                            BATCH_ID,
                            now,
                            now,
                        ),
                    )
                result["inserted"] += 1

        if dry_run:
            conn.rollback()
        else:
            conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()
    return result


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    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, default=str))
    return 0


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