""" ERGON Platform - Backend server ================================ Sirve el dashboard ERGON, expone la API REST de obras, y conecta el agente IA con Claude Sonnet 4.6 para chat streaming SSE + parser de bitacora. USO local: 1. Configurar credenciales en .env.local (gitignored): - ANTHROPIC_API_KEY=sk-ant-... - SUPABASE_URL, SUPABASE_ANON_KEY, SUPABASE_JWT_SECRET (auth bridge) - DATABASE_PATH, BLOBS_DIR, DOCS_DIR (storage) 2. Ejecutar: python servidor_local.py 3. Abrir en navegador: http://localhost:8080/dashboard.html USO production (Railway): Variables de entorno se inyectan por el servicio. PORT se asigna dinamico. El Dockerfile corre `python servidor_local.py` directamente. """ import hashlib import http.server import json import os import sys import urllib.request import urllib.error import urllib.parse import ssl import csv import re import subprocess import tempfile import shutil import uuid import io from datetime import datetime # Cargar .env.local si existe (dev). En production Railway inyecta vars directamente. try: from dotenv import load_dotenv # type: ignore _env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env.local") if os.path.exists(_env_path): load_dotenv(_env_path) except ImportError: pass # python-dotenv opcional; production usa env vars del runtime # Pillow es opcional - si falta, uploads funcionan sin EXIF GPS. try: from PIL import Image, ExifTags # type: ignore _PIL_AVAILABLE = True except ImportError: Image = None # type: ignore ExifTags = None # type: ignore _PIL_AVAILABLE = False # API de obras (sprint 1 F4) - queries SQLite desde db/api_obras.py sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "db")) # F10: generate_dg_civil y db_loader viven en BASE_DIR, que SimpleHTTPRequestHandler # no agrega al sys.path automaticamente. sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) try: import api_obras # type: ignore except ImportError: api_obras = None # Bridge auth Supabase (JWT validation + user sync) try: import supabase_auth # type: ignore except ImportError: supabase_auth = None # type: ignore # Rate limiting (60 req/min default, buckets diferenciados auth/chat/upload) try: import rate_limit # type: ignore except ImportError: rate_limit = None # type: ignore try: import generate_dg_civil # type: ignore # F10: export Excel FCAT1 desde DB import db_loader # type: ignore except ImportError: generate_dg_civil = None # type: ignore db_loader = None # type: ignore PORT = int(os.environ.get("PORT", 8080)) API_URL = "https://api.anthropic.com/v1/messages" BASE_DIR = os.path.dirname(os.path.abspath(__file__)) LOG_FILE = os.path.join(BASE_DIR, "consultas_log.csv") CONVERT_SCRIPT = os.path.join(BASE_DIR, "convert_skp_to_glb.py") BLOBS_DIR = os.path.join(BASE_DIR, "db", "blobs") DOCS_DIR = os.path.join(BASE_DIR, "db", "docs") # F8 sprint 3 MAX_FOTO_BYTES = 20 * 1024 * 1024 # 20MB por foto MAX_DOC_BYTES = 50 * 1024 * 1024 # F8: 50MB por documento (sprint 3) ALLOWED_FOTO_MIME = {"image/jpeg", "image/png", "image/webp"} # Blender path — buscar automaticamente o usar .env def find_blender(): """Buscar ejecutable de Blender (portable o instalado).""" # 1. Variable de entorno env_path = os.environ.get("BLENDER_PATH", "") if env_path and os.path.exists(env_path): return env_path # 2. .env file env_file = os.path.join(BASE_DIR, ".env") if os.path.exists(env_file): with open(env_file, "r") as f: for line in f: if line.strip().startswith("BLENDER_PATH="): p = line.split("=", 1)[1].strip().strip("\"'") if os.path.exists(p): return p # 3. Portable en subcarpeta blender/ portable = os.path.join(BASE_DIR, "blender") if os.path.isdir(portable): for item in os.listdir(portable): candidate = os.path.join(portable, item, "blender.exe") if os.path.exists(candidate): return candidate # 4. PATH del sistema if shutil.which("blender"): return shutil.which("blender") return None BLENDER_EXE = find_blender() # Patron para extraer la clasificacion de la respuesta de Claude CLASIFICACION_PATTERN = re.compile(r"\[CLASIFICACION:(OBRA|AMBIGUA|FUERA)\]") def _parse_multipart(body: bytes, content_type: str) -> dict: """Parser minimal multipart/form-data para reemplazar cgi.FieldStorage (removido en 3.13). Devuelve {name: {'filename': str|None, 'content_type': str|None, 'data': bytes}}. Solo lo que necesitamos: campos simples + un archivo. Sin soporte multipart/mixed ni content-transfer-encoding (no aplica para uploads de browser modernos). """ # Extraer boundary m = re.search(r'boundary="?([^";]+)"?', content_type, re.IGNORECASE) if not m: raise ValueError("Content-Type multipart sin boundary") boundary = m.group(1).encode("utf-8") delim = b"--" + boundary parts: dict = {} segments = body.split(delim) # segments[0] es preamble (o vacio); el ultimo es epilogue o "--\r\n" for seg in segments[1:]: if seg.startswith(b"--"): # closing boundary "--\r\n" break # cada segmento real empieza con \r\n y termina con \r\n if seg.startswith(b"\r\n"): seg = seg[2:] if seg.endswith(b"\r\n"): seg = seg[:-2] header_end = seg.find(b"\r\n\r\n") if header_end < 0: continue raw_headers = seg[:header_end].decode("utf-8", errors="replace") data = seg[header_end + 4:] headers: dict[str, str] = {} for line in raw_headers.split("\r\n"): if ":" in line: k, v = line.split(":", 1) headers[k.strip().lower()] = v.strip() disposition = headers.get("content-disposition", "") name_m = re.search(r'name="([^"]+)"', disposition) if not name_m: continue name = name_m.group(1) filename_m = re.search(r'filename="([^"]*)"', disposition) filename = filename_m.group(1) if filename_m else None ctype = headers.get("content-type") parts[name] = { "filename": filename, "content_type": ctype, "data": data, } return parts def _exif_dms_to_decimal(dms, ref): """Convierte (grados, minutos, segundos) EXIF a decimal. ref='S'/'W' niegan.""" try: deg = float(dms[0]) minu = float(dms[1]) sec = float(dms[2]) except (TypeError, ValueError, IndexError): return None dec = deg + minu / 60.0 + sec / 3600.0 if ref in ("S", "W"): dec = -dec return round(dec, 6) def _apply_demo_watermark_xlsx(xlsx_path: str) -> None: """Aplica marca de agua DEMO al Excel exportado. Implementacion: - Header centrado en cada hoja con texto DEMO + disclaimer - Fondo arena claro en filas 1-2 con texto destacado - No invasivo: deja el resto del contenido intacto Si openpyxl no esta disponible o el archivo no es xlsx valido, salta silenciosamente (log a stdout). """ try: from openpyxl import load_workbook from openpyxl.styles import Font, PatternFill, Alignment except ImportError: print(" [watermark] openpyxl no disponible, salto marca DEMO") return try: wb = load_workbook(xlsx_path) except Exception as e: print(f" [watermark] no se pudo abrir xlsx: {e}") return disclaimer = ( "DEMO - Cuenta de prueba ERGON | " "Datos no contractuales | " "Activa una suscripcion en ergonpy.com/pricing" ) fill = PatternFill(start_color="FFEDE5D7", end_color="FFEDE5D7", fill_type="solid") font = Font(name="IBM Plex Sans", size=9, bold=True, color="FFB68A3E") center = Alignment(horizontal="center", vertical="center", wrap_text=False) for ws in wb.worksheets: # Header impreso (visible al imprimir/PDF) try: ws.oddHeader.center.text = disclaimer ws.oddHeader.center.size = 9 ws.oddHeader.center.color = "B68A3E" except Exception: pass # Banner en fila 1 sobre las primeras columnas try: last_col_letter = ws.cell(row=1, column=max(ws.max_column, 6)).column_letter ws.insert_rows(1) ws.merge_cells(f"A1:{last_col_letter}1") cell = ws["A1"] cell.value = disclaimer cell.fill = fill cell.font = font cell.alignment = center ws.row_dimensions[1].height = 22 except Exception as banner_err: print(f" [watermark] WARN banner en hoja '{ws.title}': {banner_err}") try: wb.save(xlsx_path) except Exception as e: print(f" [watermark] save fallo: {e}") def _extract_exif_gps_fecha(file_bytes): """Devuelve (lat, lon, fecha_YYYY-MM-DD) desde EXIF del blob. Todo None si falla.""" if not _PIL_AVAILABLE: return (None, None, None) try: img = Image.open(io.BytesIO(file_bytes)) exif = img._getexif() if hasattr(img, "_getexif") else None if not exif: return (None, None, None) tag_by_name = {ExifTags.TAGS.get(k, k): v for k, v in exif.items()} # Fecha exif_fecha = None raw_fecha = tag_by_name.get("DateTimeOriginal") or tag_by_name.get("DateTime") if isinstance(raw_fecha, str) and len(raw_fecha) >= 10: # Formato EXIF: "YYYY:MM:DD HH:MM:SS" d = raw_fecha[:10].replace(":", "-") if re.fullmatch(r"\d{4}-\d{2}-\d{2}", d): exif_fecha = d # GPS lat = lon = None gps_raw = tag_by_name.get("GPSInfo") if gps_raw: gps_by_name = {ExifTags.GPSTAGS.get(k, k): v for k, v in gps_raw.items()} if all(k in gps_by_name for k in ("GPSLatitude", "GPSLatitudeRef", "GPSLongitude", "GPSLongitudeRef")): lat = _exif_dms_to_decimal(gps_by_name["GPSLatitude"], gps_by_name["GPSLatitudeRef"]) lon = _exif_dms_to_decimal(gps_by_name["GPSLongitude"], gps_by_name["GPSLongitudeRef"]) return (lat, lon, exif_fecha) except Exception: return (None, None, None) def log_consulta(clasificacion, query, response_text): """Escribe una linea al CSV de log de consultas.""" file_exists = os.path.exists(LOG_FILE) try: with open(LOG_FILE, "a", newline="", encoding="utf-8") as f: writer = csv.writer(f) if not file_exists: writer.writerow(["timestamp", "clasificacion", "consulta", "respuesta_preview"]) writer.writerow([ datetime.now().strftime("%Y-%m-%d %H:%M:%S"), clasificacion, query[:100], response_text[:100] ]) except Exception as e: print(f" [LOG] Error escribiendo log: {e}") # Contexto del proyecto para el agente SYSTEM_PROMPT = """Sos el Agente ERGON, un asistente de inteligencia de obra de ERGON Platform (Paraguay). Estas conectado a los datos del proyecto Obra Demo (cliente: Grupo DG Desarrollo, Asuncion). ALCANCE - LO QUE PODES RESPONDER: - Todo sobre el proyecto Obra Demo: avance, costos, cronograma, subcontratistas, alertas, earned value, proyecciones. - Analisis y recomendaciones de gestion de obra, construccion, ingenieria civil. - Comparaciones entre rubros, contratistas, periodos. - Metodologias de gestion de proyectos (PMBOK, Earned Value, Curva S, ruta critica). - Estrategias para resolver problemas especificos de la obra. - Generar informes y reportes del proyecto. FUERA DE ALCANCE - LO QUE NO DEBES RESPONDER: - Cualquier tema que NO sea sobre la obra o gestion de proyectos de construccion. - No respondas preguntas generales de conocimiento, cultura, historia, politica, deportes, etc. - No escribas codigo, no hagas traducciones, no respondas sobre otros temas tecnologicos. - No respondas preguntas personales ni actues como asistente general. - Si el usuario pregunta algo fuera de alcance, responde exactamente: "Soy el Agente ERGON y estoy especializado en el control de Obra Demo. Puedo ayudarte con avance de obra, costos, contratistas, alertas, cronograma, o generar informes del proyecto. ¿En que aspecto de la obra puedo asistirte?" DATOS ACTUALES DEL PROYECTO: - Avance General: 68% (plan 72%, desvio -4 pts) - Presupuesto Total: Gs. 34,200M - Ejecutado: Gs. 23,250M (68%) - Desvio Presupuestario: +Gs. 2,840M (+8.3%) - SPI (Cronograma): 0.95 (mejorando +0.03) - CPI (Costo): 0.92 (empeorando -0.02) - Personal Promedio/Dia: 47 personas (+5 vs mes anterior) - Dias Perdidos (Clima): 3 de 22 habiles - Contratos Activos: 28 (2 con atraso documental) - Fecha Est. Finalizacion: Jun 2025 (+18 dias vs original May 2025) - Contingencia disponible: Gs. 1,710M (5% PPTO) - Semaforo General: AMARILLO AVANCE POR RUBRO (Plan% -> Real% = Desvio): - Hormigon Armado: 95% -> 92% = -3 pts (Peso 18%, PPTO 6200M, Ejecutado 6620M, desvio +420M) - Albanileria: 80% -> 72% = -8 pts (Peso 8%, PPTO 2800M, Ejecutado 2750M, desvio -50M) **RUTA CRITICA: bloquea pintura** - Inst. Electrica: 65% -> 58% = -7 pts (Peso 12%, PPTO 3100M, Ejecutado 3410M, desvio +310M) - Cielorraso: 55% -> 50% = -5 pts (Peso 3%, PPTO 900M, Ejecutado 870M) - Pintura: 40% -> 38% = -2 pts (Peso 4%, PPTO 1200M, Ejecutado 1150M) - Carp. Aluminio: 65% -> 60% = -5 pts (Peso 10%, PPTO 4500M, Ejecutado 4680M, desvio +180M) - Carp. Madera: 50% -> 48% = -2 pts (Peso 5%, PPTO 1800M, Ejecutado 1760M) - Inst. Sanitaria: 72% -> 70% = -2 pts (Peso 8%, PPTO 2400M, Ejecutado 2380M) - Termomecanica AA: 78% -> 68% = -10 pts (Peso 7%, PPTO 2900M, Ejecutado 3380M, desvio +480M) **CRITICO** - Ascensores: 50% -> 50% = 0 pts (Peso 9%, PPTO 3800M, Ejecutado 3800M) - Pisos y Revestimientos: 55% -> 50% = -5 pts (Peso 6%, PPTO 3200M, Ejecutado 3160M) - Demolicion: 100% -> 100% = 0 pts (Peso 2%, PPTO 680M, Ejecutado 680M) ALERTAS ACTIVAS: 1. ALTA - Termomecanica AA: desvio presupuestario +16.5% (Gs. 480M). Causa: equipos importados mas caros 2. ALTA - Inst. Electrica: atraso -7 pts + sobrecosto +10% (+Gs. 310M). Causa: retrasos ANDE 3. MEDIA - TYM Ingenieria: caida asistencia -29% (de 7.3 a 5.2 personas/dia). Patron: 6>5>4>3 4. MEDIA - Albanileria: atraso -8 pts. Bloquea pintura. Puede correr fecha +1 mes 5. MEDIA - Hormigon Armado: desvio costo +6.8% (+Gs. 420M). Causa: acero +15% 6. BAJA - CPI descendente: 0.96 > 0.94 > 0.92 (2do mes a la baja) TOP SUBCONTRATISTAS (Asistencia Prom/Dia): - AGB Constructora: 12.3 (+1.5) OK - Soletanche Bachy: 8.7 (+0.8) OK - TYM Ingenieria: 5.2 (-2.1) CAIDA - CAVECON SA: 4.8 (-0.3) Leve - Carlos Benitez: 3.5 (+0.5) OK - Aislatec SRL: 3.2 (+1.2) OK - GyC Instalaciones: 4.1 (-0.5) Leve - EcoEnergia SA: 2.8 (+0.3) OK PROYECCIONES EAC: - Optimista (CPI 0.96): Gs. 35,600M (+4.1%) - Probable (CPI 0.92): Gs. 37,170M (+8.7%) - Pesimista (CPI 0.88): Gs. 38,860M (+13.6%) EARNED VALUE: - PV (Planned Value): Gs. 24,620M - EV (Earned Value): Gs. 21,390M - AC (Actual Cost): Gs. 23,250M - SV (Schedule Variance): -Gs. 3,230M - CV (Cost Variance): -Gs. 1,860M INSTRUCCIONES: - Responde SIEMPRE en espanol (sin acentos en caracteres especiales para compatibilidad). - Se conciso y directo. - Usa datos reales del proyecto, no inventes. - Si preguntan algo fuera de los datos disponibles, di que necesitas mas datos del sistema. - Sos un experto en gestion de proyectos de construccion (PMBOK, Earned Value, Curva S). - Cuando des recomendaciones, se especifico y accionable. - Firma siempre al final con "---" seguido de "Agente ERGON" en linea nueva. CLASIFICACION OBLIGATORIA DE CONSULTAS: CADA respuesta que generes DEBE empezar con EXACTAMENTE una de estas etiquetas en la primera linea: - [CLASIFICACION:OBRA] - La consulta esta directamente relacionada con el proyecto Obra Demo, gestion de obra, costos, avance, subcontratistas, cronograma, earned value, o temas de ingenieria civil y construccion. - [CLASIFICACION:AMBIGUA] - La consulta PODRIA estar relacionada pero no es clara. Ejemplos: clima general, preguntas genericas de construccion sin referencia al proyecto, temas que podrian o no aplicar a la obra. - [CLASIFICACION:FUERA] - La consulta NO tiene relacion con el proyecto ni con gestion de obras. Ejemplos: deportes, politica, cultura, programacion, preguntas personales, etc. REGLAS DE CLASIFICACION: - Para [CLASIFICACION:OBRA]: responde normalmente con toda la informacion del proyecto. - Para [CLASIFICACION:AMBIGUA]: NO respondas la consulta. En su lugar, responde SOLAMENTE con: "Esta consulta podria estar fuera del alcance del proyecto. Cada consulta consume recursos del sistema. Si su pregunta esta relacionada con Obra Demo, por favor confirme y le respondo con gusto." Si el usuario CONFIRMA que es sobre el proyecto en un mensaje posterior, clasifica como [CLASIFICACION:OBRA] y responde normalmente. - Para [CLASIFICACION:FUERA]: responde con el mensaje de fuera de alcance: "Soy el Agente ERGON y estoy especializado en el control de Obra Demo. Puedo ayudarte con avance de obra, costos, contratistas, alertas, cronograma, o generar informes del proyecto. En que aspecto de la obra puedo asistirte?" IMPORTANTE: La etiqueta [CLASIFICACION:...] debe ir SIEMPRE en la primera linea, sola, antes de cualquier otro contenido. FORMATO DE RESPUESTA: - Usa **texto** para negritas (cosas importantes, nombres, valores clave). - Usa lineas que empiecen con - para listas de items. - Usa TITULOS EN MAYUSCULAS SEGUIDOS DE : para separar secciones (ej: "RENDIMIENTO ACTUAL:") - Numera los items cuando sea una lista ordenada (1. 2. 3.) - Se estructurado y profesional. INFORMES PDF: - SOLO genera un PDF cuando el usuario use explicitamente la palabra "informe" o "reporte" pidiendo un documento. - Si el usuario solo pregunta o consulta (ej: "como va Soletanche?", "que pasa con el presupuesto?"), responde normalmente SIN ningun marcador de PDF. Es solo una respuesta conversacional. - Cuando SI pida un informe (ej: "haceme un informe de Soletanche", "necesito un reporte de rubros criticos"), genera el contenido detallado y al final agrega: [PDF_CUSTOM:Titulo del Informe] - Para "informe inversor" o "informe ejecutivo general" usa: [PDF_DOWNLOAD:inversor] - Para "reporte de alertas" usa: [PDF_DOWNLOAD:alertas] - IMPORTANTE: Preguntar NO es pedir informe. "Como va TYM?" = respuesta normal. "Haceme un informe de TYM" = respuesta + PDF. """ def get_api_key(): """Get API key from environment or .env file.""" key = os.environ.get("ANTHROPIC_API_KEY", "") if key: return key env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env") if os.path.exists(env_path): with open(env_path, "r") as f: for line in f: line = line.strip() if line.startswith("ANTHROPIC_API_KEY="): return line.split("=", 1)[1].strip().strip("\"'") return "" class DGHandler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__( *args, directory=os.path.dirname(os.path.abspath(__file__)), **kwargs ) def _check_rate_limit(self) -> bool: """Verifica rate limit del request. Si excede, emite 429 y retorna False. Bucket se decide por path (auth/chat/upload/default/static). Static no tiene limite. Configurable via env RATE_LIMIT_*_PER_MIN. """ if rate_limit is None: return True try: rate_limit.check_rate_limit( self.client_address[0] if self.client_address else "", self.headers, self.path, ) return True except rate_limit.RateLimitError as e: body = json.dumps({ "error": f"Rate limit excedido en bucket '{e.bucket}': {e.limit}/min", "retry_after_seconds": e.retry_after, }).encode("utf-8") try: self.send_response(429) self.send_header("Content-Type", "application/json") self.send_header("Retry-After", str(e.retry_after)) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) except Exception: pass return False def do_POST(self): if not self._check_rate_limit(): return parsed = urllib.parse.urlparse(self.path) # F13: rutas auth publicas (login) o dependientes de sesion (logout/register/asignar-obra) if parsed.path == "/api/auth/login": self.handle_auth_login() return if parsed.path == "/api/auth/logout": self.handle_auth_logout() return if parsed.path == "/api/auth/register": self.handle_auth_register() return if parsed.path == "/api/auth/asignar-obra": self.handle_auth_asignar_obra() return if parsed.path == "/api/chat": session = self._require_session() if session is None: return self.handle_chat() return if parsed.path == "/api/chat/stream": session = self._require_session() if session is None: return self.handle_chat_stream() return if parsed.path == "/api/log": session = self._require_session() if session is None: return self.handle_log() return if parsed.path == "/api/convert-skp": session = self._require_session() if session is None: return self.handle_skp_convert() return if parsed.path == "/api/obras": # F7 + F13: crear obra nueva, solo admin session = self._require_session(rol_requerido="admin") if session is None: return self.handle_obra_write(codigo=None, method="POST") return # F9 + F13: /api/obras/:codigo/fotos | /api/obras/:codigo/daily-log parts = [p for p in parsed.path.split("/") if p] if (len(parts) == 4 and parts[0] == "api" and parts[1] == "obras" and parts[3] == "fotos"): session = self._require_session() if session is None: return if not self._require_acceso_obra(session, parts[2]): return self.handle_foto_upload(codigo=parts[2]) return if (len(parts) == 4 and parts[0] == "api" and parts[1] == "obras" and parts[3] == "daily-log"): session = self._require_session() if session is None: return if not self._require_acceso_obra(session, parts[2]): return self.handle_daily_log_write(codigo=parts[2]) return # F8: POST /api/obras/:codigo/documentos (multipart upload) if (len(parts) == 4 and parts[0] == "api" and parts[1] == "obras" and parts[3] == "documentos"): session = self._require_session() if session is None: return if not self._require_acceso_obra(session, parts[2]): return self.handle_documento_upload(codigo=parts[2], session=session) return # F-parser: POST /api/obras/:codigo/daily-log/:id/parse # POST /api/obras/:codigo/daily-log/:id/confirmar # POST /api/obras/:codigo/daily-log/:id/rollback if (len(parts) == 6 and parts[0] == "api" and parts[1] == "obras" and parts[3] == "daily-log" and parts[5] in ("parse", "confirmar", "rollback")): session = self._require_session() if session is None: return if not self._require_acceso_obra(session, parts[2]): return try: entry_id = int(parts[4]) except ValueError: self.send_json(400, {"error": f"entry_id invalido: {parts[4]}"}) return if parts[5] == "parse": self.handle_daily_log_parse(codigo=parts[2], entry_id=entry_id) elif parts[5] == "confirmar": self.handle_daily_log_confirmar(codigo=parts[2], entry_id=entry_id, session=session) else: # rollback self.handle_daily_log_rollback(codigo=parts[2], entry_id=entry_id) return self.send_error(404) def do_DELETE(self): """F8: DELETE /api/obras/:codigo/documentos/:id (admin only).""" if not self._check_rate_limit(): return parsed = urllib.parse.urlparse(self.path) parts = [p for p in parsed.path.split("/") if p] if (len(parts) == 5 and parts[0] == "api" and parts[1] == "obras" and parts[3] == "documentos"): session = self._require_session(rol_requerido="admin") if session is None: return if not self._require_acceso_obra(session, parts[2]): return try: doc_id = int(parts[4]) except ValueError: self.send_json(400, {"error": "doc_id no numerico"}) return self.handle_documento_delete(codigo=parts[2], doc_id=doc_id) return self.send_error(404) def do_PUT(self): # F7 + F13: editar obra existente. Solo admin. if not self._check_rate_limit(): return parsed = urllib.parse.urlparse(self.path) parts = [p for p in parsed.path.split("/") if p] if len(parts) == 3 and parts[0] == "api" and parts[1] == "obras": session = self._require_session(rol_requerido="admin") if session is None: return self.handle_obra_write(codigo=parts[2], method="PUT") else: self.send_error(404) def do_GET(self): # Rutas /api/obras* delegan a api_obras; el resto cae en archivos estaticos. parsed = urllib.parse.urlparse(self.path) # Healthcheck endpoint para Railway / monitoring (siempre OK, sin rate limit) if parsed.path == "/health": self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(b'{"status":"ok","service":"ergonpy-backend"}') return # Rate limit a partir de aca (static GETs caen en bucket sin limite) if not self._check_rate_limit(): return if parsed.path == "/api/auth/me": self.handle_auth_me() return if parsed.path.startswith("/api/obras"): session = self._require_session() if session is None: return self.handle_api_obras(parsed.path, parsed.query, session) return super().do_GET() def do_OPTIONS(self): self.send_response(200) self._write_cors_headers() self.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() # ----------------------------------------------------------------------- # F13 AUTH helpers # ----------------------------------------------------------------------- def _write_cors_headers(self): """CORS permisivo con credentials. Origen echo solo si se envia, evita wildcard+credentials.""" origin = self.headers.get("Origin", "") if origin: self.send_header("Access-Control-Allow-Origin", origin) self.send_header("Access-Control-Allow-Credentials", "true") self.send_header("Vary", "Origin") else: self.send_header("Access-Control-Allow-Origin", "*") def _parse_cookie(self, name: str) -> str: raw = self.headers.get("Cookie", "") or "" for piece in raw.split(";"): piece = piece.strip() if not piece or "=" not in piece: continue k, v = piece.split("=", 1) if k.strip() == name: return v.strip() return "" def _require_session(self, rol_requerido: "str|None" = None) -> "dict|None": """ Valida la sesion del request. Acepta DOS mecanismos en paralelo: 1. Authorization: Bearer (emitido por Supabase Auth, validado HMAC) 2. Cookie ergon_session (bcrypt legacy para admin@demo.local y otros) Si ambos estan presentes, JWT gana (mas reciente, mas confiable). Responde 401/403 y devuelve None si falla. """ if api_obras is None: self.send_json(500, {"error": "api_obras no disponible"}) return None # 1. Probar JWT en Authorization header (Supabase Auth) if supabase_auth is not None: auth_header = self.headers.get("Authorization", "") if auth_header: conn = api_obras.get_conn() try: sb_session = supabase_auth.session_from_authorization_header(conn, auth_header) finally: conn.close() if sb_session is not None: if rol_requerido and sb_session.get("rol") != rol_requerido: self.send_json( 403, {"error": f"Acceso denegado. Se requiere rol '{rol_requerido}'."}, ) return None return sb_session # Si habia Authorization header pero JWT invalido, NO caer en cookie # (evita confusion). Responder 401 y salir. self.send_json(401, {"error": "Token invalido o expirado."}) return None # 2. Fallback cookie ergon_session (auth legacy bcrypt) token = self._parse_cookie("ergon_session") if not token: self.send_json(401, {"error": "No autenticado. Inicie sesion en /login.html."}) return None conn = api_obras.get_conn() try: session = api_obras.get_session(conn, token) finally: conn.close() if session is None: self._clear_session_cookie_via_json( 401, {"error": "Sesion vencida o invalida. Vuelva a iniciar sesion."}, ) return None if rol_requerido and session.get("rol") != rol_requerido: self.send_json( 403, {"error": f"Acceso denegado. Se requiere rol '{rol_requerido}'."}, ) return None return session def _require_acceso_obra(self, session: dict, codigo: str) -> bool: """True si el usuario tiene acceso a la obra. En caso contrario responde 403/404 y devuelve False.""" if not re.fullmatch(r"[A-Z0-9]{3,6}", codigo or ""): self.send_json(400, {"error": "codigo invalido. Formato: A-Z0-9, 3-6 chars"}) return False conn = api_obras.get_conn() try: if api_obras.get_obra(conn, codigo) is None: self.send_json(404, {"error": f"obra {codigo} no encontrada"}) return False if session.get("rol") == "admin": return True ok = api_obras.usuario_tiene_acceso_obra( conn, session["usuario_id"], session["rol"], codigo, ) if not ok: # Mensaje honesto sin filtrar existencia de otras obras. self.send_json( 403, {"error": f"No tiene acceso a la obra {codigo}."}, ) return ok finally: conn.close() def _user_needs_watermark(self, session: "dict|None") -> bool: """True si los exports del usuario deben llevar marca DEMO. Politica: - admin (rol global ERGON) nunca lleva marca - sin session -> sin marca (rutas publicas no llegan aca) - resto: depende de get_trial_status(user)["watermark_required"] """ if session is None: return False if session.get("rol") == "admin": return False if api_obras is None or not hasattr(api_obras, "get_trial_status"): return False conn = api_obras.get_conn() try: user = api_obras.get_usuario(conn, session.get("usuario_id")) if not user: return False status = api_obras.get_trial_status(user) return bool(status.get("watermark_required")) except Exception as e: print(f" [watermark] check fallo: {e}") return False finally: conn.close() def _set_session_cookie(self, token: str, ttl_days: int): max_age = ttl_days * 24 * 3600 cookie = ( f"ergon_session={token}; Path=/; HttpOnly; SameSite=Lax; " f"Max-Age={max_age}" ) self.send_header("Set-Cookie", cookie) def _clear_session_cookie(self): self.send_header( "Set-Cookie", "ergon_session=; Path=/; HttpOnly; SameSite=Lax; Max-Age=0", ) def _read_json_body(self, max_bytes: int = 64 * 1024) -> "dict|None": try: content_length = int(self.headers.get("Content-Length", 0)) except ValueError: self.send_json(400, {"error": "Content-Length invalido"}) return None if content_length <= 0: self.send_json(400, {"error": "Body vacio"}) return None if content_length > max_bytes: self.send_json(400, {"error": f"Body demasiado grande (max {max_bytes} bytes)"}) return None try: raw = self.rfile.read(content_length) return json.loads(raw.decode("utf-8")) except json.JSONDecodeError as e: self.send_json(400, {"error": f"JSON invalido: {e}"}) return None except Exception as e: self.send_json(400, {"error": f"Error leyendo body: {e}"}) return None def handle_auth_login(self): if api_obras is None: self.send_json(500, {"error": "api_obras no disponible"}) return if not api_obras.bcrypt_available(): self.send_json(500, {"error": "bcrypt no instalado en el servidor. pip install bcrypt"}) return payload = self._read_json_body(max_bytes=4096) if payload is None: return email = (payload.get("email") or "").strip().lower() password = payload.get("password") or "" if not email or not password: # Mensaje generico para no exponer cual campo falto. self.send_json(401, {"error": "Email o password invalidos"}) return conn = api_obras.get_conn() try: row = api_obras.get_usuario_by_email(conn, email) # Si el usuario no existe igual ejecutamos verify_password contra un hash dummy # para evitar timing-leak entre "email inexistente" y "email con password mal". # bcrypt.checkpw ya es constant-time internamente sobre el hash dado. if row is None: # Dummy check para nivelar timing. api_obras.verify_password(password, "$2b$12$" + "A" * 53) self.send_json(401, {"error": "Email o password invalidos"}) return if not api_obras.verify_password(password, row["password_hash"]): self.send_json(401, {"error": "Email o password invalidos"}) return session = api_obras.create_session(conn, row["id"]) print(f" [AUTH] login OK usuario_id={row['id']} rol={row['rol']} email={email}") self.send_response(200) self.send_header("Content-Type", "application/json") self._write_cors_headers() self._set_session_cookie(session["token"], session["ttl_days"]) body = json.dumps({ "ok": True, "usuario": { "id": row["id"], "email": row["email"], "nombre": row["nombre"], "rol": row["rol"], }, "expira_at": session["expira_at"], }).encode("utf-8") self.send_header("Content-Length", len(body)) self.end_headers() self.wfile.write(body) except Exception as e: print(f" [AUTH] Error login: {e}") self.send_json(500, {"error": str(e)}) finally: conn.close() def handle_auth_logout(self): token = self._parse_cookie("ergon_session") if api_obras is not None and token: conn = api_obras.get_conn() try: api_obras.delete_session(conn, token) finally: conn.close() self.send_response(200) self.send_header("Content-Type", "application/json") self._write_cors_headers() self._clear_session_cookie() body = json.dumps({"ok": True}).encode("utf-8") self.send_header("Content-Length", len(body)) self.end_headers() self.wfile.write(body) def handle_auth_me(self): if api_obras is None: self.send_json(500, {"error": "api_obras no disponible"}) return token = self._parse_cookie("ergon_session") if not token: self.send_json(401, {"error": "No autenticado"}) return conn = api_obras.get_conn() try: session = api_obras.get_session(conn, token) if session is None: self._clear_session_cookie_via_json(401, {"error": "Sesion vencida"}) return # Trial / tier status para el frontend (banner, modal, watermarks) trial_status = None try: user_full = api_obras.get_usuario(conn, session["usuario_id"]) if user_full and hasattr(api_obras, "get_trial_status"): trial_status = api_obras.get_trial_status(user_full) except Exception: trial_status = None self.send_json(200, { "usuario": { "id": session["usuario_id"], "email": session["email"], "nombre": session["nombre"], "rol": session["rol"], }, "expira_at": session["expira_at"], "trial": trial_status, }) finally: conn.close() def _clear_session_cookie_via_json(self, code: int, data: dict): response = json.dumps(data).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json") self._write_cors_headers() self._clear_session_cookie() self.send_header("Content-Length", len(response)) self.end_headers() self.wfile.write(response) def handle_auth_register(self): session = self._require_session(rol_requerido="admin") if session is None: return payload = self._read_json_body(max_bytes=8192) if payload is None: return conn = api_obras.get_conn() try: result = api_obras.create_usuario(conn, payload) if result.get("ok"): self.send_json(result.get("status", 201), { "ok": True, "usuario": result["usuario"], }) else: self.send_json(result.get("status", 400), {"errors": result.get("errors", ["Error desconocido"])}) finally: conn.close() def handle_auth_asignar_obra(self): session = self._require_session(rol_requerido="admin") if session is None: return payload = self._read_json_body(max_bytes=1024) if payload is None: return usuario_id = payload.get("usuario_id") obra_id = payload.get("obra_id") if not isinstance(usuario_id, int) or not isinstance(obra_id, int): self.send_json(400, {"errors": ["usuario_id y obra_id deben ser enteros."]}) return conn = api_obras.get_conn() try: result = api_obras.assign_usuario_obra(conn, usuario_id, obra_id) status = result.get("status", 201 if result.get("ok") else 400) if result.get("ok"): self.send_json(status, {"ok": True}) else: self.send_json(status, {"errors": result.get("errors", ["Error"])}) finally: conn.close() def handle_obra_write(self, codigo, method: str) -> None: """F7 CRUD PARAMETROS: POST /api/obras (crear) o PUT /api/obras/:codigo (editar).""" if api_obras is None: self.send_json(500, {"error": "api_obras module not available"}) return # Si viene codigo por path (PUT), validar formato regex antes de cualquier DB op. if codigo is not None: if not re.fullmatch(r"[A-Z0-9]{3,6}", codigo): self.send_json(400, {"errors": [ f"codigo invalido '{codigo}'. Formato esperado: 3 a 6 caracteres A-Z o 0-9." ]}) return # Leer body JSON try: content_length = int(self.headers.get("Content-Length", 0)) if content_length <= 0: self.send_json(400, {"errors": ["Body vacio"]}) return if content_length > 5 * 1024 * 1024: # 5MB guard self.send_json(400, {"errors": ["Body demasiado grande"]}) return raw = self.rfile.read(content_length) payload = json.loads(raw.decode("utf-8")) except json.JSONDecodeError as e: self.send_json(400, {"errors": [f"JSON invalido: {e}"]}) return except Exception as e: self.send_json(400, {"errors": [f"Error leyendo body: {e}"]}) return conn = api_obras.get_conn() try: result = api_obras.upsert_obra(conn, payload, codigo_existente=codigo) if result.get("ok"): status = 201 if method == "POST" else 200 obra = api_obras.get_obra(conn, result["codigo"]) self.send_json(status, { "ok": True, "codigo": result["codigo"], "created": result.get("created", False), "obra": obra, }) print(f" [API] {method} /api/obras -> {result['codigo']} ({'created' if result.get('created') else 'updated'})") else: status = result.get("status", 400) self.send_json(status, {"errors": result.get("errors", ["Error desconocido"])}) except Exception as e: print(f" [API] Error {method} /api/obras: {e}") self.send_json(500, {"errors": [str(e)]}) finally: conn.close() def handle_api_obras(self, path: str, query: str, session: dict) -> None: """Router minimo para endpoints GET /api/obras[/codigo[/recurso]]. F13: filtrado por rol aplicado. Admin ve todas; cliente solo las obras asignadas en usuarios_obras (403 si intenta acceder a otras). """ if api_obras is None: self.send_json(500, {"error": "api_obras module not available"}) return # Parse path: /api/obras | /api/obras/{codigo} | /api/obras/{codigo}/{recurso} parts = [p for p in path.split("/") if p] # ["api", "obras", ?, ?] codigo = parts[2] if len(parts) >= 3 else None recurso = parts[3] if len(parts) >= 4 else None qs = urllib.parse.parse_qs(query) if query else {} # Validacion defensiva de codigo: solo A-Z0-9 largo 3-6 (coincide con CHECK de DB) if codigo is not None: if not re.fullmatch(r"[A-Z0-9]{3,6}", codigo): self.send_json(400, {"error": "codigo invalido. Formato: A-Z0-9, 3-6 chars"}) return conn = api_obras.get_conn() try: if codigo is None: # GET /api/obras: listar filtrado por rol todas = api_obras.list_obras(conn) visibles = api_obras.filter_obras_por_usuario( todas, usuario_id=session["usuario_id"], rol=session["rol"], conn=conn, ) self.send_json(200, visibles) return # Obra debe existir para recursos anidados obra = api_obras.get_obra(conn, codigo) if obra is None: self.send_json(404, {"error": f"obra {codigo} no encontrada"}) return # F13: cliente solo accede a sus obras asignadas if session["rol"] != "admin": ok = api_obras.usuario_tiene_acceso_obra( conn, session["usuario_id"], session["rol"], codigo, ) if not ok: self.send_json(403, {"error": f"No tiene acceso a la obra {codigo}."}) return if recurso is None: self.send_json(200, obra) return # F9: sub-ruta fotos/{id}/blob sirve el archivo binario if recurso == "fotos" and len(parts) == 6 and parts[5] == "blob": try: foto_id = int(parts[4]) except ValueError: self.send_json(400, {"error": "foto_id no numerico"}) return self.serve_foto_blob(conn, codigo, foto_id) return # F10: export Excel FCAT1 desde DB -> /api/obras/:codigo/export/excel if recurso == "export" and len(parts) == 5 and parts[4] == "excel": self.serve_export_excel(codigo, session=session) return # F-parser: GET /api/obras/:codigo/daily-log/:id/borrador # Retorna el borrador almacenado sin re-llamar a Claude. if (recurso == "daily-log" and len(parts) == 6 and parts[5] == "borrador"): try: entry_id = int(parts[4]) except ValueError: self.send_json(400, {"error": "entry_id no numerico"}) return self.handle_daily_log_get_borrador(conn=conn, codigo=codigo, entry_id=entry_id) return # F8: sub-rutas documentos if recurso == "documentos": if len(parts) == 4: # GET /api/obras/:codigo/documentos (lista con filtros) docs = api_obras.list_documentos( conn, codigo, categoria=(qs.get("categoria") or [None])[0], tag=(qs.get("tag") or [None])[0], desde=(qs.get("desde") or [None])[0], hasta=(qs.get("hasta") or [None])[0], ) self.send_json(200, docs) return if len(parts) >= 5: try: doc_id = int(parts[4]) except ValueError: self.send_json(400, {"error": "doc_id no numerico"}) return if len(parts) == 5: # GET /api/obras/:codigo/documentos/:id (metadata) doc = api_obras.get_documento(conn, codigo, doc_id) if doc is None: self.send_json(404, {"error": "documento no encontrado"}) return self.send_json(200, doc) return if len(parts) == 6 and parts[5] == "blob": self.serve_documento_blob(conn, codigo, doc_id) return self.send_json(404, {"error": "ruta de documentos invalida"}) return dispatch = { "rubros": lambda: api_obras.get_rubros(conn, codigo), "subcontratistas": lambda: api_obras.get_subcontratistas(conn, codigo), "avance": lambda: api_obras.get_avance(conn, codigo), "presupuesto": lambda: api_obras.get_presupuesto(conn, codigo), "evm": lambda: api_obras.get_evm(conn, codigo), "alertas": lambda: api_obras.get_alertas(conn, codigo), "asistencia": lambda: api_obras.get_asistencia( conn, codigo, desde=(qs.get("desde") or [None])[0], hasta=(qs.get("hasta") or [None])[0]), "contratos": lambda: api_obras.get_contratos(conn, codigo), "materiales": lambda: api_obras.get_materiales(conn, codigo), "fotos": lambda: api_obras.list_fotos( conn, codigo, desde=(qs.get("desde") or [None])[0], hasta=(qs.get("hasta") or [None])[0], sector=(qs.get("sector") or [None])[0]), "daily-log": lambda: api_obras.list_daily_log( conn, codigo, desde=(qs.get("desde") or [None])[0], hasta=(qs.get("hasta") or [None])[0]), } handler = dispatch.get(recurso) if handler is None: self.send_json(404, {"error": f"recurso desconocido: {recurso}"}) return self.send_json(200, handler()) except Exception as e: print(f" [API] Error {path}: {e}") self.send_json(500, {"error": str(e)}) finally: conn.close() def handle_foto_upload(self, codigo: str) -> None: """F9 POST /api/obras/:codigo/fotos (multipart/form-data). Campos esperados: fecha (YYYY-MM-DD obligatoria), sector, piso, nota, usuario. Archivo: campo 'foto' (JPEG/PNG/WebP, <= 20MB). """ if api_obras is None: self.send_json(500, {"errors": ["api_obras module not available"]}) return if not re.fullmatch(r"[A-Z0-9]{3,6}", codigo): self.send_json(400, {"errors": [ f"codigo invalido '{codigo}'. Formato esperado: 3 a 6 caracteres A-Z o 0-9." ]}) return ctype = self.headers.get("Content-Type", "") if "multipart/form-data" not in ctype.lower(): self.send_json(400, {"errors": [ "Content-Type debe ser multipart/form-data con el archivo en el campo 'foto'." ]}) return try: content_length = int(self.headers.get("Content-Length", 0)) except ValueError: self.send_json(400, {"errors": ["Content-Length invalido"]}) return if content_length <= 0: self.send_json(400, {"errors": ["Body vacio"]}) return if content_length > MAX_FOTO_BYTES + 4096: # margen para headers multipart # Drenar body para que el cliente reciba el 400 sin ConnectionAbortedError. remaining = content_length while remaining > 0: chunk = self.rfile.read(min(65536, remaining)) if not chunk: break remaining -= len(chunk) self.send_json(400, {"errors": [ f"Archivo demasiado grande (max {MAX_FOTO_BYTES // (1024*1024)}MB). " f"Recibidos {content_length} bytes." ]}) return try: body = self.rfile.read(content_length) parts = _parse_multipart(body, ctype) except ValueError as e: self.send_json(400, {"errors": [str(e)]}) return except Exception as e: self.send_json(400, {"errors": [f"Error parseando multipart: {e}"]}) return if "foto" not in parts or not parts["foto"].get("filename"): self.send_json(400, {"errors": [ "Falta el archivo en el campo 'foto' del form-data (o filename vacio)." ]}) return foto_part = parts["foto"] file_bytes = foto_part["data"] if not file_bytes: self.send_json(400, {"errors": ["Archivo vacio"]}) return if len(file_bytes) > MAX_FOTO_BYTES: self.send_json(400, {"errors": [ f"Archivo de {len(file_bytes)} bytes excede el maximo " f"({MAX_FOTO_BYTES // (1024*1024)}MB)." ]}) return declared_mime = (foto_part.get("content_type") or "").lower() if declared_mime and declared_mime not in ALLOWED_FOTO_MIME: self.send_json(400, {"errors": [ f"Mime type '{declared_mime}' no soportado. " f"Soportados: {', '.join(sorted(ALLOWED_FOTO_MIME))}." ]}) return mime_type = declared_mime or "image/jpeg" def _text(field): p = parts.get(field) if not p: return "" return p["data"].decode("utf-8", errors="replace").strip() meta = { "fecha": _text("fecha"), "sector": _text("sector"), "piso": _text("piso"), "nota": _text("nota"), "usuario": _text("usuario"), "mime_type": mime_type, "size_bytes": len(file_bytes), } # EXIF defensivo: si Pillow falla, no aborta el upload. lat, lon, exif_fecha = _extract_exif_gps_fecha(file_bytes) meta["exif_gps_lat"] = lat meta["exif_gps_lon"] = lon meta["exif_fecha"] = exif_fecha # Resolve obra_id para el path del blob (sin tocar DB todavia). conn = api_obras.get_conn() blob_path_rel = None blob_path_abs = None try: obra = api_obras.get_obra(conn, codigo) if obra is None: self.send_json(404, {"errors": [f"obra {codigo} no encontrada"]}) return obra_id = obra["id"] # Escribir blob a disco antes del insert DB (si falla DB, limpiamos). target_dir = os.path.join(BLOBS_DIR, str(obra_id)) try: os.makedirs(target_dir, exist_ok=True) except OSError as e: self.send_json(500, {"errors": [ f"No se pudo crear directorio de blobs ({target_dir}): {e}" ]}) return ext = {"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp"}.get(mime_type, ".bin") blob_name = f"{uuid.uuid4().hex}{ext}" blob_path_abs = os.path.join(target_dir, blob_name) blob_path_rel = os.path.relpath(blob_path_abs, BASE_DIR).replace("\\", "/") try: with open(blob_path_abs, "wb") as f: f.write(file_bytes) except OSError as e: self.send_json(500, {"errors": [f"Error guardando archivo: {e}"]}) return result = api_obras.create_foto(conn, codigo, meta, blob_path_rel) if not result.get("ok"): # Rollback blob en disco try: os.remove(blob_path_abs) except OSError: pass self.send_json(result.get("status", 400), {"errors": result.get("errors", [])}) return print(f" [API] POST /api/obras/{codigo}/fotos -> id={result['foto']['id']}, " f"{len(file_bytes)} bytes, exif={'si' if lat else 'no'}") self.send_json(201, {"ok": True, "foto": result["foto"]}) except Exception as e: # Cleanup defensivo if blob_path_abs and os.path.exists(blob_path_abs): try: os.remove(blob_path_abs) except OSError: pass print(f" [API] Error foto upload: {e}") self.send_json(500, {"errors": [str(e)]}) finally: conn.close() def handle_daily_log_write(self, codigo: str) -> None: """F9 POST /api/obras/:codigo/daily-log (JSON).""" if api_obras is None: self.send_json(500, {"errors": ["api_obras module not available"]}) return if not re.fullmatch(r"[A-Z0-9]{3,6}", codigo): self.send_json(400, {"errors": [ f"codigo invalido '{codigo}'. Formato esperado: 3 a 6 caracteres A-Z o 0-9." ]}) return try: content_length = int(self.headers.get("Content-Length", 0)) if content_length <= 0: self.send_json(400, {"errors": ["Body vacio"]}) return if content_length > 256 * 1024: # daily log es texto corto self.send_json(400, {"errors": ["Body demasiado grande (max 256KB)"]}) return raw = self.rfile.read(content_length) payload = json.loads(raw.decode("utf-8")) except json.JSONDecodeError as e: self.send_json(400, {"errors": [f"JSON invalido: {e}"]}) return except Exception as e: self.send_json(400, {"errors": [f"Error leyendo body: {e}"]}) return conn = api_obras.get_conn() try: result = api_obras.create_daily_log(conn, codigo, payload) if result.get("ok"): print(f" [API] POST /api/obras/{codigo}/daily-log -> id={result['entry']['id']}") self.send_json(201, {"ok": True, "entry": result["entry"]}) else: self.send_json(result.get("status", 400), {"errors": result.get("errors", [])}) except Exception as e: print(f" [API] Error daily-log: {e}") self.send_json(500, {"errors": [str(e)]}) finally: conn.close() # ----------------------------------------------------------------------- # F-parser BITACORA IA (sprint 2026-04-23) # ----------------------------------------------------------------------- def handle_daily_log_parse(self, codigo: str, entry_id: int) -> None: """POST /api/obras/:codigo/daily-log/:id/parse. Corre el parser de bitacora sobre el entry y guarda el borrador en la DB. Devuelve el borrador para que el frontend lo presente al usuario. """ if api_obras is None: self.send_json(500, {"errors": ["api_obras no disponible"]}) return # Import diferido para evitar cargar anthropic SDK al arrancar el servidor try: from db.parse_dispatcher import trigger_parse # type: ignore except ImportError: try: import sys as _sys from pathlib import Path as _Path _sys.path.insert(0, str(_Path(__file__).resolve().parent / "db")) from parse_dispatcher import trigger_parse # type: ignore except ImportError as ie: self.send_json(500, {"errors": [f"parse_dispatcher no disponible: {ie}"]}) return conn = api_obras.get_conn() try: result = trigger_parse(conn, codigo, entry_id) status = result.get("status", 200 if result.get("ok") else 500) if result.get("ok"): borrador = result["borrador"] n_acciones = len(borrador.get("acciones", [])) print(f" [API] POST /daily-log/{entry_id}/parse -> {n_acciones} acciones") self.send_json(status, {"ok": True, "borrador": borrador}) else: self.send_json(status, {"ok": False, "error": result.get("error")}) except Exception as e: print(f" [API] Error parse: {e}") self.send_json(500, {"errors": [str(e)]}) finally: conn.close() def handle_daily_log_confirmar(self, codigo: str, entry_id: int, session: dict) -> None: """POST /api/obras/:codigo/daily-log/:id/confirmar. Aplica transaccionalmente las acciones aprobadas (posiblemente editadas) a las tablas destino (asistencia, avance, presupuesto, etc.). Body esperado: {"acciones": [{"tool": "upsert_asistencia", "input": {...}}, ...]} """ if api_obras is None: self.send_json(500, {"errors": ["api_obras no disponible"]}) return try: from db.parse_dispatcher import confirmar_acciones # type: ignore except ImportError: try: import sys as _sys from pathlib import Path as _Path _sys.path.insert(0, str(_Path(__file__).resolve().parent / "db")) from parse_dispatcher import confirmar_acciones # type: ignore except ImportError as ie: self.send_json(500, {"errors": [f"parse_dispatcher no disponible: {ie}"]}) return try: content_length = int(self.headers.get("Content-Length", 0)) if content_length <= 0: self.send_json(400, {"errors": ["Body vacio"]}) return if content_length > 512 * 1024: self.send_json(400, {"errors": ["Body demasiado grande (max 512KB)"]}) return raw = self.rfile.read(content_length) payload = json.loads(raw.decode("utf-8")) except json.JSONDecodeError as e: self.send_json(400, {"errors": [f"JSON invalido: {e}"]}) return except Exception as e: self.send_json(400, {"errors": [f"Error leyendo body: {e}"]}) return acciones = payload.get("acciones", []) usuario = (session or {}).get("nombre") or (session or {}).get("usuario") or "desconocido" conn = api_obras.get_conn() try: result = confirmar_acciones(conn, codigo, entry_id, acciones, usuario) status = result.get("status", 200 if result.get("ok") else 400) if result.get("ok"): print(f" [API] POST /daily-log/{entry_id}/confirmar -> " f"{result.get('count', 0)} filas afectadas") self.send_json(status, { "ok": True, "count": result.get("count", 0), "diff": result.get("diff", []), }) else: self.send_json(status, {"ok": False, "errors": result.get("errors", [])}) except Exception as e: print(f" [API] Error confirmar: {e}") self.send_json(500, {"errors": [str(e)]}) finally: conn.close() def handle_daily_log_rollback(self, codigo: str, entry_id: int) -> None: """POST /api/obras/:codigo/daily-log/:id/rollback. Revierte los INSERTs aplicados en el ultimo confirm. Ventana ~5 min controlada por el frontend (backend no valida tiempo).""" if api_obras is None: self.send_json(500, {"errors": ["api_obras no disponible"]}) return try: from db.parse_dispatcher import rollback_acciones # type: ignore except ImportError: try: import sys as _sys from pathlib import Path as _Path _sys.path.insert(0, str(_Path(__file__).resolve().parent / "db")) from parse_dispatcher import rollback_acciones # type: ignore except ImportError as ie: self.send_json(500, {"errors": [f"parse_dispatcher no disponible: {ie}"]}) return conn = api_obras.get_conn() try: result = rollback_acciones(conn, codigo, entry_id) status = result.get("status", 200 if result.get("ok") else 400) if result.get("ok"): print(f" [API] POST /daily-log/{entry_id}/rollback -> " f"{result.get('count_reverted', 0)} revertidas, " f"{result.get('count_skipped', 0)} skipped") self.send_json(status, { "ok": True, "reverted": result.get("reverted", []), "skipped": result.get("skipped", []), "count_reverted": result.get("count_reverted", 0), "count_skipped": result.get("count_skipped", 0), }) else: errors = result.get("errors") or [result.get("error", "rollback fallo")] self.send_json(status, {"ok": False, "errors": errors}) except Exception as e: print(f" [API] Error rollback: {e}") self.send_json(500, {"errors": [str(e)]}) finally: conn.close() def handle_daily_log_get_borrador(self, conn, codigo: str, entry_id: int) -> None: """GET /api/obras/:codigo/daily-log/:id/borrador. Devuelve el borrador almacenado (tras un reload) sin re-llamar a Claude. La conexion la provee do_GET -- no la cerramos aca.""" try: from db.parse_dispatcher import get_borrador # type: ignore except ImportError: try: import sys as _sys from pathlib import Path as _Path _sys.path.insert(0, str(_Path(__file__).resolve().parent / "db")) from parse_dispatcher import get_borrador # type: ignore except ImportError as ie: self.send_json(500, {"errors": [f"parse_dispatcher no disponible: {ie}"]}) return try: result = get_borrador(conn, codigo, entry_id) status = result.get("status", 200 if result.get("ok") else 500) if result.get("ok"): self.send_json(status, {"ok": True, "borrador": result["borrador"]}) else: self.send_json(status, {"ok": False, "error": result.get("error")}) except Exception as e: print(f" [API] Error get borrador: {e}") self.send_json(500, {"errors": [str(e)]}) # ----------------------------------------------------------------------- # F8 DOCUMENTOS (sprint 3, 2026-04-20) # ----------------------------------------------------------------------- def handle_documento_upload(self, codigo: str, session: dict) -> None: """POST /api/obras/:codigo/documentos (multipart). Campos del form: categoria (enum 8), titulo, tags (CSV), fecha_doc (YYYY-MM-DD opt), notas (opt), archivo (campo 'archivo' con file_bytes). Storage: `db/docs/{obra_id}/{uuid}.{ext}`. Dedup via sha256 informativo (no rechaza, solo reporta duplicados previos). """ if api_obras is None: self.send_json(500, {"errors": ["api_obras no disponible"]}) return ctype = self.headers.get("Content-Type", "") if "multipart/form-data" not in ctype.lower(): self.send_json(400, {"errors": [ "Content-Type debe ser multipart/form-data con el archivo en el campo 'archivo'." ]}) return try: content_length = int(self.headers.get("Content-Length", 0)) except ValueError: self.send_json(400, {"errors": ["Content-Length invalido"]}) return if content_length <= 0: self.send_json(400, {"errors": ["Body vacio"]}) return if content_length > MAX_DOC_BYTES + 4096: # margen multipart # Drenar antes del 413 para que el cliente reciba la respuesta limpia. remaining = content_length while remaining > 0: chunk = self.rfile.read(min(65536, remaining)) if not chunk: break remaining -= len(chunk) self.send_json(413, {"errors": [ f"Archivo demasiado grande (max {MAX_DOC_BYTES // (1024*1024)}MB). " f"Recibidos {content_length} bytes." ]}) return try: body = self.rfile.read(content_length) parts = _parse_multipart(body, ctype) except ValueError as e: self.send_json(400, {"errors": [str(e)]}) return except Exception as e: self.send_json(400, {"errors": [f"Error parseando multipart: {e}"]}) return if "archivo" not in parts or not parts["archivo"].get("filename"): self.send_json(400, {"errors": [ "Falta el archivo en el campo 'archivo' del form-data (o filename vacio)." ]}) return archivo_part = parts["archivo"] file_bytes = archivo_part["data"] if not file_bytes: self.send_json(400, {"errors": ["Archivo vacio"]}) return if len(file_bytes) > MAX_DOC_BYTES: self.send_json(413, {"errors": [ f"Archivo de {len(file_bytes)} bytes excede el maximo " f"({MAX_DOC_BYTES // (1024*1024)}MB)." ]}) return filename = archivo_part.get("filename") or "documento.bin" _root, ext = os.path.splitext(filename) ext = ext.lower()[:10] or ".bin" # capar longitud para evitar path abuse # Solo chars seguros en la extension if not re.fullmatch(r"\.[A-Za-z0-9]{1,10}", ext): ext = ".bin" mime_type = (archivo_part.get("content_type") or "application/octet-stream").lower() sha256_hex = hashlib.sha256(file_bytes).hexdigest() def _text(field): p = parts.get(field) if not p: return "" return p["data"].decode("utf-8", errors="replace").strip() meta = { "categoria": _text("categoria").lower(), "titulo": _text("titulo"), "tags": _text("tags"), "fecha_doc": _text("fecha_doc"), "notas": _text("notas"), "subido_por": session.get("email"), "mime_type": mime_type, "size_bytes": len(file_bytes), "sha256": sha256_hex, } conn = api_obras.get_conn() blob_path_abs = None blob_path_rel = None try: obra = api_obras.get_obra(conn, codigo) if obra is None: self.send_json(404, {"errors": [f"obra {codigo} no encontrada"]}) return obra_id = obra["id"] # Guardar blob a disco ANTES del insert DB (si falla DB, borramos blob) target_dir = os.path.join(DOCS_DIR, str(obra_id)) try: os.makedirs(target_dir, exist_ok=True) except OSError as e: self.send_json(500, {"errors": [ f"No se pudo crear directorio ({target_dir}): {e}" ]}) return blob_name = f"{uuid.uuid4().hex}{ext}" blob_path_abs = os.path.join(target_dir, blob_name) blob_path_rel = os.path.relpath(blob_path_abs, BASE_DIR).replace("\\", "/") try: with open(blob_path_abs, "wb") as f: f.write(file_bytes) except OSError as e: self.send_json(500, {"errors": [f"Error guardando archivo: {e}"]}) return result = api_obras.create_documento(conn, codigo, meta, blob_path_rel) if not result.get("ok"): # Rollback blob en disco try: os.remove(blob_path_abs) except OSError: pass self.send_json(result.get("status", 400), {"errors": result.get("errors", [])}) return payload = { "ok": True, "documento": result["documento"], } if result.get("duplicados"): payload["dedup"] = { "warning": (f"Ya hay {len(result['duplicados'])} documento(s) " "con el mismo contenido (sha256) en esta obra."), "duplicados": [ {"id": d["id"], "titulo": d["titulo"], "categoria": d["categoria"], "created_at": d["created_at"]} for d in result["duplicados"] ], } print(f" [API] POST /api/obras/{codigo}/documentos -> id={result['documento']['id']} " f"({len(file_bytes)} bytes, sha={sha256_hex[:8]}..., " f"dedup={'si' if result.get('duplicados') else 'no'})") self.send_json(201, payload) except Exception as e: if blob_path_abs and os.path.exists(blob_path_abs): try: os.remove(blob_path_abs) except OSError: pass print(f" [API] Error documento upload {codigo}: {e}") self.send_json(500, {"errors": [str(e)]}) finally: conn.close() def handle_documento_delete(self, codigo: str, doc_id: int) -> None: """DELETE /api/obras/:codigo/documentos/:id (admin only). Borra DB + blob.""" if api_obras is None: self.send_json(500, {"error": "api_obras no disponible"}) return conn = api_obras.get_conn() try: result = api_obras.delete_documento(conn, codigo, doc_id) if not result.get("ok"): self.send_json(result.get("status", 400), {"errors": result.get("errors", ["Error"])}) return blob_rel = result.get("blob_path") or "" blob_abs = os.path.normpath(os.path.join(BASE_DIR, blob_rel)) # Guard contra path traversal: solo borramos si esta dentro de DOCS_DIR if blob_abs.startswith(os.path.normpath(DOCS_DIR)) and os.path.exists(blob_abs): try: os.remove(blob_abs) except OSError as e: print(f" [API] WARN no pude borrar blob {blob_abs}: {e}") self.send_response(204) self._write_cors_headers() self.send_header("Content-Length", "0") self.end_headers() print(f" [API] DELETE /api/obras/{codigo}/documentos/{doc_id} OK") finally: conn.close() def serve_documento_blob(self, conn, codigo: str, doc_id: int) -> None: """GET /api/obras/:codigo/documentos/:id/blob -> binario.""" doc = api_obras.get_documento(conn, codigo, doc_id) if doc is None: self.send_json(404, {"error": "documento no encontrado"}) return blob_rel = doc.get("blob_path") or "" blob_abs = os.path.normpath(os.path.join(BASE_DIR, blob_rel)) # Guard path traversal if not blob_abs.startswith(os.path.normpath(DOCS_DIR)): self.send_json(500, {"error": "blob_path invalido"}) return if not os.path.exists(blob_abs): self.send_json(404, {"error": "archivo de documento no existe en disco"}) return try: with open(blob_abs, "rb") as f: data = f.read() except OSError as e: self.send_json(500, {"error": f"Error leyendo blob: {e}"}) return # Deducir nombre de descarga a partir de titulo + extension del blob titulo = doc.get("titulo") or "documento" safe_titulo = re.sub(r'[\\/:*?"<>|]+', "_", titulo) _r, blob_ext = os.path.splitext(blob_abs) fname = f"{safe_titulo}{blob_ext or ''}" mime = doc.get("mime_type") or "application/octet-stream" self.send_response(200) self.send_header("Content-Type", mime) self.send_header( "Content-Disposition", f'attachment; filename="{fname}"', ) self._write_cors_headers() self.send_header("Content-Length", len(data)) self.end_headers() self.wfile.write(data) def serve_export_excel(self, codigo: str, session: dict | None = None) -> None: """F10 GET /api/obras/:codigo/export/excel -> xlsx attachment. Genera el Excel FCAT1 al vuelo desde la data de la obra en ergon.db usando `generate_dg_civil.generate()` con `db_loader.load_cfg_from_db`. Si el usuario esta en periodo de prueba o el plan exige marca de agua, el archivo lleva header "DEMO - Cuenta de prueba ERGON" en todas las hojas. """ if generate_dg_civil is None or db_loader is None: self.send_json(500, {"error": ( "modulo generate_dg_civil o db_loader no disponible. " "Verifique que ambos esten en el mismo directorio que servidor_local.py." )}) return tmp_dir = None try: cfg = generate_dg_civil.build_cfg(obra_codigo=codigo) # Nombre estable para Content-Disposition (sin depender de que cfg lo calcule). corte = cfg.get("corte") corte_str = corte.isoformat() if hasattr(corte, "isoformat") else "corte" fname = f"{codigo}_Corte_{corte_str}.xlsx" tmp_dir = tempfile.mkdtemp(prefix="dg_excel_") out_path = os.path.join(tmp_dir, fname) generate_dg_civil.generate(cfg, out_path) if not os.path.exists(out_path): self.send_json(500, {"error": "Generacion fallo: archivo no creado"}) return # Marca DEMO si corresponde try: if self._user_needs_watermark(session): _apply_demo_watermark_xlsx(out_path) fname = "DEMO_" + fname except Exception as wm_err: print(f" [API] WARN watermark falla (continuando sin marca): {wm_err}") with open(out_path, "rb") as f: data = f.read() self.send_response(200) self.send_header( "Content-Type", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) self.send_header( "Content-Disposition", f'attachment; filename="{fname}"', ) self._write_cors_headers() self.send_header("Content-Length", len(data)) self.end_headers() self.wfile.write(data) print(f" [API] GET /api/obras/{codigo}/export/excel -> {fname} ({len(data)} bytes)") except db_loader.ObraNotFound as e: self.send_json(404, {"error": str(e)}) except Exception as e: print(f" [API] Error export excel {codigo}: {e}") self.send_json(500, {"error": f"Error generando Excel: {e}"}) finally: if tmp_dir and os.path.exists(tmp_dir): try: shutil.rmtree(tmp_dir, ignore_errors=True) except OSError: pass def serve_foto_blob(self, conn, codigo: str, foto_id: int) -> None: """F9 GET /api/obras/:codigo/fotos/:id/blob -> binario imagen.""" foto = api_obras.get_foto(conn, codigo, foto_id) if foto is None: self.send_json(404, {"error": "foto no encontrada"}) return blob_rel = foto.get("blob_path") or "" blob_abs = os.path.normpath(os.path.join(BASE_DIR, blob_rel)) # Guard contra path traversal: el archivo debe vivir dentro de BLOBS_DIR. if not blob_abs.startswith(os.path.normpath(BLOBS_DIR)): self.send_json(500, {"error": "blob_path invalido"}) return if not os.path.exists(blob_abs): self.send_json(404, {"error": "archivo de foto no existe en disco"}) return try: with open(blob_abs, "rb") as f: data = f.read() except OSError as e: self.send_json(500, {"error": f"Error leyendo blob: {e}"}) return mime = foto.get("mime_type") or "image/jpeg" self.send_response(200) self.send_header("Content-Type", mime) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Cache-Control", "public, max-age=3600") self.send_header("Content-Length", len(data)) self.end_headers() self.wfile.write(data) def handle_chat(self): api_key = get_api_key() if not api_key: self.send_json(500, { "error": "ANTHROPIC_API_KEY no configurada. Ejecute: set ANTHROPIC_API_KEY=sk-ant-..." }) return content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body) messages = data.get("messages", []) payload = json.dumps({ "model": "claude-sonnet-4-20250514", "max_tokens": 2048, "system": SYSTEM_PROMPT, "messages": messages }).encode("utf-8") req = urllib.request.Request( API_URL, data=payload, headers={ "Content-Type": "application/json", "x-api-key": api_key, "anthropic-version": "2023-06-01" }, method="POST" ) # Extraer la consulta del usuario para logging user_query = "" for msg in reversed(messages): if msg.get("role") == "user": user_query = msg.get("content", "") if isinstance(user_query, list): user_query = " ".join( p.get("text", "") for p in user_query if p.get("type") == "text" ) break try: ctx = ssl.create_default_context() with urllib.request.urlopen(req, context=ctx, timeout=60) as resp: result = json.loads(resp.read().decode("utf-8")) text = result.get("content", [{}])[0].get("text", "Sin respuesta") # Parsear y extraer la clasificacion clasificacion = "OBRA" # default si Claude no incluyo tag match = CLASIFICACION_PATTERN.search(text) if match: clasificacion = match.group(1) # Remover la etiqueta de clasificacion antes de enviar al frontend clean_text = CLASIFICACION_PATTERN.sub("", text).strip() # Limpiar salto de linea extra al inicio si quedo clean_text = re.sub(r"^\n+", "", clean_text) # Log de la consulta log_consulta(clasificacion, user_query, clean_text) print(f" [CLAS] {clasificacion} | {user_query[:60]}") self.send_json(200, { "response": clean_text, "clasificacion": clasificacion }) except urllib.error.HTTPError as e: error_body = e.read().decode("utf-8", errors="replace") print(f"API Error {e.code}: {error_body}") self.send_json(e.code, {"error": f"Error API: {error_body}"}) except Exception as e: print(f"Error: {e}") self.send_json(500, {"error": str(e)}) def handle_chat_stream(self): """ Streaming del chat (SSE al cliente; stream=true a Anthropic). Lee upstream por lineas, detecta 'data: {...}' de content_block_delta, reenvia al cliente como eventos SSE propios: event: delta -> {"text": "..."} event: done -> {"clasificacion": "...", "clean_text": "..."} event: error -> {"error": "..."} El cliente acumula los delta en pantalla; al done, reemplaza con clean_text (sin el tag CLASIFICACION) y lo pushea a conversationHistory. """ api_key = get_api_key() if not api_key: self.send_json(500, {"error": "ANTHROPIC_API_KEY no configurada"}) return content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) try: data = json.loads(body) except json.JSONDecodeError: self.send_json(400, {"error": "JSON invalido"}) return messages = data.get("messages", []) payload = json.dumps({ "model": "claude-sonnet-4-20250514", "max_tokens": 2048, "system": SYSTEM_PROMPT, "messages": messages, "stream": True, }).encode("utf-8") req = urllib.request.Request( API_URL, data=payload, headers={ "Content-Type": "application/json", "x-api-key": api_key, "anthropic-version": "2023-06-01", }, method="POST", ) user_query = "" for msg in reversed(messages): if msg.get("role") == "user": user_query = msg.get("content", "") if isinstance(user_query, list): user_query = " ".join( p.get("text", "") for p in user_query if p.get("type") == "text" ) break # Headers SSE al cliente self.send_response(200) self.send_header("Content-Type", "text/event-stream; charset=utf-8") self.send_header("Cache-Control", "no-cache, no-transform") self.send_header("Connection", "keep-alive") self.send_header("X-Accel-Buffering", "no") # por si hay nginx self.end_headers() def write_sse(event_name: str, data_dict: dict) -> None: frame = f"event: {event_name}\ndata: {json.dumps(data_dict, ensure_ascii=False)}\n\n" try: self.wfile.write(frame.encode("utf-8")) self.wfile.flush() except (BrokenPipeError, ConnectionResetError): # Cliente se desconecto mid-stream pass buffer_parts: list[str] = [] try: ctx = ssl.create_default_context() with urllib.request.urlopen(req, context=ctx, timeout=60) as upstream: # Anthropic emite SSE; leemos linea por linea for raw_line in upstream: line = raw_line.decode("utf-8", errors="replace").rstrip("\r\n") if not line or not line.startswith("data: "): continue try: evt = json.loads(line[6:]) except json.JSONDecodeError: continue etype = evt.get("type") if etype == "content_block_delta": delta_text = evt.get("delta", {}).get("text", "") if delta_text: buffer_parts.append(delta_text) write_sse("delta", {"text": delta_text}) elif etype == "message_stop": break elif etype == "error": write_sse("error", {"error": str(evt.get("error", "upstream error"))}) return # Procesar texto completo: extraer clasificacion + limpiar full_text = "".join(buffer_parts) if buffer_parts else "Sin respuesta" clasificacion = "OBRA" m = CLASIFICACION_PATTERN.search(full_text) if m: clasificacion = m.group(1) clean_text = CLASIFICACION_PATTERN.sub("", full_text).strip() clean_text = re.sub(r"^\n+", "", clean_text) log_consulta(clasificacion, user_query, clean_text) print(f" [CLAS] {clasificacion} | {user_query[:60]} (stream)") write_sse("done", { "clasificacion": clasificacion, "clean_text": clean_text, }) except urllib.error.HTTPError as e: err_body = e.read().decode("utf-8", errors="replace") print(f"API Error {e.code} (stream): {err_body}") write_sse("error", {"error": f"HTTP {e.code}: {err_body[:200]}"}) except Exception as e: print(f"Stream error: {e}") write_sse("error", {"error": str(e)}) def handle_log(self): """Endpoint para que el frontend registre consultas en modo simulado.""" try: content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body) clasificacion = data.get("clasificacion", "OBRA") query = data.get("query", "") response_text = data.get("response", "") # Validar clasificacion if clasificacion not in ("OBRA", "AMBIGUA", "FUERA"): clasificacion = "OBRA" log_consulta(clasificacion, query, response_text) print(f" [LOG-SIM] {clasificacion} | {query[:60]}") self.send_json(200, {"status": "ok"}) except Exception as e: print(f" [LOG] Error en /api/log: {e}") self.send_json(500, {"error": str(e)}) def handle_skp_convert(self): """Convierte .skp a .glb usando Blender headless.""" if not BLENDER_EXE: self.send_json(500, { "error": "Blender no encontrado. Instalar en subcarpeta blender/ o configurar BLENDER_PATH en .env" }) return # Leer el archivo subido (raw binary body) content_length = int(self.headers.get("Content-Length", 0)) if content_length == 0 or content_length > 200 * 1024 * 1024: # Max 200MB self.send_json(400, {"error": "Archivo vacio o demasiado grande (max 200MB)"}) return file_data = self.rfile.read(content_length) # Guardar en temp tmp_dir = tempfile.mkdtemp(prefix="dg_skp_") input_skp = os.path.join(tmp_dir, "model.skp") output_glb = os.path.join(tmp_dir, "model.glb") try: with open(input_skp, "wb") as f: f.write(file_data) print(f" [SKP] Convirtiendo {content_length / 1024:.0f} KB...") # Llamar a Blender headless result = subprocess.run( [ BLENDER_EXE, "-b", "--python", CONVERT_SCRIPT, "--", input_skp, output_glb ], capture_output=True, text=True, timeout=120 ) if result.returncode != 0: error_msg = result.stderr[-500:] if result.stderr else "Error desconocido" print(f" [SKP] ERROR: {error_msg}") self.send_json(500, {"error": f"Blender error: {error_msg}"}) return if not os.path.exists(output_glb): self.send_json(500, {"error": "Conversion fallo: no se genero archivo .glb"}) return # Leer y devolver el .glb with open(output_glb, "rb") as f: glb_data = f.read() print(f" [SKP] OK: {len(glb_data) / 1024:.0f} KB generados") self.send_response(200) self.send_header("Content-Type", "model/gltf-binary") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Content-Length", len(glb_data)) self.end_headers() self.wfile.write(glb_data) except subprocess.TimeoutExpired: print(" [SKP] TIMEOUT: conversion excedio 120s") self.send_json(500, {"error": "Conversion timeout (modelo muy grande?)"}) except Exception as e: print(f" [SKP] Error: {e}") self.send_json(500, {"error": str(e)}) finally: # Cleanup temp files try: shutil.rmtree(tmp_dir, ignore_errors=True) except Exception: pass def send_json(self, code, data): response = json.dumps(data).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Content-Length", len(response)) self.end_headers() self.wfile.write(response) def log_message(self, format, *args): msg = format % args if "GET /api/" in msg or "POST /api/" in msg: print(f" [API] {msg}") elif "404" in msg or "500" in msg: print(f" [ERR] {msg}") def main(): api_key = get_api_key() print() print("=" * 60) print(" DG INGENIERIA - Servidor Local") print(" Plataforma de Control de Obras v2.0") print("=" * 60) print() if api_key: print(f" API Key: ...{api_key[-8:]}") print(f" Modelo: claude-sonnet-4-20250514") print(f" Agente IA: CONECTADO") else: print(" API Key: NO CONFIGURADA") print(" Para conectar el agente IA:") print(" set ANTHROPIC_API_KEY=sk-ant-...") print(" (o crear archivo .env)") print() print(" El dashboard funcionara sin IA (respuestas simuladas)") print() if BLENDER_EXE: print(f" Blender: {BLENDER_EXE}") print(f" SKP>GLB: HABILITADO") else: print(" Blender: NO ENCONTRADO") print(" SKP>GLB: DESHABILITADO (instalar Blender en subcarpeta blender/)") print() db_path = os.path.join(BASE_DIR, "db", "ergon.db") if api_obras is not None and os.path.exists(db_path): print(f" DB ERGON: {db_path}") print(f" API obras: HABILITADA (GET /api/obras, /api/obras/:codigo, ...)") else: print(" DB ERGON: NO ENCONTRADA") print(" API obras: DESHABILITADA (ejecutar db/init_db.py + db/seed_torres_dg_demo.py)") print() print(f" Servidor: http://localhost:{PORT}") print(f" Dashboard: http://localhost:{PORT}/dashboard.html") print() print(" Presione Ctrl+C para detener") print("=" * 60) print() # Init DB on first boot (idempotente). En Railway con volume vacio, # esto crea schema + migrations + admin user. En local con DB existente, # retorna sin tocar nada. try: from init_railway_db import ensure_db_initialized # type: ignore init_result = ensure_db_initialized() print(f" DB init: {init_result.get('action', 'unknown')} - {init_result.get('msg', '')}") except ImportError: print(" DB init: SKIP (init_railway_db no disponible)") except Exception as e: print(f" DB init: WARN {e}") print() server = http.server.HTTPServer(("", PORT), DGHandler) try: server.serve_forever() except KeyboardInterrupt: print("\n Servidor detenido.") server.server_close() if __name__ == "__main__": main()