Yan Xell
167 lines
6.3 KiB
Python
167 lines
6.3 KiB
Python
"""queries/iscrizioni.py — corsi intraziendali (Intraz: rpa_intra_ftp_json + rpa_intra_steps_status)"""
|
|
|
|
from .db import query, e, dt, badge
|
|
|
|
|
|
|
|
_STEP_BADGE = {
|
|
'done': lambda: badge('ok', 'done'),
|
|
'processing': lambda: badge('proc', 'processing'),
|
|
'pending': lambda: badge('warn', 'pending'),
|
|
'error': lambda: badge('err', 'error'),
|
|
'skipped': lambda: badge('info', 'skipped'),
|
|
}
|
|
|
|
def _sb(v) -> str:
|
|
fn = _STEP_BADGE.get(v)
|
|
return fn() if fn else (badge('warn', 'pending') if not v else e(v))
|
|
|
|
|
|
def get_corsi(db_path: str) -> list:
|
|
return query(db_path, """
|
|
SELECT
|
|
f.id,
|
|
f.rpa_process_id,
|
|
f.company_code,
|
|
m.ragione_sociale AS sorgente,
|
|
json_extract(f.azienda, '$.ragione_sociale') AS azienda_nome,
|
|
json_extract(f.corso, '$.IDsessione') AS sessione_id,
|
|
json_extract(f.corso, '$.descrizione') AS corso_descrizione,
|
|
json_extract(f.corso, '$.quando_start') AS corso_data,
|
|
json_extract(f.corso, '$.n_iscritti') AS n_iscritti,
|
|
s.step_ftp, s.err_ftp,
|
|
s.step_bc, s.err_bc,
|
|
s.step_iscrizione, s.err_iscrizione,
|
|
s.step_odv, s.err_odv,
|
|
s.step_sharepoint, s.err_sharepoint,
|
|
s.step_report, s.err_report,
|
|
s.step_email, s.err_email,
|
|
f.created_at
|
|
FROM rpa_intra_ftp_json f
|
|
LEFT JOIN rpa_mapping_aziende m ON m.company_code = f.company_code
|
|
LEFT JOIN rpa_intra_steps_status s ON s.rpa_intra_ftp_json_id = f.id
|
|
ORDER BY f.id DESC
|
|
""")
|
|
|
|
|
|
_ALL_STEPS = ('step_ftp', 'step_bc', 'step_iscrizione', 'step_odv',
|
|
'step_sharepoint', 'step_report', 'step_email')
|
|
_ALL_ERR_COLS = tuple(s.replace('step_', 'err_') for s in _ALL_STEPS)
|
|
|
|
|
|
def _err_note(r: dict) -> str:
|
|
parts = []
|
|
for step, err_col in zip(_ALL_STEPS, _ALL_ERR_COLS):
|
|
val = r.get(err_col)
|
|
if val:
|
|
parts.append(f"{step.replace('step_', '')}: {val}")
|
|
return " | ".join(parts)
|
|
|
|
|
|
def _step_cell(step_val, err_val) -> str:
|
|
badge_html = _sb(step_val)
|
|
if err_val:
|
|
return f'{badge_html}<br><small class="step-err" title="{e(err_val)}">{_clip(err_val, 30)}</small>'
|
|
return badge_html
|
|
|
|
|
|
def get_stats(db_path: str) -> dict:
|
|
rows = get_corsi(db_path)
|
|
return {
|
|
'isc_total': len(rows),
|
|
'isc_ok': sum(1 for r in rows if all(r.get(s) in ('done', 'skipped') for s in _ALL_STEPS)),
|
|
'isc_errore': sum(1 for r in rows if any(r.get(s) == 'error' for s in _ALL_STEPS)),
|
|
'isc_pending': sum(1 for r in rows if _row_status(r) in ('pending', 'processing', 'skipped')),
|
|
}
|
|
|
|
|
|
def _row_status(r: dict) -> str:
|
|
if any(r.get(s) == 'error' for s in _ALL_STEPS):
|
|
return 'error'
|
|
if all(r.get(s) in ('done', 'skipped') for s in _ALL_STEPS):
|
|
return 'done'
|
|
if any(r.get(s) == 'processing' for s in _ALL_STEPS):
|
|
return 'processing'
|
|
if any(r.get(s) == 'skipped' for s in _ALL_STEPS):
|
|
return 'skipped'
|
|
return 'pending'
|
|
|
|
|
|
|
|
|
|
def _fth(label: str, col: str) -> str:
|
|
return f'<th>{label}<button class="col-filter-btn" data-col="{col}" title="Filtra {label}">⚙</button></th>'
|
|
|
|
|
|
def _clip(text: str, n: int = 20) -> str:
|
|
safe = e(text)
|
|
if len(text) <= n:
|
|
return safe
|
|
return (
|
|
f'<span class="txt-short">{e(text[:n])}'
|
|
f'<button class="btn-expand" title="{safe}">⋯</button></span>'
|
|
f'<span class="txt-full" hidden>{safe}</span>'
|
|
)
|
|
|
|
|
|
def render_table(rows: list) -> str:
|
|
if not rows:
|
|
return '<p class="empty">Nessun corso trovato.</p>'
|
|
|
|
trs = ''.join(
|
|
f'<tr'
|
|
f' data-status="{_row_status(r)}"'
|
|
f' data-sorgente="{e(r.get("sorgente") or "")}"'
|
|
f' data-sessione="{e(r.get("sessione_id") or "")}"'
|
|
f' data-azienda="{e(r.get("azienda_nome") or "")}"'
|
|
f' data-corso="{e(r.get("corso_descrizione") or "")}"'
|
|
f' data-data="{e(r.get("corso_data") or "")}"'
|
|
f' data-ftp="{e(r.get("step_ftp") or "pending")}"'
|
|
f' data-bc="{e(r.get("step_bc") or "pending")}"'
|
|
f' data-iscrizione="{e(r.get("step_iscrizione") or "pending")}"'
|
|
f' data-odv="{e(r.get("step_odv") or "pending")}"'
|
|
f' data-sharepoint="{e(r.get("step_sharepoint") or "pending")}"'
|
|
f' data-report="{e(r.get("step_report") or "pending")}"'
|
|
f' data-email="{e(r.get("step_email") or "pending")}"'
|
|
f'>'
|
|
f'<td>{e(r.get("id"))}</td>'
|
|
f'<td>{e(r.get("sorgente"))}</td>'
|
|
f'<td>{e(r.get("azienda_nome"))}</td>'
|
|
f'<td>{e(r.get("sessione_id"))}</td>'
|
|
f'<td class="nc">{_clip(r.get("corso_descrizione") or "")}</td>'
|
|
f'<td>{e(r.get("corso_data"))}</td>'
|
|
f'<td>{e(r.get("n_iscritti"))}</td>'
|
|
f'<td class="step-td">{_step_cell(r.get("step_ftp"), r.get("err_ftp"))}</td>'
|
|
f'<td class="step-td">{_step_cell(r.get("step_bc"), r.get("err_bc"))}</td>'
|
|
f'<td class="step-td">{_step_cell(r.get("step_iscrizione"), r.get("err_iscrizione"))}</td>'
|
|
f'<td class="step-td">{_step_cell(r.get("step_odv"), r.get("err_odv"))}</td>'
|
|
f'<td class="step-td">{_step_cell(r.get("step_sharepoint"), r.get("err_sharepoint"))}</td>'
|
|
f'<td class="step-td">{_step_cell(r.get("step_report"), r.get("err_report"))}</td>'
|
|
f'<td class="step-td">{_step_cell(r.get("step_email"), r.get("err_email"))}</td>'
|
|
f'</tr>'
|
|
for r in rows
|
|
)
|
|
return (
|
|
'<table><thead><tr>'
|
|
f'<th>#</th>'
|
|
f'{_fth("Sorgente", "sorgente")}'
|
|
f'{_fth("Azienda", "azienda")}'
|
|
f'{_fth("Sessione", "sessione")}'
|
|
f'{_fth("Corso", "corso")}'
|
|
f'{_fth("Data", "data")}'
|
|
f'<th>N.</th>'
|
|
f'{_fth("FTP", "ftp")}'
|
|
f'{_fth("BC", "bc")}'
|
|
f'{_fth("Iscrizione", "iscrizione")}'
|
|
f'{_fth("ODV", "odv")}'
|
|
f'{_fth("SharePoint", "sharepoint")}'
|
|
f'{_fth("Report", "report")}'
|
|
f'{_fth("Email", "email")}'
|
|
f'</tr></thead><tbody>{trs}</tbody></table>'
|
|
)
|
|
|
|
|
|
# backward-compat alias used by Reports.py
|
|
def get_iscrizioni(db_path: str) -> list:
|
|
return get_corsi(db_path)
|