# -*- coding: utf-8 -*-
"""只读查询层测试（离线）: uv run python tests/test_web_queries.py"""

import os
import sys
import tempfile
from pathlib import Path

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.stdout.reconfigure(encoding="utf-8")

from reaction_board.config import ConfigStore  # noqa: E402
from reaction_board.database import Database  # noqa: E402
from reaction_board.repository import Repository  # noqa: E402
from reaction_board.timeutil import day_bounds  # noqa: E402
from reaction_board.web.cache import DASHBOARD_CACHE  # noqa: E402
from reaction_board.web.queries import DashboardQueries  # noqa: E402
from tests.web_fixtures import DATES, build_db, write_config  # noqa: E402

FIXED_NOW = 1_793_459_000  # 2026-10-31 前后


def setup(tmp: str):
    build_db(tmp)
    config = write_config(tmp)
    db = Database(Path(tmp) / "app.db")
    db.initialize()
    store = ConfigStore(config)
    queries = DashboardQueries(db, store, clock=lambda: FIXED_NOW)
    return db, queries


def test_overview():
    with tempfile.TemporaryDirectory() as tmp:
        db, q = setup(tmp)
        try:
            overview = q.overview()
            assert overview["data_date"] == DATES[-1]
            assert overview["snapshot_days"] == 3
            assert overview["is_first_day"] is False
            assert overview["initial_import_count"] == 2
            assert overview["counts"]["active"] == 2
            assert overview["counts"]["unfaved"] == 1
            assert overview["counts"]["removed"] == 1
            assert overview["counts"]["active_ups"] == 2
            assert overview["counts"]["candidates_pending"] == 1
            assert overview["totals"]["view"] == 1200 + 5200 + 120
            assert overview["interactions"] > 0
            assert overview["view_delta"] == (100 + 100 + 0)  # 101/102 的日增量
            assert overview["snapshot_health"]["total"] == 3
            assert overview["folders"][0]["media_count"] == 3
            assert overview["folders"][0]["ratio"] < 1
            assert overview["fans"]["follower"] == 1020
            assert overview["fans"]["delta_1d"] == 10
            assert overview["preview"] is False
            assert overview["years"] == [2026]
            assert "snapshot" in overview["tasks"]
        finally:
            db.close()


def test_video_listing_filters_and_sorting():
    with tempfile.TemporaryDirectory() as tmp:
        db, q = setup(tmp)
        try:
            heat = q.list_videos(sort="heat")
            aids = [item["aid"] for item in heat["items"]]
            assert 104 not in aids, "removed 视频不应进入榜单"
            # 综合热度榜应用最小播放门槛 2000：101（1200）与 103（120）被排除
            assert aids == [102], aids
            assert heat["items"][0]["heat_percentile"] == 100.0

            views = q.list_videos(sort="view")
            assert [item["aid"] for item in views["items"]] == [102, 101, 103]

            interactions = q.list_videos(sort="interaction")
            assert interactions["items"][0]["aid"] == 101

            added = q.list_videos(sort="added", within_days=1)
            assert [item["aid"] for item in added["items"]] == [103]

            published = q.list_videos(sort="published", within_days=2)
            assert {item["aid"] for item in published["items"]} == {101, 103}

            searched = q.list_videos(sort="view", q="UP B")
            assert [item["aid"] for item in searched["items"]] == [103]
            searched = q.list_videos(sort="view", q="视频 102")
            assert [item["aid"] for item in searched["items"]] == [102]

            folder = q.list_videos(sort="view", folder_id=3700023351)
            assert folder["total"] == 3
            folder = q.list_videos(sort="view", folder_id=999)
            assert folder["total"] == 0

            page = q.list_videos(sort="view", page=2, page_size=1)
            assert page["total"] == 3
            assert page["pages"] == 3
            assert [item["aid"] for item in page["items"]] == [101]
        finally:
            db.close()


def test_growth_and_cold_start():
    with tempfile.TemporaryDirectory() as tmp:
        db, q = setup(tmp)
        try:
            growth = q.list_videos(sort="growth", growth_days=2)
            assert growth["available"] is True
            assert {item["aid"] for item in growth["items"]} == {101, 102}
            first = next(item for item in growth["items"] if item["aid"] == 101)
            assert first["growth"]["available"] is True
            assert first["growth"]["delta"] == 200
            assert first["growth"]["gap_days"] == 2

            rate = q.list_videos(sort="growth_rate", growth_days=2)
            assert {item["aid"] for item in rate["items"]} == {101, 102}

            # 只有一个快照日时增长不可用，不能返回假 0
            for day in DATES[:-1]:
                db.write(lambda conn, day=day: conn.execute(
                    "DELETE FROM snapshot WHERE stat_date = ?", (day,)
                ))
            DASHBOARD_CACHE.clear()
            cold = q.list_videos(sort="growth", growth_days=2)
            assert cold["available"] is False
            assert cold["items"] == []
            assert "两天" in cold["reason"]
        finally:
            db.close()


def test_video_detail_and_trend():
    with tempfile.TemporaryDirectory() as tmp:
        db, q = setup(tmp)
        try:
            detail = q.video_detail(101)
            assert detail["aid"] == 101
            assert detail["metrics"]["view"] == 1200
            assert detail["growth"]["delta"] == 100
            assert detail["snapshot_count"] == 3
            assert detail["folder_title"] == "2026"
            assert detail["url"].endswith("BV101")
            assert q.video_detail(999999) is None

            trend = q.video_trend(101, days=3)
            assert trend["dates"] == list(DATES)
            assert trend["series"]["view"] == [1000, 1100, 1200]
            assert trend["available"] is True
            assert len(trend["points"]) == 3

            # 请求区间早于首次快照时，从首次快照开始，不制造空白历史
            wide = q.video_trend(101, days=30)
            assert wide["dates"] == list(DATES)

            # 删除中间一天，折线在缺失日期应为 None，而不是插值
            db.write(lambda conn: conn.execute(
                "DELETE FROM snapshot WHERE aid=101 AND stat_date=?",
                ("2026-10-30",),
            ))
            trend = q.video_trend(101, days=3)
            assert trend["series"]["view"] == [1000, None, 1200]

            # 失败快照也视为缺失
            db.write(lambda conn: conn.execute(
                "UPDATE snapshot SET ok=0, view=NULL WHERE aid=101 AND stat_date=?",
                (DATES[-1],),
            ))
            trend = q.video_trend(101, days=3)
            assert trend["series"]["view"][-1] is None
        finally:
            db.close()


def test_up_queries():
    with tempfile.TemporaryDirectory() as tmp:
        db, q = setup(tmp)
        try:
            ups = q.list_ups()
            assert ups["sort"] == "avg_views"
            assert ups["total"] == 2
            assert ups["items"][0]["up_mid"] == 1
            assert ups["items"][0]["video_count"] == 2
            assert ups["items"][0]["total_views"] == 6400
            assert ups["items"][0]["avg_views"] == 3200.0

            one = q.list_ups(min_videos=2)
            assert one["total"] == 1

            detail = q.up_detail(1)
            assert detail["video_count"] == 2
            assert detail["total_views"] == 6400
            assert detail["blacklisted"] is False
            assert detail["series"]["dates"] == list(DATES)
            assert detail["series"]["views"] == [6000, 6200, 6400]
            assert detail["space_url"].endswith("/1")

            blacklisted = q.up_detail(2)
            assert blacklisted["blacklisted"] is True
            assert blacklisted["video_count"] == 1
            assert q.up_detail(999999) is None
        finally:
            db.close()


def test_query_cache_invalidation():
    with tempfile.TemporaryDirectory() as tmp:
        db, q = setup(tmp)
        try:
            DASHBOARD_CACHE.clear()
            first = q.overview()
            db.write(lambda conn: conn.execute(
                "UPDATE snapshot SET \"view\" = 999999 WHERE aid=101 AND stat_date=?",
                (DATES[-1],),
            ))
            cached = q.overview()
            assert cached["totals"]["view"] == first["totals"]["view"]

            repo = Repository(db)
            repo.bump_cache_version("videos")
            refreshed = q.overview()
            assert refreshed["totals"]["view"] > first["totals"]["view"]

            fans_first = q.fans(days=3)
            db.write(lambda conn: conn.execute(
                "UPDATE up_stat SET follower=999999 WHERE mid=37754047 AND stat_date=?",
                (DATES[-1],),
            ))
            fans_cached = q.fans(days=3)
            assert fans_cached["follower"] == fans_first["follower"]

            repo.bump_cache_version("fans")
            fans_refreshed = q.fans(days=3)
            assert fans_refreshed["follower"] == 999999
        finally:
            db.close()


def test_fans_custom_range_modes():
    with tempfile.TemporaryDirectory() as tmp:
        db, q = setup(tmp)
        try:
            daily = q.fans(start="2026-10-29", end="2026-10-31")
            assert daily["mode"] == "daily"
            assert daily["range_start"] == "2026-10-29"
            assert daily["range_end"] == "2026-10-31"
            assert daily["dates"] == list(DATES)
            assert daily["series"] == [1000, 1010, 1020]

            # 超过一年触发按月峰值模式
            db.write(lambda conn: conn.execute(
                "INSERT OR REPLACE INTO up_stat"
                "(stat_date, mid, follower, following, captured_at, ok) "
                "VALUES ('2025-01-01', 37754047, 500, NULL, 1, 1)"
            ))
            DASHBOARD_CACHE.clear()
            monthly = q.fans(start="2025-01-01", end="2026-10-31")
            assert monthly["mode"] == "monthly"
            assert monthly["span_days"] == 669
            assert monthly["dates"][0] == "2025-01-01"
            assert monthly["series"][0] == 500
            assert monthly["dates"][-1] == "2026-10-31"
            assert monthly["series"][-1] == 1020

            # 同月多条时取峰值日期，并优先较晚日期
            db.write(lambda conn: conn.execute(
                "INSERT OR REPLACE INTO up_stat"
                "(stat_date, mid, follower, following, captured_at, ok) "
                "VALUES ('2026-10-15', 37754047, 5000, NULL, 1, 1), "
                "('2026-10-16', 37754047, 5000, NULL, 1, 1)"
            ))
            DASHBOARD_CACHE.clear()
            monthly = q.fans(start="2025-01-01", end="2026-10-31")
            assert monthly["dates"][-1] == "2026-10-16"
            assert monthly["series"][-1] == 5000

            try:
                q.fans(start="2026-11-01", end="2026-10-01")
            except ValueError:
                pass
            else:
                raise AssertionError("start 晚于 end 应报错")
        finally:
            db.close()


def test_fans_candidates_system_and_export():
    with tempfile.TemporaryDirectory() as tmp:
        db, q = setup(tmp)
        try:
            fans = q.fans(days=3)
            assert fans["mid"] == 37754047
            assert fans["name"] == "咻咻满"
            assert fans["dates"] == list(DATES)
            assert fans["series"] == [1000, 1010, 1020]
            assert fans["delta_1d"] == 10
            assert fans["delta_7d"] is None
            assert fans["delta_cum"] == 20
            assert fans["options"][0]["mid"] == 37754047
            wide = q.fans(days=30)
            assert wide["dates"] == list(DATES)

            # “最近记录”只返回最近 14 天，早于窗口的记录不进入表格
            db.write(lambda conn: conn.execute(
                "INSERT OR REPLACE INTO up_stat"
                "(stat_date, mid, follower, following, captured_at, ok) "
                "VALUES (?, ?, ?, NULL, 1, 1)",
                ("2026-10-01", 37754047, 900),
            ))
            DASHBOARD_CACHE.clear()
            recent = q.fans(days=90)
            assert recent["recent_days"] == 14
            assert [point["stat_date"] for point in recent["recent_points"]] == list(DATES)
            deltas = {point["stat_date"]: point["delta"] for point in recent["recent_points"]}
            assert deltas["2026-10-29"] == 100  # 对比 2026-10-01 的 900
            assert deltas["2026-10-30"] == 10
            assert deltas["2026-10-31"] == 10
            one_day = q.fans(days=90, recent_days=1)
            assert [point["stat_date"] for point in one_day["recent_points"]] == [DATES[-1]]

            # 缺失前一天时不能把跨多天的变化标成“日增”
            db.write(lambda conn: conn.execute(
                "DELETE FROM up_stat WHERE mid=37754047 AND stat_date=?",
                ("2026-10-30",),
            ))
            DASHBOARD_CACHE.clear()
            gapped = q.fans(days=3)
            assert gapped["delta_1d"] is None
            assert gapped["delta_7d"] is None

            candidates = q.candidates()
            assert candidates["total"] == 2
            assert candidates["pending"] == 1
            assert candidates["items"][0]["aid"] == 201
            assert 1 <= candidates["items"][0]["days_left"] <= 7
            assert candidates["items"][1]["decision"] == "rejected"

            system = q.system()
            assert system["data_date"] == DATES[-1]
            assert system["snapshot_days"] == 3
            assert system["blacklist"][0]["video_count"] == 1
            assert system["tasks"]["snapshot"]["ok"] is True
            assert system["errors"] == []
            assert system["preview"] is False
            assert system["next_runs"]["snapshot"] > 0
            assert system["next_runs"]["folder_sync"] > FIXED_NOW
            assert system["next_runs"]["discover"] > FIXED_NOW
        finally:
            db.close()


def test_duration_efficiency_queries():
    with tempfile.TemporaryDirectory() as tmp:
        db, q = setup(tmp)
        try:
            db.write(lambda conn: conn.executemany(
                "UPDATE video SET duration = ?, pubdate = ? WHERE aid = ?",
                [
                    (9000, 1_756_684_800, 101),  # 2025-09-01 前后，2.5 小时
                    (7200, 1_756_771_200, 102),  # 2 小时
                    (300, 1_756_857_600, 103),   # 5 分钟，应被排除
                ],
            ))
            DASHBOARD_CACHE.clear()

            efficiency = q.list_videos(sort="efficiency")
            assert [item["aid"] for item in efficiency["items"]] == [102]
            first = efficiency["items"][0]
            assert first["duration"] == 7200
            assert first["content_hours"] == 2.0
            assert first["efficiency"] == 2600.0  # 原始播放/小时，详情页口径
            assert abs(first["efficiency_score"] - 2207.5) < 0.1
            assert efficiency["duration_filter"] == {
                "min_duration_seconds": 600, "maturity_days": 2,
                "min_view": 2000, "shrink_minutes": 60,
            }

            # 播放不足 2000 的视频不进入效率榜（101 只有 1200）
            assert all(item["aid"] != 101 for item in efficiency["items"])

            interaction = q.list_videos(sort="interaction_efficiency")
            assert [item["aid"] for item in interaction["items"]] == [102]
            assert abs(interaction["items"][0]["interaction_efficiency_score"] - 159.36) < 0.1

            # 综合热度榜也应用最小播放门槛
            heat = q.list_videos(sort="heat")
            assert [item["aid"] for item in heat["items"]] == [102]

            # 时长缺失不计入时长效率榜
            ups = q.list_ups(sort="efficiency")
            assert ups["sort"] == "efficiency"
            up_a = next(item for item in ups["items"] if item["up_mid"] == 1)
            assert up_a["total_duration"] == int(9000 + 7200)
            assert up_a["content_hours"] == 4.5
            assert up_a["views_per_hour"] == round(6400 / 4.5, 1)
            assert up_a["efficiency"] is not None
            assert up_a["interaction_efficiency"] is not None

            by_duration = q.list_ups(sort="duration")
            assert [item["up_mid"] for item in by_duration["items"]][0] == 1

            db.write(lambda conn: conn.execute(
                "UPDATE video SET duration = NULL WHERE aid = 101"
            ))
            DASHBOARD_CACHE.clear()
            missing = q.list_videos(sort="efficiency")
            assert [item["aid"] for item in missing["items"]] == [102]

            # 发布未满 2 天的视频不进入效率榜
            db.write(lambda conn: conn.execute(
                "UPDATE video SET duration = 7200, pubdate = ? WHERE aid = 102",
                (int(day_bounds("2026-10-30")[0]),),
            ))
            DASHBOARD_CACHE.clear()
            fresh = q.list_videos(sort="efficiency")
            assert fresh["items"] == []
            assert fresh["available"] is False
            assert "发布满 2 天" in fresh["reason"]
            assert "播放 ≥ 2000" in fresh["reason"]
        finally:
            db.close()


def test_year_filter_and_published_sort():
    with tempfile.TemporaryDirectory() as tmp:
        db, q = setup(tmp)
        try:
            by_year = q.list_videos(sort="view", year=2026)
            assert [item["aid"] for item in by_year["items"]] == [102, 101, 103]
            assert by_year["year"] == 2026

            none = q.list_videos(sort="view", year=2025)
            assert none["items"] == []
            assert none["total"] == 0

            published = q.list_videos(sort="published")
            assert [item["aid"] for item in published["items"]] == [103, 101, 102]

            added = q.list_videos(sort="added")
            assert added["total"] == 3

            ups = q.list_ups(sort="efficiency", year=2026)
            assert ups["year"] == 2026 and ups["total"] == 2
            ups_none = q.list_ups(sort="views", year=2025)
            assert ups_none["total"] == 0
        finally:
            db.close()


def test_about_payload():
    with tempfile.TemporaryDirectory() as tmp:
        db, q = setup(tmp)
        try:
            about = q.about()
            assert about["data_date"] == DATES[-1]
            assert about["snapshot_days"] == 3
            assert about["snapshot_first_date"] == DATES[0]
            assert about["fans_start_date"] == DATES[0]
            assert about["schema_version"] == 4
            assert about["analytics"] == {
                "min_duration_seconds": 600,
                "maturity_days": 2,
                "min_view_for_ranking": 2000,
                "video_shrink_minutes": 60,
                "shrink_minutes": None,
            }
            assert about["heat_weights"]["view"] == 0.30
            assert about["followers"]["videos_total"] == 3
            assert about["followers"]["videos_covered"] == 0
            assert about["followers"]["videos_missing"] == 3
            assert about["folders"][0]["media_id"] == 3700023351
            assert about["preview"] is False
        finally:
            db.close()


def main_test():
    test_overview()
    test_video_listing_filters_and_sorting()
    test_growth_and_cold_start()
    test_video_detail_and_trend()
    test_up_queries()
    test_query_cache_invalidation()
    test_duration_efficiency_queries()
    test_year_filter_and_published_sort()
    test_about_payload()
    test_fans_custom_range_modes()
    test_fans_candidates_system_and_export()
    print("OK: Web 查询层总览/榜单/详情/UP/时长效率/粉丝/候选/系统测试全部通过")


if __name__ == "__main__":
    main_test()