674 lines
19 KiB
Python
Executable File
674 lines
19 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Agenda diaria y evidencia de gobernanza UCEPSA v0.3.12."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
import psycopg2
|
|
from psycopg2.extras import Json, RealDictCursor
|
|
|
|
|
|
VERSION = "0.3.12"
|
|
|
|
|
|
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
|
|
|
|
|
|
def connect():
|
|
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 psycopg2.connect(
|
|
host=env_first(
|
|
"PGHOST",
|
|
default="mv_ucepsa_postgres_hot",
|
|
),
|
|
port=int(
|
|
env_first("PGPORT", default="5432")
|
|
or "5432"
|
|
),
|
|
dbname=database,
|
|
user=user,
|
|
password=password,
|
|
connect_timeout=10,
|
|
application_name=(
|
|
"context_daily_governance_v0312"
|
|
),
|
|
)
|
|
|
|
|
|
def json_safe(value: Any) -> Any:
|
|
if isinstance(value, dict):
|
|
return {
|
|
str(key): json_safe(item)
|
|
for key, item in value.items()
|
|
}
|
|
if isinstance(value, (list, tuple, set)):
|
|
return [
|
|
json_safe(item)
|
|
for item in value
|
|
]
|
|
if isinstance(value, (datetime, date)):
|
|
return value.isoformat()
|
|
if isinstance(value, Decimal):
|
|
return str(value)
|
|
return value
|
|
|
|
|
|
def print_json(value: Any) -> None:
|
|
print(
|
|
json.dumps(
|
|
json_safe(value),
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
)
|
|
|
|
|
|
def fetch_agenda(cursor, limit: int) -> list[dict[str, Any]]:
|
|
cursor.execute(
|
|
"""
|
|
SELECT *
|
|
FROM
|
|
mv_reports_ucepsa_prod
|
|
.li_context_governance_daily_agenda_v1
|
|
ORDER BY agenda_rank
|
|
LIMIT %s
|
|
""",
|
|
(limit,),
|
|
)
|
|
return [
|
|
dict(row)
|
|
for row in cursor.fetchall()
|
|
]
|
|
|
|
|
|
def fetch_summary(cursor) -> dict[str, Any]:
|
|
cursor.execute(
|
|
"""
|
|
SELECT *
|
|
FROM
|
|
mv_reports_ucepsa_prod
|
|
.li_context_governance_daily_summary_v1
|
|
"""
|
|
)
|
|
row = cursor.fetchone()
|
|
return dict(row) if row else {}
|
|
|
|
|
|
def fetch_decision(
|
|
cursor,
|
|
decision_group_key: str,
|
|
) -> dict[str, Any]:
|
|
cursor.execute(
|
|
"""
|
|
SELECT *
|
|
FROM
|
|
mv_reports_ucepsa_prod
|
|
.li_context_governance_decision_evidence_pack_v1
|
|
WHERE decision_group_key = %s
|
|
""",
|
|
(decision_group_key,),
|
|
)
|
|
row = cursor.fetchone()
|
|
if not row:
|
|
raise RuntimeError(
|
|
"No existe la decisión "
|
|
f"{decision_group_key!r}"
|
|
)
|
|
return dict(row)
|
|
|
|
|
|
def agenda_command(limit: int) -> None:
|
|
with connect() as connection:
|
|
with connection.cursor(
|
|
cursor_factory=RealDictCursor,
|
|
) as cursor:
|
|
print_json({
|
|
"version": VERSION,
|
|
"summary": fetch_summary(cursor),
|
|
"agenda": fetch_agenda(cursor, limit),
|
|
})
|
|
|
|
|
|
def show_command(decision_group_key: str) -> None:
|
|
with connect() as connection:
|
|
with connection.cursor(
|
|
cursor_factory=RealDictCursor,
|
|
) as cursor:
|
|
print_json(
|
|
fetch_decision(
|
|
cursor,
|
|
decision_group_key,
|
|
)
|
|
)
|
|
|
|
|
|
def snapshot_payload(
|
|
agenda: list[dict[str, Any]],
|
|
) -> str:
|
|
payload = json.dumps(
|
|
json_safe(agenda),
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
return hashlib.sha256(payload).hexdigest()
|
|
|
|
|
|
def snapshot_command(
|
|
generated_by: str,
|
|
notes: str | None,
|
|
limit: int,
|
|
) -> None:
|
|
with connect() as connection:
|
|
with connection.cursor(
|
|
cursor_factory=RealDictCursor,
|
|
) as cursor:
|
|
agenda = fetch_agenda(cursor, limit)
|
|
summary = fetch_summary(cursor)
|
|
local_date = (
|
|
agenda[0]["agenda_local_date"]
|
|
if agenda
|
|
else summary.get(
|
|
"agenda_local_date"
|
|
)
|
|
)
|
|
|
|
if local_date is None:
|
|
raise RuntimeError(
|
|
"No se pudo resolver la fecha local."
|
|
)
|
|
|
|
state_hash = snapshot_payload(agenda)
|
|
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO
|
|
mv_loss_intelligence
|
|
.context_governance_agenda_snapshots (
|
|
tenant,
|
|
site,
|
|
agenda_local_date,
|
|
generated_by,
|
|
item_count,
|
|
source_state_hash,
|
|
notes,
|
|
official_eligible
|
|
)
|
|
VALUES (
|
|
'ucepsa',
|
|
'ucepsa_onpremise',
|
|
%s,
|
|
%s,
|
|
%s,
|
|
%s,
|
|
%s,
|
|
false
|
|
)
|
|
RETURNING *
|
|
""",
|
|
(
|
|
local_date,
|
|
generated_by,
|
|
len(agenda),
|
|
state_hash,
|
|
notes,
|
|
),
|
|
)
|
|
snapshot = dict(cursor.fetchone())
|
|
|
|
for row in agenda:
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO
|
|
mv_loss_intelligence
|
|
.context_governance_agenda_snapshot_items (
|
|
snapshot_id,
|
|
item_rank,
|
|
decision_group_key,
|
|
decision_status,
|
|
max_severity,
|
|
machine_id,
|
|
decision_family,
|
|
decision_title,
|
|
question_text,
|
|
recommended_action,
|
|
review_owner,
|
|
first_seen_at,
|
|
last_seen_at,
|
|
ended_at,
|
|
pending_age_hours,
|
|
attention_status,
|
|
review_due_at,
|
|
pending_root_case_count,
|
|
reviewed_root_case_count,
|
|
linked_root_case_count,
|
|
linked_episode_count,
|
|
root_case_ids,
|
|
episode_ids,
|
|
production_orders,
|
|
odoo_workorder_ids,
|
|
known_resolution_classifications,
|
|
evidence_json,
|
|
official_eligible
|
|
)
|
|
VALUES (
|
|
%s, %s, %s, %s, %s,
|
|
%s, %s, %s, %s, %s,
|
|
%s, %s, %s, %s, %s,
|
|
%s, %s, %s, %s, %s,
|
|
%s, %s, %s, %s, %s,
|
|
%s, %s, false
|
|
)
|
|
""",
|
|
(
|
|
snapshot["snapshot_id"],
|
|
row["agenda_rank"],
|
|
row["decision_group_key"],
|
|
row["decision_status"],
|
|
row["max_severity"],
|
|
row["machine_id"],
|
|
row["decision_family"],
|
|
row["decision_title"],
|
|
row["question_text"],
|
|
row["recommended_action"],
|
|
row["review_owner"],
|
|
row["first_seen_at"],
|
|
row["last_seen_at"],
|
|
row["ended_at"],
|
|
row["pending_age_hours"],
|
|
row["attention_status"],
|
|
row["review_due_at"],
|
|
row["pending_root_case_count"],
|
|
row["reviewed_root_case_count"],
|
|
row["linked_root_case_count"],
|
|
row["linked_episode_count"],
|
|
row["root_case_ids"],
|
|
row["episode_ids"],
|
|
row["production_orders"],
|
|
row["odoo_workorder_ids"],
|
|
row[
|
|
"known_resolution_classifications"
|
|
],
|
|
Json({
|
|
"incident_codes":
|
|
row["incident_codes"],
|
|
"incident_scopes":
|
|
row["incident_scopes"],
|
|
"operator_names":
|
|
row["operator_names"],
|
|
"episode_evidence":
|
|
json_safe(
|
|
row["episode_evidence"]
|
|
),
|
|
}),
|
|
),
|
|
)
|
|
|
|
print_json({
|
|
"version": VERSION,
|
|
"status": "snapshot_created",
|
|
"snapshot": snapshot,
|
|
})
|
|
|
|
|
|
def fetch_snapshot(
|
|
cursor,
|
|
snapshot_id: int,
|
|
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
|
cursor.execute(
|
|
"""
|
|
SELECT *
|
|
FROM
|
|
mv_loss_intelligence
|
|
.context_governance_agenda_snapshots
|
|
WHERE snapshot_id = %s
|
|
""",
|
|
(snapshot_id,),
|
|
)
|
|
header = cursor.fetchone()
|
|
if not header:
|
|
raise RuntimeError(
|
|
f"No existe el snapshot {snapshot_id}"
|
|
)
|
|
|
|
cursor.execute(
|
|
"""
|
|
SELECT *
|
|
FROM
|
|
mv_loss_intelligence
|
|
.context_governance_agenda_snapshot_items
|
|
WHERE snapshot_id = %s
|
|
ORDER BY item_rank
|
|
""",
|
|
(snapshot_id,),
|
|
)
|
|
items = [
|
|
dict(row)
|
|
for row in cursor.fetchall()
|
|
]
|
|
return dict(header), items
|
|
|
|
|
|
def array_text(values: list[Any] | None) -> str:
|
|
if not values:
|
|
return "—"
|
|
return ", ".join(str(value) for value in values)
|
|
|
|
|
|
def dt_text(value: Any) -> str:
|
|
if value is None:
|
|
return "—"
|
|
if isinstance(value, datetime):
|
|
return value.strftime("%d/%m/%Y %H:%M:%S %z")
|
|
return str(value)
|
|
|
|
|
|
def markdown_document(
|
|
title_date: Any,
|
|
generated_at: Any,
|
|
generated_by: str,
|
|
summary: dict[str, Any],
|
|
items: list[dict[str, Any]],
|
|
snapshot_id: int | None = None,
|
|
notes: str | None = None,
|
|
) -> str:
|
|
lines = [
|
|
f"# Revisión diaria de gobernanza — {title_date}",
|
|
"",
|
|
f"- Generada: {dt_text(generated_at)}",
|
|
f"- Generada por: {generated_by}",
|
|
f"- Snapshot: {snapshot_id if snapshot_id is not None else 'vista actual'}",
|
|
"- Modo: SHADOW",
|
|
"- Ledger oficial: desactivado",
|
|
]
|
|
|
|
if notes:
|
|
lines.extend([
|
|
f"- Notas: {notes}",
|
|
])
|
|
|
|
lines.extend([
|
|
"",
|
|
"## Resumen",
|
|
"",
|
|
f"- Decisiones pendientes: {summary.get('pending_decision_count', len(items))}",
|
|
f"- Críticas: {summary.get('pending_critical_decision_count', '—')}",
|
|
f"- Requieren atención: {summary.get('attention_decision_count', '—')}",
|
|
f"- Vencidas: {summary.get('overdue_decision_count', '—')}",
|
|
f"- Casos raíz pendientes: {summary.get('pending_root_case_count', '—')}",
|
|
f"- Evidencias enlazadas: {summary.get('linked_episode_count', '—')}",
|
|
"",
|
|
"## Agenda",
|
|
"",
|
|
])
|
|
|
|
if not items:
|
|
lines.append(
|
|
"No hay decisiones humanas pendientes."
|
|
)
|
|
return "\n".join(lines) + "\n"
|
|
|
|
for item in items:
|
|
rank = (
|
|
item.get("agenda_rank")
|
|
or item.get("item_rank")
|
|
)
|
|
lines.extend([
|
|
f"### {rank}. {item['max_severity']} · {item.get('machine_id') or 'GLOBAL'}",
|
|
"",
|
|
f"**{item['decision_title']}**",
|
|
"",
|
|
f"- Estado: {item['decision_status']}",
|
|
f"- Atención: {item['attention_status']}",
|
|
f"- Responsable: {item['review_owner']}",
|
|
f"- Inicio: {dt_text(item['first_seen_at'])}",
|
|
f"- Última evidencia: {dt_text(item['last_seen_at'])}",
|
|
f"- Antigüedad pendiente: {item['pending_age_hours']} h",
|
|
f"- Límite de revisión: {dt_text(item['review_due_at'])}",
|
|
f"- Casos raíz: {array_text(item.get('root_case_ids'))}",
|
|
f"- Episodios: {array_text(item.get('episode_ids'))}",
|
|
f"- Órdenes: {array_text(item.get('production_orders'))}",
|
|
f"- Workorders: {array_text(item.get('odoo_workorder_ids'))}",
|
|
f"- Resolución conocida: {array_text(item.get('known_resolution_classifications'))}",
|
|
"",
|
|
"**Pregunta**",
|
|
"",
|
|
item["question_text"],
|
|
"",
|
|
"**Acción recomendada**",
|
|
"",
|
|
item["recommended_action"],
|
|
"",
|
|
])
|
|
|
|
evidence = item.get("episode_evidence")
|
|
if evidence is None:
|
|
evidence_json = item.get(
|
|
"evidence_json",
|
|
{},
|
|
)
|
|
evidence = evidence_json.get(
|
|
"episode_evidence",
|
|
[],
|
|
)
|
|
|
|
if evidence:
|
|
lines.extend([
|
|
"**Evidencia técnica resumida**",
|
|
"",
|
|
])
|
|
for row in evidence:
|
|
lines.append(
|
|
"- "
|
|
f"Episodio {row.get('episode_id')}: "
|
|
f"{row.get('incident_title')}; "
|
|
f"{row.get('latest_detail') or 'sin detalle adicional'}"
|
|
)
|
|
lines.append("")
|
|
|
|
lines.extend([
|
|
f"`{item['decision_group_key']}`",
|
|
"",
|
|
])
|
|
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def export_markdown_command(
|
|
snapshot_id: int | None,
|
|
generated_by: str,
|
|
) -> None:
|
|
with connect() as connection:
|
|
with connection.cursor(
|
|
cursor_factory=RealDictCursor,
|
|
) as cursor:
|
|
if snapshot_id is not None:
|
|
header, items = fetch_snapshot(
|
|
cursor,
|
|
snapshot_id,
|
|
)
|
|
summary = {
|
|
"pending_decision_count":
|
|
header["item_count"],
|
|
"pending_critical_decision_count":
|
|
sum(
|
|
1
|
|
for row in items
|
|
if row["max_severity"] ==
|
|
"CRITICAL"
|
|
),
|
|
"attention_decision_count":
|
|
sum(
|
|
1
|
|
for row in items
|
|
if row["attention_status"] ==
|
|
"ATTENTION"
|
|
),
|
|
"overdue_decision_count":
|
|
sum(
|
|
1
|
|
for row in items
|
|
if row["attention_status"] ==
|
|
"OVERDUE"
|
|
),
|
|
"pending_root_case_count":
|
|
sum(
|
|
row[
|
|
"pending_root_case_count"
|
|
]
|
|
for row in items
|
|
),
|
|
"linked_episode_count":
|
|
sum(
|
|
row[
|
|
"linked_episode_count"
|
|
]
|
|
for row in items
|
|
),
|
|
}
|
|
document = markdown_document(
|
|
title_date=
|
|
header["agenda_local_date"],
|
|
generated_at=
|
|
header["generated_at"],
|
|
generated_by=
|
|
header["generated_by"],
|
|
summary=summary,
|
|
items=items,
|
|
snapshot_id=snapshot_id,
|
|
notes=header["notes"],
|
|
)
|
|
else:
|
|
items = fetch_agenda(cursor, 1000)
|
|
summary = fetch_summary(cursor)
|
|
document = markdown_document(
|
|
title_date=
|
|
summary.get(
|
|
"agenda_local_date"
|
|
),
|
|
generated_at=summary.get(
|
|
"observed_at",
|
|
datetime.now(),
|
|
),
|
|
generated_by=generated_by,
|
|
summary=summary,
|
|
items=items,
|
|
)
|
|
|
|
sys.stdout.write(document)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description=(
|
|
"Agenda diaria, evidencia y snapshots "
|
|
"de gobernanza UCEPSA."
|
|
)
|
|
)
|
|
sub = parser.add_subparsers(
|
|
dest="command",
|
|
required=True,
|
|
)
|
|
|
|
agenda = sub.add_parser("agenda")
|
|
agenda.add_argument(
|
|
"--limit",
|
|
type=int,
|
|
default=50,
|
|
)
|
|
|
|
show = sub.add_parser("show")
|
|
show.add_argument(
|
|
"--decision-group-key",
|
|
required=True,
|
|
)
|
|
|
|
snapshot = sub.add_parser("snapshot")
|
|
snapshot.add_argument(
|
|
"--generated-by",
|
|
required=True,
|
|
)
|
|
snapshot.add_argument("--notes")
|
|
snapshot.add_argument(
|
|
"--limit",
|
|
type=int,
|
|
default=50,
|
|
)
|
|
|
|
export = sub.add_parser("export-markdown")
|
|
export.add_argument(
|
|
"--snapshot-id",
|
|
type=int,
|
|
)
|
|
export.add_argument(
|
|
"--generated-by",
|
|
default="MESAVAULT",
|
|
)
|
|
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
|
|
if args.command == "agenda":
|
|
agenda_command(args.limit)
|
|
return 0
|
|
|
|
if args.command == "show":
|
|
show_command(
|
|
args.decision_group_key
|
|
)
|
|
return 0
|
|
|
|
if args.command == "snapshot":
|
|
snapshot_command(
|
|
generated_by=args.generated_by,
|
|
notes=args.notes,
|
|
limit=args.limit,
|
|
)
|
|
return 0
|
|
|
|
if args.command == "export-markdown":
|
|
export_markdown_command(
|
|
snapshot_id=args.snapshot_id,
|
|
generated_by=args.generated_by,
|
|
)
|
|
return 0
|
|
|
|
raise RuntimeError("Comando no soportado")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|