"""HDFS → Kafka export and full hadoop-lake pipeline orchestration."""

from __future__ import annotations

import csv
import io
import json
import os
import time
from typing import Any

from aiokafka import AIOKafkaProducer
from webhdfs_util import open_bytes

KAFKA_BOOTSTRAP = os.getenv("KAFKA_BOOTSTRAP", "10.0.21.36:9092")
DEFAULT_TRINO_TABLE = os.getenv("HADOOP_EXPORT_TABLE", "iceberg.hadoop.historical_sales_hdfs")
DEFAULT_HDFS_FILE = os.getenv("HADOOP_EXPORT_HDFS", "/data/historical/sales_orders/year=2020/part-0.csv")

_last_hdfs_export: dict[str, Any] = {"ts": 0.0, "rows": 0, "topic": None}


def hdfs_export_snapshot() -> dict[str, Any]:
    age = time.time() - float(_last_hdfs_export.get("ts") or 0)
    return {
        **(_last_hdfs_export or {}),
        "recent": age < 120,
        "age_s": round(age, 1) if _last_hdfs_export.get("ts") else None,
    }


def _read_hdfs_csv(path: str, limit: int) -> tuple[list[str], list[list[str]]]:
    raw = open_bytes(path).decode("utf-8", errors="replace")
    rows = list(csv.reader(io.StringIO(raw)))
    if not rows:
        return [], []
    header = [c.strip() for c in rows[0]]
    data = [[cell.strip() for cell in row] for row in rows[1: 1 + limit]]
    return header, data


def _read_trino_table(table: str, limit: int) -> tuple[list[str], list[list[Any]]]:
    from sql_console import _run_trino

    result = _run_trino(f"SELECT * FROM {table} LIMIT {limit}", limit)
    if not result.get("ok"):
        raise RuntimeError(result.get("error") or "Trino query failed")
    return list(result.get("columns") or []), list(result.get("rows") or [])


async def export_hdfs_to_kafka(
    path: str | None = None,
    topic: str = "hdfs.historical.sales",
    limit: int = 2000,
    source: str = "trino",
    table: str | None = None,
    feed: Any = None,
) -> dict[str, Any]:
    global _last_hdfs_export
    header: list[str] = []
    data: list[list[Any]] = []
    src_label = source

    try:
        if source == "trino":
            tbl = table or DEFAULT_TRINO_TABLE
            header, data = _read_trino_table(tbl, limit)
            src_label = tbl
        else:
            p = path or DEFAULT_HDFS_FILE
            header, data = _read_hdfs_csv(p, limit)
            src_label = p
    except Exception as exc:
        if source == "trino" and path:
            try:
                header, data = _read_hdfs_csv(path, limit)
                src_label = path
            except Exception:
                return {"ok": False, "error": str(exc)[:400]}
        else:
            return {"ok": False, "error": str(exc)[:400]}

    if not data:
        return {"ok": False, "error": "No data rows to export"}

    producer = AIOKafkaProducer(
        bootstrap_servers=KAFKA_BOOTSTRAP,
        value_serializer=lambda v: json.dumps(v, default=str).encode("utf-8"),
    )
    await producer.start()
    rows_sent = 0
    try:
        for i, row in enumerate(data):
            payload = {
                header[j] if j < len(header) else f"col_{j}": row[j] if j < len(row) else None
                for j in range(max(len(header), len(row)))
            }
            payload["_source"] = "hadoop"
            payload["_origin"] = src_label
            payload["_row"] = i + 1
            await producer.send_and_wait(topic, payload)
            rows_sent += 1
    finally:
        await producer.stop()

    _last_hdfs_export = {"ts": time.time(), "rows": rows_sent, "topic": topic, "source": src_label}
    if feed:
        feed("hadoop-ranger", f"[hadoop→kafka] {rows_sent} rows from {src_label} → {topic}", "info")
    return {"ok": True, "rows_sent": rows_sent, "topic": topic, "source": src_label, "columns": header}