import io
import json
import uuid
from pathlib import Path

import pandas as pd
from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, UploadFile, status
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.core import jobs
from app.core.database import get_db
from app.core.permissions import deny_vendor, require_permission, require_portal_action, require_portal_item, resolve_portal_grants
from app.core.security import AuthContext, get_auth_context
from app.crud.base import CRUDBase
from app.crud import record_sql
from app.crud import entity_storage, field_indexes
from app.crud.access_policy import access_clause, require_record_access
from app.crud.audit import write_audit_log
from app.crud.change_log import changed_keys, fields_diff, values_diff
from app.crud.dataset_query import attach_user_names, filter_records, parse_filters_param, prepare_filters
from app.crud import record_import
from app.crud.formula import compute_formula_fields
from app.crud.scope import check_value_in_scope, is_restricted_for
from app.crud import field_constraints
from app.crud.validation import validate_field_values
from app.models.master import Master, MasterRecord
from app.routers.uploads import UPLOAD_ROOT
from app.schemas.master import (
    BulkImportResult,
    BulkImportRowError,
    ImportColumnsResult,
    MasterCreate,
    MasterRead,
    MasterRecordCreate,
    MasterRecordRead,
    MasterRecordUpdate,
    MasterUpdate,
    RecordImportStarted,
    RecordImportStatus,
    StorageChange,
    StorageInfo,
    StoragePreview,
)

router = APIRouter(prefix="/masters", tags=["masters"])
crud = CRUDBase[Master, MasterCreate, MasterUpdate](Master)
records_crud = CRUDBase[MasterRecord, MasterRecordCreate, MasterRecordUpdate](MasterRecord)


async def _count_fields(db: AsyncSession, tenant_id: uuid.UUID, master_id: uuid.UUID) -> dict:
    counts = await record_sql.capped_source_counts(db, "master", tenant_id, [master_id])
    return _as_fields(counts.get(master_id))


def _as_fields(count: tuple[int, bool] | None) -> dict:
    total, capped = count or (0, False)
    return {"record_count": total, "record_count_capped": capped}


@router.get("", response_model=list[MasterRead])
async def list_masters(
    skip: int = 0,
    limit: int = 100,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "view")),
) -> list[MasterRead]:
    masters = await crud.list(db, ctx.tenant_id, skip=skip, limit=limit)
    if ctx.vendor_id is not None:
        grants = await resolve_portal_grants(db, ctx.tenant_id, ctx.user_id)
        allowed = grants.get("masters", set())
        masters = [m for m in masters if str(m.id).lower() in allowed]
    counts = await record_sql.capped_source_counts(db, "master", ctx.tenant_id, [m.id for m in masters])
    tables = await entity_storage.tables_by_master(db, ctx.tenant_id)
    return [
        MasterRead.model_validate(m).model_copy(update={**_as_fields(counts.get(m.id)), "storage_tier": _tier(tables.get(str(m.id)))})
        for m in masters
    ]


@router.get("/{master_id}", response_model=MasterRead)
async def get_master(
    master_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "view")),
    __: AuthContext = Depends(require_portal_item("masters", "master_id")),
) -> MasterRead:
    master = await crud.get(db, ctx.tenant_id, master_id)
    count = await _count_fields(db, ctx.tenant_id, master_id)
    tier = _tier(await entity_storage.entity_table(db, ctx.tenant_id, master_id))
    return MasterRead.model_validate(master).model_copy(update={**count, "storage_tier": tier})


@router.post("", response_model=MasterRead, status_code=status.HTTP_201_CREATED)
async def create_master(
    payload: MasterCreate,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "create")),
    __: AuthContext = Depends(deny_vendor()),
) -> MasterRead:
    # storage_tier isn't a Master column — left unset so create() skips it.
    master = await crud.create(db, ctx.tenant_id, MasterCreate.model_validate(payload.model_dump(exclude={"storage_tier"})), created_by=ctx.user_id)
    if payload.storage_tier == "table":
        # The table is created in this transaction: if it can't be, the
        # datasource isn't created either.
        try:
            await entity_storage.promote(db, ctx.tenant_id, ctx.user_id, entity_storage.master_source(master))
        except entity_storage.StorageError as exc:
            raise _storage_http(exc) from exc
    await field_indexes.sync_field_indexes(db, "master", master.id, master.field_schema)
    await write_audit_log(
        db, ctx.tenant_id, ctx.user_id, "master.create", "master", master.id, {"name": master.name, "storage_tier": payload.storage_tier},
    )
    return MasterRead.model_validate(master).model_copy(update={"storage_tier": payload.storage_tier})


@router.patch("/{master_id}", response_model=MasterRead)
async def update_master(
    master_id: uuid.UUID,
    payload: MasterUpdate,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "edit")),
    __: AuthContext = Depends(deny_vendor()),
) -> MasterRead:
    current = await crud.get(db, ctx.tenant_id, master_id)
    before = {"name": current.name, "field_schema": list(current.field_schema), "scope_field": current.scope_field}
    master = await crud.update(db, ctx.tenant_id, master_id, payload)
    # A table datasource's table follows its fields — or the edit is refused
    # (and rolled back) when existing records don't fit the change.
    try:
        await entity_storage.migrate(db, ctx.tenant_id, ctx.user_id, entity_storage.master_source(master))
    except entity_storage.StorageError as exc:
        raise _storage_http(exc) from exc
    await field_indexes.sync_field_indexes(db, "master", master.id, master.field_schema)
    after = {"name": master.name, "field_schema": master.field_schema, "scope_field": master.scope_field}
    await write_audit_log(
        db, ctx.tenant_id, ctx.user_id, "master.update", "master", master.id,
        {"name": master.name, "changed": changed_keys(before, after, ["name", "scope_field"]), **fields_diff(before["field_schema"], after["field_schema"])},
    )
    count = await _count_fields(db, ctx.tenant_id, master_id)
    tier = _tier(await entity_storage.entity_table(db, ctx.tenant_id, master_id))
    return MasterRead.model_validate(master).model_copy(update={**count, "storage_tier": tier})


async def _published_forms_referencing_master(db: AsyncSession, tenant_id: uuid.UUID, master_id: uuid.UUID) -> int:
    """FR-MST-007: block deleting a master a published form's dropdown still
    points at. field_definitions is a JSONB array of field objects; walk it
    with jsonb_array_elements rather than a plain @> containment check,
    since dropdown fields are nested inside that array, not top-level keys."""
    stmt = text(
        """
        SELECT count(DISTINCT f.form_id)
        FROM forms f, jsonb_array_elements(f.field_definitions) elem
        WHERE f.tenant_id = :tenant_id
          AND f.status = 'published'
          AND elem->>'master_id' = :master_id
        """
    )
    result = await db.execute(stmt, {"tenant_id": str(tenant_id), "master_id": str(master_id)})
    return result.scalar_one()


async def _masters_referencing_master(db: AsyncSession, tenant_id: uuid.UUID, master_id: uuid.UUID) -> list[str]:
    """Blocks deleting a Master that another Master's schema links to via a
    "reference" field (e.g. deleting Country while State.country still
    points at it). Same jsonb_array_elements pattern as the forms check
    below, since reference fields live inside the field_schema array, not
    as a top-level column.

    Note: this only protects the *schema* relationship. It does not check
    whether any existing record's field_values actually points at a
    specific record being deleted from master_records — record-level
    referential integrity (e.g. blocking deletion of a single Country
    record while States still reference that record) isn't enforced here.
    """
    stmt = text(
        """
        SELECT DISTINCT m.name
        FROM masters m, jsonb_array_elements(m.field_schema) elem
        WHERE m.tenant_id = :tenant_id
          AND m.id::text != :master_id
          AND elem->>'reference_master_id' = :master_id
        """
    )
    result = await db.execute(stmt, {"tenant_id": str(tenant_id), "master_id": str(master_id)})
    return [row[0] for row in result.all()]


@router.delete("/{master_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_master(
    master_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "delete")),
    __: AuthContext = Depends(deny_vendor()),
) -> None:
    await crud.get(db, ctx.tenant_id, master_id)
    referencing = await _published_forms_referencing_master(db, ctx.tenant_id, master_id)
    if referencing:
        raise HTTPException(
            status_code=409,
            detail=f"Master is referenced by {referencing} published form(s); cannot delete (FR-MST-007).",
        )
    referencing_masters = await _masters_referencing_master(db, ctx.tenant_id, master_id)
    if referencing_masters:
        raise HTTPException(
            status_code=409,
            detail=f"Master is linked from other datasource(s): {', '.join(referencing_masters)}; cannot delete.",
        )
    deleted = await crud.get(db, ctx.tenant_id, master_id)
    name = deleted.name
    if await entity_storage.entity_table(db, ctx.tenant_id, master_id):
        try:
            await entity_storage.demote(db, ctx.tenant_id, ctx.user_id, master_id, reason="Datasource deleted")
        except entity_storage.StorageError as exc:
            raise _storage_http(exc) from exc
    await crud.delete(db, ctx.tenant_id, master_id)
    await field_indexes.sync_field_indexes(db, "master", master_id, [])
    await write_audit_log(db, ctx.tenant_id, ctx.user_id, "master.delete", "master", master_id, {"name": name})


# ---- Storage tier (app/crud/entity_storage.py) ------------------------------------


def _tier(row) -> str:
    return "flexible" if row is None else "building" if row.state == "building" else "table"


def _storage_http(exc: entity_storage.StorageError) -> HTTPException:
    detail: dict = {"code": "storage", "message": exc.message}
    if exc.issues:
        detail["issues"] = exc.issues
    return HTTPException(status_code=exc.status, detail=detail)


@router.get("/{master_id}/storage", response_model=StorageInfo)
async def get_storage(
    master_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "view")),
    __: AuthContext = Depends(deny_vendor()),
) -> dict:
    master = await crud.get(db, ctx.tenant_id, master_id)
    return await entity_storage.describe(db, ctx.tenant_id, entity_storage.master_source(master))


@router.get("/{master_id}/storage/preview", response_model=StoragePreview)
async def preview_storage(
    master_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "view")),
    __: AuthContext = Depends(deny_vendor()),
) -> dict:
    master = await crud.get(db, ctx.tenant_id, master_id)
    return await entity_storage.preview(db, ctx.tenant_id, entity_storage.master_source(master))


@router.post("/{master_id}/storage", response_model=StorageInfo)
async def change_storage(
    master_id: uuid.UUID,
    payload: StorageChange,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "edit")),
    __: AuthContext = Depends(deny_vendor()),
) -> dict:
    master = await crud.get(db, ctx.tenant_id, master_id)
    try:
        if payload.tier == "table":
            await entity_storage.promote(db, ctx.tenant_id, ctx.user_id, entity_storage.master_source(master))
        else:
            await entity_storage.demote(db, ctx.tenant_id, ctx.user_id, master_id)
    except entity_storage.StorageError as exc:
        raise _storage_http(exc) from exc
    await write_audit_log(db, ctx.tenant_id, ctx.user_id, "master.storage", "master", master_id, {"name": master.name, "tier": payload.tier})
    return await entity_storage.describe(db, ctx.tenant_id, entity_storage.master_source(master))


# ---- Master records -------------------------------------------------------------

def _record_matches_search(field_values: dict, q: str) -> bool:
    """Case-insensitive substring match across every value in a record's
    field_values — OR semantics across all fields (a Data Grid's search box
    shouldn't require picking which column to search). Mirrors
    app/crud/dataset_query.py's _eval_operator's own "contains" semantics
    (str(expected).lower() in str(actual).lower()) applied per-field there;
    here it's applied across every key instead of one pinned field name.
    Matches raw stored values only — a "reference" field's resolved display
    text isn't searched, same pre-existing limitation _eval_operator's own
    "contains" filter already has."""
    needle = q.lower()
    return any(needle in str(v).lower() for v in field_values.values() if v is not None)


_RECORD_ORDER = (MasterRecord.created_at, MasterRecord.id)


def _source(master: Master) -> record_sql.RecordSource:
    return record_sql.RecordSource("master", master.id, master.field_schema, master=master)


class _RecordQuery:
    """This caller's visible records of one Master — row scope, search box
    (q), parent-record filter and a Data List block's own filters — as one
    SQL WHERE (app/crud/record_sql.py). The one code path both
    list_master_records and count_master_records use, so list and count can
    never disagree about which rows exist.

    filter_field/filter_value — a Page's Data List/Grid block scoping itself
    to one parent record's incoming ?recordId= (PageBlockDef.record_filter_field,
    see app/schemas/navigation.py).

    When every part is exact, the page and count come straight from SQL.
    Otherwise (a search or filter SQL can only narrow) the narrowed rows are
    finished in Python with the original semantics and paged there."""

    @classmethod
    async def build(
        cls,
        db: AsyncSession,
        ctx: AuthContext,
        master: Master,
        q: str | None,
        filter_field: str | None,
        filter_value: str | None,
        filters: str | None,
    ) -> "_RecordQuery":
        try:
            parsed = parse_filters_param(filters)
        except ValueError as exc:
            raise HTTPException(status_code=422, detail=str(exc)) from exc
        self = cls()
        self.db, self.ctx, self.master = db, ctx, master
        self.q, self.filter_field, self.filter_value, self.parsed = q, filter_field, filter_value, parsed
        src = record_sql.RecordSource("master", master.id, master.field_schema, master=master)
        self.where, self.exact = src.base_where(ctx.tenant_id), True
        allowed = await is_restricted_for(db, ctx.tenant_id, ctx.user_id, master)
        if allowed is not None:
            self.where.append(record_sql.scope_clause(src.fv, master.scope_field, allowed))
        restriction = await access_clause(db, ctx.tenant_id, ctx.user_id, src, "view")
        if restriction is not None:
            self.where.append(restriction)
        parts = []
        if q:
            parts.append(record_sql.search_clause(src.fv, q, src.field_defs))
        if filter_field and filter_value is not None:
            parts.append(record_sql.str_equals_clause(src.fv, filter_field, filter_value))
        for part in parts:
            self.where.append(part.sql)
            self.exact = self.exact and part.exact
        if parsed:
            # A Page Data List block's own filters (e.g. created_by equals
            # {{current_user.id}}) — same semantics as every dataset query.
            resolved = await prepare_filters(db, ctx.tenant_id, ctx.user_id, parsed, src.field_defs)
            clauses, exact = record_sql.filter_clauses(src, resolved, ctx.tenant_id)
            self.where += clauses
            self.exact = self.exact and exact
        return self

    async def _in_python(self) -> list[MasterRecord]:
        stmt = select(MasterRecord).where(*self.where).order_by(*_RECORD_ORDER)
        visible = [row[0] for row in await record_sql.read_capped(self.db, stmt)]
        if self.q:
            visible = [r for r in visible if _record_matches_search(r.field_values, self.q)]
        if self.filter_field and self.filter_value is not None:
            visible = [r for r in visible if str(r.field_values.get(self.filter_field)) == self.filter_value]
        if self.parsed:
            declared = {f["name"] for f in self.master.field_schema}
            visible = await filter_records(self.db, self.ctx.tenant_id, self.ctx.user_id, visible, self.parsed, declared, self.master.field_schema)
        return visible

    async def page(self, skip: int, limit: int, after: uuid.UUID | None = None) -> list[MasterRecord]:
        """A page of records. `after` is the last record of the page before,
        and doesn't get slower deep into a large source; if that record has
        gone since, `skip` still finds about the same place."""
        if not self.exact:
            visible = await self._in_python()
            at = next((i for i, r in enumerate(visible) if r.id == after), None) if after else None
            start = skip if at is None else at + 1
            return visible[start : start + limit]
        stmt = select(MasterRecord).where(*self.where).order_by(*_RECORD_ORDER).limit(limit)
        position = None
        if after:
            position = (
                await self.db.execute(
                    select(*_RECORD_ORDER).where(MasterRecord.tenant_id == self.ctx.tenant_id, MasterRecord.master_id == self.master.id, MasterRecord.id == after)
                )
            ).first()
        stmt = stmt.where(record_sql.after_clause(_RECORD_ORDER, tuple(position))) if position else stmt.offset(skip)
        return list((await self.db.execute(stmt)).scalars().all())

    async def count(self) -> dict[str, int | bool]:
        if not self.exact:
            return {"total": len(await self._in_python()), "capped": False}
        total, capped = await record_sql.capped_count(self.db, MasterRecord, self.where)
        return {"total": total, "capped": capped}


@router.get("/{master_id}/records", response_model=list[MasterRecordRead])
async def list_master_records(
    master_id: uuid.UUID,
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=record_sql.MAX_PAGE_SIZE),
    after: uuid.UUID | None = None,
    q: str | None = None,
    filter_field: str | None = None,
    filter_value: str | None = None,
    filters: str | None = None,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "view")),
    __: AuthContext = Depends(require_portal_item("masters", "master_id")),
) -> list[MasterRecord]:
    master = await crud.get(db, ctx.tenant_id, master_id)
    query = await _RecordQuery.build(db, ctx, master, q, filter_field, filter_value, filters)
    page = await query.page(skip, limit, after)
    await attach_user_names(db, ctx.tenant_id, page)
    return page


@router.get("/{master_id}/records/count")
async def count_master_records(
    master_id: uuid.UUID,
    q: str | None = None,
    filter_field: str | None = None,
    filter_value: str | None = None,
    filters: str | None = None,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "view")),
    __: AuthContext = Depends(require_portal_item("masters", "master_id")),
) -> dict[str, int | bool]:
    """The total-matching-row count a paginated caller (Page Designer's Data
    List block, in its live-pagination/search mode — see PageBlockDef.
    live_pagination) needs to render "page N of M" / know when Next should
    disable — list_master_records itself never returns one (its
    response_model is a bare list, and changing that would break every
    existing caller), so this is a small sibling endpoint instead. `capped`
    means there are more than `total` (record_sql.COUNT_CAP)."""
    master = await crud.get(db, ctx.tenant_id, master_id)
    query = await _RecordQuery.build(db, ctx, master, q, filter_field, filter_value, filters)
    return await query.count()


@router.post("/{master_id}/records", response_model=MasterRecordRead, status_code=status.HTTP_201_CREATED)
async def create_master_record(
    master_id: uuid.UUID,
    payload: MasterRecordCreate,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "create")),
    __: AuthContext = Depends(require_portal_action("masters", "master_id", "create")),
) -> MasterRecord:
    master = await crud.get(db, ctx.tenant_id, master_id)
    payload.field_values = await field_constraints.prepare_create(db, ctx.tenant_id, ctx.user_id, master, payload.field_values)
    errors = await validate_field_values(db, ctx.tenant_id, master.field_schema, payload.field_values)
    errors += await field_constraints.unique_errors(db, ctx.tenant_id, master, payload.field_values)
    if errors:
        raise HTTPException(status_code=422, detail=errors)
    if not await check_value_in_scope(db, ctx.tenant_id, ctx.user_id, master, payload.field_values):
        raise HTTPException(status_code=422, detail=[f"'{master.scope_field}' is outside your assigned scope."])
    payload.field_values = await field_constraints.finish_create(db, ctx.tenant_id, master, {
        **payload.field_values,
        **compute_formula_fields(master.field_schema, payload.field_values),
    })
    record = await records_crud.create(db, ctx.tenant_id, payload, master_id=master_id, created_by=ctx.user_id)
    await write_audit_log(db, ctx.tenant_id, ctx.user_id, "record.create", "master_record", record.id, {"master_id": str(master_id)})
    return record


@router.patch("/{master_id}/records/{record_id}", response_model=MasterRecordRead)
async def update_master_record(
    master_id: uuid.UUID,
    record_id: uuid.UUID,
    payload: MasterRecordUpdate,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "edit")),
    __: AuthContext = Depends(require_portal_action("masters", "master_id", "edit")),
) -> MasterRecord:
    master = await crud.get(db, ctx.tenant_id, master_id)
    existing = await records_crud.get(db, ctx.tenant_id, record_id)
    if existing.master_id != master_id or existing.is_deleted:
        raise HTTPException(status_code=404, detail="MasterRecord not found")
    if not await check_value_in_scope(db, ctx.tenant_id, ctx.user_id, master, existing.field_values):
        raise HTTPException(status_code=404, detail="MasterRecord not found")
    await require_record_access(db, ctx.tenant_id, ctx.user_id, _source(master), record_id, "edit")
    before = dict(existing.field_values)
    payload.field_values = field_constraints.keep_autonumbers(master, payload.field_values, existing.field_values)
    errors = await validate_field_values(db, ctx.tenant_id, master.field_schema, payload.field_values)
    errors += await field_constraints.unique_errors(db, ctx.tenant_id, master, payload.field_values, record_id=record_id)
    if errors:
        raise HTTPException(status_code=422, detail=errors)
    if not await check_value_in_scope(db, ctx.tenant_id, ctx.user_id, master, payload.field_values):
        raise HTTPException(status_code=422, detail=[f"'{master.scope_field}' is outside your assigned scope."])
    payload.field_values = {
        **payload.field_values,
        **compute_formula_fields(master.field_schema, payload.field_values),
    }
    record = await records_crud.update(db, ctx.tenant_id, record_id, payload, updated_by=ctx.user_id)
    await write_audit_log(
        db, ctx.tenant_id, ctx.user_id, "record.update", "master_record", record.id,
        {"master_id": str(master_id), "changes": values_diff(before, record.field_values)},
    )
    return record


@router.delete("/{master_id}/records/{record_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_master_record(
    master_id: uuid.UUID,
    record_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "delete")),
    __: AuthContext = Depends(require_portal_action("masters", "master_id", "delete")),
) -> None:
    master = await crud.get(db, ctx.tenant_id, master_id)
    existing = await records_crud.get(db, ctx.tenant_id, record_id)
    if existing.master_id != master_id or existing.is_deleted:
        raise HTTPException(status_code=404, detail="MasterRecord not found")
    if not await check_value_in_scope(db, ctx.tenant_id, ctx.user_id, master, existing.field_values):
        raise HTTPException(status_code=404, detail="MasterRecord not found")
    await require_record_access(db, ctx.tenant_id, ctx.user_id, _source(master), record_id, "delete")
    await records_crud.soft_delete(db, ctx.tenant_id, record_id, updated_by=ctx.user_id)
    await write_audit_log(db, ctx.tenant_id, ctx.user_id, "record.delete", "master_record", record_id, {"master_id": str(master_id)})


# ---- Bulk import (FR-MST-004 / FR-MST-005) -------------------------------------------------------------

def _read_import_file(filename: str | None, raw: bytes) -> pd.DataFrame:
    filename = (filename or "").lower()
    try:
        if filename.endswith(".xlsx") or filename.endswith(".xls"):
            return pd.read_excel(io.BytesIO(raw))
        return pd.read_csv(io.BytesIO(raw))
    except Exception as exc:  # pandas raises various parser-specific errors
        raise HTTPException(status_code=400, detail=f"Could not parse file: {exc}") from exc


@router.post("/{master_id}/import-columns", response_model=ImportColumnsResult)
async def get_import_columns(
    master_id: uuid.UUID,
    file: UploadFile = File(...),
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "create")),
    __: AuthContext = Depends(require_portal_action("masters", "master_id", "import")),
) -> ImportColumnsResult:
    await crud.get(db, ctx.tenant_id, master_id)
    # A large file's first lines are enough, for a CSV.
    raw = await file.read(1024 * 1024) if not record_import.is_excel(file.filename) else await file.read()
    if not record_import.is_excel(file.filename):
        raw = raw[: raw.rfind(b"\n") + 1] or raw
    df = _read_import_file(file.filename, raw)
    return ImportColumnsResult(columns=[str(c) for c in df.columns])


@router.post("/{master_id}/bulk-import", response_model=BulkImportResult)
async def bulk_import_master_records(
    master_id: uuid.UUID,
    file: UploadFile = File(...),
    column_mapping: str = Form(..., description='JSON object: {"source column": "master field name"}'),
    commit: bool = Form(False),
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "create")),
    __: AuthContext = Depends(require_portal_action("masters", "master_id", "import")),
) -> BulkImportResult:
    master = await crud.get(db, ctx.tenant_id, master_id)

    try:
        mapping: dict[str, str] = json.loads(column_mapping)
    except json.JSONDecodeError as exc:
        raise HTTPException(status_code=400, detail="column_mapping must be valid JSON") from exc
    if (file.size or 0) > record_import.MAX_DIRECT_IMPORT_BYTES:
        raise HTTPException(status_code=413, detail="This file is too large to check here — start a background import instead.")

    df = _read_import_file(file.filename, await file.read())
    rows = record_import.clean_rows(df, mapping)
    # `seen`: unique values across the file.
    valid_rows, failed = await record_import.check_rows(db, ctx.tenant_id, ctx.user_id, master, rows, 1, {})
    row_errors = [BulkImportRowError(**f) for f in failed]

    if commit and valid_rows:
        db.add_all(
            [
                MasterRecord(tenant_id=ctx.tenant_id, master_id=master_id, field_values=await field_constraints.finish_create(db, ctx.tenant_id, master, row), created_by=ctx.user_id)
                for row in valid_rows
            ]
        )
        await db.flush()
        await write_audit_log(db, ctx.tenant_id, ctx.user_id, "record.import", "master", master_id, {"rows": len(valid_rows)})

    return BulkImportResult(
        total_rows=len(df),
        valid_rows=len(valid_rows),
        committed=commit and bool(valid_rows),
        row_errors=row_errors,
    )


@router.post("/{master_id}/imports", response_model=RecordImportStarted, status_code=status.HTTP_202_ACCEPTED)
async def start_record_import(
    master_id: uuid.UUID,
    file: UploadFile = File(...),
    column_mapping: str = Form(..., description='JSON object: {"source column": "master field name"}'),
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "create")),
    __: AuthContext = Depends(require_portal_action("masters", "master_id", "import")),
) -> RecordImportStarted:
    """Imports a file of any size in the background (app/crud/record_import.py).
    Follow it with GET /masters/{id}/imports/{job_id}."""
    await crud.get(db, ctx.tenant_id, master_id)
    try:
        mapping = json.loads(column_mapping)
    except json.JSONDecodeError as exc:
        raise HTTPException(status_code=400, detail="column_mapping must be valid JSON") from exc
    if not mapping or not isinstance(mapping, dict):
        raise HTTPException(status_code=422, detail="Match at least one column to a field.")
    suffix = Path(file.filename or "").suffix.lower()
    if suffix not in (".csv", ".xlsx", ".xls"):
        raise HTTPException(status_code=422, detail="Choose a CSV or Excel file.")
    limit = record_import.MAX_EXCEL_BYTES if record_import.is_excel(file.filename) else record_import.MAX_FILE_BYTES
    folder = UPLOAD_ROOT / str(ctx.tenant_id) / "imports"
    folder.mkdir(parents=True, exist_ok=True)
    path = folder / f"{uuid.uuid4().hex}{suffix}"
    written = 0
    with path.open("wb") as out:
        while chunk := await file.read(1024 * 1024):
            written += len(chunk)
            if written > limit:
                out.close()
                path.unlink(missing_ok=True)
                raise HTTPException(status_code=413, detail=f"This file is over the {limit // (1024 * 1024):,} MB limit for its type.")
            out.write(chunk)
    job_id = await jobs.enqueue(
        db,
        record_import.JOB_KIND,
        {
            "master_id": str(master_id), "user_id": str(ctx.user_id), "path": str(path), "filename": file.filename, "mapping": mapping,
            "done_rows": 0, "imported": 0, "failed": 0, "errors": [],
        },
        tenant_id=ctx.tenant_id,
    )
    return RecordImportStarted(job_id=job_id)


@router.get("/{master_id}/imports/{job_id}", response_model=RecordImportStatus)
async def get_record_import(
    master_id: uuid.UUID,
    job_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("masters", "view")),
    __: AuthContext = Depends(require_portal_item("masters", "master_id")),
) -> RecordImportStatus:
    row = (
        await db.execute(
            text("SELECT status, payload, result, last_error FROM jobs WHERE id = :id AND tenant_id = :tenant AND kind = :kind"),
            {"id": job_id, "tenant": ctx.tenant_id, "kind": record_import.JOB_KIND},
        )
    ).first()
    if row is None or row.payload["master_id"] != str(master_id):
        raise HTTPException(status_code=404, detail="Import not found")
    p = row.payload
    return RecordImportStatus(
        status=row.status, filename=p["filename"], done_rows=p["done_rows"], imported=p["imported"], failed=p["failed"], errors=p["errors"],
        message=row.result if row.status == "done" else row.last_error,
    )