from __future__ import annotations

import json
import os
from datetime import date, datetime
from pathlib import Path
from zoneinfo import ZoneInfo

from gyxx_flow.catalog import ScheduleEntry, WorkflowCatalog, WorkflowEntry
from gyxx_flow.scheduler_service import (
    PythonScheduler,
    SchedulerInstanceLock,
    SubprocessWorkflowLauncher,
    latest_due_slot,
)

PROJECT_ROOT = Path(__file__).resolve().parents[1]
SHANGHAI = ZoneInfo("Asia/Shanghai")


class FakeProcess:
    def __init__(self, pid: int = 1234) -> None:
        self.pid = pid
        self.return_code = None

    def poll(self):
        return self.return_code


class FakeLauncher:
    def __init__(self) -> None:
        self.launched = []

    def launch(self, workflow_id: str, slot: datetime, business_date: date):
        process = FakeProcess(1000 + len(self.launched))
        self.launched.append((workflow_id, slot, business_date, process))
        return process

    def close(self, process) -> None:
        pass


def test_latest_due_slot_supports_all_schedule_kinds() -> None:
    now = datetime(2026, 7, 28, 10, 5, tzinfo=SHANGHAI)
    assert latest_due_slot(ScheduleEntry("a", "daily", "10:00"), now) == datetime(2026, 7, 28, 10, 0, tzinfo=SHANGHAI)
    assert latest_due_slot(ScheduleEntry("a", "weekly", "09:00", days=("Monday",)), now) == datetime(2026, 7, 27, 9, 0, tzinfo=SHANGHAI)
    assert latest_due_slot(ScheduleEntry("a", "monthly", "08:00", day_of_month=1), now) == datetime(2026, 7, 1, 8, 0, tzinfo=SHANGHAI)
    assert latest_due_slot(ScheduleEntry("a", "interval_days", "11:00", every_days=3, anchor_date="2026-07-25"), now) == datetime(2026, 7, 25, 11, 0, tzinfo=SHANGHAI)


def test_latest_due_slot_chooses_latest_of_multiple_daily_times() -> None:
    schedule = ScheduleEntry(
        "a",
        "daily",
        "08:00",
        at_times=("08:00", "16:00", "22:00"),
    )

    assert latest_due_slot(
        schedule,
        datetime(2026, 7, 28, 7, 59, tzinfo=SHANGHAI),
    ) == datetime(2026, 7, 27, 22, 0, tzinfo=SHANGHAI)
    assert latest_due_slot(
        schedule,
        datetime(2026, 7, 28, 8, 5, tzinfo=SHANGHAI),
    ) == datetime(2026, 7, 28, 8, 0, tzinfo=SHANGHAI)
    assert latest_due_slot(
        schedule,
        datetime(2026, 7, 28, 16, 5, tzinfo=SHANGHAI),
    ) == datetime(2026, 7, 28, 16, 0, tzinfo=SHANGHAI)
    assert latest_due_slot(
        schedule,
        datetime(2026, 7, 28, 22, 5, tzinfo=SHANGHAI),
    ) == datetime(2026, 7, 28, 22, 0, tzinfo=SHANGHAI)


def test_tick_launches_due_workflow_once_and_persists_slot(tmp_path: Path) -> None:
    catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
    launcher = FakeLauncher()
    scheduler = PythonScheduler(catalog, tmp_path, launcher=launcher, misfire_grace_seconds=600)
    now = datetime(2026, 7, 28, 10, 5, tzinfo=SHANGHAI)

    started = scheduler.tick(now)
    scheduler.tick(now)

    assert {item[0] for item in launcher.launched} == {
        "content.summary.weekly",
        "product.persona.daily",
        "product.ecommerce_costs.daily",
        "product.tmall_baibu_apply",
    }
    assert set(started) == {item[0] for item in launcher.launched}
    state = json.loads((tmp_path / "state" / "scheduler" / "state.json").read_text(encoding="utf-8"))
    assert state["jobs"]["product.persona.daily"]["slot"].startswith("2026-07-28T10:00:00")


def test_completed_process_is_recorded_and_not_relaunched(tmp_path: Path) -> None:
    catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
    launcher = FakeLauncher()
    scheduler = PythonScheduler(catalog, tmp_path, launcher=launcher, misfire_grace_seconds=600)
    now = datetime(2026, 7, 28, 23, 5, tzinfo=SHANGHAI)
    scheduler.tick(now)
    process = next(item[3] for item in launcher.launched if item[0] == "product.alert.daily")
    process.return_code = 0

    scheduler.tick(now)

    state = scheduler.status()
    assert state["jobs"]["product.alert.daily"]["status"] == "success"
    assert sum(item[0] == "product.alert.daily" for item in launcher.launched) == 1


def test_multiple_daily_slots_launch_once_without_overlap(tmp_path: Path) -> None:
    workflow = WorkflowEntry(
        workflow_id="shop.price_appeal",
        module="shop_intelligence",
        trigger="scheduled",
        entry="run.py",
        source_task_name="price appeal",
    )
    schedule = ScheduleEntry(
        workflow_id=workflow.workflow_id,
        kind="daily",
        at="08:00",
        at_times=("08:00", "16:00", "22:00"),
    )
    catalog = WorkflowCatalog(
        timezone="Asia/Shanghai",
        workflows=(workflow,),
        schedules=(schedule,),
    )
    launcher = FakeLauncher()
    scheduler = PythonScheduler(
        catalog,
        tmp_path,
        launcher=launcher,
        misfire_grace_seconds=600,
    )

    scheduler.tick(datetime(2026, 7, 28, 8, 5, tzinfo=SHANGHAI))
    scheduler.tick(datetime(2026, 7, 28, 8, 5, tzinfo=SHANGHAI))
    morning_process = launcher.launched[0][3]
    morning_process.return_code = 0
    scheduler.tick(datetime(2026, 7, 28, 8, 6, tzinfo=SHANGHAI))

    scheduler.tick(datetime(2026, 7, 28, 16, 5, tzinfo=SHANGHAI))
    afternoon_process = launcher.launched[1][3]
    scheduler.tick(datetime(2026, 7, 28, 22, 5, tzinfo=SHANGHAI))
    assert len(launcher.launched) == 2

    afternoon_process.return_code = 0
    scheduler.tick(datetime(2026, 7, 28, 22, 5, tzinfo=SHANGHAI))
    scheduler.tick(datetime(2026, 7, 28, 22, 5, tzinfo=SHANGHAI))

    assert [item[1].hour for item in launcher.launched] == [8, 16, 22]
    assert scheduler.status()["jobs"][workflow.workflow_id]["slot"].startswith(
        "2026-07-28T22:00:00"
    )


def test_content_metrics_schedule_has_initial_and_retry_slots(tmp_path: Path) -> None:
    workflow = WorkflowEntry(
        workflow_id="content.metrics.daily",
        module="content_marketing",
        trigger="scheduled",
        entry="run_all.py",
        source_task_name="content metrics daily",
    )
    schedule = ScheduleEntry(
        workflow_id=workflow.workflow_id,
        kind="daily",
        at="01:00",
        at_times=("01:00", "05:00"),
    )
    catalog = WorkflowCatalog(
        timezone="Asia/Shanghai",
        workflows=(workflow,),
        schedules=(schedule,),
    )
    launcher = FakeLauncher()
    scheduler = PythonScheduler(
        catalog,
        tmp_path,
        launcher=launcher,
        misfire_grace_seconds=600,
    )

    scheduler.tick(datetime(2026, 9, 2, 1, 5, tzinfo=SHANGHAI))
    launcher.launched[0][3].return_code = 0
    scheduler.tick(datetime(2026, 9, 2, 1, 6, tzinfo=SHANGHAI))
    scheduler.tick(datetime(2026, 9, 2, 5, 5, tzinfo=SHANGHAI))

    assert [item[1].strftime("%H:%M") for item in launcher.launched] == [
        "01:00",
        "05:00",
    ]


def test_misfire_older_than_grace_is_not_launched(tmp_path: Path) -> None:
    catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
    launcher = FakeLauncher()
    scheduler = PythonScheduler(catalog, tmp_path, launcher=launcher, misfire_grace_seconds=60)
    scheduler.tick(datetime(2026, 7, 28, 10, 2, tzinfo=SHANGHAI))
    assert launcher.launched == []


def test_configured_schedules_disable_cookie_refresh_job() -> None:
    catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")

    assert len(catalog.schedules) == 29
    assert [
        schedule.workflow_id for schedule in catalog.schedules if not schedule.enabled
    ] == [
        "content.metrics.backfill",
        "content.marketing_report.daily",
        "content.relogin.weekly",
        "supply.replenishment.weekly",
    ]


def test_dry_run_reports_due_without_writing_state(tmp_path: Path) -> None:
    catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
    launcher = FakeLauncher()
    scheduler = PythonScheduler(catalog, tmp_path, launcher=launcher, misfire_grace_seconds=600, dry_run=True)
    due = scheduler.tick(datetime(2026, 7, 28, 10, 5, tzinfo=SHANGHAI))
    assert "product.persona.daily" in due
    assert launcher.launched == []
    assert not (tmp_path / "state" / "scheduler" / "state.json").exists()


def test_restart_does_not_overlap_a_still_running_workflow(tmp_path: Path) -> None:
    state_path = tmp_path / "state" / "scheduler" / "state.json"
    state_path.parent.mkdir(parents=True)
    state_path.write_text(json.dumps({
        "schema_version": 1, "updated_at": None,
        "jobs": {"product.persona.daily": {
            "slot": "2026-07-27T10:00:00+08:00", "status": "running", "pid": os.getpid(),
        }},
    }), encoding="utf-8")
    launcher = FakeLauncher()
    scheduler = PythonScheduler(
        WorkflowCatalog.load(PROJECT_ROOT / "config"), tmp_path,
        launcher=launcher, misfire_grace_seconds=600,
    )

    scheduler.tick(datetime(2026, 7, 28, 10, 5, tzinfo=SHANGHAI))

    assert "product.persona.daily" not in {item[0] for item in launcher.launched}


def test_scheduler_instance_lock_rejects_second_owner(tmp_path: Path) -> None:
    lock_path = tmp_path / "service.lock"
    with SchedulerInstanceLock(lock_path):
        try:
            with SchedulerInstanceLock(lock_path):
                raise AssertionError("second lock must not be acquired")
        except RuntimeError as exc:
            assert "already running" in str(exc)
    assert not lock_path.exists()


def test_schedule_business_date_offset_uses_two_days_before(tmp_path: Path) -> None:
    catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
    launcher = FakeLauncher()
    scheduler = PythonScheduler(
        catalog,
        tmp_path,
        launcher=launcher,
        misfire_grace_seconds=600,
    )

    scheduler.tick(datetime(2026, 8, 1, 9, 5, tzinfo=SHANGHAI))

    launched = next(
        item for item in launcher.launched
        if item[0] == "shop.jd_self_operated.daily"
    )
    assert launched[1].date() == date(2026, 8, 1)
    assert launched[2] == date(2026, 7, 30)


def test_product_daily_schedule_passes_previous_business_day(tmp_path: Path) -> None:
    catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
    launcher = FakeLauncher()
    scheduler = PythonScheduler(
        catalog,
        tmp_path,
        launcher=launcher,
        misfire_grace_seconds=600,
    )

    scheduler.tick(datetime(2026, 8, 1, 8, 45, tzinfo=SHANGHAI))

    product_launch = next(
        item for item in launcher.launched if item[0] == "product.daily"
    )
    assert product_launch[1].date() == date(2026, 8, 1)
    assert product_launch[2] == date(2026, 7, 31)


def test_subprocess_launcher_passes_explicit_business_date_and_execute(
    monkeypatch,
    tmp_path: Path,
) -> None:
    captured: dict[str, object] = {}

    class Process:
        pid = 9090

        def poll(self):
            return None

    def fake_popen(argv, **kwargs):
        captured["argv"] = argv
        captured["kwargs"] = kwargs
        return Process()

    monkeypatch.setattr("gyxx_flow.scheduler_service.subprocess.Popen", fake_popen)
    launcher = SubprocessWorkflowLauncher(tmp_path, tmp_path / "var")
    process = launcher.launch(
        "shop.jd_self_operated.daily",
        datetime(2026, 8, 1, 10, 0, tzinfo=SHANGHAI),
        date(2026, 7, 31),
    )
    launcher.close(process)

    assert captured["argv"][-4:] == [
        "shop.jd_self_operated.daily",
        "--date",
        "2026-07-31",
        "--execute",
    ]
    assert captured["kwargs"]["env"]["GYXX_SCHEDULE_SLOT"] == "2026-08-01T10:00:00+08:00"
    assert captured["kwargs"]["env"]["GYXX_SCHEDULE_TIME"] == "10:00"
    assert "--scheduled" not in captured["argv"]


def test_subprocess_launcher_selects_tmall_no_hyperlink_scope(
    monkeypatch,
    tmp_path: Path,
) -> None:
    captured: dict[str, object] = {}

    class Process:
        pid = 9091

        def poll(self):
            return None

    def fake_popen(argv, **kwargs):
        captured["argv"] = argv
        return Process()

    monkeypatch.setattr("gyxx_flow.scheduler_service.subprocess.Popen", fake_popen)
    launcher = SubprocessWorkflowLauncher(tmp_path, tmp_path / "var")
    process = launcher.launch(
        "product.tmall_baibu_apply",
        datetime(2026, 8, 31, 18, 0, tzinfo=SHANGHAI),
        date(2026, 8, 31),
    )
    launcher.close(process)

    assert captured["argv"][-6:] == [
        "product.tmall_baibu_apply",
        "--date",
        "2026-08-31",
        "--tmall-baibu-import-mode",
        "without_hyperlinks",
        "--execute",
    ]