from datetime import datetime
import json
import os


def test_build_defined_ppt_jobs_uses_latest_date():
    from services.ppt_auto_generation_service import build_defined_ppt_jobs

    jobs = build_defined_ppt_jobs(latest_date="2026-05-11")
    by_type = {job.report_type: job for job in jobs}

    assert list(by_type) == [
        "daily", "weekly", "monthly", "quarterly", "half_yearly", "annual", "ttm",
        "strategy", "competitor", "competitor_v4", "promo", "promo_compare",
        "forecast_pre_event", "vendor", "category", "customer", "new_product",
        "market_intel", "price_elasticity",
    ]
    assert by_type["daily"].sub_arg == "2026/05/11"
    assert by_type["monthly"].sub_arg == "2026/05"
    assert by_type["quarterly"].sub_arg == "2026/Q2"
    assert by_type["half_yearly"].sub_arg == "2026/H1"
    assert by_type["annual"].sub_arg == "2026"
    assert by_type["strategy"].sub_arg == "2026/05/01-2026/05/11"
    assert by_type["market_intel"].sub_arg == "2026/05/11 起一週"
    assert by_type["competitor"].sub_arg == "monthly"
    assert by_type["promo"].sub_arg == "2026/05/05-2026/05/11"
    assert by_type["promo"].expected_params["label"] == "2026/05/05~2026/05/11"
    assert by_type["strategy"].expected_params == {
        "report_type": "strategy",
        "start": "2026/05/01",
        "end": "2026/05/11",
        "label": "2026/05 月策略（截至 05/11）",
    }


def test_auto_generation_respects_disabled_flag(monkeypatch):
    monkeypatch.setenv("PPT_AUTO_GENERATION_ENABLED", "false")

    from services.ppt_auto_generation_service import generate_defined_ppt_reports

    result = generate_defined_ppt_reports(report_types=["daily"])

    assert result["ok"] is False
    assert result["status"] == "disabled"


def test_dry_run_does_not_generate(monkeypatch):
    monkeypatch.setenv("PPT_AUTO_GENERATION_ENABLED", "true")

    from services import ppt_auto_generation_service as svc

    monkeypatch.setattr(svc, "_latest_sales_date", lambda: "2026-05-11")

    result = svc.generate_defined_ppt_reports(
        report_types=["daily", "monthly"],
        dry_run=True,
    )

    assert result["ok"] is True
    assert result["status"] == "planned"
    assert [job["report_type"] for job in result["jobs"]] == ["daily", "monthly"]


def test_coverage_marks_ready_from_database(monkeypatch):
    from services import ppt_auto_generation_service as svc

    class _Rows:
        def fetchall(self):
            return [
                (
                    "daily",
                    json.dumps({"report_type": "daily", "date": "2026/05/11"}),
                    "/tmp/ocbot_daily_ok.pptx",
                    datetime(2026, 5, 11, 20, 30),
                ),
                (
                    "monthly",
                    json.dumps({"report_type": "monthly", "month": "2026/05"}),
                    "/tmp/ocbot_monthly_ok.pptx",
                    datetime(2026, 5, 11, 20, 30),
                ),
            ]

    class _Session:
        def execute(self, *_args, **_kwargs):
            return _Rows()

        def close(self):
            return None

    monkeypatch.setattr(svc, "get_session", lambda: _Session())
    monkeypatch.setattr(svc, "_latest_sales_date", lambda: "2026-05-11")
    monkeypatch.setenv("PPT_AUTO_GENERATION_ENABLED", "true")

    result = svc.get_defined_report_coverage(
        month_start=datetime(2026, 5, 1),
        month_end=datetime(2026, 6, 1),
        reports_dir="/tmp/does-not-exist-for-test",
        report_types=["daily", "monthly", "weekly"],
    )

    by_key = {item["key"]: item for item in result["items"]}
    assert by_key["daily"]["ready"] is True
    assert by_key["daily"]["status"] == "ready"
    assert by_key["daily"]["status_label"] == "已產出"
    assert by_key["monthly"]["ready"] is True
    assert by_key["weekly"]["ready"] is False
    assert by_key["weekly"]["status"] == "missing"
    assert by_key["weekly"]["status_label"] == "待排程補齊"
    assert result["missing_count"] == 1


def test_due_schedule_kinds_include_periodic_boundaries():
    from services.ppt_auto_generation_service import get_due_schedule_kinds

    assert get_due_schedule_kinds(datetime(2026, 5, 18)) == ["daily", "weekly"]
    assert get_due_schedule_kinds(datetime(2026, 7, 1)) == [
        "daily",
        "monthly",
        "quarterly",
        "half_yearly",
    ]
    assert get_due_schedule_kinds(datetime(2026, 1, 1)) == [
        "daily",
        "monthly",
        "quarterly",
        "half_yearly",
        "annual",
    ]


def test_latest_schedule_occurrence_replays_missed_slots():
    from services.ppt_auto_generation_service import get_latest_schedule_occurrence

    assert get_latest_schedule_occurrence("daily", datetime(2026, 6, 6, 14, 0)) == datetime(2026, 6, 5, 20, 30)
    assert get_latest_schedule_occurrence("daily", datetime(2026, 6, 6, 21, 0)) == datetime(2026, 6, 6, 20, 30)
    assert get_latest_schedule_occurrence("weekly", datetime(2026, 6, 6, 14, 0)) == datetime(2026, 6, 1, 20, 40)
    assert get_latest_schedule_occurrence("monthly", datetime(2026, 6, 6, 14, 0)) == datetime(2026, 6, 1, 20, 50)
    assert get_latest_schedule_occurrence("quarterly", datetime(2026, 6, 6, 14, 0)) == datetime(2026, 4, 1, 21, 0)
    assert get_latest_schedule_occurrence("half_yearly", datetime(2026, 6, 6, 14, 0)) == datetime(2026, 1, 1, 21, 10)
    assert get_latest_schedule_occurrence("annual", datetime(2026, 6, 6, 14, 0)) == datetime(2026, 1, 1, 21, 20)


def test_ppt_catchup_plan_marks_missing_after_missed_slot(monkeypatch):
    from services import ppt_auto_generation_service as svc

    class _Rows:
        def fetchall(self):
            return []

    class _Session:
        def execute(self, *_args, **_kwargs):
            return _Rows()

        def close(self):
            return None

    monkeypatch.setattr(svc, "get_session", lambda: _Session())
    monkeypatch.setattr(svc, "_latest_sales_date", lambda: "2026-06-04")

    plan = svc.get_scheduled_ppt_catchup_plan(
        now=datetime(2026, 6, 6, 14, 0),
        schedule_kinds=["daily"],
    )

    assert plan[0]["scheduled_at"] == "2026-06-05 20:30"
    assert plan[0]["missing_report_types"] == ["daily"]
    assert plan[0]["ready"] is False


def test_ppt_catchup_plan_uses_existing_exact_report(monkeypatch, tmp_path):
    from services import ppt_auto_generation_service as svc

    pptx = tmp_path / "ocbot_daily_ok.pptx"
    pptx.write_bytes(b"pptx")

    class _Rows:
        def fetchall(self):
            return [
                (
                    "daily",
                    json.dumps({"report_type": "daily", "date": "2026/06/04"}),
                    str(pptx),
                )
            ]

    class _Session:
        def execute(self, *_args, **_kwargs):
            return _Rows()

        def close(self):
            return None

    monkeypatch.setattr(svc, "get_session", lambda: _Session())
    monkeypatch.setattr(svc, "_latest_sales_date", lambda: "2026-06-04")

    plan = svc.get_scheduled_ppt_catchup_plan(
        now=datetime(2026, 6, 6, 14, 0),
        schedule_kinds=["daily"],
    )

    assert plan[0]["ready_report_types"] == ["daily"]
    assert plan[0]["missing_report_types"] == []
    assert plan[0]["ready"] is True


def test_ppt_catchup_generates_only_missing_types(monkeypatch):
    from services import ppt_auto_generation_service as svc

    calls = []

    monkeypatch.setattr(
        svc,
        "get_scheduled_ppt_catchup_plan",
        lambda **_kwargs: [{
            "schedule_kind": "weekly",
            "scheduled_at": "2026-06-01 20:40",
            "missing_report_types": ["market_intel"],
        }],
    )

    def fake_generate_defined_ppt_reports(**kwargs):
        calls.append(kwargs)
        return {"ok": True, "ready": 1, "errors": 0, "jobs": [{"report_type": "market_intel"}]}

    monkeypatch.setattr(svc, "generate_defined_ppt_reports", fake_generate_defined_ppt_reports)

    result = svc.catch_up_scheduled_ppt_reports()

    assert result["status"] == "completed"
    assert result["generated_kinds"] == ["weekly"]
    assert calls == [{
        "report_types": ["market_intel"],
        "schedule_kind": "weekly",
        "force": False,
    }]


def test_ppt_catchup_background_queues_without_main_loop_block(monkeypatch):
    from services import ppt_auto_generation_service as svc

    calls = []
    threads = []

    class _Thread:
        def __init__(self, *, target, name, daemon):
            self.target = target
            self.name = name
            self.daemon = daemon
            threads.append(self)

        def start(self):
            self.target()

    def fake_catchup(**kwargs):
        calls.append(kwargs)
        return {"ok": True, "status": "skipped", "runs": []}

    monkeypatch.setattr(svc.threading, "Thread", _Thread)
    monkeypatch.setattr(svc, "catch_up_scheduled_ppt_reports", fake_catchup)

    result = svc.start_scheduled_ppt_catchup_background(schedule_kinds=["daily", "weekly"])

    assert result["status"] == "queued"
    assert result["schedule_kinds"] == ["daily", "weekly"]
    assert threads[0].name == "ppt-auto-generation-catchup"
    assert threads[0].daemon is True
    assert calls == [{
        "now": None,
        "force": False,
        "schedule_kinds": ["daily", "weekly"],
    }]


def test_scheduled_generation_sets_fast_fallback_env(monkeypatch, tmp_path):
    from routes import openclaw_bot_routes as bot_routes
    from services import ppt_auto_generation_service as svc

    output = tmp_path / "ocbot_market_intel_ok.pptx"
    output.write_bytes(b"pptx")
    observed = {}
    job = svc.build_defined_ppt_jobs(
        latest_date="2026-05-11",
        report_types=["market_intel"],
    )[0]

    monkeypatch.delenv("PPT_SCHEDULED_FAST_FALLBACK", raising=False)
    monkeypatch.delenv("MCP_FAST_STATIC_FALLBACK", raising=False)
    monkeypatch.setattr(svc, "_expire_matching_ppt_cache", lambda _job: 0)
    monkeypatch.setattr(bot_routes, "send_message", lambda *_args, **_kwargs: None, raising=False)

    def fake_generate_ppt_cmd(*_args, **_kwargs):
        observed["ppt_fast"] = os.getenv("PPT_SCHEDULED_FAST_FALLBACK")
        observed["mcp_fast"] = os.getenv("MCP_FAST_STATIC_FALLBACK")
        return str(output)

    monkeypatch.setattr(bot_routes, "_generate_ppt_cmd", fake_generate_ppt_cmd)

    path, invalidated = svc._generate_job(job, schedule_kind="weekly")

    assert path == str(output)
    assert invalidated == 0
    assert observed == {"ppt_fast": "true", "mcp_fast": "true"}
    assert os.getenv("PPT_SCHEDULED_FAST_FALLBACK") is None
    assert os.getenv("MCP_FAST_STATIC_FALLBACK") is None


def test_schedule_cadence_status_exposes_all_periodic_contracts():
    from services.ppt_auto_generation_service import get_schedule_cadence_status

    cadences = get_schedule_cadence_status([
        {"key": "daily", "ready": True},
        {"key": "weekly", "ready": False},
        {"key": "market_intel", "ready": True},
    ])
    by_key = {cadence["key"]: cadence for cadence in cadences}

    assert [cadence["schedule_text"] for cadence in cadences] == [
        "每日 20:30",
        "每週一 20:40",
        "每月 1 日 20:50",
        "每季首日 21:00",
        "每半年首日 21:10",
        "每年 1/1 21:20",
    ]
    assert by_key["weekly"]["report_types"] == ["weekly", "market_intel"]
    assert by_key["weekly"]["ready_count"] == 1
    assert by_key["weekly"]["missing_report_types"] == ["weekly"]
    assert by_key["weekly"]["missing_report_labels"] == ["週報"]
    assert by_key["weekly"]["status_label"] == "已完成 1/2"
    assert by_key["weekly"]["coverage_text"] == "1/2"
    assert "TTM 滾動 12 月" in by_key["monthly"]["report_labels"]


def test_scheduled_generation_uses_profile_without_generating(monkeypatch):
    from services import ppt_auto_generation_service as svc

    calls = []

    def fake_generate_defined_ppt_reports(**kwargs):
        calls.append(kwargs)
        return {"ok": True, "ready": 2, "errors": 0, "jobs": [{"report_type": "weekly"}]}

    monkeypatch.setattr(svc, "generate_defined_ppt_reports", fake_generate_defined_ppt_reports)

    result = svc.generate_scheduled_ppt_reports(schedule_kind="weekly", force=True)

    assert result["ok"] is True
    assert result["schedule_kinds"] == ["weekly"]
    assert calls == [{
        "report_types": ("weekly", "market_intel"),
        "schedule_kind": "weekly",
        "force": True,
    }]


def test_force_regeneration_expires_only_matching_cache(monkeypatch):
    from services import ppt_auto_generation_service as svc

    captured = {}

    class _Result:
        rowcount = 1

    class _Session:
        def execute(self, sql, params):
            captured["sql"] = str(sql)
            captured["params"] = dict(params)
            return _Result()

        def commit(self):
            captured["committed"] = True

        def close(self):
            captured["closed"] = True

    monkeypatch.setattr(svc, "get_session", lambda: _Session())
    monkeypatch.setattr(svc, "_normalize_ppt_cache_parameters", lambda params: "normalized-cache-key")

    job = svc.build_defined_ppt_jobs(latest_date="2026-05-11", report_types=["daily"])[0]
    affected = svc._expire_matching_ppt_cache(job)

    assert affected == 1
    assert "UPDATE ppt_reports" in captured["sql"]
    assert "report_type = :report_type" in captured["sql"]
    assert "parameters = :parameters" in captured["sql"]
    assert captured["params"]["report_type"] == "daily"
    assert captured["params"]["parameters"] == "normalized-cache-key"
    assert captured["committed"] is True
    assert captured["closed"] is True


def test_ppt_observability_single_regeneration_forces_cache_refresh():
    from pathlib import Path

    js = Path("web/static/js/observability-charts.js").read_text(encoding="utf-8")

    assert "force: Boolean(force)" in js
    assert "triggerGeneration(false, [reportType], singleButton, label, true)" in js


def test_ppt_preview_reports_missing_converter(monkeypatch, tmp_path):
    from services import ppt_preview_service as svc

    pptx = tmp_path / "demo.pptx"
    pptx.write_bytes(b"fake pptx")
    monkeypatch.setattr(svc.shutil, "which", lambda _name: None)

    result = svc.build_ppt_preview(pptx, cache_dir=tmp_path / "cache")

    assert result.ok is False
    assert "LibreOffice" in (result.error or "")


def test_ppt_preview_cache_info_is_read_only(tmp_path):
    from pathlib import Path
    from services import ppt_preview_service as svc

    pptx = tmp_path / "demo.pptx"
    pptx.write_bytes(b"fake pptx")
    cache_dir = tmp_path / "cache"

    miss = svc.get_ppt_preview_cache_info(pptx, cache_dir=cache_dir)
    assert miss.pdf_path
    assert miss.cache_exists is False

    Path(miss.pdf_path).parent.mkdir(parents=True, exist_ok=True)
    Path(miss.pdf_path).write_bytes(b"%PDF-1.4\n")

    hit = svc.get_ppt_preview_cache_info(pptx, cache_dir=cache_dir)
    assert hit.cache_exists is True
    assert hit.cache_size_kb is not None