""" Registro de comprobantes y códigos de seguimiento. Cada comprobante que llega por correo recibe un ``codigo_seguimiento`` (por ejemplo ``ABR-261019-00042``) que se comunica al proveedor y que sirve para identificarlo si lo reenvía. El registro también guarda el resultado del proceso (aceptado, rechazado o error) y el estado de la notificación enviada. Un comprobante se identifica por (RUC emisor, tipo, serie, número): si vuelve a llegar, en el mismo correo o en otro, conserva su código. Los correos que no traen ningún comprobante identificable (sin adjuntos, sin XML válido…) se registran aparte con ``registrar_correo``: una fila por correo, sin serie ni número. El esquema se crea en ``database_utilities.init_db``. """ import json import sqlite3 from contextlib import contextmanager from datetime import datetime import database_utilities PREFIJO_CODIGO = "ABR" ESTADOS = ("pendiente", "aceptado", "rechazado", "error") ESTADOS_NOTIFICACION = ("pendiente", "enviada", "fallida", "omitida") # ============================================================================ @contextmanager def _conexion(): """Conexión con filas tipo diccionario; confirma al salir o revierte si hubo error.""" conn = sqlite3.connect(database_utilities.DB_NAME) conn.row_factory = sqlite3.Row try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() def _ahora() -> str: return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def _a_dict(fila: sqlite3.Row | None) -> dict | None: if fila is None: return None registro = dict(fila) if registro.get("observaciones") is not None: registro["observaciones"] = json.loads(registro["observaciones"]) return registro def _insertar(conn: sqlite3.Connection, email_id: str, ruc_emisor, tipo, serie, numero) -> dict: """Inserta la fila y le asigna el código, que se deriva de su ``id`` (único por construcción).""" ahora = _ahora() cursor = conn.execute( """INSERT INTO comprobantes (email_id, ruc_emisor, tipo, serie, numero, creado_en, actualizado_en) VALUES (?, ?, ?, ?, ?, ?, ?)""", (email_id, ruc_emisor, tipo, serie, numero, ahora, ahora), ) codigo = f"{PREFIJO_CODIGO}-{datetime.now():%y%m%d}-{cursor.lastrowid:05d}" conn.execute("UPDATE comprobantes SET codigo_seguimiento = ? WHERE id = ?", (codigo, cursor.lastrowid)) return _a_dict(conn.execute("SELECT * FROM comprobantes WHERE id = ?", (cursor.lastrowid,)).fetchone()) # ============================================================================ def registrar_comprobante(email_id: str, ruc_emisor: str, tipo: str, serie: str, numero: str) -> dict: """ Registra un comprobante, o recupera el registro existente si ya había llegado. Si el comprobante ya existía y llega en un correo distinto, se conserva su código, se actualiza ``email_id`` al correo más reciente y se incrementa ``veces_recibido``. Si es el mismo correo (reproceso), no cambia nada. Returns: dict — La fila de ``comprobantes`` (con ``codigo_seguimiento``) más la clave ``reenvio``: True si el comprobante ya había llegado en otro correo. """ with _conexion() as conn: fila = conn.execute( "SELECT * FROM comprobantes WHERE ruc_emisor = ? AND tipo = ? AND serie = ? AND numero = ?", (ruc_emisor, tipo, serie, numero), ).fetchone() if fila is None: registro = _insertar(conn, email_id, ruc_emisor, tipo, serie, numero) registro["reenvio"] = False return registro reenvio = fila["email_id"] != email_id if reenvio: conn.execute( """UPDATE comprobantes SET email_id = ?, veces_recibido = veces_recibido + 1, actualizado_en = ? WHERE id = ?""", (email_id, _ahora(), fila["id"]), ) registro = _a_dict(conn.execute("SELECT * FROM comprobantes WHERE id = ?", (fila["id"],)).fetchone()) registro["reenvio"] = reenvio return registro def registrar_correo(email_id: str) -> dict: """ Registra un correo que no trae ningún comprobante identificable (una fila por correo, sin RUC, serie ni número) y devuelve su registro. Idempotente. """ with _conexion() as conn: fila = conn.execute( "SELECT * FROM comprobantes WHERE email_id = ? AND serie IS NULL", (email_id,) ).fetchone() if fila is not None: return _a_dict(fila) return _insertar(conn, email_id, None, None, None, None) def obtener(codigo_seguimiento: str) -> dict | None: """Retorna el registro con ese código, o None si no existe.""" with _conexion() as conn: return _a_dict(conn.execute( "SELECT * FROM comprobantes WHERE codigo_seguimiento = ?", (codigo_seguimiento,) ).fetchone()) def actualizar_resultado(codigo_seguimiento: str, estado: str, observaciones: list | None = None, doc_entry: int | None = None) -> None: """ Guarda el resultado del proceso de un comprobante. Args: codigo_seguimiento: Código devuelto por ``registrar_comprobante``/``registrar_correo``. estado: Uno de ``ESTADOS``. observaciones: Lista serializable a JSON (motivos de rechazo o detalle del error). doc_entry: DocEntry del draft creado en SAP, si corresponde. Raises: ValueError: Si ``estado`` no es válido. LookupError: Si no existe el código. """ if estado not in ESTADOS: raise ValueError(f"Estado inválido: {estado!r} (válidos: {', '.join(ESTADOS)})") with _conexion() as conn: cursor = conn.execute( """UPDATE comprobantes SET estado = ?, observaciones = ?, doc_entry = ?, actualizado_en = ? WHERE codigo_seguimiento = ?""", (estado, None if observaciones is None else json.dumps(observaciones, ensure_ascii=False), doc_entry, _ahora(), codigo_seguimiento), ) if cursor.rowcount == 0: raise LookupError(f"No existe el código de seguimiento {codigo_seguimiento}") def registrar_notificacion(codigo_seguimiento: str, estado: str, error: str | None = None) -> None: """ Guarda el estado de la notificación enviada al proveedor. Args: codigo_seguimiento: Código del registro. estado: Uno de ``ESTADOS_NOTIFICACION``. error: Motivo, si la notificación falló o se omitió. Raises: ValueError: Si ``estado`` no es válido. LookupError: Si no existe el código. """ if estado not in ESTADOS_NOTIFICACION: raise ValueError(f"Estado de notificación inválido: {estado!r} (válidos: {', '.join(ESTADOS_NOTIFICACION)})") ahora = _ahora() with _conexion() as conn: cursor = conn.execute( """UPDATE comprobantes SET notificacion_estado = ?, notificacion_error = ?, notificacion_fecha = ?, actualizado_en = ? WHERE codigo_seguimiento = ?""", (estado, error, ahora, ahora, codigo_seguimiento), ) if cursor.rowcount == 0: raise LookupError(f"No existe el código de seguimiento {codigo_seguimiento}")