"""
v2 pipeline orchestrator.

Same scrape step as the original pipeline (the scrapers are unchanged), but the
sink is the Baskit Ops ingestion API instead of load_mysql.py:

    per retailer: scrape -> push_to_ops (open/products/prices/close)

  python run_pipeline_v2.py run          # one full cycle (all retailers)
  python run_pipeline_v2.py push         # push only (skip scraping)
  python run_pipeline_v2.py --rotate     # continuous rotate loop (production)
  python run_pipeline_v2.py --schedule   # alias for --rotate (legacy flag)
  python run_pipeline_v2.py abort-stale  # close stuck open ingest runs
"""
from __future__ import annotations

import argparse
import logging
import time
from concurrent.futures import Future, ThreadPoolExecutor
from datetime import datetime, timezone
from typing import Any

import ops_config as cfg
from ingest_client import IngestClient, IngestError
import alerts
import push_to_ops  # noqa: E402

LOG = logging.getLogger("baskit.pipeline.v2")


def _base_pipeline():
    """Lazy-load parent orchestrator (needs `schedule` + scrapers) only when scraping."""
    import run_pipeline as base_pipeline  # noqa: WPS433
    return base_pipeline


def _ts() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")


def _retailers_in_order() -> list[str]:
    """Retailers to rotate, in BASKIT_RETAILERS order, skipping unknown names."""
    return [r for r in cfg.RETAILERS if r in push_to_ops.RETAILER_QUERIES]


def push_retailer(retailer: str) -> None:
    """Push one retailer to the ingestion API."""
    push_to_ops.push_retailer(IngestClient(), retailer)


def push_all() -> None:
    """Push every configured retailer to the ingestion API."""
    for retailer in _retailers_in_order():
        try:
            push_retailer(retailer)
        except IngestError:
            pass


def run_retailer(retailer: str, refresh: bool | None = None, scrape: bool = True) -> None:
    """Scrape one retailer, then push its SQLite snapshot to Ops."""
    base_pipeline = _base_pipeline()
    started = time.monotonic()
    use_refresh = base_pipeline.config.REFRESH if refresh is None else refresh
    LOG.info("[%s] v2 retailer start: %s (scrape=%s refresh=%s)", _ts(), retailer, scrape, use_refresh)
    if retailer not in push_to_ops.RETAILER_QUERIES:
        LOG.warning("unknown retailer %r — skipped", retailer)
        return
    try:
        if scrape:
            scrape_msg = f"Scraping {retailer} catalogue"
            alerts.track_job(retailer, "scrape", message=scrape_msg)
            try:
                with alerts.heartbeat_keepalive(
                    300.0, retailer=retailer, stage="scrape", message=scrape_msg,
                ):
                    base_pipeline.scrape_retailer(retailer, use_refresh)
            except Exception as exc:
                alerts.track_job(retailer, "error", message=f"Scrape failed: {exc}")
                alerts.report(
                    f"scrape failed for {retailer}",
                    stage="scrape", retailer=retailer, exc=exc,
                )
                raise
        alerts.track_job(retailer, "push", message=f"Pushing {retailer} to Ops")
        try:
            push_retailer(retailer)
        except Exception as exc:
            alerts.track_job(retailer, "error", message=f"Push failed: {exc}")
            raise
        elapsed = time.monotonic() - started
        LOG.info("[%s] v2 retailer done: %s in %.0fs", _ts(), retailer, elapsed)
        alerts.report_success(retailer, "cycle", f"scrape+push completed in {elapsed:.0f}s")
    finally:
        alerts.clear_job(retailer)


def run_cycle(refresh: bool | None = None, scrape: bool = True) -> None:
    """Run all retailers sequentially (scrape + push each)."""
    base_pipeline = _base_pipeline()
    started = time.monotonic()
    retailers = _retailers_in_order()
    LOG.info("[%s] v2 full cycle start (retailers=%s scrape=%s)", _ts(), ",".join(retailers), scrape)
    cfg.require_service_key()
    base_pipeline.config.ensure_dirs()
    use_refresh = base_pipeline.config.REFRESH if refresh is None else refresh
    for retailer in retailers:
        try:
            run_retailer(retailer, refresh=use_refresh, scrape=scrape)
        except Exception:
            LOG.exception("[%s] retailer cycle failed — continuing", retailer)
    LOG.info("[%s] v2 full cycle done in %.0fs", _ts(), time.monotonic() - started)


def cmd_run(args: argparse.Namespace) -> None:
    cfg.require_service_key()
    _base_pipeline().config.ensure_dirs()
    run_cycle(refresh=args.refresh or None, scrape=True)


def cmd_push(args: argparse.Namespace) -> None:
    cfg.require_service_key()
    _base_pipeline().config.ensure_dirs()
    run_cycle(scrape=False)


def cmd_abort_stale(args: argparse.Namespace) -> None:
    """Abort open ingest runs older than --older-than hours (default 6)."""
    cfg.require_service_key()
    older_than_h = args.older_than if args.older_than is not None else 6.0
    cutoff = time.time() - older_than_h * 3600
    client = IngestClient()
    runs = client.list_runs().get("runs", [])
    aborted = 0
    for r in runs:
        if r.get("status") != "open":
            continue
        started = str(r.get("started_at") or "")
        try:
            # Ops stores local wall time without TZ — treat as UTC-ish for age.
            ts = datetime.strptime(started[:19], "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)
        except ValueError:
            continue
        if ts.timestamp() > cutoff:
            continue
        rid = int(r["id"])
        note = f"Aborted by worker: open longer than {older_than_h:g}h (started {started})"
        try:
            client.abort_run(rid, note=note)
            print(f"aborted run #{rid} ({started})")
            aborted += 1
        except IngestError as exc:
            print(f"failed to abort run #{rid}: {exc}")
    print(f"done — aborted {aborted} open run(s)")


def cmd_rotate(args: argparse.Namespace) -> None:
    """
    Continuous rotate with bounded parallelism.

    Start retailers in BASKIT_RETAILERS order. If a job is still running after
    BASKIT_ROTATE_START_AFTER_SEC (default 45 min), start the next retailer so
    long PnP scrapes do not block Checkers/Woolworths. Cap concurrent jobs at
    BASKIT_ROTATE_MAX_PARALLEL (default 2).
    """
    retailers = _retailers_in_order()
    if not retailers:
        raise SystemExit("No retailers configured in BASKIT_RETAILERS")
    cooldown = args.cooldown if args.cooldown is not None else _base_pipeline().config.ROTATE_COOLDOWN_SEC
    max_parallel = (
        args.max_parallel if args.max_parallel is not None
        else _base_pipeline().config.ROTATE_MAX_PARALLEL
    )
    start_after = (
        args.start_after if args.start_after is not None
        else _base_pipeline().config.ROTATE_START_AFTER_SEC
    )
    max_parallel = max(1, int(max_parallel))
    start_after = max(0, int(start_after))
    chain = " -> ".join(retailers + [retailers[0]])
    LOG.info(
        "starting rotate loop: %s (max_parallel=%s start_after=%ss cooldown=%ss)",
        chain, max_parallel, start_after, cooldown,
    )
    cfg.require_service_key()
    _base_pipeline().config.ensure_dirs()
    use_refresh = args.refresh or None

    next_idx = 0
    # retailer -> {future, started_at}
    active: dict[str, dict[str, Any]] = {}

    def _reap() -> None:
        done = [name for name, slot in active.items() if slot["future"].done()]
        for name in done:
            fut: Future = active.pop(name)["future"]
            try:
                fut.result()
            except Exception:
                LOG.exception("[%s] retailer cycle failed — continuing rotate loop", name)

    def _should_launch() -> bool:
        if len(active) >= max_parallel:
            return False
        if not active:
            return True
        oldest = min(float(slot["started_at"]) for slot in active.values())
        return (time.monotonic() - oldest) >= start_after

    def _next_idle_retailer() -> str | None:
        nonlocal next_idx
        for _ in range(len(retailers)):
            name = retailers[next_idx % len(retailers)]
            next_idx += 1
            if name not in active:
                return name
        return None

    with ThreadPoolExecutor(max_workers=max_parallel, thread_name_prefix="v2-rot") as pool:
        while True:
            _reap()

            launched = False
            while _should_launch():
                candidate = _next_idle_retailer()
                if candidate is None:
                    break
                LOG.info(
                    "[%s] scheduling %s (%d/%d in flight)",
                    _ts(), candidate, len(active) + 1, max_parallel,
                )
                fut = pool.submit(
                    run_retailer,
                    candidate,
                    use_refresh if use_refresh is not None else None,
                    True,
                )
                active[candidate] = {"future": fut, "started_at": time.monotonic()}
                launched = True
                # After launching into a non-empty pool, wait for start_after
                # before considering another concurrent slot.
                if len(active) > 1 or start_after > 0:
                    break

            if not active:
                if cooldown > 0:
                    alerts.set_status("cooldown", message=f"Waiting {cooldown}s before next wave")
                    LOG.info("cooldown %ss before next wave", cooldown)
                    time.sleep(cooldown)
                elif not launched:
                    time.sleep(1)
                continue

            time.sleep(5)


def main() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )
    p = argparse.ArgumentParser(description="Baskit v2 pipeline (scrape -> ops ingestion API)")
    p.add_argument(
        "--rotate", action="store_true",
        help="continuous rotate loop: scrape+push one retailer, then the next",
    )
    p.add_argument(
        "--schedule", action="store_true",
        help="alias for --rotate (legacy hourly flag name)",
    )
    p.add_argument(
        "--cooldown", type=int, default=None,
        help="seconds between empty waves (default: BASKIT_ROTATE_COOLDOWN_SEC)",
    )
    p.add_argument(
        "--max-parallel", type=int, default=None,
        help="max concurrent retailers (default: BASKIT_ROTATE_MAX_PARALLEL)",
    )
    p.add_argument(
        "--start-after", type=int, default=None,
        help="seconds before starting next retailer while one is still running "
             "(default: BASKIT_ROTATE_START_AFTER_SEC)",
    )
    p.add_argument("--refresh", action="store_true", help="force --refresh on scrapers this cycle")
    sub = p.add_subparsers(dest="cmd")
    sub.add_parser("run", help="scrape then push all retailers").set_defaults(func=cmd_run)
    sub.add_parser("push", help="push only (no scrape)").set_defaults(func=cmd_push)
    abort_p = sub.add_parser(
        "abort-stale",
        help="abort open ingest runs older than N hours (cleanup stuck pushes)",
    )
    abort_p.add_argument("--older-than", type=float, default=6.0, help="hours (default 6)")
    abort_p.set_defaults(func=cmd_abort_stale)
    args = p.parse_args()

    if args.rotate or args.schedule:
        if args.schedule and not args.rotate:
            LOG.info("--schedule is now an alias for --rotate (continuous retailer loop)")
        cmd_rotate(args)
    elif args.cmd in (None, "run"):
        cmd_run(args)
    elif args.cmd == "push":
        cmd_push(args)
    elif args.cmd == "abort-stale":
        cmd_abort_stale(args)
    else:
        args.func(args)


if __name__ == "__main__":
    main()