generate_html_report.py

"""
Generate an HTML report from a migration JSON report file.

Usage:
    python generate_html_report.py reports/sample_app_migrate_sql_to_snowflake_dev_to_dev_migration_report.json
    python generate_html_report.py reports/sample_app_migrate_sql_to_snowflake_dev_to_dev_migration_report.json --open
"""
from __future__ import annotations

import argparse
import json
import webbrowser
from collections import defaultdict
from datetime import datetime
from pathlib import Path

STATUS_ORDER = {"fail": 0, "warn": 1, "skip": 2, "pass": 3}
STATUS_COLOR = {
    "pass": ("#d4edda", "#155724"),
    "fail": ("#f8d7da", "#721c24"),
    "warn": ("#fff3cd", "#856404"),
    "skip": ("#e2e3e5", "#383d41"),
}
CHECK_ORDER = ["row_count", "schema", "min_max", "checksum_sample", "constraints"]


def _badge(status: str) -> str:
    bg, fg = STATUS_COLOR.get(status, ("#e2e3e5", "#383d41"))
    return f'<span style="background:{bg};color:{fg};padding:2px 8px;border-radius:4px;font-size:0.8em;font-weight:600;text-transform:uppercase">{status}</span>'


def _short_table(label: str) -> str:
    """Extract just the source table name from 'DB.schema.Table -> DB.schema.Table'."""
    src = label.split("->")[0].strip()
    return src.split(".")[-1]


def _summary_cards(results: list[dict]) -> str:
    counts: dict[str, dict[str, int]] = {}
    for r in results:
        c = r["check_name"]
        s = r["status"]
        counts.setdefault(c, {"pass": 0, "fail": 0, "warn": 0, "skip": 0})
        counts[c][s] = counts[c].get(s, 0) + 1

    cards = ""
    for check in CHECK_ORDER:
        if check not in counts:
            continue
        c = counts[check]
        total = sum(c.values())
        fail = c.get("fail", 0)
        warn = c.get("warn", 0)
        passed = c.get("pass", 0)
        color = "#721c24" if fail else ("#856404" if warn else "#155724")
        bg = "#f8d7da" if fail else ("#fff3cd" if warn else "#d4edda")
        label = check.replace("_", " ").title()
        cards += f"""
        <div style="background:{bg};border-radius:8px;padding:16px 20px;min-width:160px;text-align:center">
          <div style="font-size:0.75em;color:{color};font-weight:600;text-transform:uppercase;letter-spacing:1px">{label}</div>
          <div style="font-size:2em;font-weight:700;color:{color}">{passed}/{total}</div>
          <div style="font-size:0.75em;color:{color}">passed{"&nbsp;·&nbsp;" + str(fail) + " failed" if fail else ""}{"&nbsp;·&nbsp;" + str(warn) + " warned" if warn else ""}</div>
        </div>"""
    return cards


def _check_section(check_name: str, results: list[dict]) -> str:
    label = check_name.replace("_", " ").title()
    sorted_results = sorted(results, key=lambda r: STATUS_ORDER.get(r["status"], 9))

    rows = ""
    for r in sorted_results:
        bg, fg = STATUS_COLOR.get(r["status"], ("#fff", "#000"))
        table = _short_table(r["table"])
        full_table = r["table"]
        src_val = r.get("source_value") or ""
        tgt_val = r.get("target_value") or ""
        details = r.get("details", "")

        # Truncate long hash values in source/target display
        def _fmt(v: str) -> str:
            v = str(v)
            return f'<span title="{v}">{v[:16]}…</span>' if len(v) > 20 else v

        # Split details into main part and known_difference note
        kd_note = ""
        display_details = details
        if details.startswith("[known_difference]"):
            parts = details.split(" | ", 1)
            kd_note = f'<div style="margin-top:4px;font-size:0.78em;color:#856404">⚠ {parts[0]}</div>'
            display_details = parts[1] if len(parts) > 1 else ""

        rows += f"""
        <tr style="background:{bg}">
          <td style="padding:8px 12px;font-weight:600;color:{fg};white-space:nowrap" title="{full_table}">{table}</td>
          <td style="padding:8px 12px;text-align:center">{_badge(r["status"])}</td>
          <td style="padding:8px 12px;font-family:monospace;font-size:0.82em">{_fmt(src_val)}</td>
          <td style="padding:8px 12px;font-family:monospace;font-size:0.82em">{_fmt(tgt_val)}</td>
          <td style="padding:8px 12px;font-size:0.82em;color:#444;max-width:500px;word-break:break-word">
            {display_details}{kd_note}
          </td>
        </tr>"""

    fail_count = sum(1 for r in results if r["status"] == "fail")
    warn_count = sum(1 for r in results if r["status"] == "warn")
    header_bg = "#f8d7da" if fail_count else ("#fff3cd" if warn_count else "#d4edda")
    header_fg = "#721c24" if fail_count else ("#856404" if warn_count else "#155724")

    section_id = f"section_{check_name}"
    return f"""
    <div style="margin-bottom:32px">
      <div style="background:{header_bg};color:{header_fg};padding:10px 16px;border-radius:6px 6px 0 0;
                  display:flex;justify-content:space-between;align-items:center;cursor:pointer"
           onclick="toggle('{section_id}')">
        <span style="font-weight:700;font-size:1.05em">{label}</span>
        <span style="font-size:0.85em">{len(results)} tables
          {f'&nbsp;·&nbsp;<b>{fail_count} failed</b>' if fail_count else ''}
          {f'&nbsp;·&nbsp;{warn_count} warned' if warn_count else ''}
        </span>
      </div>
      <div id="{section_id}">
        <table style="width:100%;border-collapse:collapse;font-size:0.9em">
          <thead>
            <tr style="background:#f0f0f0;font-size:0.8em;text-transform:uppercase;letter-spacing:0.5px">
              <th style="padding:8px 12px;text-align:left">Table</th>
              <th style="padding:8px 12px;text-align:center">Status</th>
              <th style="padding:8px 12px;text-align:left">Source Value</th>
              <th style="padding:8px 12px;text-align:left">Target Value</th>
              <th style="padding:8px 12px;text-align:left">Details</th>
            </tr>
          </thead>
          <tbody>{rows}</tbody>
        </table>
      </div>
    </div>"""


def _get_connection_labels(report_path: Path, data: dict) -> tuple[str, str]:
    """Try to read server/account from app yaml. Falls back to empty string."""
    app = data.get("app", "")
    left_env = data.get("left_env", "")
    right_env = data.get("right_env", "")
    try:
        import yaml  # type: ignore
        yaml_path = report_path.parent.parent / "configs" / "apps" / f"{app}.yaml"
        cfg = yaml.safe_load(yaml_path.read_text())
        envs = cfg.get("environments", {})

        def _server(env: str, side: str) -> str:
            systems = envs.get(env, {}).get("systems", {})
            # left = sqlserver, right = snowflake (or both snowflake for promote)
            sys_cfg = systems.get(side) or next(iter(systems.values()), {})
            conn = sys_cfg.get("connection", {})
            return conn.get("server") or conn.get("account") or ""

        left_server = _server(left_env, "sqlserver") or _server(left_env, list(
            envs.get(left_env, {}).get("systems", {}).keys() or [""])[0])
        right_server = _server(right_env, "snowflake") or _server(right_env, list(
            envs.get(right_env, {}).get("systems", {}).keys() or [""])[0])
        return left_server, right_server
    except Exception:
        return "", ""


def generate(report_path: Path) -> str:
    data = json.loads(report_path.read_text())
    results: list[dict] = data.get("table_results", [])
    summary = data.get("summary", {})
    generated_at = datetime.now().strftime("%Y-%m-%d %H:%M")
    left_server, right_server = _get_connection_labels(report_path, data)

    # Group by check_name
    by_check: dict[str, list[dict]] = defaultdict(list)
    for r in results:
        by_check[r["check_name"]].append(r)

    total = summary.get("table_total", len(results))
    failed = summary.get("table_failed", 0)
    warned = sum(1 for r in results if r["status"] == "warn")
    passed = sum(1 for r in results if r["status"] == "pass")

    skipped = sum(1 for r in results if r["status"] == "skip")
    overall_color = "#721c24" if failed else ("#856404" if warned else "#155724")
    overall_bg = "#f8d7da" if failed else ("#fff3cd" if warned else "#d4edda")
    overall_label = "Status: FAILURES DETECTED" if failed else ("Status: WARNINGS DETECTED" if warned else "Status: PASS")

    summary_cards = _summary_cards(results)

    sections = ""
    for check in CHECK_ORDER:
        if check in by_check:
            sections += _check_section(check, by_check[check])
    # Any extra checks not in ORDER
    for check, items in by_check.items():
        if check not in CHECK_ORDER:
            sections += _check_section(check, items)

    return f"""<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <title>Migration Report — {data.get("app","").upper()} {data.get("flow","")}</title>
  <style>
    body {{ font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif; margin:0; background:#f5f5f5; color:#222; }}
    .container {{ max-width:1200px; margin:0 auto; padding:24px; }}
    table {{ border-collapse:collapse; }}
    tbody tr:hover {{ filter:brightness(0.97); }}
  </style>
</head>
<body>
<div class="container">

  <!-- Header -->
  <div style="background:#1a1a2e;color:#fff;padding:24px 28px;border-radius:8px;margin-bottom:24px">
    <div style="font-size:0.8em;opacity:0.6;text-transform:uppercase;letter-spacing:1px">Migration Validation Report</div>
    <div style="font-size:1.6em;font-weight:700;margin:4px 0">
      {data.get("app","").upper()} &mdash; {data.get("flow","").replace("_"," ").title()}
    </div>
    <div style="font-size:0.85em;opacity:0.7">
      {data.get("left_env","").upper()}{f" ({left_server})" if left_server else ""}
      → {data.get("right_env","").upper()}{f" ({right_server})" if right_server else ""}
      &nbsp;·&nbsp; Generated {generated_at}
    </div>
  </div>

  <!-- Overall status -->
  <div style="background:{overall_bg};color:{overall_color};padding:14px 20px;border-radius:8px;
              margin-bottom:24px;font-weight:700;font-size:1.05em;display:flex;justify-content:space-between">
    <span>{overall_label}</span>
    <span>{passed} passed &nbsp;·&nbsp; {warned} warned &nbsp;·&nbsp; {failed} failed &nbsp;·&nbsp; {skipped} skipped &nbsp;·&nbsp; {total} total checks</span>
  </div>

  <!-- Summary cards by check type -->
  <div style="display:flex;gap:16px;flex-wrap:wrap;margin-bottom:32px">
    {summary_cards}
  </div>

  <!-- Check sections -->
  {sections}

</div>
<script>
  function toggle(id) {{
    var el = document.getElementById(id);
    el.style.display = el.style.display === 'none' ? '' : 'none';
  }}
</script>
</body>
</html>"""


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("report", help="Path to migration JSON report")
    parser.add_argument("--open", action="store_true", help="Open in browser after generating")
    args = parser.parse_args()

    report_path = Path(args.report)
    out_path = report_path.with_suffix(".html")
    out_path.write_text(generate(report_path), encoding="utf-8")
    print(f"Report written to: {out_path}")

    if args.open:
        webbrowser.open(out_path.resolve().as_uri())


if __name__ == "__main__":
    main()