# FILE: _St/Analyze8780/test_analyze8780_module.py | ROLE: Analyze8780 API·UI 회귀 테스트
from __future__ import annotations

import json
import sys
import threading
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
for path in (ROOT, Path(__file__).resolve().parent):
    if str(path) not in sys.path:
        sys.path.insert(0, str(path))

from Analyze8780.analyze8780_repository import AnalyzeRepository
from Analyze8780.analyze8780_main import build_server
from Shared.storage.sort_storage import connect_database, record_downloaded_file


def request_json(url: str, *, method: str = "GET", payload: dict | None = None) -> dict:
    body = json.dumps(payload).encode("utf-8") if payload is not None else None
    req = urllib.request.Request(url, data=body, method=method)
    if body is not None:
        req.add_header("Content-Type", "application/json")
    with urllib.request.urlopen(req, timeout=5) as response:
        return json.loads(response.read().decode("utf-8"))


def test_repository_and_http_api(tmp_path: Path) -> None:
    st_root = tmp_path / "_St"
    st_root.mkdir()
    db_path = st_root / "Shared" / "data" / "sort_local.sqlite3"
    repo = AnalyzeRepository(st_root, db_path)
    a = repo.save_observation({
        "username": "alpha", "shortcode": "A1", "views": 100, "likes": 10,
        "post_url": "https://www.instagram.com/explore/",
        "video_url": "https://cdn.example/A1.mp4",
        "observed_at": "2026-07-19T09:02:00Z",
    })
    repo.save_observation({
        "username": "alpha", "shortcode": "A1", "views": 160, "likes": 14,
        "observed_at": "2026-07-19T09:12:00Z",
    })
    assert repo.accounts()["total"] == 1
    assert repo.account_posts(a["account_id"], sort_by="views_delta")["items"][0]["views_delta"] == 60
    media_dir = tmp_path / "_StDown" / "instagram" / "alpha"
    media_dir.mkdir(parents=True)
    media_file = media_dir / "179_original.mp4"
    media_file.write_bytes(b"video")
    record_downloaded_file(
        st_root=st_root, file_path=media_file, source_mode="MANUAL_SINGLE",
        shortcode="A1", owner_username="alpha", asset_kind="VIDEO",
        account_path="alpha", event_uuid="analyze-original-file",
        file_size=media_file.stat().st_size,
    )
    image_file = media_dir / "179_original.jpg"
    image_file.write_bytes(b"image")
    record_downloaded_file(
        st_root=st_root, file_path=image_file, source_mode="MANUAL_SINGLE",
        shortcode="A1", owner_username="alpha", asset_kind="IMAGE",
        account_path="alpha", event_uuid="analyze-original-image",
        post_url="https://www.instagram.com/reel/A1/",
        video_url="https://cdn.example/A1.mp4",
        file_size=image_file.stat().st_size,
    )

    server = build_server(st_root, db_path, "127.0.0.1", 0)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    host, port = server.server_address
    base = f"http://{host}:{port}"
    try:
        health = request_json(base + "/api/health")
        assert health["ok"] is True and health["service"] == "Analyze8780"
        assert health["version"] == "2.3.8.287"
        accounts = request_json(base + "/api/accounts?page=1&page_size=50")
        assert accounts["total"] == 1
        queue_saved = request_json(base + "/api/collection-accounts", method="PUT", payload={
            "items": [
                {"username": "alpha", "position": 1},
                {"username": "https://www.instagram.com/explore/", "position": 2},
                {"username": "beta", "position": 3},
            ]
        })
        assert queue_saved["ok"] is True and queue_saved["total"] == 2
        queue_loaded = request_json(base + "/api/collection-accounts")
        assert [item["username"] for item in queue_loaded["items"]] == ["alpha", "beta"]

        batch = request_json(base + "/api/observations/batch", method="POST", payload={
            "items": [{
                "username": "beta", "shortcode": "B1", "views": 0, "views_available": 0, "likes": 25,
                "comments": 8, "reposts": 4, "share_count": 999,
                "sort_by": "comments", "sort_value": 8, "source_kind": "LOCAL",
                "observed_at": "2026-07-25T10:07:00Z",
            }]
        })
        assert batch["ok"] is True and batch["saved_count"] == 1 and batch["skipped_count"] == 0 and batch["failed_count"] == 0
        unchanged = request_json(base + "/api/observations/batch", method="POST", payload={
            "items": [{
                "username": "beta", "shortcode": "B1", "views": 0, "views_available": 0, "likes": 25,
                "comments": 8, "reposts": 4, "sort_by": "likes", "sort_value": 25, "source_kind": "LOCAL",
                "observed_at": "2026-07-25T10:17:00Z",
            }]
        })
        assert unchanged["ok"] is True and unchanged["saved_count"] == 0 and unchanged["skipped_count"] == 1
        with connect_database(db_path) as conn:
            counts = conn.execute("""
                SELECT
                    (SELECT COUNT(*) FROM post_stat_history h JOIN posts p ON p.id=h.post_id WHERE p.shortcode='B1') AS history_count,
                    (SELECT COUNT(*) FROM post_stat_10m b JOIN posts p ON p.id=b.post_id WHERE p.shortcode='B1') AS bucket_count
            """).fetchone()
            assert counts["history_count"] == 1
            assert counts["bucket_count"] == 1
        changed = request_json(base + "/api/observations/batch", method="POST", payload={
            "items": [{
                "username": "beta", "shortcode": "B1", "views": 0, "views_available": 0, "likes": 26,
                "comments": 8, "reposts": 4, "sort_by": "likes", "sort_value": 26, "source_kind": "LOCAL",
                "observed_at": "2026-07-25T10:27:00Z",
            }]
        })
        assert changed["ok"] is True and changed["saved_count"] == 1 and changed["skipped_count"] == 0
        with connect_database(db_path) as conn:
            row = conn.execute("""
                SELECT s.reposts, s.views_available, s.sort_by, s.sort_value, s.source_kind, b.bucket_at
                FROM posts p
                JOIN post_stat_latest s ON s.post_id=p.id
                JOIN post_stat_10m b ON b.post_id=p.id
                WHERE p.shortcode='B1'
            """).fetchone()
            assert row["reposts"] == 4
            assert row["views_available"] == 0
            assert row["sort_by"] == "likes" and row["sort_value"] == 26 and row["source_kind"] == "LOCAL"
            assert row["bucket_at"] == "2026-07-25T10:20:00Z"
            counts = conn.execute("""
                SELECT
                    (SELECT COUNT(*) FROM post_stat_history h JOIN posts p ON p.id=h.post_id WHERE p.shortcode='B1') AS history_count,
                    (SELECT COUNT(*) FROM post_stat_10m b JOIN posts p ON p.id=b.post_id WHERE p.shortcode='B1') AS bucket_count
            """).fetchone()
            assert counts["history_count"] == 2
            assert counts["bucket_count"] == 2

        posts = request_json(base + f"/api/accounts/{a['account_id']}/posts?sort_by=views_delta")
        assert posts["items"][0]["views_per_hour"] == 360
        assert posts["items"][0]["post_url"] == "https://www.instagram.com/reel/A1/"
        assert posts["items"][0]["folder_name"] == "alpha"
        assert posts["items"][0]["file_name"] == "179_original.mp4"
        assert posts["items"][0]["has_mp4"] == 1
        assert posts["items"][0]["has_jpg"] == 1

        accounts_after = request_json(base + "/api/accounts?page=1&page_size=50")
        beta = next(item for item in accounts_after["items"] if item["username"] == "beta")
        beta_posts = request_json(base + f"/api/accounts/{beta['id']}/posts?sort_by=likes")
        assert beta_posts["total"] == 1
        assert beta_posts["items"][0]["shortcode"] == "B1"
        assert beta_posts["items"][0]["likes"] == 26
        assert beta_posts["items"][0]["has_mp4"] == 0
        assert beta_posts["items"][0]["has_jpg"] == 0
        deleted = request_json(base + f"/api/accounts/{a['account_id']}", method="DELETE")
        assert deleted["deleted"] is True
        tombstones = request_json(base + "/api/deleted/accounts")
        assert tombstones["items"][0]["username"] == "alpha"
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)


def test_runner_uses_py312() -> None:
    runner = (Path(__file__).resolve().parent / "run_Analyze8780.bat").read_text(encoding="utf-8")
    assert "call py312 analyze8780_main.py --host 127.0.0.1 --port 8780" in runner


def test_web_shows_saved_folder_and_original_filename() -> None:
    html = (Path(__file__).resolve().parent / "web" / "index.html").read_text(encoding="utf-8")
    assert "저장 폴더" in html
    assert "원본 파일" in html
    assert "p.folder_name" in html and "p.file_name" in html

def test_web_account_list_search_database_and_post_paging() -> None:
    html = (Path(__file__).resolve().parent / "web" / "index.html").read_text(encoding="utf-8")
    assert "grid-template-columns:minmax(200px,240px)" in html
    assert 'id="accountSearch"' in html
    assert 'id="accountPager"' not in html
    assert "const ACCOUNT_PAGE_SIZE=10000" in html
    assert "page_size=${ACCOUNT_PAGE_SIZE}" in html
    assert 'id="searchHelpButton"' in html and 'id="searchHelp"' in html
    assert "abc%" in html and "%abc%" in html and "matchAccountQuery" in html
    assert 'id="databaseSelect"' in html
    assert "/api/database-files" in html and "/api/database/open" in html
    assert 'id="postPageSize"' in html
    for value in ("50", "100", "200", "10000"):
        assert f'<option value="{value}"' in html
    assert "qs('#postPageSize').onchange=()=>loadPosts(1)" in html
    assert "account-row" in html and "account-summary" in html
    assert "selectedAccountData" in html and "renderAccountDetail" in html
    assert "new URLSearchParams(location.search).get('account')" in html
    assert "history.replaceState" in html
    assert "좌측 계정을 클릭하세요." in html
    config = json.loads((Path(__file__).resolve().parent / "analyze8780_config.json").read_text(encoding="utf-8"))
    assert config["account_page_size"] == 10000
    assert config["post_page_sizes"] == [50, 100, 200, 10000]
    assert config["chart_history_limit"] == 50


def test_web_apexcharts_visible_history_and_tall_image() -> None:
    html = (Path(__file__).resolve().parent / "web" / "index.html").read_text(encoding="utf-8")
    assert 'https://cdn.jsdelivr.net/npm/apexcharts' in html
    assert "const CHART_HISTORY_LIMIT=50" in html
    assert "stats?limit=${CHART_HISTORY_LIMIT}" in html
    assert "new IntersectionObserver" in html
    assert "shared:true" in html and "intersect:false" in html and "followCursor:true" in html
    assert "name:'조회수'" in html and "name:'좋아요'" in html
    assert "name:'댓글'" in html and "name:'공유·리포스트'" in html
    assert "colors:['#2563eb','#ef4444','#f59e0b','#10b981']" in html
    assert ".card-media{position:relative;width:100%;height:clamp(360px,46vw,560px)" in html
    assert ".card-media img{display:block;width:100%;height:100%;object-fit:contain}" in html
    assert "통계 이력" not in html
    assert "오늘 조회" not in html  # dynamic period chips, not fixed fake values
    assert "periodDelta" in html and "최근 50개 통계 기록 범위 내 조회수 증가량" in html


def test_web_download_badge_and_safe_post_link() -> None:
    html = (Path(__file__).resolve().parent / "web" / "index.html").read_text(encoding="utf-8")
    assert 'target="_blank" rel="noopener noreferrer"' in html
    assert 'target="_blank" rel="noopener"' not in html
    assert "downloadBadge" in html
    assert "kinds.push('JPG')" in html and "kinds.push('MP4')" in html
    assert "kinds.join(' · ')" in html
    assert "j.items.forEach" in html


def test_large_page_size_and_descending_sort(tmp_path: Path) -> None:
    st_root = tmp_path / "_St"
    st_root.mkdir()
    repo = AnalyzeRepository(st_root)
    account_id = 0
    for index in range(205):
        result = repo.save_observation({
            "username": "many",
            "shortcode": f"P{index:03d}",
            "views": index,
            "likes": index // 2,
            "comments": index // 3,
            "reposts": index // 4,
            "observed_at": f"2026-08-01T{index % 24:02d}:{(index * 10) % 60:02d}:00Z",
        })
        account_id = result["account_id"]
    accounts = repo.accounts(page=1, page_size=10000)
    assert accounts["page_size"] == 10000 and accounts["total"] == 1
    posts = repo.account_posts(account_id, page=1, page_size=10000, sort_by="views")
    assert posts["page_size"] == 10000 and posts["total"] == 205
    assert len(posts["items"]) == 205
    assert posts["items"][0]["shortcode"] == "P204"
    assert posts["items"][-1]["shortcode"] == "P000"


def test_database_file_management_http_api(tmp_path: Path) -> None:
    st_root = tmp_path / "_St"
    st_root.mkdir()
    server = build_server(st_root, None, "127.0.0.1", 0)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    host, port = server.server_address
    base = f"http://{host}:{port}"
    try:
        saved = request_json(base + "/api/collection-accounts", method="PUT", payload={"items": [{"username": "alpha"}]})
        assert saved["db_filename"] == "sort_local.sqlite3"
        copied = request_json(base + "/api/database/save-as", method="POST", payload={"filename": "backup"})
        assert copied["action"] == "save_as" and copied["db_filename"] == "backup.sqlite3"
        assert [item["username"] for item in copied["items"]] == ["alpha"]
        fresh = request_json(base + "/api/database/new", method="POST", payload={"filename": "fresh"})
        assert fresh["db_filename"] == "fresh.sqlite3" and fresh["items"] == []
        opened = request_json(base + "/api/database/open", method="POST", payload={"filename": "backup.sqlite3"})
        assert opened["db_filename"] == "backup.sqlite3"
        assert [item["username"] for item in opened["items"]] == ["alpha"]
        files = request_json(base + "/api/database-files")
        assert files["active_filename"] == "backup.sqlite3"
        assert {"sort_local.sqlite3", "backup.sqlite3", "fresh.sqlite3"}.issubset(set(files["files"]))
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)
