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"
BOOTSTRAP_SCRIPT = "bootstrap-v2.sh"
SENSITIVE_KEY_RE = re.compile(
    r"(SECRET|TOKEN|PASSWORD|PASS|KEY|CREDENTIAL|AUTH|API_KEY|PRIVATE)",
    re.IGNORECASE,
)


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

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

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


def get_latest_script_job(script_name: str):
    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
        """,
        (script_name, f'%"{script_name}"%'),
    ).fetchall()

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


def get_latest_preflight_job():
    return get_latest_script_job(PREFLIGHT_SCRIPT)


def get_latest_bootstrap_job():
    return get_latest_script_job(BOOTSTRAP_SCRIPT)


def get_job_log_entries(job_id: int) -> list[dict[str, Any]]:
    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 [dict(row) for row in rows]


def _redact_sensitive_text(value: str) -> str:
    redacted_lines = []
    for line in value.splitlines():
        env_match = re.match(r"^(\s*(?:export\s+)?([A-Z_][A-Z0-9_]*)\s*=\s*)(.*)$", line)
        if env_match and SENSITIVE_KEY_RE.search(env_match.group(2)):
            redacted_lines.append(f"{env_match.group(1)}[redacted]")
            continue
        json_match = re.match(r'^(\s*["\']?([^"\':=]+)["\']?\s*[:=]\s*)(.*)$', line)
        if json_match and SENSITIVE_KEY_RE.search(json_match.group(2)):
            redacted_lines.append(f"{json_match.group(1)}[redacted]")
            continue
        redacted_lines.append(line)
    return "\n".join(redacted_lines)


def get_job_log_output(job_id: int, stream: str | None = None, include_stream: bool = True) -> str:
    rows = get_job_log_entries(job_id)
    lines = []
    for row in rows:
        row_stream = row.get("stream") or "log"
        if stream and row_stream != stream:
            continue
        message = _redact_sensitive_text(row.get("message") or "")
        lines.append(f"[{row_stream}] {message}" if include_stream else message)
    return "\n".join(lines)


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 _result_value(job: dict[str, Any], keys: tuple[str, ...]) -> Any:
    data = _result_data(job)
    for key in keys:
        if key in data:
            return data.get(key)
    return None


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,
    }


def summarize_script_run(job: dict[str, Any] | None) -> dict[str, Any]:
    if not job:
        return {
            "job": None,
            "return_code": None,
            "stdout": "",
            "stderr": "",
        }

    return_code = _result_value(job, ("returncode", "return_code", "exit_code", "exitCode"))
    stdout = _result_value(job, ("stdout", "out"))
    stderr = _result_value(job, ("stderr", "err"))

    if stdout is None:
        stdout = get_job_log_output(int(job["id"]), stream="stdout", include_stream=False)
    if stderr is None:
        stderr = get_job_log_output(int(job["id"]), stream="stderr", include_stream=False)

    error_text = job.get("error_text") or ""
    if error_text:
        stderr = f"{stderr}\n{error_text}" if stderr else error_text

    return {
        "job": job,
        "return_code": return_code,
        "stdout": _redact_sensitive_text(str(stdout or "")),
        "stderr": _redact_sensitive_text(str(stderr or "")),
    }