ewoooc/scripts/check_observability_pages.py

#!/usr/bin/env python3
"""Production smoke check for AI observability pages.

The goal is to catch broken observability pages quickly after UI, route,
schema, or deployment changes. It verifies the ten war-room pages return HTTP
200 and do not expose raw framework/database errors to the user.
"""

from __future__ import annotations

import argparse
import sys
import urllib.error
import urllib.request


PAGES = [
    ("/observability/overview", "總覽"),
    ("/observability/agent_orchestration", "Agent"),
    ("/observability/business_intel", "商業"),
    ("/observability/host_health", "主機"),
    ("/observability/ai_calls", "AI 呼叫"),
    ("/observability/budget", "預算"),
    ("/observability/promotion_review", "晉升"),
    ("/observability/rag_queries", "RAG"),
    ("/observability/quality_trend", "品質"),
    ("/observability/ppt_audit_history", "PPT"),
]

EXPECTED_MARKERS = {
    "/observability/overview": ["觀測台總覽", "主機健康", "AI 呼叫"],
    "/observability/agent_orchestration": ["Agent 編排矩陣", "LLM", "MCP", "RAG"],
    "/observability/business_intel": ["商業面 × AI", "AI", "競品"],
    "/observability/host_health": ["主機健康", "Ollama", "AutoHeal"],
    "/observability/ai_calls": ["AI 呼叫", "Provider", "RAG"],
    "/observability/budget": ["預算控管", "force", "throttle"],
    "/observability/promotion_review": ["RAG 晉升審核", "Promotion", "ai_insights"],
    "/observability/rag_queries": ["RAG 召回詳情", "最近 50", "hits"],
    "/observability/quality_trend": ["反饋趨勢", "Caller", "蒸餾"],
    "/observability/ppt_audit_history": ["PPT 視覺審核", "AiderHeal", "audit"],
}

MIN_HTML_BYTES = 2500

GLOBAL_REQUIRED_MARKERS = [
    "momo-observability-mode",
    "observability-system.css",
    "AI 觀測台",
    "momo-obs-link",
]

ASSET_CHECKS = [
    (
        "/static/css/observability-system.css",
        "觀測台 CSS",
        [
            "--obs-title-size",
            "--obs-value-size",
            ".momo-observability-mode",
            ".obs-chart-frame",
            ".obs-modal-preview",
        ],
    ),
]

ERROR_NEEDLES = [
    "Traceback",
    "UndefinedError",
    "ProgrammingError",
    "Internal Server Error",
    'relation "',
    "relation "",
    "查詢失敗:",
]


def fetch_page(base_url: str, path: str, timeout: int) -> tuple[int, str]:
    request = urllib.request.Request(
        base_url.rstrip("/") + path,
        headers={"User-Agent": "momo-observability-smoke/1.0"},
    )
    with urllib.request.urlopen(request, timeout=timeout) as response:
        html = response.read().decode("utf-8", "ignore")
        return response.status, html


def main() -> int:
    parser = argparse.ArgumentParser(description="Smoke check AI observability pages")
    parser.add_argument("--base-url", default="https://mo.wooo.work")
    parser.add_argument("--timeout", type=int, default=12)
    args = parser.parse_args()

    failed = False
    print(f"Observability page smoke: {args.base_url.rstrip('/')}")

    for path, label in PAGES:
        try:
            status, html = fetch_page(args.base_url, path, args.timeout)
        except urllib.error.HTTPError as exc:
            print(f"- {label}: HTTP {exc.code}, FAIL")
            failed = True
            continue
        except Exception as exc:
            print(f"- {label}: {type(exc).__name__}: {exc}, FAIL")
            failed = True
            continue

        found = [needle for needle in ERROR_NEEDLES if needle in html]
        missing_markers = [
            marker for marker in EXPECTED_MARKERS.get(path, [])
            if marker not in html
        ]
        missing_global_markers = [
            marker for marker in GLOBAL_REQUIRED_MARKERS
            if marker not in html
        ]
        too_small = len(html.encode("utf-8")) < MIN_HTML_BYTES
        if status != 200 or found or missing_markers or missing_global_markers or too_small:
            issues = []
            if status != 200:
                issues.append("bad_status")
            if found:
                issues.extend(found)
            if missing_markers:
                issues.append(f"missing_markers={missing_markers}")
            if missing_global_markers:
                issues.append(f"missing_global_markers={missing_global_markers}")
            if too_small:
                issues.append(f"html_too_small={len(html.encode('utf-8'))}B")
            print(f"- {label}: HTTP {status}, issues={issues}, FAIL")
            failed = True
        else:
            print(f"- {label}: HTTP {status}, issues=none")

    for path, label, markers in ASSET_CHECKS:
        try:
            status, body = fetch_page(args.base_url, path, args.timeout)
        except urllib.error.HTTPError as exc:
            print(f"- {label}: HTTP {exc.code}, FAIL")
            failed = True
            continue
        except Exception as exc:
            print(f"- {label}: {type(exc).__name__}: {exc}, FAIL")
            failed = True
            continue

        missing_markers = [marker for marker in markers if marker not in body]
        if status != 200 or missing_markers:
            issues = []
            if status != 200:
                issues.append("bad_status")
            if missing_markers:
                issues.append(f"missing_asset_markers={missing_markers}")
            print(f"- {label}: HTTP {status}, issues={issues}, FAIL")
            failed = True
        else:
            print(f"- {label}: HTTP {status}, markers=ok")

    if failed:
        print("Observability page smoke: FAIL")
        return 1

    print("Observability page smoke: PASS")
    return 0


if __name__ == "__main__":
    sys.exit(main())