TDI-Dashboard/Tool/queries/logs.py

"""queries/logs.py — rpa_logs"""

from .db import query, count_q, e, dt, badge


def get_logs(db_path: str) -> list:
    return query(
        db_path,
        "SELECT l.*, COALESCE(p.process_name, 'unknown') AS process_name "
        "FROM rpa_logs l "
        "LEFT JOIN rpa_process p ON p.id = l.rpa_process_id "
        "WHERE date(l.date_log) >= date('now', '-10 days') "
        "ORDER BY l.date_log DESC LIMIT 500"
    )


def get_last_run_ids(db_path: str, n: int = 20) -> list:
    """Returns the last N distinct rpa_process_id values with process_name and start_run date, DESC."""
    return query(
        db_path,
        "SELECT l.rpa_process_id, COALESCE(p.process_name, 'unknown') AS process_name, "
        "COALESCE(p.start_run, '') AS start_run "
        "FROM rpa_logs l "
        "LEFT JOIN rpa_process p ON p.id = l.rpa_process_id "
        "WHERE l.rpa_process_id IS NOT NULL "
        "AND date(l.date_log) >= date('now', '-10 days') "
        "GROUP BY l.rpa_process_id ORDER BY l.rpa_process_id DESC LIMIT ?",
        (n,)
    )


def get_errors_by_process(db_path: str) -> list:
    return query(
        db_path,
        "SELECT process_name, COUNT(*) AS error_count "
        "FROM rpa_process "
        "WHERE completed = 0 AND process_name IS NOT NULL "
        "GROUP BY process_name "
        "ORDER BY error_count DESC LIMIT 15"
    )


def get_logs_by_run_count(db_path: str, run_id: int, search: str = '') -> int:
    if search:
        like = f'%{search}%'
        rows = query(
            db_path,
            "SELECT COUNT(*) AS n FROM rpa_logs l "
            "LEFT JOIN rpa_process p ON p.id = l.rpa_process_id "
            "WHERE l.rpa_process_id=? AND (l.log LIKE ? OR COALESCE(p.process_name,'') LIKE ?)",
            (run_id, like, like)
        )
    else:
        rows = query(db_path, "SELECT COUNT(*) AS n FROM rpa_logs WHERE rpa_process_id=?", (run_id,))
    return rows[0]['n'] if rows else 0


def get_logs_by_run(db_path: str, run_id: int, page: int = 1, per_page: int = 50, search: str = '') -> list:
    offset = (page - 1) * per_page
    if search:
        like = f'%{search}%'
        return query(
            db_path,
            "SELECT l.*, COALESCE(p.process_name, 'unknown') AS process_name "
            "FROM rpa_logs l "
            "LEFT JOIN rpa_process p ON p.id = l.rpa_process_id "
            "WHERE l.rpa_process_id=? AND (l.log LIKE ? OR COALESCE(p.process_name,'') LIKE ?) "
            "ORDER BY l.date_log DESC LIMIT ? OFFSET ?",
            (run_id, like, like, per_page, offset)
        )
    return query(
        db_path,
        "SELECT l.*, COALESCE(p.process_name, 'unknown') AS process_name "
        "FROM rpa_logs l "
        "LEFT JOIN rpa_process p ON p.id = l.rpa_process_id "
        "WHERE l.rpa_process_id=? ORDER BY l.date_log DESC LIMIT ? OFFSET ?",
        (run_id, per_page, offset)
    )


def get_stats(db_path: str, date_from: str = None, date_to: str = None) -> dict:
    conds, params = [], []
    if date_from:
        conds.append("date(date_log) >= ?")
        params.append(date_from)
    if date_to:
        conds.append("date(date_log) <= ?")
        params.append(date_to)
    where = (" WHERE " + " AND ".join(conds)) if conds else ""
    and_pfx = (" AND" if conds else " WHERE")
    t = tuple(params)
    return {
        'log_errors': count_q(
            db_path,
            f"SELECT COUNT(*) as n FROM rpa_logs{where}{and_pfx} type='error'",
            t
        ),
        'log_warnings': count_q(
            db_path,
            f"SELECT COUNT(*) as n FROM rpa_logs{where}{and_pfx} type='warning'",
            t
        ),
    }


def render_table(rows: list, offset: int = 0) -> str:
    if not rows:
        return '<p class="empty">Nessun log trovato.</p>'

    def _badge(tipo):
        m = {'info': ('log', 'Info'), 'warning': ('warn', 'Warning'), 'error': ('err', 'Error')}
        cls, lbl = m.get(tipo, ('', ''))
        return badge(cls, lbl) if cls else e(tipo)

    def _num(i):
        return f'<td class="row-num">{offset + i + 1}</td>'

    trs = ''.join(
        f'<tr data-type="{e(r.get("type", ""))}"'
        f' data-process-name="{e(r.get("process_name", "unknown"))}"'
        f' data-date="{str(r.get("date_log") or "")[:10]}">'
        f'{_num(i)}'
        f'<td>{dt(r.get("date_log"))}</td>'
        f'<td>{_badge(r.get("type"))}</td>'
        f'<td class="nc">{e(r.get("log"))}</td>'
        f'</tr>'
        for i, r in enumerate(rows)
    )
    return (
        '<table><thead><tr>'
        '<th class="row-num">#</th><th>Data</th><th>Tipo</th><th>Messaggio</th>'
        f'</tr></thead><tbody>{trs}</tbody></table>'
    )