descarga_xml_sunat/historial_manager.py

import sqlite3
from pathlib import Path
from datetime import datetime, timedelta
from logger_manager import get_logger

DB_PATH = Path(__file__).parent / "historial.db"
logger = get_logger(__name__)
# ============================================================================
def _conn():
    return sqlite3.connect(str(DB_PATH))
# ============================================================================
def init_db():
    with _conn() as conn:
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("""
            CREATE TABLE IF NOT EXISTS historial (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                fecha TEXT NOT NULL,
                docentry INTEGER NOT NULL,
                docnum TEXT,
                ruc TEXT NOT NULL,
                tipo TEXT NOT NULL,
                serie TEXT NOT NULL,
                numero TEXT NOT NULL,
                objtype INTEGER NOT NULL DEFAULT 18,
                es_retry INTEGER NOT NULL DEFAULT 0,
                intento_num INTEGER,
                consulta_status INTEGER,
                pdf_resultado TEXT,
                xml_resultado TEXT,
                resultado TEXT NOT NULL,
                detalle TEXT,
                json_path TEXT
            )
        """)
        conn.execute("CREATE INDEX IF NOT EXISTS idx_historial_docentry ON historial(docentry)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_historial_fecha ON historial(fecha)")
# ============================================================================
def registrar_intento(docentry, docnum, ruc, tipo, serie, numero, objtype, es_retry,
                       intento_num, consulta_status, pdf_resultado, xml_resultado,
                       resultado, detalle="", json_path=None):
    ahora = datetime.now().isoformat()
    with _conn() as conn:
        conn.execute("""
            INSERT INTO historial
                (fecha, docentry, docnum, ruc, tipo, serie, numero, objtype, es_retry,
                 intento_num, consulta_status, pdf_resultado, xml_resultado, resultado,
                 detalle, json_path)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (ahora, docentry, docnum, ruc, tipo, serie, numero, objtype, int(es_retry),
              intento_num, consulta_status, pdf_resultado, xml_resultado, resultado,
              str(detalle)[:500] if detalle else detalle, json_path))
# ============================================================================
def limpiar_historial_antiguo(dias=180):
    limite = (datetime.now() - timedelta(days=dias)).isoformat()
    with _conn() as conn:
        cursor = conn.execute("DELETE FROM historial WHERE fecha < ?", (limite,))
        count = cursor.rowcount
    if count > 0:
        logger.info(f"Se eliminaron {count} registros de historial de más de {dias} días.")
    return count
# ============================================================================
def historial_docentry(docentry) -> list[dict]:
    with _conn() as conn:
        conn.row_factory = sqlite3.Row
        cursor = conn.execute(
            "SELECT * FROM historial WHERE docentry = ? ORDER BY id ASC", (docentry,)
        )
        return [dict(row) for row in cursor.fetchall()]
# ============================================================================
def buscar_intentos(ruc=None, serie=None, numero=None, tipo=None) -> list[dict]:
    """Busca intentos registrados filtrando por cualquier combinación de ruc/serie/numero/tipo."""
    condiciones = []
    params = []
    if ruc:
        condiciones.append("ruc = ?")
        params.append(ruc)
    if serie:
        condiciones.append("serie = ?")
        params.append(serie)
    if numero:
        condiciones.append("numero = ?")
        params.append(str(numero))
    if tipo:
        condiciones.append("tipo = ?")
        params.append(tipo)

    where = " AND ".join(condiciones) if condiciones else "1=1"
    with _conn() as conn:
        conn.row_factory = sqlite3.Row
        cursor = conn.execute(f"SELECT * FROM historial WHERE {where} ORDER BY fecha ASC", params)
        return [dict(row) for row in cursor.fetchall()]
# ============================================================================
if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="Buscar intentos de descarga en historial.db")
    parser.add_argument("--ruc", help="RUC del emisor")
    parser.add_argument("--serie", help="Serie del comprobante (ej: F001)")
    parser.add_argument("--numero", help="Número del comprobante")
    parser.add_argument("--tipo", help="Tipo de comprobante (01, 03, 07, 08...)")
    args = parser.parse_args()

    if not any([args.ruc, args.serie, args.numero, args.tipo]):
        parser.error("Indica al menos un filtro: --ruc, --serie, --numero o --tipo")

    filas = buscar_intentos(ruc=args.ruc, serie=args.serie, numero=args.numero, tipo=args.tipo)

    if not filas:
        print("No se encontraron intentos con esos filtros.")
    else:
        print(f"{len(filas)} intento(s) encontrado(s):\n")
        for f in filas:
            print(
                f"{f['fecha']}  DocEntry={f['docentry']}  "
                f"{f['ruc']}-{f['tipo']}-{f['serie']}-{f['numero']}  "
                f"intento={f['intento_num']}  retry={'sí' if f['es_retry'] else 'no'}  "
                f"status={f['consulta_status']}  resultado={f['resultado']}  "
                f"pdf={f['pdf_resultado']}  xml={f['xml_resultado']}  detalle={f['detalle']}"
            )
# ============================================================================