GeorgeandClaude Sonnet 5
c2503745bf
- Nuevo logger_manager.py (reemplaza logger_utilities.py): logging con formato de árbol, rotación diaria a sunat.log, y consola por stdout (antes iba por stderr y se duplicaba en logs/error.log del launchd). - Nuevo historial_manager.py: registra cada intento de consulta/descarga en historial.db para auditoría, con limpieza automática de registros de más de 180 días. - consulta_cpe.py: corrige el id de notas de crédito/débito (usa F7/F8, no 07/08) y agrega reintentos ante errores 5xx. - hana_manager.py: suma consulta a ORPC para notas de crédito (Indicator 07) y soporte de ObjType para actualizar la tabla correcta (OPCH/ORPC) al marcar PDF/XML descargado. - gestor_fallas.py / main.py: propagan objtype end-to-end y dejan constancia de cada intento en historial_manager. - Elimina login_api.py y reporte_fallas.py (código no usado / script manual sin uso reciente). - Actualiza .gitignore (sunat-venv/, historial.db*) y docs/ para reflejar el estado real del proyecto. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
399 lines
17 KiB
Python
399 lines
17 KiB
Python
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
import signal
|
|
import random
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from dotenv import load_dotenv
|
|
import urllib3
|
|
|
|
# ============================================================================
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
load_dotenv()
|
|
|
|
from logger_manager import get_logger, setup_logging, limpiar_logs_antiguos, LOG_FILE
|
|
from hana_manager import obtener_pendientes, marcar_pdf_descargado, marcar_xml_descargado, cerrar_conexion
|
|
from gestor_fallas import init_db, registrar_falla, eliminar_falla, obtener_fallas_pendientes, obtener_docentries_falla, total_fallas
|
|
from historial_manager import init_db as init_historial_db, registrar_intento, limpiar_historial_antiguo
|
|
from smb_manager import sincronizar_a_smb
|
|
from consulta_cpe import (
|
|
TokenExpiredError, load_token, get_headers, consultar_cpe,
|
|
descargar_archivo, cpe_id,
|
|
API_HOST, BASE_PATH, DOWNLOAD_DIR
|
|
)
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
# ============================================================================
|
|
# CONFIGURACIÓN DE HORARIOS (desde scheduler.json)
|
|
# ============================================================================
|
|
SCHEDULER_FILE = Path(__file__).parent / "scheduler.json"
|
|
|
|
|
|
def _cargar_scheduler():
|
|
with open(SCHEDULER_FILE, encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
horarios = [tuple(h) for h in data["HORARIOS"]]
|
|
return data["DIAS_ACTIVOS"], horarios
|
|
|
|
|
|
try:
|
|
DIAS_ACTIVOS, HORARIOS = _cargar_scheduler()
|
|
except Exception as e:
|
|
logger.warning(f"Error cargando {SCHEDULER_FILE}: {e}. Usando defaults.")
|
|
# (hora_inicio, min_inicio, hora_fin, min_fin, intervalo_segundos, modo)
|
|
HORARIOS = [
|
|
(6, 30, 13, 15, 300, "rapido"), # 06:30 - 13:15 -> cada 5 minutos
|
|
(13, 15, 14, 0, 3600, "full"), # 13:15 - 14:00 -> cada 1 hora
|
|
(14, 0, 17, 30, 300, "rapido"), # 14:00 - 17:30 -> cada 5 minutos
|
|
(17, 30, 21, 30, 3600, "full"), # 17:30 - 21:30 -> cada 1 hora
|
|
]
|
|
# 0=lun, 1=mar, 2=mié, 3=jue, 4=vie, 5=sáb, 6=dom
|
|
DIAS_ACTIVOS = [0, 1, 2, 3, 4]
|
|
|
|
_running = True
|
|
|
|
# ─── Colores ────────────────────────────────────────────
|
|
color_verde = "\033[92m"
|
|
color_amarillo = "\033[93m"
|
|
color_rojo = "\033[91m"
|
|
color_fin = "\033[0m"
|
|
|
|
# ============================================================================
|
|
def ic(ok, texto_ok, texto_fail):
|
|
"""Icono + texto según estado."""
|
|
icono = "✅" if ok else "⚠️"
|
|
color = color_verde if ok else color_amarillo
|
|
return f"{icono} {color}{texto_ok if ok else texto_fail}{color_fin}"
|
|
|
|
# ============================================================================
|
|
def obtener_intervalo(dt: datetime) -> tuple[int, str] | None:
|
|
minutos = dt.hour * 60 + dt.minute
|
|
for hi, mi, hf, mf, intervalo, modo in HORARIOS:
|
|
inicio = hi * 60 + mi
|
|
fin = hf * 60 + mf
|
|
if inicio <= minutos < fin:
|
|
return (intervalo, modo)
|
|
return None
|
|
|
|
# ============================================================================
|
|
def segundos_hasta_proximo_inicio(dt: datetime) -> int:
|
|
target = dt.replace(hour=6, minute=30, second=0, microsecond=0)
|
|
if dt >= target:
|
|
target += timedelta(days=1)
|
|
return int((target - dt).total_seconds())
|
|
|
|
# ============================================================================
|
|
def segundos_hasta_proximo_dia_activo(dt: datetime) -> int:
|
|
for dias in range(1, 8):
|
|
prox = dt + timedelta(days=dias)
|
|
if prox.weekday() in DIAS_ACTIVOS:
|
|
target = prox.replace(hour=6, minute=30, second=0, microsecond=0)
|
|
return int((target - dt).total_seconds())
|
|
return 86400
|
|
|
|
# ============================================================================
|
|
def obtener_token():
|
|
"""Obtiene o renueva el token si es necesario."""
|
|
token = load_token()
|
|
if token:
|
|
logger.info("Token existente válido, reutilizando")
|
|
return token
|
|
|
|
logger.info("Token no disponible o expirado. Obteniendo nuevo...")
|
|
|
|
from get_token_headless import get_token_headless as obtener_headless, firefox_instalado
|
|
if not firefox_instalado():
|
|
logger.warning("Firefox no instalado. Saltando headless...")
|
|
else:
|
|
try:
|
|
if obtener_headless():
|
|
logger.info("Token obtenido vía get_token_headless (headless)")
|
|
return load_token()
|
|
except Exception as e:
|
|
logger.warning(f"Headless falló: {e}")
|
|
|
|
logger.warning("Headless no disponible. Intentando get_token.py con navegador visible...")
|
|
from get_token import get_token as obtener_token_playwright
|
|
obtener_token_playwright(headless=False, timeout=120000)
|
|
for _ in range(30):
|
|
tk = load_token()
|
|
if tk:
|
|
logger.info("Token obtenido vía Playwright (get_token.py)")
|
|
return tk
|
|
time.sleep(1)
|
|
logger.warning("No se detectó token tras 30s. Continuando...")
|
|
return load_token()
|
|
|
|
# ============================================================================
|
|
def procesar_lote(token):
|
|
"""Procesa documentos pendientes desde HANA."""
|
|
logger.info("Buscando documentos pendientes en HANA...")
|
|
docs_hana = obtener_pendientes()
|
|
fallas_docentries = obtener_docentries_falla()
|
|
docs_nuevos = [d for d in docs_hana if d.get("DocEntry") not in fallas_docentries]
|
|
docs_retry_raw = obtener_fallas_pendientes()
|
|
docs_retry = []
|
|
for d in docs_retry_raw:
|
|
docs_retry.append({
|
|
"DocEntry": d["docentry"],
|
|
"DocNum": d.get("numero", ""),
|
|
"LicTradNum": d["ruc"],
|
|
"Indicator": d["tipo"],
|
|
"FolioPref": d["serie"],
|
|
"FolioNum": d["numero"],
|
|
"ObjType": d.get("objtype", 18),
|
|
"Intentos": d.get("intentos", 1),
|
|
"tienePDF": "N",
|
|
"tieneXML": "N",
|
|
})
|
|
total_fallas_count = total_fallas()
|
|
|
|
if not docs_nuevos and not docs_retry:
|
|
logger.info("No hay documentos pendientes")
|
|
return
|
|
|
|
logger.info(f"{len(docs_hana)} en HANA | {len(docs_nuevos)} nuevos | {len(docs_retry)} reintentos | {total_fallas_count} en fallas")
|
|
DOWNLOAD_DIR.mkdir(exist_ok=True)
|
|
|
|
# ========================================================================
|
|
def _procesar(doc, idx, total, es_retry=False):
|
|
nonlocal token
|
|
if not _running:
|
|
return
|
|
|
|
ruc = (doc.get("LicTradNum") or "").strip()
|
|
tipo_cpe = (doc.get("Indicator") or "").strip()
|
|
serie = (doc.get("FolioPref") or "").strip()
|
|
numero = doc.get("FolioNum") or 0
|
|
docentry = doc.get("DocEntry") or 0
|
|
docnum = doc.get("DocNum") or 0
|
|
obj_type = doc.get("ObjType") or 18
|
|
intento_num = doc.get("Intentos") or 1
|
|
tiene_pdf = (doc.get("tienePDF") or "N").strip()
|
|
tiene_xml = (doc.get("tieneXML") or "N").strip()
|
|
|
|
if not ruc or not tipo_cpe or not serie or not numero:
|
|
logger.warning(f"DocEntry={docentry} DocNum={docnum}: datos incompletos, saltando")
|
|
return
|
|
|
|
ultimo = idx == total - 1
|
|
tag = "🔄" if es_retry else "📄"
|
|
cid = cpe_id(ruc, tipo_cpe, serie, numero)
|
|
|
|
task_cm = logger.last_task if ultimo else logger.task
|
|
with task_cm(f"[{idx+1}/{total}] {ruc}-{tipo_cpe}-{serie}-{numero} [DocEntry={docentry} | DocNum={docnum}]", emoji=tag) as doc_log:
|
|
for intento_token in range(2):
|
|
ok_pdf = True
|
|
ok_xml = True
|
|
hubo_error = False
|
|
|
|
try:
|
|
time.sleep(random.uniform(1.0, 2.0))
|
|
data, status_code, json_path = consultar_cpe(token, ruc, tipo_cpe, serie, numero, descargar=False, verbose=False)
|
|
|
|
if not data:
|
|
doc_log.error(f"Consulta: {color_rojo}Error {status_code}{color_fin}", _last=True)
|
|
registrar_falla(docentry, ruc, tipo_cpe, serie, numero, obj_type, f"Error HTTP {status_code}")
|
|
registrar_intento(docentry, docnum, ruc, tipo_cpe, serie, numero, obj_type, es_retry,
|
|
intento_num, status_code, "no_aplica", "no_aplica", "error",
|
|
f"Error HTTP {status_code}", None)
|
|
return
|
|
|
|
doc_log.info(f"Consulta: {color_verde}✅ OK{color_fin}")
|
|
|
|
# ─── PDF ───
|
|
if tiene_pdf == "Y":
|
|
pdf_msg = f"{color_verde}✅ ya tenía{color_fin}"
|
|
else:
|
|
time.sleep(random.uniform(1.0, 2.0))
|
|
ok_pdf = descargar_archivo(token, cid, "01",
|
|
datetime.now().strftime("%Y%m%d_%H%M%S"),
|
|
verbose=False)
|
|
if ok_pdf:
|
|
dl_url = f"{API_HOST}{BASE_PATH}/{cid}/01"
|
|
local_path = str(DOWNLOAD_DIR / f"{cid[:-2] if cid.endswith('-2') else cid}.pdf")
|
|
marcar_pdf_descargado(docentry, dl_url, local_path, obj_type)
|
|
pdf_msg = f"{color_verde}📥 descargado{color_fin}"
|
|
else:
|
|
pdf_msg = f"{color_amarillo}⚠️ error{color_fin}"
|
|
hubo_error = True
|
|
|
|
doc_log.info(f"PDF: {pdf_msg}")
|
|
|
|
# ─── XML ───
|
|
if tiene_xml == "Y":
|
|
xml_msg = f"{color_verde}✅ ya tenía{color_fin}"
|
|
else:
|
|
time.sleep(random.uniform(1.0, 2.0))
|
|
ok_xml = descargar_archivo(token, cid, "02",
|
|
datetime.now().strftime("%Y%m%d_%H%M%S"),
|
|
verbose=False)
|
|
if ok_xml:
|
|
dl_url = f"{API_HOST}{BASE_PATH}/{cid}/02"
|
|
local_path = str(DOWNLOAD_DIR / f"{cid[:-2] if cid.endswith('-2') else cid}.xml")
|
|
marcar_xml_descargado(docentry, dl_url, local_path, obj_type)
|
|
xml_msg = f"{color_verde}📥 descargado{color_fin}"
|
|
else:
|
|
xml_msg = f"{color_amarillo}⚠️ error{color_fin}"
|
|
hubo_error = True
|
|
|
|
doc_log.info(f"XML: {xml_msg}")
|
|
doc_log.info(f"HANA: {ic(ok_pdf and ok_xml, '✅ actualizado', '⚠️ pendiente')}")
|
|
|
|
if ok_pdf and ok_xml:
|
|
if es_retry:
|
|
doc_log.info(f"SQLite: {color_amarillo}🗑️ quitando registro{color_fin}")
|
|
eliminar_falla(docentry)
|
|
doc_log.info(f"Resultado: {color_verde}✅ OK{color_fin}", _last=True)
|
|
detalle_historial = "OK"
|
|
else:
|
|
fallos = [s for s, ok in [("PDF", ok_pdf), ("XML", ok_xml)] if not ok]
|
|
doc_log.warning(f"Resultado: {color_amarillo}⚠️ parcial{color_fin} ({', '.join(fallos)})", _last=True)
|
|
detalle_historial = f"falló: {', '.join(fallos)}"
|
|
if hubo_error:
|
|
registrar_falla(docentry, ruc, tipo_cpe, serie, numero, obj_type, detalle_historial)
|
|
|
|
pdf_resultado = "ya_tenia" if tiene_pdf == "Y" else ("descargado" if ok_pdf else "error")
|
|
xml_resultado = "ya_tenia" if tiene_xml == "Y" else ("descargado" if ok_xml else "error")
|
|
registrar_intento(docentry, docnum, ruc, tipo_cpe, serie, numero, obj_type, es_retry,
|
|
intento_num, status_code, pdf_resultado, xml_resultado,
|
|
"ok" if (ok_pdf and ok_xml) else "parcial",
|
|
detalle_historial, json_path)
|
|
|
|
break
|
|
|
|
except TokenExpiredError:
|
|
if intento_token == 0:
|
|
doc_log.info("Token expirado, renovando...")
|
|
nuevo_token = obtener_token()
|
|
if nuevo_token:
|
|
token = nuevo_token
|
|
doc_log.info("Token renovado, reintentando...")
|
|
continue
|
|
doc_log.error(f"Resultado: {color_rojo}Token no renovado{color_fin}", _last=True)
|
|
else:
|
|
doc_log.error(f"Resultado: {color_rojo}Token expiró{color_fin}", _last=True)
|
|
registrar_falla(docentry, ruc, tipo_cpe, serie, numero, obj_type, "Token expiró")
|
|
registrar_intento(docentry, docnum, ruc, tipo_cpe, serie, numero, obj_type, es_retry,
|
|
intento_num, None, "no_aplica", "no_aplica", "error", "Token expiró", None)
|
|
break
|
|
|
|
except Exception as e:
|
|
doc_log.error(f"Resultado: {color_rojo}error{color_fin} ({e})", _last=True)
|
|
registrar_falla(docentry, ruc, tipo_cpe, serie, numero, obj_type, str(e))
|
|
registrar_intento(docentry, docnum, ruc, tipo_cpe, serie, numero, obj_type, es_retry,
|
|
intento_num, None, "no_aplica", "no_aplica", "error", str(e), None)
|
|
break
|
|
|
|
# ========================================================================
|
|
|
|
for idx, doc in enumerate(docs_nuevos):
|
|
_procesar(doc, idx, len(docs_nuevos), es_retry=False)
|
|
|
|
for idx, doc in enumerate(docs_retry):
|
|
_procesar(doc, idx, len(docs_retry), es_retry=True)
|
|
|
|
# ============================================================================
|
|
def signal_handler(signum, frame):
|
|
global _running
|
|
logger.warning("Señal de parada recibida. Deteniendo...")
|
|
_running = False
|
|
|
|
# ============================================================================
|
|
def main():
|
|
global _running
|
|
signal.signal(signal.SIGINT, signal_handler)
|
|
signal.signal(signal.SIGTERM, signal_handler)
|
|
|
|
setup_logging()
|
|
limpiar_logs_antiguos(dias=7)
|
|
init_db()
|
|
init_historial_db()
|
|
limpiar_historial_antiguo(dias=180)
|
|
logger.info(f"Log: {LOG_FILE}")
|
|
|
|
logger.info("=" * 60)
|
|
logger.info("DAEMON DESCARGA XML/PDF SUNAT")
|
|
logger.info("=" * 60)
|
|
logger.info(f"Días activos: {DIAS_ACTIVOS} (0=lun, 6=dom)")
|
|
logger.info("Horarios:")
|
|
for hi, mi, hf, mf, intervalo, modo in HORARIOS:
|
|
logger.info(f"{hi:02d}:{mi:02d} - {hf:02d}:{mf:02d} | cada {intervalo}s | {modo}")
|
|
|
|
token = None
|
|
|
|
while _running:
|
|
ahora = datetime.now()
|
|
|
|
if ahora.weekday() not in DIAS_ACTIVOS:
|
|
espera = segundos_hasta_proximo_dia_activo(ahora)
|
|
prox = ahora + timedelta(seconds=espera)
|
|
logger.info(f"Día no activo. Siguiente inicio: {prox.strftime('%a %Y-%m-%d %H:%M:%S')}")
|
|
for _ in range(int(espera)):
|
|
if not _running:
|
|
break
|
|
time.sleep(1)
|
|
continue
|
|
|
|
intervalo_info = obtener_intervalo(ahora)
|
|
if not intervalo_info:
|
|
espera = segundos_hasta_proximo_inicio(ahora)
|
|
prox = ahora + timedelta(seconds=espera)
|
|
logger.info(f"Fuera de horario. Siguiente inicio: {prox.strftime('%H:%M:%S')}")
|
|
for _ in range(int(espera)):
|
|
if not _running:
|
|
break
|
|
time.sleep(1)
|
|
continue
|
|
|
|
intervalo, modo = intervalo_info
|
|
|
|
logger.info("#" * 60)
|
|
logger.info(f"CICLO: {ahora.strftime('%Y-%m-%d %H:%M:%S')} | modo={modo} | intervalo={intervalo}s")
|
|
logger.info("#" * 60)
|
|
|
|
if not token:
|
|
token = obtener_token()
|
|
if not token:
|
|
logger.error("No se pudo obtener token. Esperando 60s...")
|
|
time.sleep(60)
|
|
continue
|
|
|
|
procesar_lote(token)
|
|
|
|
try:
|
|
sincronizar_a_smb(str(DOWNLOAD_DIR))
|
|
except Exception as e:
|
|
logger.warning(f"Error en sincronización SMB: {e}")
|
|
|
|
# Verificar si el token sigue válido para el próximo ciclo
|
|
from consulta_cpe import TOKEN_FILE as tf
|
|
if tf.exists():
|
|
try:
|
|
data = json.loads(tf.read_text())
|
|
expires = data.get("expires_at", 0)
|
|
remaining = expires - time.time()
|
|
if remaining < 300:
|
|
logger.warning(f"Token por expirar ({int(remaining)}s). Se renovará en próximo ciclo.")
|
|
token = None
|
|
except:
|
|
pass
|
|
|
|
logger.info(f"Esperando {intervalo}s hasta próximo ciclo...")
|
|
for _ in range(intervalo):
|
|
if not _running:
|
|
break
|
|
time.sleep(1)
|
|
|
|
cerrar_conexion()
|
|
logger.info("Daemon detenido.")
|
|
|
|
# ============================================================================
|
|
if __name__ == "__main__":
|
|
main()
|
|
# ============================================================================
|