""" ERGON -- Parser de bitacora diaria. Lee una entrada de daily_log (texto libre, usualmente transcripcion de voz), llama a Claude con tool use para extraer acciones estructuradas, y devuelve un borrador que el usuario debe confirmar antes de que el sistema grabe. Contract: parse_bitacora(conn, codigo, entry_id) -> dict { "ok": bool, "entry_id": int, "acciones": [ # una por tool call que hizo Claude {"tool": str, "input": dict, "id": str} ], "meta": { "modelo": str, "prompt_version": str, "procesado_at": iso, "usage": {"input_tokens": int, "output_tokens": int}, "texto_original": str, "preguntas_pendientes": list[str], # extraidas del texto libre de Claude "resumen_modelo": str # lo que Claude escribio ademas de las tools } } Importante: - El parser NO escribe en la DB. Solo propone. El endpoint /confirmar hace el upsert real tras la aprobacion del usuario. - No hace loop agentic: toma los tool_use de Claude como borrador final (stop_reason=="tool_use" o "end_turn" indistintamente). - Inyecta contexto dinamico: rubros activos + subcontratistas de la obra + fecha real del sistema (scar_009). Asi Claude no alucina nombres. """ from __future__ import annotations import json import os import sqlite3 from datetime import date, datetime, timezone from pathlib import Path from typing import Any, Optional try: import anthropic _HAS_SDK = True except ImportError: _HAS_SDK = False # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- MODEL = "claude-sonnet-4-6" # mejor precision en extraccion + cache-friendly (minimo 1024 tokens) MAX_TOKENS = 2048 PROMPT_VERSION = "v3" # --------------------------------------------------------------------------- # Tools -- 1 por tabla operativa del schema. Mismos nombres de campo que la DB. # --------------------------------------------------------------------------- TOOLS: list[dict] = [ { "name": "upsert_asistencia", "description": ( "Registrar asistencia diaria de personal en obra. " "Usar cuando el usuario menciona cantidad de personas que trabajaron en un dia, " "o reporta lluvia/clima con impacto en la jornada, o declara un dia perdido." ), "input_schema": { "type": "object", "properties": { "fecha": { "type": "string", "description": "Fecha ISO YYYY-MM-DD del dia registrado. " "'hoy', 'ayer' deben resolverse contra la fecha real del sistema inyectada en el system prompt." }, "total_personal": { "type": "integer", "description": "Cantidad total de personas que trabajaron ese dia (>=0)" }, "lluvia_mm": { "type": "number", "description": "Milimetros de lluvia registrados ese dia. Opcional (default 0)" }, "dia_perdido": { "type": "boolean", "description": "True si la jornada se considero perdida (no se trabajo)." }, "notas": {"type": "string", "description": "Notas libres opcionales"}, "confianza": { "type": "object", "properties": { "fecha": {"type": "number"}, "total_personal": {"type": "number"}, "lluvia_mm": {"type": "number"}, "dia_perdido": {"type": "number"} }, "description": "Confianza 0.0-1.0 por campo extraido" } }, "required": ["fecha", "total_personal"] } }, { "name": "upsert_avance_mensual", "description": ( "Registrar avance fisico mensual (acumulado) de un rubro. " "Usar cuando el usuario reporta porcentaje de avance de un rubro especifico. " "El rubro debe coincidir con uno de la lista de rubros activos de la obra inyectada." ), "input_schema": { "type": "object", "properties": { "rubro_nombre": { "type": "string", "description": "Nombre del rubro EXACTAMENTE como aparece en la lista de rubros activos de la obra." }, "mes": { "type": "string", "description": "Mes en formato YYYY-MM. Por default el mes actual." }, "plan_acum_pct": { "type": "number", "description": "Plan acumulado al fin de mes, en % (0-100). Opcional." }, "real_acum_pct": { "type": "number", "description": "Real acumulado al fin de mes, en % (0-100)." }, "confianza": { "type": "object", "properties": { "rubro_nombre": {"type": "number"}, "real_acum_pct": {"type": "number"} } } }, "required": ["rubro_nombre", "mes", "real_acum_pct"] } }, { "name": "upsert_presupuesto_mensual", "description": ( "Registrar monto real (AC) o plan presupuestario mensual por rubro. " "Usar cuando el usuario reporta un pago, desembolso o presupuesto por rubro. " "Distinguir entre 'plan' (presupuesto) y 'real' (gasto ejecutado)." ), "input_schema": { "type": "object", "properties": { "rubro_nombre": {"type": "string", "description": "Rubro al que imputa el gasto (match con rubros activos)"}, "mes": {"type": "string", "description": "Mes YYYY-MM. Default: mes actual."}, "concepto": { "type": "string", "enum": ["plan", "real"], "description": "'real' = gasto ejecutado (AC). 'plan' = presupuesto." }, "monto": { "type": "number", "description": "Monto en la moneda de la obra (ej. Gs para PYG). Usar numero absoluto." }, "proveedor": { "type": "string", "description": "Proveedor o subcontratista que recibio el pago. Match con subcontratistas conocidos si es posible." }, "confianza": { "type": "object", "properties": { "rubro_nombre": {"type": "number"}, "monto": {"type": "number"}, "proveedor": {"type": "number"} } } }, "required": ["rubro_nombre", "mes", "concepto", "monto"] } }, { "name": "upsert_contrato", "description": ( "Registrar o actualizar un contrato, orden de trabajo (OT), adenda u otro documento contractual. " "Usar cuando el usuario menciona emision, firma, vencimiento o cambio de estado de un documento formal." ), "input_schema": { "type": "object", "properties": { "tipo": { "type": "string", "enum": ["Contrato principal", "OT", "Adenda", "Nota de pedido", "Otro"], "description": "Tipo de documento contractual" }, "descripcion": {"type": "string", "description": "Descripcion breve del contrato"}, "subcontratista_nombre": { "type": "string", "description": "Subcontratista asociado. Match con subcontratistas activos." }, "fecha_emision": {"type": "string", "description": "ISO YYYY-MM-DD"}, "fecha_firma": {"type": "string", "description": "ISO YYYY-MM-DD o null si pendiente"}, "fecha_vence": {"type": "string", "description": "ISO YYYY-MM-DD o null si no aplica"}, "estado": { "type": "string", "enum": ["PENDIENTE", "FIRMADO", "OBSERVADO", "VENCIDO", "N/A"], "description": "Estado actual. 'PENDIENTE' = emitido pero sin firmar; 'FIRMADO' = firmado OK." }, "monto": {"type": "number", "description": "Monto contratado (opcional)"}, "confianza": { "type": "object", "properties": { "tipo": {"type": "number"}, "subcontratista_nombre": {"type": "number"}, "estado": {"type": "number"} } } }, "required": ["tipo", "descripcion", "fecha_emision", "estado"] } }, { "name": "upsert_material", "description": ( "Registrar movimiento de material (recibido en obra o consumido). " "Usar cuando el usuario menciona entrega, recepcion o consumo de material con cantidad." ), "input_schema": { "type": "object", "properties": { "item": {"type": "string", "description": "Nombre del material (ej. 'Hormigon H-21', 'Cemento', 'Varilla 12mm')"}, "unidad": {"type": "string", "description": "Unidad de medida (m3, kg, bolsa, unidad, etc.)"}, "recibido_delta": {"type": "number", "description": "Cantidad recibida (si aplica)"}, "consumido_delta": {"type": "number", "description": "Cantidad consumida (si aplica)"}, "proveedor": {"type": "string", "description": "Proveedor (opcional)"}, "confianza": { "type": "object", "properties": { "item": {"type": "number"}, "unidad": {"type": "number"}, "cantidad": {"type": "number"} } } }, "required": ["item", "unidad"] } }, { "name": "upsert_subcontratista", "description": ( "Agregar un subcontratista/proveedor nuevo a la obra. " "SOLO usar si el nombre NO aparece en la lista de subcontratistas conocidos inyectada. " "Si ya existe, no llamar a esta tool." ), "input_schema": { "type": "object", "properties": { "nombre": {"type": "string"}, "rubro_nombre": { "type": "string", "description": "Rubro al que pertenece (match con rubros activos)" }, "contacto": {"type": "string", "description": "Telefono, email o persona de contacto"}, "confianza": { "type": "object", "properties": {"nombre": {"type": "number"}, "rubro_nombre": {"type": "number"}} } }, "required": ["nombre"] } }, ] # --------------------------------------------------------------------------- # Contexto dinamico de la obra # --------------------------------------------------------------------------- def _fetch_contexto_obra(conn: sqlite3.Connection, codigo: str) -> Optional[dict]: """Query rapido: obra + rubros + subcontratistas + cortes recientes para inyectar al system prompt de Claude.""" obra = conn.execute( "SELECT id, codigo, nombre, cliente, moneda, fecha_corte_actual " "FROM obras WHERE codigo = ?", (codigo,) ).fetchone() if not obra: return None obra_id = obra["id"] rubros = [ {"nombre": r["nombre"], "peso_pct": r["peso_pct"]} for r in conn.execute( "SELECT nombre, peso_pct FROM rubros_obra WHERE obra_id = ? ORDER BY orden", (obra_id,) ).fetchall() ] subs = [ {"nombre": s["nombre"]} for s in conn.execute( "SELECT nombre FROM subcontratistas WHERE obra_id = ? ORDER BY orden", (obra_id,) ).fetchall() ] return { "obra_id": obra_id, "obra_codigo": obra["codigo"], "obra_nombre": obra["nombre"], "obra_cliente": obra["cliente"], "moneda": obra["moneda"], "fecha_corte": obra["fecha_corte_actual"], "rubros": rubros, "subcontratistas": subs, } def _fetch_entry(conn: sqlite3.Connection, entry_id: int) -> Optional[dict]: row = conn.execute( "SELECT id, obra_id, fecha, autor, actividades, observaciones, clima " "FROM daily_log WHERE id = ?", (entry_id,) ).fetchone() if not row: return None return { "id": row["id"], "obra_id": row["obra_id"], "fecha": row["fecha"], "autor": row["autor"], "actividades": row["actividades"] or "", "observaciones": row["observaciones"] or "", "clima": row["clima"] or "", } # --------------------------------------------------------------------------- # System prompt # --------------------------------------------------------------------------- def build_system_prompt(ctx: dict, hoy_iso: str) -> str: """Arma el system prompt con contexto dinamico de la obra + regla de no-adivinar.""" rubros_list = "\n".join(f" - {r['nombre']} (peso {r['peso_pct']}%)" for r in ctx["rubros"]) subs_list = "\n".join(f" - {s['nombre']}" for s in ctx["subcontratistas"]) if not subs_list: subs_list = " (ninguno cargado todavia)" return f"""Eres el parser de bitacora diaria de ERGON, plataforma de inteligencia de obra. Tu tarea: leer el texto libre de una entrada de bitacora (posible transcripcion de voz de un fiscalizador o capataz) y extraer acciones estructuradas para cargar al sistema mediante las tools disponibles. # Contexto de la obra - Obra: {ctx['obra_codigo']} -- {ctx['obra_nombre']} - Cliente: {ctx['obra_cliente']} - Moneda: {ctx['moneda']} - Fecha de corte actual: {ctx['fecha_corte']} - Fecha REAL del sistema HOY: {hoy_iso} # Rubros activos de la obra (usa SOLO estos nombres en rubro_nombre) {rubros_list} # Subcontratistas conocidos (ya cargados) {subs_list} # Reglas criticas 1. **No adivines.** Si un campo no esta claro en el texto, NO lo completes. Si necesitas que lo aclare el usuario, escribe tu pregunta en el mensaje final y NO llames a la tool parcialmente. 2. **Confianza por campo explicita.** En cada tool call, incluye el objeto 'confianza' con un numero 0.0-1.0 por campo clave. 0.9+ = seguro; 0.7-0.9 = probable; <0.7 = dudoso. 3. **Montos requieren moneda explicita**. Cualquier pago o presupuesto debe llevar unidad monetaria explicita en el texto: "guaranies", "gs", "dolares", "USD", etc. Si solo dice "10 millones" / "10M" / "10 mil" SIN que la palabra "guaranies" (o moneda equivalente) aparezca en el texto, NO dispares upsert_presupuesto_mensual y pregunta la moneda, aunque la obra este configurada en PYG. Esta regla protege contra errores de interpretacion. 4. **Fechas relativas**. "hoy" = {hoy_iso}. "ayer" = un dia antes de {hoy_iso}. "el martes" = martes de la semana actual. NUNCA inventes fechas. 5. **Hecho vs plan**. Solo pasado carga datos. "Manana viene el cemento" NO dispara ninguna tool: solo se menciona en el mensaje final como nota. 6. **Rubros**. En 'rubro_nombre' usa EXACTAMENTE uno de los nombres listados arriba. Si el texto menciona un rubro ambiguo (ej. "hormigon" puede ser "Hormigon Armado" o parte de otro), da tu mejor match con confianza <=0.85 y pide confirmacion. 7. **Subcontratistas nuevos**. Solo llama a upsert_subcontratista si el nombre NO aparece en la lista de subcontratistas conocidos. Si aparece, usalo como referencia en otras tools sin crearlo de nuevo. 8. **Duplicados**. Si el texto repite una misma informacion dos veces, dispara la tool UNA sola vez. 9. **Sin datos = sin tools**. Si el texto es conversacional pero no reporta datos concretos (ej. "hoy fue un buen dia, todo normal"), NO llames ninguna tool. Solo escribe un breve resumen en el mensaje final. 10. **Balance cautela vs accion -- esto es critico.** La cautela NO significa sobre-preguntar. Aplica: - Si un campo REQUERIDO esta expresado completamente en el texto (ej. "diez millones de guaranies" = 10.000.000 Gs; "52 personas" = 52; "3000 metros cubicos" = 3000 m3), **dispara la tool con confianza 0.85+. NO pidas confirmacion.** - Solo pide clarificacion cuando: (a) un campo REQUERIDO es realmente ambiguo (ej. "10" sin unidad, "pagamos mucho"), (b) hay conflicto evidente entre dos datos del mismo texto, (c) una unidad critica falta (ej. monto sin moneda implicita cuando la obra acepta varias). - Para campos OPCIONALES que no estan en el texto, **NO preguntes**. Simplemente no los incluyas en la tool call. Ej: si la tool upsert_contrato tiene 'monto' opcional y el texto no lo menciona, dispara la tool sin monto. - Preguntar cuando el dato YA ESTA en el texto genera fricción innecesaria y es peor que adivinar. # Formato de respuesta - Llamar las tools necesarias (puede ser 0, 1 o varias). - Ademas, escribir un mensaje final breve con: - Un resumen de lo que entendiste en 1-2 lineas. - Las preguntas pendientes, si hay campos con baja confianza. - Notas sobre eventos futuros mencionados (si los hay). Prompt version: {PROMPT_VERSION}""" # --------------------------------------------------------------------------- # Orquestador # --------------------------------------------------------------------------- def parse_bitacora( conn: sqlite3.Connection, codigo: str, entry_id: int, api_key: Optional[str] = None, hoy_override: Optional[str] = None, ) -> dict: """Orquesta el parse. Retorna borrador JSON-serializable.""" if not _HAS_SDK: return {"ok": False, "error": "anthropic SDK no instalado"} ctx = _fetch_contexto_obra(conn, codigo) if not ctx: return {"ok": False, "error": f"obra {codigo} no encontrada"} entry = _fetch_entry(conn, entry_id) if not entry: return {"ok": False, "error": f"entry {entry_id} no encontrado"} if entry["obra_id"] != ctx["obra_id"]: return {"ok": False, "error": "entry no pertenece a la obra indicada"} hoy_iso = hoy_override or date.today().isoformat() system_prompt = build_system_prompt(ctx, hoy_iso) # Texto de la bitacora a parsear user_text = ( f"Bitacora del {entry['fecha']}" + (f" (autor: {entry['autor']})" if entry["autor"] else "") + ":\n\n" ) if entry["actividades"]: user_text += f"ACTIVIDADES: {entry['actividades']}\n\n" if entry["observaciones"]: user_text += f"OBSERVACIONES: {entry['observaciones']}\n\n" if entry["clima"]: user_text += f"CLIMA: {entry['clima']}\n" key = api_key or os.environ.get("ANTHROPIC_API_KEY") if not key: return {"ok": False, "error": "ANTHROPIC_API_KEY no configurada"} client = anthropic.Anthropic(api_key=key) try: resp = client.messages.create( model=MODEL, max_tokens=MAX_TOKENS, system=[ # System cacheable: stable a traves de multiples parses {"type": "text", "text": system_prompt, "cache_control": {"type": "ephemeral"}} ], tools=TOOLS, messages=[{"role": "user", "content": user_text}], ) except anthropic.APIError as e: return {"ok": False, "error": f"API error: {e}"} except Exception as e: return {"ok": False, "error": f"Unexpected: {e}"} # Extraer tool_uses + texto libre acciones = [] resumen_partes = [] for block in resp.content: if block.type == "tool_use": acciones.append({ "id": block.id, "tool": block.name, "input": dict(block.input) if block.input else {}, }) elif block.type == "text": if block.text.strip(): resumen_partes.append(block.text.strip()) resumen_modelo = "\n\n".join(resumen_partes) preguntas = _extract_preguntas(resumen_modelo) return { "ok": True, "entry_id": entry_id, "acciones": acciones, "meta": { "modelo": MODEL, "prompt_version": PROMPT_VERSION, "procesado_at": datetime.now(timezone.utc).isoformat(), "usage": { "input_tokens": resp.usage.input_tokens, "output_tokens": resp.usage.output_tokens, "cache_creation_input_tokens": getattr(resp.usage, "cache_creation_input_tokens", 0) or 0, "cache_read_input_tokens": getattr(resp.usage, "cache_read_input_tokens", 0) or 0, }, "texto_original": user_text, "preguntas_pendientes": preguntas, "resumen_modelo": resumen_modelo, "stop_reason": resp.stop_reason, }, } def _extract_preguntas(texto: str) -> list[str]: """Extrae lineas que parecen preguntas (terminan con '?').""" if not texto: return [] preguntas = [] for linea in texto.splitlines(): linea = linea.strip().lstrip("-•").strip() if linea.endswith("?") and len(linea) > 5: preguntas.append(linea) return preguntas # --------------------------------------------------------------------------- # CLI (para pruebas manuales) # --------------------------------------------------------------------------- def _main_cli(): import argparse ap = argparse.ArgumentParser(description="Parse de bitacora ERGON") ap.add_argument("codigo", help="Codigo de obra (ej. TDG1)") ap.add_argument("entry_id", type=int, help="ID de la entrada daily_log") ap.add_argument("--hoy", help="Override fecha de hoy (YYYY-MM-DD)") args = ap.parse_args() from api_obras import get_conn conn = get_conn() r = parse_bitacora(conn, args.codigo, args.entry_id, hoy_override=args.hoy) print(json.dumps(r, indent=2, ensure_ascii=False)) if __name__ == "__main__": _main_cli()