appfactory-portal/app/db/migration_readiness.py

import json
import re
from typing import Any

from app.db.database import get_connection
from app.db.migrations import run_migrations

PREFLIGHT_SCRIPT = "preflight-check.sh"


def _job_matches_script(row: dict[str, Any]) -> bool:
    if row.get("target_id") == PREFLIGHT_SCRIPT:
        return True

    try:
        payload = json.loads(row.get("payload_json") or "{}")
    except (TypeError, ValueError):
        return False

    return payload.get("script_name") == PREFLIGHT_SCRIPT


def get_latest_preflight_job():
    run_migrations()
    con = get_connection()

    rows = con.execute(
        """
        SELECT *
        FROM jobs
        WHERE type = 'run_script'
          AND (
            target_id = ?
            OR payload_json LIKE ?
          )
        ORDER BY id DESC
        LIMIT 20
        """,
        (PREFLIGHT_SCRIPT, f'%"{PREFLIGHT_SCRIPT}"%'),
    ).fetchall()

    con.close()
    for row in rows:
        job = dict(row)
        if _job_matches_script(job):
            return job
    return None


def get_job_log_output(job_id: int) -> str:
    run_migrations()
    con = get_connection()

    rows = con.execute(
        """
        SELECT stream, message
        FROM job_logs
        WHERE job_id = ?
        ORDER BY id
        """,
        (job_id,),
    ).fetchall()

    con.close()
    return "\n".join(f"[{row['stream'] or 'log'}] {row['message'] or ''}" for row in rows)


def _count_token(text: str, token: str) -> int:
    return len(re.findall(rf"(?<![A-Z0-9_]){re.escape(token)}(?![A-Z0-9_])", text.upper()))


def _result_data(job: dict[str, Any]) -> dict[str, Any]:
    try:
        value = json.loads(job.get("result_json") or "{}")
    except (TypeError, ValueError):
        return {}
    return value if isinstance(value, dict) else {}


def _first_int(data: dict[str, Any], keys: tuple[str, ...]) -> int | None:
    for key in keys:
        value = data.get(key)
        if value is None:
            continue
        try:
            return int(value)
        except (TypeError, ValueError):
            continue
    return None


def _readiness_from_script_output(result_data: dict[str, Any], combined_text: str) -> bool:
    ready_value = result_data.get("ready")
    if isinstance(ready_value, bool):
        return ready_value

    readiness_value = str(
        result_data.get("readiness")
        or result_data.get("migration_readiness")
        or result_data.get("migrationReadiness")
        or ""
    ).strip().upper()

    if readiness_value == "READY":
        return True
    if readiness_value == "NOT READY":
        return False

    upper_text = combined_text.upper()
    if re.search(r"(?<![A-Z0-9_])NOT\s+READY(?![A-Z0-9_])", upper_text):
        return False
    if re.search(r"(?<![A-Z0-9_])READY(?![A-Z0-9_])", upper_text):
        return True

    return False


def summarize_readiness(job: dict[str, Any] | None) -> dict[str, Any]:
    if not job:
        return {
            "ready": False,
            "ok_count": 0,
            "warn_count": 0,
            "fail_count": 0,
            "raw_output": "",
            "job": None,
        }

    raw_output = get_job_log_output(int(job["id"]))
    result_data = _result_data(job)
    combined_text = "\n".join(
        value
        for value in (
            raw_output,
            job.get("result_json") or "",
            job.get("error_text") or "",
        )
        if value
    )

    ok_count = _first_int(result_data, ("ok", "ok_count", "okCount"))
    warn_count = _first_int(result_data, ("warn", "warn_count", "warnCount", "warnings"))
    fail_count = _first_int(result_data, ("fail", "fail_count", "failCount", "failed", "errors"))

    if ok_count is None:
        ok_count = _count_token(combined_text, "OK")
    if warn_count is None:
        warn_count = _count_token(combined_text, "WARN")
    if fail_count is None:
        fail_count = _count_token(combined_text, "FAIL")

    return {
        "ready": _readiness_from_script_output(result_data, combined_text),
        "ok_count": ok_count,
        "warn_count": warn_count,
        "fail_count": fail_count,
        "raw_output": raw_output,
        "job": job,
    }