#!/usr/bin/env python3 """Constructor persistente de casos raíz de revisión v0.3.10.""" from __future__ import annotations import argparse import json import os import signal import sys import time from dataclasses import dataclass from datetime import datetime, timedelta from typing import Any, Iterable import psycopg2 from psycopg2.extras import RealDictCursor VERSION = "0.3.10" BUILDER_NAME = "context_review_case_builder_v0310" POLICY_CODE = "CONTEXT_ROOT_CASE_QUEUE_V1" RUNNING = True def stop_handler(signum: int, frame: Any) -> None: del signum, frame global RUNNING RUNNING = False signal.signal(signal.SIGTERM, stop_handler) signal.signal(signal.SIGINT, stop_handler) def env_first(*names: str, default: str | None = None) -> str | None: for name in names: value = os.getenv(name) if value not in (None, ""): return value return default @dataclass(frozen=True) class Config: tenant: str site: str pg_host: str pg_port: int pg_database: str pg_user: str pg_password: str def load_config() -> Config: database = env_first("PGDATABASE", "POSTGRES_DB") user = env_first("PGUSER", "POSTGRES_USER") password = env_first("PGPASSWORD", "POSTGRES_PASSWORD") missing = [ name for name, value in ( ("PGDATABASE", database), ("PGUSER", user), ("PGPASSWORD", password), ) if not value ] if missing: raise RuntimeError( "Faltan variables: " + ", ".join(missing) ) return Config( tenant=env_first( "TENANT_OVERRIDE", "TENANT", default="ucepsa", ) or "ucepsa", site=env_first( "SITE_OVERRIDE", "SITE", default="ucepsa_onpremise", ) or "ucepsa_onpremise", pg_host=env_first( "PGHOST", default="mv_ucepsa_postgres_hot", ) or "mv_ucepsa_postgres_hot", pg_port=int( env_first("PGPORT", default="5432") or "5432" ), pg_database=str(database), pg_user=str(user), pg_password=str(password), ) def connect(config: Config): return psycopg2.connect( host=config.pg_host, port=config.pg_port, dbname=config.pg_database, user=config.pg_user, password=config.pg_password, connect_timeout=10, application_name=BUILDER_NAME, ) def severity_rank(value: str | None) -> int: return { "WARNING": 1, "CRITICAL": 2, }.get(value or "", 0) def case_family(episode: dict[str, Any]) -> str: code = str(episode["incident_code"]) if code == "MULTIPLE_OPEN_SESSIONS": return "MULTIPLE_OPEN_SESSIONS" if code == "RUNNING_WITHOUT_SHOPFLOOR_SESSION": return "UNCONFIRMED_ACTIVITY" if code in { "ORDER_WITHOUT_SHOPFLOOR_SESSION", "RUNNING_WITH_ORDER_NO_SHOPFLOOR_SESSION", }: return "ORDER_WITHOUT_SESSION" if code in { "OPEN_SESSION_ORDER_MISMATCH", "WORKORDER_SESSION_STATE_MISMATCH", }: return "ORDER_SESSION_MISMATCH" if code.startswith("SHOPFLOOR_SYNC_"): return "SYNC_HEALTH" return code def family_owner(family: str) -> str: if family == "MULTIPLE_OPEN_SESSIONS": return "JOINT" if family in { "UNCONFIRMED_ACTIVITY", "ORDER_WITHOUT_SESSION", }: return "PRODUCTION" return "TECHNICAL" def family_title(family: str) -> str: return { "MULTIPLE_OPEN_SESSIONS": "Varias órdenes Shop Floor abiertas en una máquina", "UNCONFIRMED_ACTIVITY": "Actividad de máquina sin contexto confirmado", "ORDER_WITHOUT_SESSION": "Orden Odoo sin sesión Shop Floor utilizable", "ORDER_SESSION_MISMATCH": "Orden Odoo y sesión Shop Floor no coinciden", "SYNC_HEALTH": "Problema de sincronización Shop Floor", }.get( family, family.replace("_", " ").title(), ) def question_text( family: str, machine_id: str | None, orders: list[str], ) -> str: machine = machine_id or "la instalación" order_text = ( ", ".join(orders) if orders else "las órdenes implicadas" ) if family == "MULTIPLE_OPEN_SESSIONS": return ( f"¿Cuál de {order_text} era la orden realmente " f"ejecutada en {machine} y qué debe hacerse con " "las demás?" ) if family == "UNCONFIRMED_ACTIVITY": return ( f"¿La actividad de {machine} correspondía a " "sistema antiguo, prueba o ajuste, material " "residual o un inicio omitido en Shop Floor?" ) if family == "ORDER_WITHOUT_SESSION": return ( f"¿La orden publicada en {machine} llegó a " "iniciarse realmente en Shop Floor?" ) if family == "ORDER_SESSION_MISMATCH": return ( f"¿Qué orden estaba realmente ejecutándose en " f"{machine}?" ) return ( f"¿Cuál es la causa operativa del caso detectado " f"en {machine}?" ) def recommended_action(family: str) -> str: return { "MULTIPLE_OPEN_SESSIONS": "Confirmar la orden real y corregir en Odoo las órdenes abiertas por error.", "UNCONFIRMED_ACTIVITY": "Clasificar el intervalo y crear contexto explícito solo cuando exista confirmación humana.", "ORDER_WITHOUT_SESSION": "Confirmar si la orden era real o una transición breve de Odoo.", "ORDER_SESSION_MISMATCH": "Corregir Odoo o el mapeo antes de atribuir producción o paros.", "SYNC_HEALTH": "Resolver el problema técnico antes de utilizar datos nuevos.", }.get( family, "Revisar la evidencia técnica y documentar una única resolución.", ) def episode_interval( episode: dict[str, Any], ) -> tuple[datetime, datetime]: start = episode["first_seen_at"] end = ( episode.get("ended_at") or episode.get("last_seen_at") or start ) return start, end def cluster_episodes( episodes: list[dict[str, Any]], grouping_window_s: int, ) -> list[list[dict[str, Any]]]: grouped: dict[ tuple[str | None, str], list[dict[str, Any]], ] = {} for episode in episodes: family = case_family(episode) grouped.setdefault( ( episode.get("machine_id"), family, ), [], ).append(episode) clusters: list[list[dict[str, Any]]] = [] grace = timedelta(seconds=grouping_window_s) for (_, family), family_episodes in grouped.items(): ordered = sorted( family_episodes, key=lambda row: ( row["first_seen_at"], row["episode_id"], ), ) if family != "MULTIPLE_OPEN_SESSIONS": clusters.extend([[row] for row in ordered]) continue current: list[dict[str, Any]] = [] current_end: datetime | None = None for row in ordered: start, end = episode_interval(row) if ( not current or current_end is None or start <= current_end + grace ): current.append(row) current_end = ( end if current_end is None else max(current_end, end) ) continue clusters.append(current) current = [row] current_end = end if current: clusters.append(current) return clusters def existing_case_for_cluster( cursor, episode_ids: list[int], ) -> int | None: cursor.execute( """ SELECT DISTINCT case_id FROM mv_loss_intelligence .context_review_case_evidence WHERE episode_id = ANY(%s::bigint[]) ORDER BY case_id """, (episode_ids,), ) rows = cursor.fetchall() if len(rows) > 1: raise RuntimeError( "Las evidencias del mismo cluster ya pertenecen " f"a varios casos: {rows!r}" ) return ( int(rows[0]["case_id"]) if rows else None ) def case_key( family: str, machine_id: str | None, episode_ids: list[int], ) -> str: first_episode = min(episode_ids) if family == "MULTIPLE_OPEN_SESSIONS": return ( "ROOT:MULTIPLE_OPEN_SESSIONS:" f"{machine_id or 'GLOBAL'}:" f"{first_episode}" ) return f"ROOT:EPISODE:{first_episode}" def mirrored_review_state( cluster: list[dict[str, Any]], ) -> tuple[str, str | None, str | None, datetime | None, str | None]: statuses = { str(row["review_status"]) for row in cluster } if "PENDING" in statuses: return "PENDING", None, None, None, None reviewed = [ row for row in cluster if row["review_status"] == "REVIEWED" ] if reviewed: latest = max( reviewed, key=lambda row: ( row.get("reviewed_at") or row["last_seen_at"] ), ) classification = latest.get( "review_classification" ) mapped = ( classification if classification in { "AUTHORIZED_LEGACY_PRODUCTION", "TEST_OR_SETUP", "RESIDUAL_MATERIAL", "ODOO_START_OMITTED", "DATA_ISSUE", "NOT_RELEVANT", "OTHER", } else "OTHER" ) return ( "REVIEWED", mapped, latest.get("reviewed_by"), latest.get("reviewed_at"), latest.get("review_notes"), ) if statuses == {"DISMISSED"}: return ( "DISMISSED", "NOT_RELEVANT", None, None, "Todas las evidencias estaban descartadas.", ) return "NOT_REQUIRED", None, None, None, None def evidence_role(episode: dict[str, Any]) -> str: scope = episode.get("incident_scope") if scope == "MACHINE": return "MACHINE_ROOT" if scope == "SESSION": return "SESSION_EVIDENCE" return "PRIMARY_EPISODE" def process_cycle( config: Config, dry_run: bool = False, ) -> dict[str, Any]: with connect(config) as connection: with connection.cursor( cursor_factory=RealDictCursor, ) as cursor: cursor.execute( """ SELECT pg_try_advisory_xact_lock( hashtext(%s) ) AS locked """, ( ( f"{config.tenant}:" f"{config.site}:" f"{BUILDER_NAME}" ), ), ) if not cursor.fetchone()["locked"]: return { "version": VERSION, "status": "skipped_lock_busy", } cursor.execute( """ SELECT builder_interval_s, grouping_window_s, history_lookback_days FROM mv_loss_intelligence .context_review_case_policies WHERE tenant = %s AND site = %s AND policy_code = %s AND active LIMIT 1 """, ( config.tenant, config.site, POLICY_CODE, ), ) policy = cursor.fetchone() if not policy: raise RuntimeError( f"No existe política activa {POLICY_CODE!r}" ) cursor.execute( """ SELECT * FROM mv_reports_ucepsa_prod .li_shopfloor_context_incident_history_v1 WHERE review_required AND ( episode_status = 'OPEN' OR first_seen_at >= now() - ( %s * interval '1 day' ) ) ORDER BY machine_id NULLS FIRST, first_seen_at, episode_id """, ( int( policy["history_lookback_days"] ), ), ) episodes = [ dict(row) for row in cursor.fetchall() ] clusters = cluster_episodes( episodes, int(policy["grouping_window_s"]), ) preview = [] link_count = 0 for cluster in clusters: ids = sorted( int(row["episode_id"]) for row in cluster ) family = case_family(cluster[0]) machine_id = cluster[0].get( "machine_id" ) orders = sorted({ str(row["production_order"]) for row in cluster if row.get("production_order") }) first_seen = min( row["first_seen_at"] for row in cluster ) last_seen = max( row["last_seen_at"] for row in cluster ) open_rows = [ row for row in cluster if row["episode_status"] == "OPEN" ] ended_at = ( None if open_rows else max( row.get("ended_at") or row["last_seen_at"] for row in cluster ) ) max_severity = max( ( str(row["max_severity"]) for row in cluster ), key=severity_rank, ) review_status, classification, reviewed_by, reviewed_at, notes = ( mirrored_review_state(cluster) ) existing_case_id = existing_case_for_cluster( cursor, ids, ) key = case_key( family, machine_id, ids, ) item = { "case_id": existing_case_id, "case_key": key, "case_family": family, "machine_id": machine_id, "episode_ids": ids, "case_status": ( "OPEN" if open_rows else "CLOSED" ), "first_seen_at": first_seen, "last_seen_at": last_seen, "ended_at": ended_at, "max_severity": max_severity, "review_status": review_status, "orders": orders, } preview.append(item) if dry_run: continue owner = family_owner(family) title = family_title(family) question = question_text( family, machine_id, orders, ) action = recommended_action(family) if existing_case_id is None: cursor.execute( """ INSERT INTO mv_loss_intelligence .context_review_cases ( tenant, site, case_key, case_family, root_incident_code, machine_id, case_title, question_text, recommended_action, case_status, max_severity, first_seen_at, last_seen_at, ended_at, evidence_count, review_owner, review_required, review_status, resolution_classification, reviewed_by, reviewed_at, review_notes, official_eligible ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 0, %s, true, %s, %s, %s, %s, %s, false ) RETURNING case_id """, ( config.tenant, config.site, key, family, str( cluster[0][ "incident_code" ] ), machine_id, title, question, action, ( "OPEN" if open_rows else "CLOSED" ), max_severity, first_seen, last_seen, ended_at, owner, review_status, classification, reviewed_by, reviewed_at, notes, ), ) case_id = int( cursor.fetchone()["case_id"] ) else: case_id = existing_case_id cursor.execute( """ UPDATE mv_loss_intelligence .context_review_cases SET case_family = %s, root_incident_code = %s, machine_id = %s, case_title = %s, question_text = %s, recommended_action = %s, case_status = %s, max_severity = %s, first_seen_at = LEAST( first_seen_at, %s ), last_seen_at = GREATEST( last_seen_at, %s ), ended_at = %s, review_owner = %s, updated_at = now() WHERE case_id = %s """, ( family, str( cluster[0][ "incident_code" ] ), machine_id, title, question, action, ( "OPEN" if open_rows else "CLOSED" ), max_severity, first_seen, last_seen, ended_at, owner, case_id, ), ) for row in cluster: cursor.execute( """ INSERT INTO mv_loss_intelligence .context_review_case_evidence ( case_id, episode_id, evidence_role ) VALUES (%s, %s, %s) ON CONFLICT (episode_id) DO UPDATE SET evidence_role = EXCLUDED.evidence_role WHERE mv_loss_intelligence .context_review_case_evidence .case_id = EXCLUDED.case_id """, ( case_id, int(row["episode_id"]), evidence_role(row), ), ) link_count += cursor.rowcount cursor.execute( """ UPDATE mv_loss_intelligence .context_review_cases c SET evidence_count = ( SELECT COUNT(*) FROM mv_loss_intelligence .context_review_case_evidence e WHERE e.case_id = c.case_id ), updated_at = now() WHERE c.case_id = %s """, (case_id,), ) if dry_run: return { "version": VERSION, "status": "dry_run", "episode_count": len(episodes), "case_count": len(clusters), "cases": preview, "builder_interval_s": int( policy["builder_interval_s"] ), } cursor.execute( """ INSERT INTO mv_loss_intelligence .context_review_case_builder_state ( tenant, site, builder_name, last_success_at, last_episode_count, last_case_count, last_link_count, last_error, updated_at ) VALUES ( %s, %s, %s, now(), %s, %s, %s, NULL, now() ) ON CONFLICT ( tenant, site, builder_name ) DO UPDATE SET last_success_at = EXCLUDED.last_success_at, last_episode_count = EXCLUDED.last_episode_count, last_case_count = EXCLUDED.last_case_count, last_link_count = EXCLUDED.last_link_count, last_error = NULL, updated_at = now() """, ( config.tenant, config.site, BUILDER_NAME, len(episodes), len(clusters), link_count, ), ) return { "version": VERSION, "status": "ok", "episode_count": len(episodes), "case_count": len(clusters), "link_count": link_count, "builder_interval_s": int( policy["builder_interval_s"] ), } def record_error( config: Config, error: Exception, ) -> None: try: with connect(config) as connection: with connection.cursor() as cursor: cursor.execute( """ INSERT INTO mv_loss_intelligence .context_review_case_builder_state ( tenant, site, builder_name, last_error, updated_at ) VALUES ( %s, %s, %s, %s, now() ) ON CONFLICT ( tenant, site, builder_name ) DO UPDATE SET last_error = EXCLUDED.last_error, updated_at = now() """, ( config.tenant, config.site, BUILDER_NAME, str(error)[:4000], ), ) except Exception as state_error: print( json.dumps( { "status": "builder_error_state_write_failed", "error": str(error), "state_error": str(state_error), }, ensure_ascii=False, ), file=sys.stderr, flush=True, ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Construye casos raíz SHADOW a partir " "de episodios técnicos." ) ) mode = parser.add_mutually_exclusive_group( required=True ) mode.add_argument("--once", action="store_true") mode.add_argument("--watch", action="store_true") parser.add_argument("--dry-run", action="store_true") parser.add_argument("--interval", type=int) return parser.parse_args() def main() -> int: args = parse_args() config = load_config() while RUNNING: try: result = process_cycle( config, dry_run=args.dry_run, ) print( json.dumps( result, ensure_ascii=False, default=str, ), flush=True, ) if args.once or args.dry_run: return 0 interval = ( args.interval or int( result.get( "builder_interval_s", 15, ) ) ) for _ in range(max(interval, 1)): if not RUNNING: break time.sleep(1) except Exception as error: record_error(config, error) print( json.dumps( { "version": VERSION, "status": "error", "error": str(error), }, ensure_ascii=False, ), file=sys.stderr, flush=True, ) if args.once or args.dry_run: return 1 for _ in range( max(args.interval or 15, 1) ): if not RUNNING: break time.sleep(1) return 0 if __name__ == "__main__": raise SystemExit(main())