"""Server-side integrity of a Form submission's values — what the form in the
browser already enforces, checked again where it can't be bypassed:

  * only the form's own fields are stored — on create, anything else is
    dropped; on update, keys the form doesn't declare keep their stored
    values (tasks write extra keys onto submissions, see crud/tasks.py's
    data_update) and can't be added or changed by the caller;
  * every value has its field type's shape (a number field holds a number,
    a checkbox a boolean, a multi-select a list, a date an ISO date, ...);
    numeric and boolean text from API clients is normalized, not rejected;
  * a choice holds one of its options — a static option, or a live record
    of the linked datasource — after the field's option filters and
    cascading parent, exactly as the form offered them;
  * a user field holds an active user of its declared pool, a file field an
    upload of this tenant.

Option filters whose values use a page/URL parameter can't be checked here
(the browser's URL isn't something the server can trust or know), so those
filters are skipped; filters on the signed-in user ("{{current_user.id}}")
or their session ("{{session.office_id}}") are enforced.

On update only the values that changed are re-checked, so a submission whose
form later dropped an option or record can still be edited.
"""

import re
import uuid
from datetime import date, datetime, time
from typing import Any

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

from app.core.form_validation import GEOLOCATION_ERROR, field_errors_detail, parse_geolocation
from app.crud import record_sql
from app.crud.user_picker import list_picker_users
from app.models.identity import User
from app.models.master import MasterRecord
from app.models.upload import UploadedFile

# Layout markers and buttons carry no value; formula values are computed on
# the server (callers drop and recompute those).
NO_VALUE_TYPES = {"section", "panel", "tabs", "accordion", "wizard", "row", "column", "group_end", "button"}

_TEXT_TYPES = {"text", "textarea", "password", "email", "url", "rich_text", "markdown", "html", "code", "json", "color_picker", "signature", "combobox", "barcode"}
_BOOLEAN_TYPES = {"checkbox", "toggle", "consent"}
_NUMBER_TYPES = {"number", "currency", "percentage", "slider", "rating"}
_SINGLE_CHOICE_TYPES = {"radio", "dropdown", "autocomplete", "lookup"}
_FILE_TYPES = {"file_upload", "image_upload", "attachment", "camera"}
_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
_TOKEN_RE = re.compile(r"\{\{\s*([^}]+?)\s*\}\}")
_PHONE_RE = re.compile(r"^\+(\d{1,4})(?:\s+([\d\s\-().]*))?$")
_E164_RE = re.compile(r"^\+\d{8,15}$")
_ADDRESS_PARTS = ("line1", "line2", "city", "state", "postal_code", "country")


def _empty(value: Any) -> bool:
    return value is None or value == "" or value == []


def _number(value: Any) -> tuple[bool, Any]:
    if isinstance(value, bool):
        return False, value
    if isinstance(value, (int, float)):
        return True, value
    if isinstance(value, str):
        try:
            n = float(value.strip())
        except ValueError:
            return False, value
        return True, int(n) if n.is_integer() and re.fullmatch(r"\s*[+-]?\d+\s*", value) else n
    return False, value


def _iso(value: Any, kind: str) -> bool:
    if not isinstance(value, str):
        return False
    try:
        if kind == "date":
            date.fromisoformat(value[:10])
        elif kind == "time":
            time.fromisoformat(value)
        else:
            datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return False
    return True


def split_choices(value: Any) -> list[str]:
    """An autocomplete with allow_multiple: "Kochi, Thrissur" -> its choices."""
    return [part.strip() for part in str(value).split(",") if part.strip()] if value not in (None, "") else []


def _shape(field: dict, value: Any) -> tuple[str | None, Any]:
    """(error, normalized value) for one non-empty value."""
    t = field.get("type")
    if t == "autocomplete" and field.get("allow_multiple"):
        if not isinstance(value, str):
            return "must be text, choices separated by commas", value
        return None, ", ".join(dict.fromkeys(split_choices(value))) or None
    if t == "embedded_form":
        # The embedded form's answers (saved as its own submission by the
        # router). A stored submission id is only ever kept, never sent anew.
        if not isinstance(value, dict):
            return "must be the embedded form's answers", value
        return None, value if any(not _empty(v) for v in value.values()) else None
    if t in _TEXT_TYPES:
        if not isinstance(value, str):
            return "must be text", value
        if t == "email" and not _EMAIL_RE.match(value.strip()):
            return "must be an email address", value
        return None, value
    if t in _NUMBER_TYPES:
        ok, n = _number(value)
        return (None, n) if ok else ("must be a number", value)
    if t == "phone":
        return normalize_phone(field, value)
    if t == "address":
        if not isinstance(value, dict) or set(value) - set(_ADDRESS_PARTS) or not all(isinstance(v, str) or v is None for v in value.values()):
            return "must be an address", value
        clean = {k: (value.get(k) or "").strip() for k in _ADDRESS_PARTS if (value.get(k) or "").strip()}
        return None, clean or None
    if t == "matrix":
        rows, options = field.get("matrix_rows") or [], field.get("options") or []
        if not isinstance(value, dict):
            return "must be an answer for each statement", value
        clean = {}
        for row, answer in value.items():
            if answer in (None, ""):
                continue
            if row not in rows or answer not in options:
                return "has an answer that isn't one of the choices", value
            clean[row] = answer
        return None, clean or None
    if t == "ranking":
        options = field.get("options") or []
        if not isinstance(value, list) or not all(isinstance(v, str) for v in value):
            return "must be the options in order", value
        if sorted(value) != sorted(options):
            return "must order every option exactly once", value
        return None, value
    if t in _BOOLEAN_TYPES:
        if isinstance(value, bool):
            return None, value
        if isinstance(value, str) and value.strip().lower() in ("true", "false"):
            return None, value.strip().lower() == "true"
        return "must be true or false", value
    if t in ("date", "time", "datetime"):
        return (None, value) if _iso(value, t) else (f"must be an ISO {t}", value)
    if t == "daterange":
        if not isinstance(value, dict) or set(value) - {"from", "to"}:
            return "must be a date range", value
        if any(v not in (None, "") and not _iso(v, "date") for v in value.values()):
            return "must be a date range", value
        return None, value
    if t == "duration":
        if not isinstance(value, dict) or set(value) - {"amount", "unit"}:
            return "must be a duration", value
        if value.get("amount") not in (None, ""):
            ok, n = _number(value["amount"])
            if not ok:
                return "must be a duration", value
            value = {**value, "amount": n}
        return None, value
    if t in _SINGLE_CHOICE_TYPES or t in ("user",) or t in _FILE_TYPES:
        return (None, value) if isinstance(value, str) else ("must be a single choice", value)
    if t == "treeselect":
        if isinstance(value, str) or (isinstance(value, list) and all(isinstance(v, str) for v in value)):
            return None, value
        return "must be a choice", value
    if t == "multi_attachment":
        if isinstance(value, list) and all(isinstance(v, str) for v in value):
            return None, value
        return "must be a list of uploaded files", value
    if t in ("multiselect", "tags"):
        if isinstance(value, list) and all(isinstance(v, str) for v in value):
            return None, value
        return "must be a list of choices", value
    if t == "subform":
        if not isinstance(value, list) or not all(isinstance(r, dict) for r in value):
            return "must be a list of line items", value
        return None, value
    if t == "geolocation":
        return (None, value) if parse_geolocation(value) is not None else (GEOLOCATION_ERROR, value)
    if t == "dynamic_grid":
        if not isinstance(value, list) or not all(isinstance(r, dict) for r in value):
            return "must be a list of rows", value
        columns = {c.get("name"): c.get("type") for c in field.get("grid_columns") or []}
        rows = []
        for row in value:
            clean = {}
            for name, cell in row.items():
                if name not in columns:
                    continue  # a column the grid doesn't have
                if _empty(cell):
                    clean[name] = cell
                    continue
                kind = columns[name]
                if kind == "number":
                    ok, cell = _number(cell)
                    if not ok:
                        return f"column '{name}' must be a number", value
                elif kind == "checkbox" and not isinstance(cell, bool):
                    return f"column '{name}' must be true or false", value
                elif kind == "date" and not _iso(cell, "date"):
                    return f"column '{name}' must be an ISO date", value
                elif kind == "text" and not isinstance(cell, str):
                    return f"column '{name}' must be text", value
                clean[name] = cell
            rows.append(clean)
        return None, rows
    # label / hidden (filled by prefill from any field type) and anything
    # newer than this module: any JSON value.
    return None, value


def normalize_phone(field: dict, value: Any) -> tuple[str | None, Any]:
    """"+<dial code> <number>" with the number's digits kept, e.g.
    "+91 98470 12345" -> "+91 9847012345". A bare number takes the field's
    default_country; "+919847012345" (E.164, no space) is kept as it is.
    Just a dial code (nothing typed) is blank."""
    if not isinstance(value, str):
        return "must be a phone number", value
    raw = value.strip()
    if not raw.startswith("+"):
        code = (field.get("default_country") or "").strip()
        if not code:
            return "must start with the country code, e.g. +91 98470 12345", value
        raw = f"{code} {raw}"
    if _E164_RE.match(raw):
        return None, raw
    m = _PHONE_RE.match(raw)
    if not m:
        return "must be a country code and a number, e.g. +91 98470 12345", value
    digits = re.sub(r"\D", "", m.group(2) or "")
    if not digits:
        return None, None
    if not 4 <= len(digits) <= 14:
        return "must be a phone number of 4 to 14 digits after the country code", value
    return None, f"+{m.group(1)} {digits}"


def _multi_attachment_error(field: dict, value: list[str]) -> str | None:
    if field.get("max_files") and len(value) > field["max_files"]:
        return f"takes at most {field['max_files']} file{'s' if field['max_files'] != 1 else ''}"
    try:
        [uuid.UUID(v) for v in value]
    except ValueError:
        return "lists a file that isn't uploaded"
    return None


# ---- Option filters (frontend/src/lib/optionFilters.ts, on the server) ---------------------------


async def _template_context(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, names: set[str]) -> dict[str, str]:
    """The trusted values option-filter tokens can name: the signed-in user
    and their session. Page/URL parameters are absent (unverifiable)."""
    from app.models.identity import Tenant

    user = await db.get(User, user_id) if user_id else None
    if user is None or user.tenant_id != tenant_id:
        return {}
    ctx: dict[str, str] = {
        "current_user.id": str(user.id),
        "current_user.name": user.name or "",
        "current_user.email": user.email or "",
        "current_user.vendor_id": str(user.vendor_id) if user.vendor_id else "",
        "current_user.tenant_id": str(user.tenant_id),
    }
    if any(n.startswith("session.") for n in names):
        from app.crud.login_session import build_login_session_data
        from app.schemas.identity import LoginSessionSettings

        tenant = await db.get(Tenant, tenant_id)
        try:
            settings = LoginSessionSettings.model_validate((tenant.login_session_settings if tenant else None) or {}).model_copy(update={"task_id": None})
            session = await build_login_session_data(db, tenant, user, settings) if tenant else {}
        except ValueError:
            session = {}
        for key, value in session.items():
            ctx[f"session.{key}"] = "" if value is None else str(value)
    return ctx


def _resolve_filters(filters: list[dict], ctx: dict[str, str]) -> list[dict] | None:
    """The filters the server can enforce, resolved. None = a filter resolved
    blank without skip_if_empty, so no option passes (as in the browser)."""
    out = []
    for f in filters:
        template = f.get("value") or ""
        tokens = _TOKEN_RE.findall(template)
        if any(t not in ctx and not t.startswith(("current_user.", "session.")) for t in tokens):
            continue  # names a page/URL parameter — unverifiable here
        value = _TOKEN_RE.sub(lambda m: ctx.get(m.group(1), ""), template).strip()
        missing = any(not ctx.get(t, "").strip() for t in tokens)
        needs_value = f.get("operator") not in ("is_empty", "is_not_empty")
        if needs_value and (value == "" or missing):
            if f.get("skip_if_empty"):
                continue
            return None
        out.append({"field": f.get("field") or "", "operator": f.get("operator"), "value": value})
    return out


def _compare(actual: str, expected: str) -> float:
    try:
        if actual.strip() and expected.strip():
            return float(actual) - float(expected)
    except ValueError:
        pass
    return (actual > expected) - (actual < expected)


def _matches(actual: Any, f: dict) -> bool:
    empty = actual is None or actual == "" or (isinstance(actual, list) and not actual)
    op = f["operator"]
    if op == "is_empty":
        return empty
    if op == "is_not_empty":
        return not empty
    if empty:
        return False
    if isinstance(actual, list):
        return any(_matches(v, f) for v in actual)
    text = str(actual).lower() if isinstance(actual, bool) else str(actual)
    expected = f["value"]
    if op == "equals":
        return text == expected
    if op == "not_equals":
        return text != expected
    if op == "contains":
        return expected.lower() in text.lower()
    if op == "in":
        return text in [v.strip() for v in expected.split(",")]
    if op in ("gt", "gte", "lt", "lte"):
        c = _compare(text, expected)
        return {"gt": c > 0, "gte": c >= 0, "lt": c < 0, "lte": c <= 0}[op]
    return True


async def _query_option_values(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, field: dict, filters: list[dict] | None = None) -> set[str] | None:
    """The values a Data Query-backed choice field offers (cached like any
    run of the query) — narrowed by the field's option filters, already
    resolved — or None when the query can't be run."""
    from app.crud import data_queries

    try:
        result = await data_queries.run_for_user(db, tenant_id, user_id, field["options_query_id"], {})
    except (LookupError, PermissionError, data_queries.DataQueryError):
        return None
    column = field.get("options_value_field")
    rows = [r for r in result.rows if all(_matches(r.get(f["field"]), f) for f in filters or [])]
    return {str(r.get(column)) for r in rows if r.get(column) is not None}


# ---- The check -----------------------------------------------------------------------------------


async def check_submission_values(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID | None,
    field_defs: list[dict],
    values: dict,
    previous: dict | None = None,
    locked: set[str] | None = None,
) -> dict:
    """The values to store, normalized — or HTTPException 422 with per-field
    errors. `previous` is the stored values when updating a submission.
    `locked` names fields the caller may not change (read-only or hidden to
    them — see app/crud/field_access.py): on update they keep their stored
    value whatever was sent."""
    by_name = {f["name"]: f for f in field_defs if f.get("type") not in NO_VALUE_TYPES}
    # An autonumber is the server's to give (on submit), never the caller's.
    numbers = {n for n, f in by_name.items() if f.get("type") == "autonumber"}
    locked = (locked or set()) | numbers
    if previous is None:
        clean = {k: v for k, v in values.items() if k in by_name and k not in numbers}
    else:
        # Undeclared keys keep their stored values; declared ones take the
        # new value unless locked.
        clean = {k: v for k, v in previous.items() if k not in by_name}
        for name in by_name:
            source = previous if name in locked else values
            if name in source:
                clean[name] = source[name]

    errors: dict[str, str] = {}
    to_check: list[tuple[dict, Any]] = []
    for name, field in by_name.items():
        if field.get("type") in ("formula", "autonumber") or name not in clean or _empty(clean[name]):
            continue
        if previous is not None and clean[name] == previous.get(name):
            continue  # unchanged — not re-checked
        error, normalized = _shape(field, clean[name])
        if error:
            errors[name] = error
            continue
        clean[name] = normalized
        if not _empty(normalized):
            to_check.append((field, normalized))

    # Line items: each row checked like a small form of its own.
    for field, value in list(to_check):
        if field.get("type") != "subform":
            continue
        rows = []
        for i, row in enumerate(value):
            try:
                rows.append(await check_submission_values(db, tenant_id, user_id, field.get("subform_fields") or [], row))
            except HTTPException as exc:
                detail = exc.detail[0]["msg"] if isinstance(exc.detail, list) and exc.detail else str(exc.detail)
                errors[field["name"]] = f"row {i + 1} — {detail}"
                break
        clean[field["name"]] = rows
    to_check = [(f, v) for f, v in to_check if f.get("type") != "subform"]

    # Choices, people and files — batched lookups.
    ctx: dict[str, str] | None = None
    record_ids: dict[str, set[uuid.UUID]] = {}
    pending_choice: list[tuple[dict, list[str]]] = []
    for field, value in to_check:
        t = field.get("type")
        if t in _SINGLE_CHOICE_TYPES or t in ("multiselect", "treeselect"):
            chosen = value if isinstance(value, list) else split_choices(value) if field.get("allow_multiple") else [value]
            pending_choice.append((field, chosen))
            if field.get("master_id"):
                for v in chosen:
                    try:
                        record_ids.setdefault(str(field["master_id"]), set()).add(uuid.UUID(v))
                    except ValueError:
                        errors[field["name"]] = "isn't one of the available options"

    records: dict[str, dict] = {}
    for master_id, ids in record_ids.items():
        rows = await db.execute(
            select(MasterRecord.id, MasterRecord.field_values).where(
                MasterRecord.tenant_id == tenant_id,
                MasterRecord.master_id == uuid.UUID(master_id),
                MasterRecord.is_deleted.is_(False),
                record_sql.ids_any(MasterRecord.id, sorted(ids)),
            )
        )
        for rid, fv in rows.all():
            records[str(rid)] = {"id": str(rid), **(fv or {})}

    query_options: dict[str, set[str] | None] = {}
    for field, chosen in pending_choice:
        name = field["name"]
        if name in errors:
            continue
        filters = field.get("option_filters") or []
        resolved: list[dict] | None = []
        if filters:
            if ctx is None:
                names = {t for fd in field_defs for f in fd.get("option_filters") or [] for t in _TOKEN_RE.findall(f.get("value") or "")}
                ctx = await _template_context(db, tenant_id, user_id, names)
            resolved = _resolve_filters(filters, ctx)
        for v in chosen:
            if field.get("master_id"):
                record = records.get(v)
                ok = record is not None
                if ok and field.get("cascade_parent_field") and field.get("cascade_match_field"):
                    ok = record.get(field["cascade_match_field"]) == clean.get(field["cascade_parent_field"])
                if ok and resolved is not None:
                    ok = all(_matches(record.get(f["field"]) if f["field"] != "id" else record["id"], f) for f in resolved)
                elif resolved is None:
                    ok = False
            elif field.get("options_query_id"):
                if name not in query_options:
                    query_options[name] = await _query_option_values(db, tenant_id, user_id, field, resolved if filters else [])
                ok = resolved is not None and query_options[name] is not None and str(v) in query_options[name]
            else:
                options = field.get("options") or []
                ok = v in options and resolved is not None and all(_matches(v, f) for f in resolved)
            if not ok:
                errors[name] = "isn't one of the available options"
                break

    # A signed-out visitor (a public form) may only attach files uploaded to
    # a public form — never one of the workspace's own (app/routers/public_forms.py).
    _public_only = [UploadedFile.public_form_id.is_not(None), UploadedFile.created_by.is_(None)] if user_id is None else []
    for field, value in to_check:
        name, t = field["name"], field.get("type")
        if name in errors:
            continue
        if t == "user":
            role_id = field.get("user_role_id")
            pool = await list_picker_users(db, tenant_id, field.get("user_type") or "any", uuid.UUID(str(role_id)) if role_id else None)
            if value not in {str(u.id) for u in pool}:
                errors[name] = "isn't one of the people this field offers"
        elif t in _FILE_TYPES:
            try:
                file_id = uuid.UUID(value)
            except ValueError:
                errors[name] = "isn't an uploaded file"
                continue
            found = await db.execute(select(UploadedFile.id).where(UploadedFile.tenant_id == tenant_id, UploadedFile.id == file_id, *_public_only))
            if found.scalar_one_or_none() is None:
                errors[name] = "isn't an uploaded file"
        elif t == "multi_attachment":
            errors_for = _multi_attachment_error(field, value)
            if errors_for:
                errors[name] = errors_for
                continue
            ids = [uuid.UUID(v) for v in value]
            found = dict((await db.execute(select(UploadedFile.id, UploadedFile.size_bytes).where(UploadedFile.tenant_id == tenant_id, record_sql.ids_any(UploadedFile.id, ids), *_public_only))).all())
            if len(found) < len(set(ids)):
                errors[name] = "lists a file that isn't uploaded"
            elif field.get("max_file_size_mb") and any(size > field["max_file_size_mb"] * 1024 * 1024 for size in found.values()):
                errors[name] = f"has a file larger than {field['max_file_size_mb']:g} MB"

    if errors:
        # Same sentence style as app/core/form_validation.py's messages.
        sentences = {k: f"{m[0].upper()}{m[1:]}{'' if m.endswith('.') else '.'}" for k, m in errors.items()}
        raise HTTPException(status_code=422, detail=field_errors_detail(field_defs, sentences))
    return clean