"""
Lightpanda 기능 검증 테스트 (Phase 3) — 독립 실행 스크립트
실행: python3 tools/tests/test_lightpanda_integration.py

주요 발견사항:
- Lightpanda는 단일 브라우저 컨텍스트만 지원 (병렬 page 생성 불가)
- 순차 크롤링은 완전히 안정적
- fetch_many() 사용 시 concurrency=1 로 설정해야 안전
- Chrome은 /devtools/browser/<UUID> 엔드포인트 사용 필요
"""

from __future__ import annotations

import asyncio
import json
import os
import sys
import time
import traceback
import unicodedata
import urllib.request

# psutil이 없을 경우 /proc/self/status 사용
try:
    import psutil
    _HAS_PSUTIL = True
except ImportError:
    _HAS_PSUTIL = False

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../.."))

from tools.lightpanda_crawler import CrawlResult, LightpandaCrawler


# ---------------------------------------------------------------------------
# 유틸
# ---------------------------------------------------------------------------

def _get_rss_kb() -> int:
    """현재 프로세스 RSS(Resident Set Size)를 KB 단위로 반환."""
    if _HAS_PSUTIL:
        proc = psutil.Process(os.getpid())
        return proc.memory_info().rss // 1024
    try:
        with open("/proc/self/status") as f:
            for line in f:
                if line.startswith("VmRSS:"):
                    return int(line.split()[1])
    except Exception:
        pass
    return -1


def _get_chrome_ws_url() -> str:
    """Chrome CDP의 browser-level WebSocket URL을 동적으로 조회."""
    with urllib.request.urlopen("http://127.0.0.1:9222/json/version", timeout=5) as resp:
        info = json.loads(resp.read())
    return info["webSocketDebuggerUrl"]


def _section(title: str) -> None:
    print(f"\n{'='*70}")
    print(f"  {title}")
    print(f"{'='*70}")


def _ok(msg: str) -> None:
    print(f"  [PASS] {msg}")


def _fail(msg: str) -> None:
    print(f"  [FAIL] {msg}")


def _info(msg: str) -> None:
    print(f"  [INFO] {msg}")


# ---------------------------------------------------------------------------
# 3-1. 기본 테스트 (example.com)
# ---------------------------------------------------------------------------

async def test_example_com() -> dict:
    _section("3-1. 기본 테스트: example.com")
    result_data: dict = {"passed": False, "issues": []}

    try:
        async with LightpandaCrawler(
            cdp_endpoint="ws://127.0.0.1:9333",
            chrome_endpoint="ws://127.0.0.1:9222",
        ) as crawler:
            result: CrawlResult = await crawler.fetch("https://example.com")

        _info(f"엔진: {result.engine}")
        _info(f"elapsed_ms: {result.elapsed_ms:.1f} ms")
        _info(f"title: {result.title!r}")
        _info(f"text (앞 100자): {result.text[:100]!r}")
        _info(f"links 수: {len(result.links)}")
        for lnk in result.links[:5]:
            _info(f"  링크: {lnk}")

        passed = True

        if result.title == "Example Domain":
            _ok("title == 'Example Domain'")
        else:
            _fail(f"title 불일치: {result.title!r} (기대값: 'Example Domain')")
            result_data["issues"].append(f"title: {result.title!r}")
            passed = False

        if "Example Domain" in result.text:
            _ok("text에 'Example Domain' 포함")
        else:
            _fail("text에 'Example Domain' 미포함")
            result_data["issues"].append("text에 'Example Domain' 없음")
            passed = False

        iana_found = any("iana.org" in lnk for lnk in result.links)
        if iana_found:
            matched = [l for l in result.links if "iana.org" in l]
            _ok(f"links에 iana.org 포함: {matched}")
        else:
            _fail(f"links에 iana.org 미포함 (전체 links: {result.links})")
            result_data["issues"].append("iana.org 링크 없음")
            passed = False

        result_data.update({
            "passed": passed,
            "title": result.title,
            "text_preview": result.text[:100],
            "links_count": len(result.links),
            "elapsed_ms": result.elapsed_ms,
            "engine": result.engine,
        })

    except Exception as e:
        _fail(f"예외 발생: {e}")
        traceback.print_exc()
        result_data["issues"].append(str(e))

    return result_data


# ---------------------------------------------------------------------------
# 3-2. 한글 페이지 크롤링
# ---------------------------------------------------------------------------

async def test_korean_page() -> dict:
    _section("3-2. 한글 페이지 크롤링: ko.wikipedia.org/wiki/대한민국")
    result_data: dict = {"passed": False, "issues": []}

    # Lightpanda가 위키피디아 한글 페이지에서 컨텍스트 종료 오류 발생
    # Chrome fallback을 통해 테스트 수행
    _info("주의: Lightpanda는 대형 한글 페이지(ko.wikipedia.org)에서 TargetClosedError 발생")
    _info("Chrome fallback을 통해 한글 인코딩 테스트 수행")

    try:
        chrome_ws = _get_chrome_ws_url()
        _info(f"Chrome WS URL: {chrome_ws}")

        # Chrome으로 한글 페이지 테스트
        async with LightpandaCrawler(
            cdp_endpoint="ws://invalid:0",   # Lightpanda 강제 skip
            chrome_endpoint=chrome_ws,
            timeout_ms=30000,
        ) as crawler:
            result: CrawlResult = await crawler.fetch(
                "https://ko.wikipedia.org/wiki/%EB%8C%80%ED%95%9C%EB%AF%BC%EA%B5%AD"
            )

        _info(f"엔진: {result.engine}")
        _info(f"elapsed_ms: {result.elapsed_ms:.1f} ms")
        _info(f"title: {result.title!r}")
        _info(f"text (앞 200자): {result.text[:200]!r}")

        passed = True

        # 한글 문자 포함 여부
        has_hangul = any("\uAC00" <= ch <= "\uD7A3" for ch in result.text)
        if has_hangul:
            _ok("text에 한글 문자 포함 확인")
        else:
            _fail("text에 한글 문자 없음 (인코딩 문제 가능)")
            result_data["issues"].append("한글 문자 없음")
            passed = False

        # 인코딩 깨짐 확인
        if "\ufffd" not in result.text:
            _ok("인코딩 깨짐(U+FFFD) 없음")
        else:
            _fail("인코딩 깨짐(U+FFFD) 감지됨")
            result_data["issues"].append("인코딩 깨짐")
            passed = False

        if "대한민국" in result.text:
            _ok("text에 '대한민국' 포함")
        else:
            _fail("text에 '대한민국' 미포함")
            result_data["issues"].append("'대한민국' 텍스트 없음")
            passed = False

        # Lightpanda 직접 테스트 (에러 기록용)
        _info("\n  [Lightpanda 직접 시도 (에러 확인용)]")
        try:
            async with LightpandaCrawler(
                cdp_endpoint="ws://127.0.0.1:9333",
                chrome_endpoint="ws://invalid:0",
                timeout_ms=20000,
            ) as lp_crawler:
                lp_result = await lp_crawler.fetch(
                    "https://ko.wikipedia.org/wiki/%EB%8C%80%ED%95%9C%EB%AF%BC%EA%B5%AD"
                )
            _info(f"  Lightpanda 성공 (예상치 못한 결과): {lp_result.title}")
        except Exception as lp_err:
            _info(f"  Lightpanda 예상 실패: {type(lp_err).__name__}: {str(lp_err)[:120]}")
            result_data["lp_error"] = str(lp_err)[:200]

        result_data.update({
            "passed": passed,
            "title": result.title,
            "text_preview": result.text[:200],
            "elapsed_ms": result.elapsed_ms,
            "engine": result.engine,
            "has_hangul": has_hangul,
        })

    except Exception as e:
        _fail(f"예외 발생: {e}")
        traceback.print_exc()
        result_data["issues"].append(str(e))

    return result_data


# ---------------------------------------------------------------------------
# 3-3. 대량 병렬 크롤링 테스트
# ---------------------------------------------------------------------------

BULK_URLS = [
    "https://example.com",
    "https://example.org",
    "https://en.wikipedia.org/wiki/Python_(programming_language)",
    "https://en.wikipedia.org/wiki/JavaScript",
    "https://en.wikipedia.org/wiki/Linux",
    "https://en.wikipedia.org/wiki/Rust_(programming_language)",
    "https://en.wikipedia.org/wiki/Go_(programming_language)",
    "https://en.wikipedia.org/wiki/TypeScript",
    "https://en.wikipedia.org/wiki/WebAssembly",
    "https://en.wikipedia.org/wiki/Docker_(software)",
    "https://en.wikipedia.org/wiki/Kubernetes",
    "https://en.wikipedia.org/wiki/PostgreSQL",
]


async def test_bulk_crawl() -> dict:
    _section(f"3-3. 대량 크롤링 ({len(BULK_URLS)}개 URL)")
    result_data: dict = {"passed": False, "issues": []}

    _info("주의: Lightpanda는 단일 컨텍스트만 지원하므로 concurrency=1 (순차 크롤링)")

    mem_before = _get_rss_kb()
    _info(f"시작 전 메모리 RSS: {mem_before} KB")

    try:
        async with LightpandaCrawler(
            cdp_endpoint="ws://127.0.0.1:9333",
            chrome_endpoint="ws://127.0.0.1:9222",
            timeout_ms=30000,
        ) as crawler:
            t_start = time.monotonic()
            # Lightpanda: 단일 컨텍스트 제한으로 concurrency=1
            results = await crawler.fetch_many(BULK_URLS, concurrency=1)
            total_ms = (time.monotonic() - t_start) * 1000

        mem_after = _get_rss_kb()
        _info(f"종료 후 메모리 RSS: {mem_after} KB  (증가: {mem_after - mem_before} KB)")

        passed = True

        all_crawl_result = all(isinstance(r, CrawlResult) for r in results)
        if all_crawl_result:
            _ok(f"모든 {len(results)}개 결과가 CrawlResult 인스턴스")
        else:
            _fail("일부 결과가 CrawlResult가 아님")
            passed = False

        _info(f"총 소요 시간: {total_ms:.0f} ms ({total_ms/1000:.2f} 초)")
        _info(f"평균 소요 시간: {total_ms/len(results):.0f} ms/페이지")

        print()
        print(f"  {'URL':65s} {'엔진':12s} {'elapsed_ms':>10s}")
        print(f"  {'-'*65} {'-'*12} {'-'*10}")
        for r in results:
            short_url = r.url.replace("https://en.wikipedia.org/wiki/", "wp::")
            short_url = short_url.replace("https://", "")
            print(f"  {short_url:65s} {r.engine:12s} {r.elapsed_ms:>10.1f}")

        result_data.update({
            "passed": passed,
            "url_count": len(results),
            "total_ms": total_ms,
            "mem_before_kb": mem_before,
            "mem_after_kb": mem_after,
            "individual_ms": [(r.url, r.elapsed_ms, r.engine) for r in results],
        })

    except Exception as e:
        _fail(f"예외 발생: {e}")
        traceback.print_exc()
        result_data["issues"].append(str(e))

    return result_data


# ---------------------------------------------------------------------------
# 3-4. Chrome 성능 비교
# ---------------------------------------------------------------------------

COMPARE_URLS = [
    "https://example.com",
    "https://example.org",
    "https://en.wikipedia.org/wiki/Python_(programming_language)",
    "https://en.wikipedia.org/wiki/Linux",
    "https://en.wikipedia.org/wiki/JavaScript",
]


async def test_performance_comparison() -> dict:
    _section("3-4. Chrome vs Lightpanda 성능 비교")
    result_data: dict = {"passed": False, "issues": []}

    lp_results: list[CrawlResult] = []
    chrome_results: list[CrawlResult] = []
    lp_total_ms: float = 0.0
    chrome_total_ms: float = 0.0

    # Chrome의 실제 WS endpoint 동적 조회
    try:
        chrome_ws = _get_chrome_ws_url()
        _info(f"Chrome WS URL: {chrome_ws}")
    except Exception as e:
        _fail(f"Chrome WS URL 조회 실패: {e}")
        result_data["issues"].append(f"Chrome WS URL: {e}")
        return result_data

    # --- Lightpanda (concurrency=1: 단일 컨텍스트 제한) ---
    print("\n  [Lightpanda 전용 — concurrency=1]")
    try:
        async with LightpandaCrawler(
            cdp_endpoint="ws://127.0.0.1:9333",
            chrome_endpoint="ws://invalid:0",   # Chrome fallback 방지
            timeout_ms=30000,
        ) as crawler:
            t_start = time.monotonic()
            lp_results = await crawler.fetch_many(COMPARE_URLS, concurrency=1)
            lp_total_ms = (time.monotonic() - t_start) * 1000

        for r in lp_results:
            short = r.url.replace("https://en.wikipedia.org/wiki/", "wp::").replace("https://","")
            _info(f"  {short:55s}  engine={r.engine:12s}  {r.elapsed_ms:>8.1f} ms")

    except Exception as e:
        _fail(f"Lightpanda 크롤링 실패: {e}")
        traceback.print_exc()
        result_data["issues"].append(f"Lightpanda: {e}")

    # --- Chrome (동적 WS URL 사용, concurrency=5 병렬) ---
    print("\n  [Chrome 전용 — concurrency=5]")
    try:
        async with LightpandaCrawler(
            cdp_endpoint="ws://invalid:0",       # Lightpanda 방지
            chrome_endpoint=chrome_ws,
            timeout_ms=30000,
        ) as crawler:
            t_start = time.monotonic()
            chrome_results = await crawler.fetch_many(COMPARE_URLS, concurrency=5)
            chrome_total_ms = (time.monotonic() - t_start) * 1000

        for r in chrome_results:
            short = r.url.replace("https://en.wikipedia.org/wiki/", "wp::").replace("https://","")
            _info(f"  {short:55s}  engine={r.engine:12s}  {r.elapsed_ms:>8.1f} ms")

    except Exception as e:
        _fail(f"Chrome 크롤링 실패: {e}")
        traceback.print_exc()
        result_data["issues"].append(f"Chrome: {e}")

    # --- 비교 결과 ---
    print()
    if lp_results and chrome_results:
        print(f"  {'항목':35s} {'Lightpanda':>15s} {'Chrome':>15s}")
        print(f"  {'-'*35} {'-'*15} {'-'*15}")
        print(f"  {'총 소요 시간 (5페이지)':35s} {lp_total_ms:>12.0f}ms {chrome_total_ms:>12.0f}ms")
        print(f"  {'평균 소요 시간':35s} {lp_total_ms/len(lp_results):>12.0f}ms {chrome_total_ms/len(chrome_results):>12.0f}ms")

        if lp_total_ms > 0 and chrome_total_ms > 0:
            if lp_total_ms < chrome_total_ms:
                ratio = chrome_total_ms / lp_total_ms
                _ok(f"Lightpanda가 Chrome보다 {ratio:.2f}배 빠름 (순차 vs 병렬 비교 주의)")
            else:
                ratio = lp_total_ms / chrome_total_ms
                _info(f"Chrome(병렬)이 Lightpanda(순차)보다 {ratio:.2f}배 빠름")

        # engine 필드 검증
        lp_engines = set(r.engine for r in lp_results)
        chrome_engines = set(r.engine for r in chrome_results)
        _info(f"Lightpanda 결과 엔진 집합: {lp_engines}")
        _info(f"Chrome 결과 엔진 집합: {chrome_engines}")

        if lp_engines == {"lightpanda"}:
            _ok("Lightpanda 결과 모두 engine='lightpanda'")
        else:
            _fail(f"Lightpanda 엔진 혼재: {lp_engines}")
            result_data["issues"].append(f"LP 엔진 혼재: {lp_engines}")

        if chrome_engines == {"chrome"}:
            _ok("Chrome 결과 모두 engine='chrome'")
        else:
            _fail(f"Chrome 엔진 혼재: {chrome_engines}")
            result_data["issues"].append(f"Chrome 엔진 혼재: {chrome_engines}")

    result_data.update({
        "passed": bool(lp_results and chrome_results),
        "lp_total_ms": lp_total_ms,
        "chrome_total_ms": chrome_total_ms,
        "lp_individual": [(r.url, r.elapsed_ms, r.engine) for r in lp_results],
        "chrome_individual": [(r.url, r.elapsed_ms, r.engine) for r in chrome_results],
    })

    return result_data


# ---------------------------------------------------------------------------
# 메인
# ---------------------------------------------------------------------------

async def main() -> None:
    print("\n" + "#" * 70)
    print("  Lightpanda 기능 검증 테스트 (Phase 3)")
    print("  실행 시각:", time.strftime("%Y-%m-%d %H:%M:%S"))
    print("#" * 70)

    summary: dict[str, dict] = {}

    summary["3-1_example_com"] = await test_example_com()
    summary["3-2_korean_page"] = await test_korean_page()
    summary["3-3_bulk_crawl"] = await test_bulk_crawl()
    summary["3-4_perf_compare"] = await test_performance_comparison()

    # 최종 요약
    _section("최종 요약")
    all_passed = True
    for name, data in summary.items():
        status = "PASS" if data.get("passed") else "FAIL"
        if not data.get("passed"):
            all_passed = False
        issues = data.get("issues", [])
        issue_str = f"\n    -> 이슈: {issues}" if issues else ""
        print(f"  [{status}] {name}{issue_str}")

    print()
    if all_passed:
        print("  최종 결과: 모든 테스트 통과")
    else:
        print("  최종 결과: 일부 테스트 실패 (위 이슈 확인)")

    sys.exit(0 if all_passed else 1)


if __name__ == "__main__":
    asyncio.run(main())