JKTV-online/database/job_store.py

import sqlite3
from typing import Any, Dict, List, Optional

from database.paths import WEB_DB


class JobStore:
    def __init__(self, database_path=WEB_DB):
        self.database_path = str(database_path)

    def _connect(self):
        db = sqlite3.connect(self.database_path, timeout=30)
        db.row_factory = sqlite3.Row
        db.execute('PRAGMA busy_timeout = 30000')
        db.execute('PRAGMA foreign_keys = ON')
        return db

    def create_job(
        self,
        job_type: str,
        match_id: Optional[str] = None,
        input_path: Optional[str] = None,
        created_by: Optional[str] = None,
    ) -> int:
        db = self._connect()
        try:
            cursor = db.execute(
                """
                INSERT INTO etl_jobs (
                    job_type, match_id, input_path, created_by
                ) VALUES (?, ?, ?, ?)
                """,
                [job_type, match_id, input_path, created_by],
            )
            db.commit()
            return int(cursor.lastrowid)
        finally:
            db.close()

    def get_job(self, job_id: int) -> Optional[Dict[str, Any]]:
        db = self._connect()
        try:
            row = db.execute(
                'SELECT * FROM etl_jobs WHERE id = ?',
                [job_id],
            ).fetchone()
            return dict(row) if row else None
        finally:
            db.close()

    def list_jobs(
        self,
        limit: int = 20,
        status: Optional[str] = None,
        job_type: Optional[str] = None,
        offset: int = 0,
    ) -> List[Dict[str, Any]]:
        db = self._connect()
        try:
            where = []
            args = []
            if status:
                where.append('status = ?')
                args.append(status)
            if job_type:
                where.append('job_type = ?')
                args.append(job_type)
            where_sql = f"WHERE {' AND '.join(where)}" if where else ''
            args.extend([
                max(1, min(int(limit), 100)),
                max(0, int(offset)),
            ])
            rows = db.execute(
                f"""
                SELECT *
                FROM etl_jobs
                {where_sql}
                ORDER BY created_at DESC, id DESC
                LIMIT ? OFFSET ?
                """,
                args,
            ).fetchall()
            return [dict(row) for row in rows]
        finally:
            db.close()

    def get_summary(self) -> Dict[str, int]:
        db = self._connect()
        try:
            rows = db.execute(
                """
                SELECT status, COUNT(*) AS count
                FROM etl_jobs
                GROUP BY status
                """
            ).fetchall()
            summary = {
                'total': 0,
                'queued': 0,
                'running': 0,
                'succeeded': 0,
                'failed': 0,
            }
            for row in rows:
                summary[row['status']] = int(row['count'])
                summary['total'] += int(row['count'])
            return summary
        finally:
            db.close()

    def start_job(self, job_id: int, stage: str, message: str):
        db = self._connect()
        try:
            db.execute(
                """
                UPDATE etl_jobs
                SET status = 'running',
                    current_stage = ?,
                    progress = 1,
                    message = ?,
                    started_at = CURRENT_TIMESTAMP
                WHERE id = ? AND status = 'queued'
                """,
                [stage, message, job_id],
            )
            db.commit()
        finally:
            db.close()

    def update_progress(
        self,
        job_id: int,
        stage: str,
        progress: int,
        message: str,
    ):
        db = self._connect()
        try:
            db.execute(
                """
                UPDATE etl_jobs
                SET current_stage = ?, progress = ?, message = ?
                WHERE id = ?
                """,
                [stage, max(0, min(int(progress), 100)), message, job_id],
            )
            db.commit()
        finally:
            db.close()

    def append_log(self, job_id: int, text: str):
        if not text:
            return
        db = self._connect()
        try:
            db.execute(
                """
                UPDATE etl_jobs
                SET log_text = substr(log_text || ?, -100000)
                WHERE id = ?
                """,
                [text, job_id],
            )
            db.commit()
        finally:
            db.close()

    def finish_job(
        self,
        job_id: int,
        succeeded: bool,
        message: str,
        duration_seconds: float,
    ):
        status = 'succeeded' if succeeded else 'failed'
        db = self._connect()
        try:
            db.execute(
                """
                UPDATE etl_jobs
                SET status = ?,
                    current_stage = ?,
                    progress = ?,
                    message = ?,
                    finished_at = CURRENT_TIMESTAMP,
                    duration_seconds = ?
                WHERE id = ?
                """,
                [
                    status,
                    'complete' if succeeded else 'failed',
                    100 if succeeded else 0,
                    message,
                    round(float(duration_seconds), 3),
                    job_id,
                ],
            )
            db.execute(
                """
                UPDATE match_imports
                SET status = ?,
                    imported_at = CASE
                        WHEN ? = 'succeeded' THEN CURRENT_TIMESTAMP
                        ELSE imported_at
                    END,
                    updated_at = CURRENT_TIMESTAMP
                WHERE job_id = ?
                """,
                [status, status, job_id],
            )
            db.commit()
        finally:
            db.close()

    def upsert_match_import(
        self,
        match_id: str,
        content_sha256: str,
        source_path: str,
        status: str,
        job_id: int,
    ):
        db = self._connect()
        try:
            db.execute(
                """
                INSERT INTO match_imports (
                    match_id, content_sha256, source_path, status, job_id
                ) VALUES (?, ?, ?, ?, ?)
                ON CONFLICT(match_id) DO UPDATE SET
                    content_sha256 = excluded.content_sha256,
                    source_path = excluded.source_path,
                    status = excluded.status,
                    job_id = excluded.job_id,
                    updated_at = CURRENT_TIMESTAMP
                """,
                [match_id, content_sha256, source_path, status, job_id],
            )
            db.commit()
        finally:
            db.close()