"""Full vendor catalog: new layouts (index without volume, futures contracts, option chains), no-data responses,
hot/cold lake roots, the complete plan, and the options task through the vendor mock."""
from __future__ import annotations

import io
import zipfile
from datetime import UTC, date, datetime

import httpx
import pytest
from findata_market_firstrate import (
    FRDClient,
    FRDNoData,
    build_full_plan,
    dataset_id_for,
    parse_options_file,
    ticker_from_filename,
)
from findata_market_firstrate.archive import parse_bar_file
from findata_quality import validate_bars
from findata_schema import OPTIONS_SCHEMA, Adjustment, AssetClass, Interval

from tests.conftest import TEST_USERID

NOW = datetime(2026, 11, 1, 12, 0, tzinfo=UTC)


def test_index_without_volume_and_fx_two_fields():
    t, raw = parse_bar_file(b"2026-10-29 04:00:00,1136.43,1137.03,1135.97,1135.97\n2026-10-29 04:01:00,1135.89,1135.89,1134.56,1134.83\n",
                            "AEX", AssetClass.index, Interval.m1, Adjustment.raw, "firstrate", NOW)
    assert raw == 2 and t.num_rows == 2 and t.column("volume").to_pylist() == [0.0, 0.0]
    out, anomalies = validate_bars(t, Interval.m1, asset_class="index")
    assert all("zero_volume" not in f for f in out.column("quality_flags").to_pylist())
    t, _ = parse_bar_file(b"20261026,00:02:00,3.49536933,3.49536933,3.49536933,3.49536933,1\n", "AUDBRL", AssetClass.fx, Interval.m1, Adjustment.raw, "firstrate", NOW)
    assert t.column("timestamp_utc")[0].as_py() == datetime(2026, 10, 26, 4, 2, tzinfo=UTC)  # 00:02 EDT
    out, anomalies = validate_bars(t, Interval.m1, asset_class="fx")
    assert not any(a["code"] in ("outside_session", "non_trading_day") for a in anomalies)  # 24h market: no session checks
    t, _ = parse_bar_file(b"2026-10-30 00:00:00,119.071,119.46,119.04,119.045,208\n", "AAVE", AssetClass.crypto, Interval.m1, Adjustment.raw, "firstrate", NOW, source_tz="UTC")
    assert t.column("timestamp_utc")[0].as_py() == datetime(2026, 10, 30, 0, 0, tzinfo=UTC)


def test_contract_filenames_and_daily_open_interest():
    assert ticker_from_filename("A6_F25_1day.txt", contract=True) == "A6_F25"
    assert ticker_from_filename("ES_H25_1min.txt", contract=True) == "ES_H25"
    assert ticker_from_filename("ES_full_1min_contin_adj_ratio.txt") == "ES"
    t, _ = parse_bar_file(b"2024-09-11,0.6669,0.6679,0.6669,0.6679,10,5\n", "A6_F25", AssetClass.futures, Interval.d1, Adjustment.raw, "firstrate", NOW)
    assert t.num_rows == 1 and t.column("session_date")[0].as_py() == date(2024, 9, 11)


def test_option_chain_parser():
    data = b"2026-10-01,33,2026-10-16,c,8.14,7.3,8.8,0.0,1.102138,0,0,1.0,0.0,1e-05,0.0,0.0\n2026-10-01,34,2026-10-16,p,7.14,6.4,7.9,0.0,1.048606,12,3,-0.4,0.01,1e-05,-0.02,0.0\n"
    t, raw = parse_options_file(data, "AAL", "firstrate", NOW)
    assert raw == 2 and t.num_rows == 2 and t.schema.equals(OPTIONS_SCHEMA)
    r = t.to_pylist()[1]
    assert r["underlying"] == "AAL" and r["call_put"] == "p" and r["strike"] == 34.0 and r["expiry"] == date(2026, 10, 16) and r["open_interest"] == 12.0


def test_no_data_response_is_distinguished_from_errors(tmp_path):
    def handler(req: httpx.Request) -> httpx.Response:
        return httpx.Response(200, text="No update available", headers={"content-type": "text/plain", "content-length": "19"})
    c = FRDClient(TEST_USERID, min_interval_s=0, transport=httpx.MockTransport(handler))
    with pytest.raises(FRDNoData):
        c.download("full_update", {"type": "stock", "period": "week", "timeframe": "1day", "adjustment": "adj_split"}, tmp_path / "x.zip")


def test_dataset_ids_and_full_plan():
    assert dataset_id_for("stock", "1min", "adj_splitdiv") == "market_bars.stock.1m.split_div"
    assert dataset_id_for("etf", "1day", "UNADJUSTED") == "market_bars.etf.1d.raw"
    assert dataset_id_for("futures", "30min", "contin_adj_absolute") == "market_bars.futures.30m.continuous_absolute"
    assert dataset_id_for("futures", "1min", None, contracts=True) == "market_bars.futures_contracts.1m.raw"
    assert dataset_id_for("options", "1day", None) == "options.eod"
    plan = build_full_plan(today=date(2026, 11, 1))
    assert len(plan) == 798
    assert plan[0]["dataset_id"] == "market_bars.stock.1m.raw" and plan[0]["params"]["ticker_range"] == "A"
    assert sum(1 for j in plan if j["kind"] == "options") == 67 + 1
    assert len({j["dedupe_key"] for j in plan}) == len(plan)
    assert all(j["priority"] >= 10 for j in plan) and plan[-1]["kind"] == "options"
    small = build_full_plan(types=["index"], include_options=False, include_delisted=False, today=date(2026, 11, 1))
    assert len(small) == 5


def test_cold_root_placement(tmp_path, monkeypatch):
    from findata_ingest.lake import Lake
    cold = tmp_path / "ssd"
    cold.mkdir()
    with pytest.raises(RuntimeError):
        Lake(tmp_path / "hot", cold)  # marker missing → refuse (volume may be unmounted)
    (cold / ".findata-cold-root").write_text("x")
    lake = Lake(tmp_path / "hot", cold)
    assert lake.root_for("market_bars.stock.1m.raw") == tmp_path / "hot"
    assert lake.root_for("options.eod") == cold and lake.root_for("market_bars.stock.1m.split_div") == cold
    assert lake.raw_dir("firstrate", "x").is_relative_to(cold)
    assert lake.abs("curated/options.eod/base/underlying=A/period=2024q1/p.parquet").is_relative_to(cold)
    assert lake.abs("curated/market_bars.stock.1m.raw/delta/session_date=2026-10-30/p.parquet").is_relative_to(tmp_path / "hot")
    assert lake.rel(cold / "raw" / "a.zip") == "raw/a.zip"


def test_options_task_with_vendor_mock(vendor, lake, log, monkeypatch):
    from findata_core.db import connection, fetch_one
    from findata_ingest import queue
    from findata_ingest.tasks import options as O

    def opt_zip() -> bytes:
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, "w") as zf:
            zf.writestr(zipfile.ZipInfo("AAL_month_option_chain.txt", (2026, 10, 30, 0, 0, 0)), "2026-10-01,33,2026-10-16,c,8.14,7.3,8.8,0.0,1.1,0,0,1.0,0.0,1e-05,0.0,0.0\n2026-10-02,33,2026-10-16,c,8.0,7.2,8.7,0.0,1.1,1,2,1.0,0.0,1e-05,0.0,0.0\n")
            zf.writestr(zipfile.ZipInfo("SPY_month_option_chain.txt", (2026, 10, 30, 0, 0, 0)), "2026-10-01,500,2026-10-17,p,1.5,1.4,1.6,0.2,0.21,100,50,-0.3,0.01,0.1,-0.05,0.0\n")
        return buf.getvalue()
    vendor.archive_override = opt_zip()
    jid = queue.enqueue("options", "options.eod", {"period": "month"})
    job = queue.claim_next(["options"])
    r = O.run(job, lake, log)
    queue.finish(job, True, r)
    assert r["underlyings"] == 2 and r["rows"] == 3 and r["files"] == 2
    with connection() as conn:
        rows = conn.execute("SELECT symbol_min, date_min, date_max, row_count FROM lake_files WHERE dataset_id='options.eod' AND version_removed IS NULL ORDER BY 1").fetchall()
    assert [(x["symbol_min"], x["row_count"]) for x in rows] == [("AAL", 2), ("SPY", 1)]
    assert fetch_one("SELECT status FROM datasets WHERE id='options.eod'")["status"] == "active"
    # the month archive re-run supersedes the `latest` partition (no duplicates)
    job = queue.claim_next(["options"]) or (queue.enqueue("options", "options.eod", {"period": "month"}) and queue.claim_next(["options"]))
    r2 = O.run(job, lake, log)
    queue.finish(job, True, r2)
    with connection() as conn:
        live = conn.execute("SELECT count(*) AS n, sum(row_count) AS rows FROM lake_files WHERE dataset_id='options.eod' AND version_removed IS NULL").fetchone()
    assert live["n"] == 2 and live["rows"] == 3
    vendor.archive_override = None


def test_deflate64_fallback_uses_7z(tmp_path, monkeypatch):
    import zipfile as zf_mod

    from findata_market_firstrate import archive as A
    z = tmp_path / "a.zip"
    with zf_mod.ZipFile(z, "w") as zf:
        zf.writestr("X_option_chain.txt", "2026-10-01,33,2026-10-16,c,1,1,1,0,0,0,0,0,0,0,0,0\n")
    # simulate Python's zipfile refusing the compression method
    def boom(path):
        raise NotImplementedError("That compression method is not supported")
        yield  # pragma: no cover
    monkeypatch.setattr(A, "_iter_zip_members", boom)
    if not A._find_7z():
        pytest.skip("7z not installed")
    out = list(A.iter_archive_files(z))
    assert out == [("X_option_chain.txt", b"2026-10-01,33,2026-10-16,c,1,1,1,0,0,0,0,0,0,0,0,0\n")]
    assert not list(tmp_path.glob("unz-*"))  # temp dir cleaned


def test_extract_scope_from_sql():
    from datetime import date

    from findata_ingest.reader import extract_scope
    s = extract_scope("SELECT * FROM market_bars_stock_1m_raw WHERE symbol='AAPL' AND year=2024 AND session_date >= DATE '2024-03-01' AND session_date <= '2024-03-31'")
    assert s == {"symbols": ["AAPL"], "start": date(2024, 3, 1), "end": date(2024, 3, 31)}
    s = extract_scope("select count(*) from options_eod where underlying in ('SPY', 'QQQ') and trade_date between DATE '2024-03-01' and DATE '2024-03-31'")
    assert s["symbols"] == ["QQQ", "SPY"] and s["start"] == date(2024, 3, 1) and s["end"] == date(2024, 3, 31)
    assert extract_scope("select count(*) from market_bars_stock_1d_raw where session_date = DATE '2024-03-15'")["symbols"] == []
    assert extract_scope("select * from x where year in (2019, 2020)")["start"] == date(2019, 1, 1)


def test_scoped_sql_uses_catalog_files(vendor, lake, log):
    from findata_core.db import fetch_one
    from findata_ingest.reader import dataset_views, run_sql
    if fetch_one("SELECT count(*) AS n FROM lake_files WHERE dataset_id='market_bars.stock.1m.raw' AND version_removed IS NULL")["n"] == 0:
        pytest.skip("no 1m data in this test database")
    views = dataset_views()
    t, meta = run_sql("SELECT count(*) AS n FROM market_bars_stock_1m_raw WHERE symbol='AAPL' AND session_date=DATE '2026-10-30'", 10, 30, views)
    assert t.to_pylist()[0]["n"] == 900 and meta["views"] == ["market_bars_stock_1m_raw"]
    t, meta = run_sql("SELECT count(*) AS n FROM market_bars_stock_1m_raw WHERE session_date=DATE '2026-10-30'", 10, 30, views)
    assert t.to_pylist()[0]["n"] >= 900 and meta["views"] == ["market_bars_stock_1m_raw (full scan)"]