"""Scan the Hive-partitioned lake without re-reading every historical flight."""

from __future__ import annotations

import os
from pathlib import Path


def flight_window() -> int:
    return max(1, int(os.environ.get("METRIC_FLIGHT_WINDOW", "5")))


def recent_flights(data_dir: Path, dataset: str, limit: int | None = None) -> list[Path]:
    """Most recently modified flight= partitions for a dataset."""
    limit = limit or flight_window()
    base = data_dir / f"dataset={dataset}"
    if not base.is_dir():
        return []
    flights = [p for p in base.iterdir() if p.is_dir() and p.name.startswith("flight=")]
    flights.sort(key=lambda p: p.stat().st_mtime, reverse=True)
    return flights[:limit]


def parquet_reader(data_dir: Path, dataset: str, *, limit: int | None = None) -> str:
    """DuckDB read_parquet() source limited to recent flights."""
    flights = recent_flights(data_dir, dataset, limit)
    if not flights:
        path = data_dir / f"dataset={dataset}" / "**" / "*.parquet"
        return f"'{path}', hive_partitioning=true, union_by_name=true"
    if len(flights) == 1:
        return f"'{flights[0]}/**/*.parquet', hive_partitioning=true, union_by_name=true"
    inner = ", ".join(f"'{f}/**/*.parquet'" for f in flights)
    return f"[{inner}], hive_partitioning=true, union_by_name=true"


def iter_parquet_files(data_dir: Path, *, flight_limit: int | None = None) -> list[Path]:
    """Parquet paths under recent flights only — avoids full-lake rglob."""
    limit = flight_limit or flight_window()
    out: list[Path] = []
    for ds in ("telemetry", "detections", "state"):
        for flight in recent_flights(data_dir, ds, limit):
            out.extend(flight.rglob("*.parquet"))
    return out