Luca Banfi
73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
"""queries/processes.py — rpa_process"""
|
|
|
|
from .db import query, count_q, e, dt, dur, badge
|
|
|
|
|
|
def get_processes(db_path: str) -> list:
|
|
return query(
|
|
db_path,
|
|
"SELECT id, process_name, note, completed, start_run, finish_run FROM rpa_process "
|
|
"WHERE date(start_run) >= date('now', '-10 days') "
|
|
"ORDER BY id DESC LIMIT 50"
|
|
)
|
|
|
|
|
|
def get_stats(db_path: str, date_from: str = None, date_to: str = None) -> dict:
|
|
conds, params = [], []
|
|
if date_from:
|
|
conds.append("date(start_run) >= ?")
|
|
params.append(date_from)
|
|
if date_to:
|
|
conds.append("date(start_run) <= ?")
|
|
params.append(date_to)
|
|
where = (" WHERE " + " AND ".join(conds)) if conds else ""
|
|
and_pfx = (" AND" if conds else " WHERE")
|
|
t = tuple(params)
|
|
total = count_q(db_path, f"SELECT COUNT(*) as n FROM rpa_process{where}", t)
|
|
completed = count_q(db_path, f"SELECT COUNT(*) as n FROM rpa_process{where}{and_pfx} completed=1", t)
|
|
return {
|
|
'total_runs': total,
|
|
'completed_runs': completed,
|
|
'incomplete_runs': total - completed,
|
|
}
|
|
|
|
|
|
def get_schema_stats(db_path: str) -> list:
|
|
return query(db_path, """
|
|
SELECT process_name,
|
|
COUNT(*) as run_count,
|
|
SUM(CASE WHEN start_run IS NOT NULL AND finish_run IS NOT NULL
|
|
THEN CAST((julianday(finish_run) - julianday(start_run)) * 86400 AS INTEGER)
|
|
ELSE 0 END) as total_seconds
|
|
FROM rpa_process
|
|
GROUP BY process_name
|
|
ORDER BY run_count DESC
|
|
""")
|
|
|
|
|
|
def render_table(rows: list) -> str:
|
|
if not rows:
|
|
return '<p class="empty">Nessun processo trovato.</p>'
|
|
|
|
def _badge(completed):
|
|
return badge('ok', 'Completato') if completed in (1, True, '1') else badge('err', 'Incompleto')
|
|
|
|
trs = ''.join(
|
|
f'<tr data-completed="{1 if r.get("completed") in (1, True, "1") else 0}"'
|
|
f' data-process-name="{e(r.get("process_name", "unknown"))}">'
|
|
f'<td><span class="proc-name-badge">{e(r.get("process_name", "—"))}</span></td>'
|
|
f'<td>{dt(r.get("start_run"))}</td>'
|
|
f'<td>{dt(r.get("finish_run"))}</td>'
|
|
f'<td>{dur(r.get("start_run"), r.get("finish_run"))}</td>'
|
|
f'<td>{_badge(r.get("completed"))}</td>'
|
|
f'<td class="nc">{e(r.get("note"))}</td>'
|
|
f'<td><a href="/logs?run={e(r.get("id"))}" class="btn-log">→ Log</a></td>'
|
|
f'</tr>'
|
|
for r in rows
|
|
)
|
|
return (
|
|
'<table><thead><tr>'
|
|
'<th>Processo</th><th>Inizio</th><th>Fine</th><th>Durata</th><th>Stato</th><th>Note</th><th></th>'
|
|
f'</tr></thead><tbody>{trs}</tbody></table>'
|
|
)
|