Luca Banfi
86 lines
2.3 KiB
Python
86 lines
2.3 KiB
Python
"""queries/db.py — helper condiviso per query SQLite"""
|
|
|
|
import sqlite3
|
|
from datetime import datetime
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from ..config import DbType
|
|
|
|
|
|
def query(db_path: str, sql: str, params: tuple = ()) -> list:
|
|
conn = sqlite3.connect(db_path)
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute(sql, params)
|
|
return [dict(r) for r in cur.fetchall()]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def count(db_path: str, sql: str) -> int:
|
|
rows = query(db_path, sql)
|
|
return rows[0]['n'] if rows else 0
|
|
|
|
|
|
def count_q(db_path: str, sql: str, params: tuple = ()) -> int:
|
|
rows = query(db_path, sql, params)
|
|
return rows[0]['n'] if rows else 0
|
|
|
|
|
|
# ------------------------------------------------------------------ HTML utils
|
|
|
|
def e(v) -> str:
|
|
if v is None:
|
|
return '-'
|
|
return str(v).replace('&', '&').replace('<', '<').replace('>', '>').replace('"', '"')
|
|
|
|
|
|
def dt(v) -> str:
|
|
if not v:
|
|
return '-'
|
|
try:
|
|
return datetime.fromisoformat(str(v)).strftime('%d/%m/%Y %H:%M:%S')
|
|
except Exception:
|
|
return str(v)
|
|
|
|
|
|
def d(v) -> str:
|
|
if not v:
|
|
return '-'
|
|
try:
|
|
return datetime.fromisoformat(str(v)).strftime('%d/%m/%Y')
|
|
except Exception:
|
|
return str(v)
|
|
|
|
|
|
def dur(start, finish) -> str:
|
|
if not start or not finish:
|
|
return '-'
|
|
try:
|
|
s = int((datetime.fromisoformat(str(finish)) - datetime.fromisoformat(str(start))).total_seconds())
|
|
return f"{s // 60}m {s % 60}s" if s >= 60 else f"{s}s"
|
|
except Exception:
|
|
return '-'
|
|
|
|
|
|
def badge(cls: str, text: str) -> str:
|
|
return f'<span class="badge {cls}">{text}</span>'
|
|
|
|
|
|
def detect_db(db_path: str) -> 'DbType':
|
|
"""Returns DB_TYPE_REG_LOMB, DB_TYPE_INTRAZ, or DB_TYPE_UNKNOWN based on DB schema.
|
|
|
|
reg_Lomb (db_reg_lombardia) → has table 'sessione_documenti'
|
|
Intraz (db_corsi_intraziendali) → has table 'rpa_intra_api_iscrizione'
|
|
"""
|
|
from ..config import DbType
|
|
rows = query(db_path, "SELECT name FROM sqlite_master WHERE type='table'")
|
|
tables = {r['name'] for r in rows}
|
|
if 'sessione_documenti' in tables:
|
|
return DbType.REG_LOMB
|
|
if 'rpa_intra_api_iscrizione' in tables:
|
|
return DbType.INTRAZ
|
|
return DbType.UNKNOWN
|