87 lines
3.6 KiB
Python
87 lines
3.6 KiB
Python
import sqlite3
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
|
|
DB_PATH = Path(__file__).parent / "fallas_descarga.db"
|
|
# ============================================================================
|
|
def _conn():
|
|
return sqlite3.connect(str(DB_PATH))
|
|
# ============================================================================
|
|
def init_db():
|
|
with _conn() as conn:
|
|
conn.execute("""
|
|
CREATE TABLE IF NOT EXISTS fallas (
|
|
docentry INTEGER PRIMARY KEY,
|
|
ruc TEXT NOT NULL,
|
|
tipo TEXT NOT NULL,
|
|
serie TEXT NOT NULL,
|
|
numero TEXT NOT NULL,
|
|
error TEXT NOT NULL,
|
|
intentos INTEGER NOT NULL DEFAULT 1,
|
|
proximo_intento TEXT,
|
|
fecha_ultimo TEXT NOT NULL
|
|
)
|
|
""")
|
|
# ============================================================================
|
|
def _calcular_proximo_intento(intentos):
|
|
if intentos >= 20:
|
|
return None
|
|
if intentos == 1:
|
|
delta = timedelta(minutes=0)
|
|
elif intentos == 2:
|
|
delta = timedelta(minutes=5)
|
|
elif intentos == 3:
|
|
delta = timedelta(minutes=15)
|
|
elif intentos == 4:
|
|
delta = timedelta(minutes=30)
|
|
elif intentos == 5:
|
|
delta = timedelta(hours=1)
|
|
elif intentos <= 10:
|
|
delta = timedelta(hours=2)
|
|
elif intentos <= 15:
|
|
delta = timedelta(hours=6)
|
|
else:
|
|
delta = timedelta(hours=12)
|
|
return (datetime.now() + delta).isoformat()
|
|
# ============================================================================
|
|
def registrar_falla(docentry, ruc, tipo_cpe, serie, numero, error, intentos=None):
|
|
ahora = datetime.now().isoformat()
|
|
if intentos is None:
|
|
with _conn() as conn:
|
|
row = conn.execute(
|
|
"SELECT intentos FROM fallas WHERE docentry = ?", (docentry,)
|
|
).fetchone()
|
|
intentos = (row[0] + 1) if row else 1
|
|
proximo = _calcular_proximo_intento(intentos)
|
|
with _conn() as conn:
|
|
conn.execute("""
|
|
INSERT OR REPLACE INTO fallas
|
|
(docentry, ruc, tipo, serie, numero, error, intentos, proximo_intento, fecha_ultimo)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""", (docentry, ruc, tipo_cpe, serie, numero, str(error)[:200], intentos, proximo, ahora))
|
|
# ============================================================================
|
|
def eliminar_falla(docentry):
|
|
with _conn() as conn:
|
|
conn.execute("DELETE FROM fallas WHERE docentry = ?", (docentry,))
|
|
# ============================================================================
|
|
def obtener_fallas_pendientes() -> list[dict]:
|
|
ahora = datetime.now().isoformat()
|
|
with _conn() as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
cursor = conn.execute("""
|
|
SELECT * FROM fallas
|
|
WHERE (proximo_intento IS NULL OR proximo_intento <= ?)
|
|
AND intentos < 20
|
|
ORDER BY intentos ASC, fecha_ultimo ASC
|
|
""", (ahora,))
|
|
return [dict(row) for row in cursor.fetchall()]
|
|
# ============================================================================
|
|
def obtener_docentries_falla() -> set[int]:
|
|
with _conn() as conn:
|
|
cursor = conn.execute("SELECT docentry FROM fallas")
|
|
return {row[0] for row in cursor.fetchall()}
|
|
# ============================================================================
|
|
def total_fallas() -> int:
|
|
with _conn() as conn:
|
|
return conn.execute("SELECT COUNT(*) FROM fallas").fetchone()[0]
|
|
# ============================================================================ |