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

import json
import sys
import threading
import urllib.error
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 request_bytes(url: str, *, range_header: str = "") -> tuple[int, bytes, dict[str, str]]:
    req = urllib.request.Request(url)
    if range_header:
        req.add_header("Range", range_header)
    with urllib.request.urlopen(req, timeout=5) as response:
        return response.status, response.read(), dict(response.headers.items())


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 / "A1.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 / "A1.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)
    opened_folders: list[Path] = []
    server.RequestHandlerClass.repository._file_selector = lambda path: opened_folders.append(path)
    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.297"
        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/"
        post = posts["items"][0]
        assert "media_assets" not in post
        assert post["has_video"] == 1
        assert post["has_image"] == 1
        files = post["media_files"]
        assert [file["file_name"] for file in files] == ["A1.jpg", "A1.mp4"]
        assert all("id" not in file and "post_id" not in file for file in files)
        video_file = next(file for file in files if file["file_name"].endswith(".mp4"))
        assert video_file["virtual_url"] == "/_StDown/instagram/alpha/A1.mp4"
        assert video_file["absolute_path"] == str(media_file.resolve())
        status, body, headers = request_bytes(base + video_file["virtual_url"])
        assert status == 200 and body == b"video"
        assert headers["Content-Type"].startswith("video/")
        status, body, headers = request_bytes(base + video_file["virtual_url"], range_header="bytes=1-3")
        assert status == 206 and body == b"ide" and headers["Content-Range"] == "bytes 1-3/5"
        try:
            request_bytes(base + f"/api/posts/{a['post_id']}/media/1")
        except urllib.error.HTTPError as exc:
            assert exc.code == 404
        else:
            raise AssertionError("numeric media route must be removed")
        opened = request_json(
            base + "/api/open-folder",
            method="POST",
            payload={"virtual_path": video_file["virtual_url"]},
        )
        assert opened["opened"] is True and opened["file_name"] == "A1.mp4"
        assert opened_folders == [media_file.resolve()]

        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_video"] == 0
        assert beta_posts["items"][0]["has_image"] == 0
        deleted_post = request_json(base + f"/api/posts/{a['post_id']}", method="DELETE")
        assert deleted_post["deleted"] is True and deleted_post["deleted_file_count"] == 2
        assert not media_file.exists() and not image_file.exists()
        with connect_database(db_path) as conn:
            assert conn.execute("SELECT COUNT(*) FROM posts WHERE id=?", (a["post_id"],)).fetchone()[0] == 0
            assert conn.execute("SELECT COUNT(*) FROM post_stat_history WHERE post_id=?", (a["post_id"],)).fetchone()[0] == 0
            assert conn.execute("SELECT COUNT(*) FROM media_assets WHERE post_id=?", (a["post_id"],)).fetchone()[0] == 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_like_search_and_account_table_check(tmp_path: Path) -> None:
    st_root = tmp_path / "_St"
    st_root.mkdir()
    repo = AnalyzeRepository(st_root)
    for username in ("tem.duck", "team.duck", "other"):
        repo.save_observation({"username": username, "shortcode": username.replace(".", "_")})
    prefix = repo.accounts(page=1, page_size=10000, search_pattern="tem%")
    assert [item["username"] for item in prefix["items"]] == ["tem.duck"]
    assert prefix["total"] == 1 and prefix["total_all"] == 3
    contains = repo.accounts(page=1, page_size=10000, search_pattern="%duck")
    assert {item["username"] for item in contains["items"]} == {"tem.duck", "team.duck"}
    single = repo.accounts(page=1, page_size=10000, search_pattern="te_.duck")
    assert [item["username"] for item in single["items"]] == ["tem.duck"]
    exact = repo.accounts(page=1, page_size=10000, search_pattern="tem.duck")
    assert [item["username"] for item in exact["items"]] == ["tem.duck"]
    checked = repo.account_table_check("tem%")
    assert checked["ok"] is True and checked["quick_check"] == ["ok"]
    assert checked["table_name"] == "social_accounts"
    assert checked["total"] == 1 and checked["total_all"] == 3


def test_post_delete_rejects_shared_media_reference(tmp_path: Path) -> None:
    st_root = tmp_path / "_St"
    st_root.mkdir()
    repo = AnalyzeRepository(st_root)
    first = repo.save_observation({"username": "first", "shortcode": "S1"})
    second = repo.save_observation({"username": "second", "shortcode": "S2"})
    media_dir = tmp_path / "_StDown" / "instagram" / "shared"
    media_dir.mkdir(parents=True)
    media_file = media_dir / "shared.mp4"
    media_file.write_bytes(b"shared")
    record_downloaded_file(
        st_root=st_root, file_path=media_file, source_mode="MANUAL_SINGLE",
        shortcode="S1", owner_username="first", asset_kind="VIDEO",
        account_path="shared", event_uuid="shared-first", file_size=media_file.stat().st_size,
    )
    with connect_database(repo.db_path) as conn:
        asset = conn.execute(
            "SELECT storage_root_id, relative_path, file_name, folder_name FROM media_assets WHERE post_id=?",
            (first["post_id"],),
        ).fetchone()
        conn.execute(
            """
            INSERT INTO media_assets(post_id, asset_kind, media_index, storage_root_id, relative_path, file_name, folder_name, local_status)
            VALUES(?, 'VIDEO', 0, ?, ?, ?, ?, 'AVAILABLE')
            """,
            (second["post_id"], asset["storage_root_id"], asset["relative_path"], asset["file_name"], asset["folder_name"]),
        )
        conn.commit()
    try:
        repo.delete_post(first["post_id"])
    except ValueError as exc:
        assert "shared media reference detected" in str(exc)
    else:
        raise AssertionError("shared media reference must block deletion")
    assert media_file.exists()
    with connect_database(repo.db_path) as conn:
        assert conn.execute("SELECT COUNT(*) FROM posts WHERE id=?", (first["post_id"],)).fetchone()[0] == 1


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_stdown_virtual_and_absolute_paths() -> None:
    html = (Path(__file__).resolve().parent / "web" / "index.html").read_text(encoding="utf-8")
    assert "가상 경로" in html and "실제 경로" in html
    assert "/api/open-folder" in html and "virtual_path" in html
    assert "/_StDown/instagram/" not in html  # generated from API data, not hard-coded per post
    assert "/api/posts/${post.id}/media/${asset.id}" not in html
    assert "📋 경로 복사" in html and "copyLocalPath" in html
    assert 'target="_blank" rel="noopener noreferrer"' 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 'id="refreshAccounts"' not in html
    assert "0/200" not in html
    assert "const ACCOUNT_PAGE_SIZE=10000" in html
    assert "page_size=${ACCOUNT_PAGE_SIZE}&search=${encodeURIComponent(pattern)}" in html
    assert 'id="searchHelpButton"' in html and 'id="searchHelp"' in html
    assert "%단어%" in html and "임의의 한 글자" in html
    assert "SQLite <b>LIKE</b> 조건에 그대로 사용" in html
    assert "matchAccountQuery" not in html
    assert 'id="checkAccounts"' in html and "/api/accounts/check?search=" in html
    assert "social_accounts" in html and "검색 ${num(accountTotal)} / 전체 ${num(accountTotalAll)}" in html
    assert 'id="databaseSelect"' in html and 'id="databaseButtons"' in html
    assert "renderDatabaseButtons" in html and "button.onclick=()=>switchDatabase(filename)" 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 DEFAULT_SETTINGS={mediaWidth:'38%',mediaHeight:520,chartDays:0,chartLimit:50" in html
    assert "stats?limit=${uiSettings.chartLimit}" 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 "height:var(--media-height,520px)" in html
    assert ".card-media img,.card-media video{display:block;width:100%;height:100%;object-fit:contain" in html
    assert "connectNulls:false" in html
    assert "function chartValue(value)" in html
    assert "function validCssSize(value)" in html
    assert 'id="settingMediaWidth" type="text"' 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 "assetExtension(file)==='image'" in html and "assetExtension(file)==='video'" in html
    assert "kinds.push(ext||'IMG')" in html and "kinds.push(ext||'MP4')" in html
    assert "webp|avif" in html
    assert "kinds.join(' · ')" in html
    assert "result.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)


def test_stdown_virtual_path_guards_and_image_mime(tmp_path: Path) -> None:
    st_root = tmp_path / "_St"
    st_root.mkdir()
    repo = AnalyzeRepository(st_root)
    image = tmp_path / "_StDown" / "instagram" / "safe.account" / "ABC_123.webp"
    image.parent.mkdir(parents=True)
    image.write_bytes(b"RIFFfake-webp")
    assert repo.resolve_stdown_virtual_path(
        "/_StDown/instagram/safe.account/ABC_123.webp"
    ) == image.resolve()
    for invalid in (
        "/_StDown/tiktok/safe.account/ABC_123.webp",
        "/_StDown/instagram/../ABC_123.webp",
        "/_StDown/instagram/safe.account/../ABC_123.webp",
        "/_StDown/instagram/safe.account/ABC_123.exe",
        "/_StDown/instagram/safe.account/sub/ABC_123.webp",
    ):
        try:
            repo.resolve_stdown_virtual_path(invalid)
        except (ValueError, FileNotFoundError):
            pass
        else:
            raise AssertionError(f"invalid virtual path accepted: {invalid}")

    server = build_server(st_root, repo.db_path, "127.0.0.1", 0)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    host, port = server.server_address
    try:
        status, body, headers = request_bytes(
            f"http://{host}:{port}/_StDown/instagram/safe.account/ABC_123.webp"
        )
        assert status == 200 and body == b"RIFFfake-webp"
        assert headers["Content-Type"] == "image/webp"
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)


def test_account_posts_refresh_discovers_shortcode_actual_extension_without_asset_ids(tmp_path: Path) -> None:
    st_root = tmp_path / "_St"
    st_root.mkdir()
    repo = AnalyzeRepository(st_root)
    saved = repo.save_observation({
        "username": "ameliauusa",
        "shortcode": "ABC123",
        "thumbnail_url": "https://cdn.example/temporary.jpg",
    })
    local = tmp_path / "_StDown" / "instagram" / "ameliauusa" / "ABC123.avif"
    local.parent.mkdir(parents=True)
    local.write_bytes(b"avif-local-image")
    result = repo.account_posts(saved["account_id"])
    post = result["items"][0]
    assert "media_assets" not in post
    assert post["has_image"] == 1
    assert post["media_files"] == [{
        "asset_kind": "IMAGE",
        "file_name": "ABC123.avif",
        "extension": ".avif",
        "folder_name": "ameliauusa",
        "relative_path": "instagram/ameliauusa/ABC123.avif",
        "absolute_path": str(local.resolve()),
        "virtual_url": "/_StDown/instagram/ameliauusa/ABC123.avif",
    }]
    assert repo.resolve_stdown_virtual_path(post["media_files"][0]["virtual_url"]) == local.resolve()
    with connect_database(repo.db_path) as conn:
        assert conn.execute("SELECT COUNT(*) FROM media_assets WHERE post_id=?", (saved["post_id"],)).fetchone()[0] == 0
    deleted = repo.delete_post(saved["post_id"])
    assert deleted["deleted_file_count"] == 1
    assert not local.exists()