"""Bounded, idempotent FinData exports of raw 1-minute bars (<= 12 years per request, one symbol per request)."""
from __future__ import annotations

import datetime as dt
from pathlib import Path

import pyarrow.parquet as pq

from .api_client import FinDataClient, export_result_dict
from .provenance import append_jsonl, manifest_index

BARS_FIELDS = ["instrument_id", "symbol_as_reported", "venue", "asset_class", "interval", "timestamp_utc", "session_date",
               "open", "high", "low", "close", "volume", "currency", "trade_count", "vwap", "adjustment_type", "source_id",
               "ingested_at", "quality_flags", "revision_id"]


def chunk_ranges(start: dt.date, end: dt.date, max_years: int = 12) -> list[tuple[dt.date, dt.date]]:
    out, a = [], start
    while a <= end:
        b = min(end, dt.date(a.year + max_years - 1, 12, 31))
        out.append((a, b))
        a = dt.date(b.year + 1, 1, 1)
    return out


def validate_bars_parquet(path: Path) -> dict:
    meta = pq.read_metadata(path)
    schema = pq.read_schema(path)
    names = schema.names
    if names != BARS_FIELDS:
        raise ValueError(f"unexpected bars schema in {path.name}: {names}")
    md = {k.decode(): v.decode() for k, v in (schema.metadata or {}).items()}
    if md.get("findata.schema") != "bars" or md.get("findata.timestamp_convention") != "bar_open":
        raise ValueError(f"missing/unexpected FinData schema metadata in {path.name}: {md}")
    return {"rows": meta.num_rows, "schema_version": md.get("findata.schema_version")}


def export_symbol(client: FinDataClient, symbol: str, start: dt.date, end: dt.date, dataset_id: str, cache_dir: Path,
                  manifest: Path, keep: bool = False) -> list[Path]:
    """Export all chunks of one symbol; skip chunks already in the manifest when the file is still present (or keep=False
    and the chunk was already consumed). Returns the list of Parquet paths available on disk."""
    idx = manifest_index(manifest, ["symbol", "start", "end", "dataset_id"])
    paths = []
    for a, b in chunk_ranges(start, end):
        dest = cache_dir / f"{symbol}__{a}__{b}.parquet"
        key = (symbol, str(a), str(b), dataset_id)
        if dest.exists():
            paths.append(dest)
            continue
        if key in idx and idx[key].get("consumed") and not keep:
            # already processed in a previous run; downstream will find the feature rows
            continue
        res = client.export(dataset_id, [symbol], str(a), str(b), dest)
        info = validate_bars_parquet(dest)
        rec = export_result_dict(res)
        rec.update({"symbol": symbol, "rows_in_file": info["rows"], "schema_version": info["schema_version"], "consumed": False})
        append_jsonl(manifest, rec)
        paths.append(dest)
    return paths


def mark_consumed(manifest: Path, symbol: str, dataset_id: str) -> None:
    append_jsonl(manifest, {"symbol": symbol, "dataset_id": dataset_id, "consumed": True, "event": "features_written"})