"""CSV / Excel bulk-plot import (INDL-42a, INDL-42g).

Everything here runs **server-side** and treats the uploaded file and the mapped
rows as fully untrusted (OWASP API3 / A07): the same field rules as the single
Add-Plot form are re-applied per row, protected columns can never be mass-assigned,
spreadsheet cells are parsed with a hardened reader (row cap + decompressed-size
cap, no external-entity resolution), and any value written back into the
downloadable error report is formula-injection-safe (A03 / ASVS V5.3 / SEC-05).
"""
from __future__ import annotations

import csv
import io
import logging
import zipfile
from typing import Optional
from uuid import UUID

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from src.apps.plots.models.plot import Plot
from src.apps.plots.models.plot_type import PlotType
from src.apps.plots.services.plot_status_service import PlotStatusService
from src.apps.sections.models.section import Section
from src.core.exceptions import ValidationError

logger = logging.getLogger(__name__)

# ── Hard limits (SEC-04 / API4 / file-level validation rules) ────────────────
MAX_ROWS = 10_000
MAX_DECOMPRESSED_BYTES = 50 * 1024 * 1024  # 50 MB expanded XLSX ceiling
_FORMULA_TRIGGERS = ("=", "+", "-", "@", "\t", "\r")

# ── Heuristic column mapping dictionary (INDL-42g) ───────────────────────────
# target field -> the normalised synonyms that should map onto it. Matching is
# case-insensitive and ignores spaces / underscores / punctuation.
COLUMN_SYNONYMS: dict[str, list[str]] = {
    "plot_ref": ["plotid", "plotref", "plotreference", "ref", "plotno",
                 "plotnumber", "grave", "graveno", "gravenumber", "id"],
    "section": ["section", "sect", "sec", "block", "area", "garden"],
    "row": ["row", "rowno", "rownumber", "line"],
    "plot_type": ["plottype", "type", "gravetype", "kind", "category"],
    "status": ["status", "state", "availability", "condition"],
    "reserved_by": ["owner", "deedholder", "holder", "reservedby", "family",
                    "purchaser", "buyer", "ownername"],
    "price": ["price", "cost", "amount", "value", "fee"],
    "latitude": ["lat", "latitude", "gpslat", "gpslatitude", "y"],
    "longitude": ["lng", "lon", "long", "longitude", "gpslng", "gpslongitude", "x"],
    "label_manual": ["label", "displaylabel", "displayname", "name"],
    "public_description": ["description", "publicdescription", "listing",
                           "blurb", "summary"],
    "notes": ["notes", "note", "internalnote", "comment", "comments", "remarks"],
}

# Fields the importer is allowed to write (defence-in-depth alongside the ORM
# whitelist in _row_to_plot — protected cols can never be reached from a file).
IMPORTABLE_FIELDS = set(COLUMN_SYNONYMS.keys())
REQUIRED_TARGETS = ("plot_ref", "section")


def _normalise(header: str) -> str:
    return "".join(c for c in header.lower() if c.isalnum())


def suggest_mapping(columns: list[str]) -> dict[str, Optional[str]]:
    """Best-effort source-column -> target-field mapping (Step 2, AC-14).

    Deterministic, no external call, works with ANTHROPIC_API_KEY unset. Each
    target is assigned at most once (first/best source column wins).
    """
    taken: set[str] = set()
    result: dict[str, Optional[str]] = {}
    for col in columns:
        norm = _normalise(col)
        match: Optional[str] = None
        # 1) exact synonym hit
        for target, syns in COLUMN_SYNONYMS.items():
            if target in taken:
                continue
            if norm in syns:
                match = target
                break
        # 2) substring fallback (e.g. "plot ref no" contains "plotref")
        if match is None:
            for target, syns in COLUMN_SYNONYMS.items():
                if target in taken:
                    continue
                if any(s in norm or norm in s for s in syns if len(s) >= 3):
                    match = target
                    break
        if match is not None:
            taken.add(match)
        result[col] = match
    return result


def escape_formula(value: object) -> str:
    """Prefix a leading formula trigger with a single quote (A03 / SEC-05)."""
    s = "" if value is None else str(value)
    if s and s[0] in _FORMULA_TRIGGERS:
        return "'" + s
    return s


def _guard_xlsx_size(content: bytes) -> None:
    """Reject zip-bomb XLSX before openpyxl expands it (SEC-04)."""
    try:
        with zipfile.ZipFile(io.BytesIO(content)) as zf:
            total = sum(info.file_size for info in zf.infolist())
    except zipfile.BadZipFile as exc:
        raise ValidationError("The uploaded Excel file is corrupt or unreadable") from exc
    if total > MAX_DECOMPRESSED_BYTES:
        raise ValidationError(
            "The uploaded file is too large once expanded. Split it into smaller imports."
        )


def parse_file(filename: str, content: bytes) -> tuple[list[str], list[dict]]:
    """Parse a CSV/XLSX byte buffer into (columns, rows) — hardened.

    Enforces the 10,000-row cap immediately; never returns more than MAX_ROWS.
    """
    name = (filename or "").lower()
    if name.endswith(".csv"):
        columns, rows = _parse_csv(content)
    elif name.endswith((".xlsx", ".xls")):
        columns, rows = _parse_xlsx(content)
    else:
        raise ValidationError("Unsupported file type — upload CSV or Excel")

    if len(rows) > MAX_ROWS:
        raise ValidationError(f"File exceeds the {MAX_ROWS:,} row limit")
    return columns, rows


def _parse_csv(content: bytes) -> tuple[list[str], list[dict]]:
    text = content.decode("utf-8-sig", errors="replace")
    reader = csv.reader(io.StringIO(text))
    try:
        header = next(reader)
    except StopIteration:
        raise ValidationError("The uploaded file is empty")
    columns = [h.strip() for h in header]
    rows: list[dict] = []
    for raw in reader:
        if not any((c or "").strip() for c in raw):
            continue  # skip blank lines
        rows.append({columns[i]: raw[i] if i < len(raw) else "" for i in range(len(columns))})
        if len(rows) > MAX_ROWS:
            break
    return columns, rows


def _parse_xlsx(content: bytes) -> tuple[list[str], list[dict]]:
    _guard_xlsx_size(content)
    try:
        import openpyxl
    except ImportError as exc:  # pragma: no cover
        raise ValidationError("Excel import is not available on this server") from exc

    # read_only streams rows; data_only returns cached values, not formulas —
    # so an "=cmd" cell arrives as its stored text, never evaluated.
    wb = openpyxl.load_workbook(
        io.BytesIO(content), read_only=True, data_only=True
    )
    ws = wb.active
    rows_iter = ws.iter_rows(values_only=True)
    try:
        header = next(rows_iter)
    except StopIteration:
        wb.close()
        raise ValidationError("The uploaded file is empty")
    columns = [str(h).strip() if h is not None else "" for h in header]
    rows: list[dict] = []
    for raw in rows_iter:
        if raw is None or not any(c is not None and str(c).strip() for c in raw):
            continue
        record = {}
        for i in range(len(columns)):
            val = raw[i] if i < len(raw) else None
            record[columns[i]] = "" if val is None else str(val)
        rows.append(record)
        if len(rows) > MAX_ROWS:
            break
    wb.close()
    return columns, rows


class PlotImportService:
    """Validate mapped rows against the single-plot rules, then bulk-create."""

    def __init__(self, db: AsyncSession):
        self.db = db

    async def _lookup_context(self, tenant_id: UUID) -> dict:
        # Section code -> id (case-insensitive)
        secs = (await self.db.execute(
            select(Section).where(Section.tenant_id == tenant_id)
        )).scalars().all()
        section_by_code = {(s.code or "").strip().lower(): s.id for s in secs}
        section_by_name = {(s.name or "").strip().lower(): s.id for s in secs}

        types = (await self.db.execute(
            select(PlotType).where(PlotType.tenant_id == tenant_id)
        )).scalars().all()
        type_by_name = {(t.name or "").strip().lower(): t.id for t in types}

        statuses = await PlotStatusService(self.db).list_or_seed(tenant_id)
        status_by_name = {
            (s.name or "").strip().lower(): s for s in statuses
        }
        default_status = next(
            (s for s in statuses if s.status_key == "vacant"),
            next((s for s in statuses if s.is_default), None),
        )
        # Existing refs, to flag DB duplicates (AC-15).
        refs = (await self.db.execute(
            select(Plot.plot_ref).where(Plot.tenant_id == tenant_id)
        )).scalars().all()
        existing_refs = {r.lower() for r in refs}
        return {
            "section_by_code": section_by_code,
            "section_by_name": section_by_name,
            "type_by_name": type_by_name,
            "status_by_name": status_by_name,
            "default_status": default_status,
            "existing_refs": existing_refs,
        }

    @staticmethod
    def _map_row(row: dict, mapping: dict[str, str]) -> dict:
        """Project a raw source row onto whitelisted target fields only."""
        mapped: dict[str, str] = {}
        for source, target in mapping.items():
            if not target or target == "ignore" or target not in IMPORTABLE_FIELDS:
                continue
            value = row.get(source, "")
            mapped[target] = value.strip() if isinstance(value, str) else value
        return mapped

    def _validate_row(
        self, mapped: dict, ctx: dict, seen_refs: set[str]
    ) -> tuple[Optional[dict], list[str], list[str]]:
        """Return (plot_kwargs | None, errors, warnings) for one mapped row."""
        errors: list[str] = []
        warnings: list[str] = []

        plot_ref = (mapped.get("plot_ref") or "").strip()
        if not plot_ref:
            errors.append("Plot ID is required")
        elif len(plot_ref) > 50:
            errors.append("Plot ID exceeds 50 characters")
        else:
            key = plot_ref.lower()
            if key in seen_refs:
                errors.append(f'Duplicate plot ID "{plot_ref}" appears earlier in this file')
            elif key in ctx["existing_refs"]:
                errors.append(f'Duplicate plot ID "{plot_ref}" already exists')

        # Section (required) — resolved to an id
        section_id = None
        section_raw = (mapped.get("section") or "").strip()
        if not section_raw:
            errors.append("Section is required")
        else:
            section_id = (
                ctx["section_by_code"].get(section_raw.lower())
                or ctx["section_by_name"].get(section_raw.lower())
            )
            if section_id is None:
                errors.append(f'Unknown section "{section_raw}"')

        # Plot type (optional) — resolved to an id
        plot_type_id = None
        type_raw = (mapped.get("plot_type") or "").strip()
        if type_raw:
            plot_type_id = ctx["type_by_name"].get(type_raw.lower())
            if plot_type_id is None:
                warnings.append(f'Unknown plot type "{type_raw}" — left unset')

        # Status (optional) — unknown values default, not error (AC-15)
        status_raw = (mapped.get("status") or "").strip()
        status_key = None
        if status_raw:
            match = ctx["status_by_name"].get(status_raw.lower())
            if match is not None and match.status_key:
                status_key = match.status_key
            else:
                default_name = ctx["default_status"].name if ctx["default_status"] else "default"
                warnings.append(
                    f'Unrecognized status "{status_raw}" — defaulted to {default_name}'
                )
        if status_key is None:
            status_key = (
                ctx["default_status"].status_key if ctx["default_status"] else "vacant"
            ) or "vacant"

        # GPS
        lat = self._parse_decimal(mapped.get("latitude"), "Latitude", -90, 90, errors)
        lng = self._parse_decimal(mapped.get("longitude"), "Longitude", -180, 180, errors)
        price = self._parse_decimal(mapped.get("price"), "Price", 0, 9_999_999.99, errors)

        if errors:
            return None, errors, warnings

        # Cap every string field to its column width so one oversized cell can never
        # blow up the single batch flush and abort the whole import.
        kwargs = {
            "plot_ref": plot_ref,
            "section_id": section_id,
            "plot_type_id": plot_type_id,
            "status": status_key,
            "latitude": lat,
            "longitude": lng,
            "price_override": price,
            "reserved_by": (mapped.get("reserved_by") or "")[:255] or None,
            "label_manual": (mapped.get("label_manual") or "")[:50] or None,
            "public_description": (mapped.get("public_description") or "")[:180] or None,
            "notes": (mapped.get("notes") or None) or None,
        }
        return kwargs, errors, warnings

    @staticmethod
    def _parse_decimal(raw, label, lo, hi, errors) -> Optional[float]:
        if raw is None or (isinstance(raw, str) and not raw.strip()):
            return None
        try:
            val = float(str(raw).replace(",", "").replace("$", "").strip())
        except ValueError:
            errors.append(f"{label} is not a valid number")
            return None
        if not (lo <= val <= hi):
            errors.append(f"{label} must be between {lo} and {hi}")
            return None
        return val

    async def import_rows(
        self, tenant_id: UUID, rows: list[dict], mapping: dict[str, str]
    ) -> dict:
        """Validate + create every row. Row failures are isolated, never fatal.

        Returns {created, skipped, failed:[{row, plot_ref, reason}], warnings:[...]}.
        """
        if len(rows) > MAX_ROWS:
            raise ValidationError(f"File exceeds the {MAX_ROWS:,} row limit")

        # Required mapping present? (AC-14 server-side re-check — A07)
        targets = {t for t in mapping.values() if t and t != "ignore"}
        missing = [t for t in REQUIRED_TARGETS if t not in targets]
        if missing:
            raise ValidationError(
                "Map the required fields before continuing: "
                + ", ".join("Plot ID" if m == "plot_ref" else "Section" for m in missing)
            )

        ctx = await self._lookup_context(tenant_id)
        seen_refs: set[str] = set()
        created = 0
        skipped = 0
        failed: list[dict] = []
        warnings: list[dict] = []

        for idx, raw_row in enumerate(rows, start=1):
            mapped = self._map_row(raw_row, mapping)
            kwargs, errors, row_warnings = self._validate_row(mapped, ctx, seen_refs)
            if errors:
                failed.append({
                    "row": idx,
                    "plot_ref": escape_formula(mapped.get("plot_ref")),
                    "reason": "; ".join(errors),
                })
                skipped += 1
                continue

            seen_refs.add(kwargs["plot_ref"].lower())
            if row_warnings:
                warnings.append({"row": idx, "plot_ref": kwargs["plot_ref"],
                                 "messages": row_warnings})

            # Imported plots never carry drawn geometry — point marker only.
            plot = Plot(tenant_id=tenant_id, is_veteran_section=False, **kwargs)
            self.db.add(plot)
            created += 1

        await self.db.flush()
        logger.info(
            "plots.import actor_tenant=%s created=%s skipped=%s total=%s",
            tenant_id, created, skipped, len(rows),
        )
        return {
            "created": created,
            "skipped": skipped,
            "failed": failed,
            "warnings": warnings,
            "total": len(rows),
        }

    @staticmethod
    def build_error_report_csv(failed: list[dict]) -> str:
        """Formula-safe CSV of failed rows (A03 / SEC-05)."""
        buf = io.StringIO()
        writer = csv.writer(buf)
        writer.writerow(["Row", "Plot ID", "Reason"])
        for f in failed:
            writer.writerow([
                f.get("row", ""),
                escape_formula(f.get("plot_ref", "")),
                escape_formula(f.get("reason", "")),
            ])
        return buf.getvalue()