from __future__ import annotations

import json
import logging
import re
from typing import Any, Dict, List, Optional

logger = logging.getLogger(__name__)


def _coerce_route_handler_result_to_dict(result: Any) -> Dict[str, Any]:
    """
    `ask_question` / `multi_agent_ask_question` return plain dicts in many paths, but the
    training-driven multi-agent pipeline returns `JSONResponse` (same bytes the HTTP client
    would get). Talkback must deserialize to a dict so metadata merges and persistence work.
    """
    if isinstance(result, dict):
        return result
    if result is None:
        return {
            "success": False,
            "message": "Handler returned no response",
            "details": {},
        }
    body = getattr(result, "body", None)
    if callable(body):
        try:
            body = body()
        except Exception:
            body = None
    if isinstance(body, (bytes, bytearray, memoryview)):
        try:
            parsed = json.loads(bytes(body).decode("utf-8"))
            if isinstance(parsed, dict):
                return parsed
            return {
                "success": False,
                "message": "Handler returned non-object JSON",
                "details": {"parsed": parsed},
            }
        except Exception as e:
            logger.warning("Talkback could not parse handler JSON body: %s", e)
            return {
                "success": False,
                "message": "Could not parse handler JSON response",
                "details": {"error": str(e)},
            }
    return {
        "success": False,
        "message": "Handler returned an unsupported response type",
        "details": {"handler_type": type(result).__name__},
    }


def _parse_sql_preview_for_meta(sql_val: Any, max_len: int = 400) -> Optional[str]:
    if sql_val is None:
        return None
    if isinstance(sql_val, (dict, list)):
        try:
            s = json.dumps(sql_val, ensure_ascii=False, default=str)
        except Exception:
            s = str(sql_val)
    else:
        s = str(sql_val).strip()
    if not s:
        return None
    if len(s) > max_len:
        return s[:max_len] + "…"
    return s


def _df_row_count_and_columns(details: Dict[str, Any]) -> tuple[Optional[int], Optional[List[str]]]:
    df = details.get("df")
    if not isinstance(df, dict):
        return None, None
    if "headers" in df and isinstance(df["headers"], list):
        cols = [str(c) for c in df["headers"][:24]]
        data = df.get("data")
        n = len(data) if isinstance(data, list) else None
        return n, cols
    if "agents" in df and isinstance(df["agents"], list):
        cols: List[str] = []
        n = 0
        for ag in df["agents"][:8]:
            if not isinstance(ag, dict):
                continue
            if ag.get("headers"):
                cols.extend([str(x) for x in ag["headers"][:12]])
            d = ag.get("data")
            if isinstance(d, list):
                n += len(d)
        return (n if n else None, cols[:24] or None)
    return None, None


def _infer_datasource_from_details(details: Dict[str, Any]) -> Optional[str]:
    for k in ("datasource", "data_source"):
        v = details.get(k)
        if isinstance(v, str) and v.strip():
            return v.strip()
    ep = details.get("api_endpoint")
    if isinstance(ep, list) and len(ep) > 0:
        return "API"
    if isinstance(ep, str) and ep.strip() and ep.strip() not in ("[]", "N", "None", ""):
        return "API"
    sql = details.get("sql")
    if sql and str(sql).strip() and str(sql).strip() not in ("[]", "N", "None", ""):
        return "Database"
    fr = details.get("file_results")
    if fr and isinstance(fr, list) and len(fr) > 0:
        return "File"
    return None


def _collect_api_names(details: Dict[str, Any]) -> List[str]:
    names: List[str] = []
    ep = details.get("api_endpoint")
    if isinstance(ep, list):
        names.extend([str(x)[:120] for x in ep[:8]])
    elif isinstance(ep, str) and ep.strip() and ep.strip() not in ("[]", "N", ""):
        names.append(ep[:160])
    df = details.get("df")
    if isinstance(df, dict) and isinstance(df.get("agents"), list):
        for ag in df["agents"][:8]:
            if isinstance(ag, dict) and ag.get("api_name"):
                names.append(str(ag["api_name"])[:120])
    # de-dupe preserving order
    seen = set()
    out: List[str] = []
    for n in names:
        if n and n not in seen:
            seen.add(n)
            out.append(n)
    return out[:10]


def _ordinal_to_index(question: str) -> Optional[int]:
    q = str(question or "").strip().lower()
    if not q:
        return None
    q = q.replace("-", " ")
    compact = re.sub(r"\s+", " ", q)

    patterns: List[tuple[str, int]] = [
        (r"\b(first|1st|one|option\s*1|number\s*1)\b", 0),
        (r"\b(second|2nd|two|option\s*2|number\s*2)\b", 1),
        (r"\b(third|3rd|three|option\s*3|number\s*3)\b", 2),
        (r"\b(fourth|4th|four|option\s*4|number\s*4)\b", 3),
        (r"\b(fifth|5th|five|option\s*5|number\s*5)\b", 4),
    ]
    for pat, idx in patterns:
        if re.search(pat, compact):
            return idx

    numeric_patterns = [
        r"\b(?:option|number)\s*(\d{1,2})\b",
        r"\b(\d{1,2})(?:st|nd|rd|th)\b",
        r"^\s*(\d{1,2})\s*[\.\),]?\s*$",
        r"^\s*(\d{1,2})\s*$",
    ]
    for pat in numeric_patterns:
        m = re.search(pat, compact)
        if not m:
            continue
        try:
            n = int(m.group(1))
        except Exception:
            continue
        if 1 <= n <= 50:
            return n - 1
    return None


def _looks_like_clarification_reply(question_text: str) -> bool:
    """
    Heuristic to decide whether current input is likely answering a prior clarification prompt
    (agent/source disambiguation), instead of being a brand-new question.
    """
    q = str(question_text or "").strip()
    if not q:
        return False
    ql = q.lower()

    if _ordinal_to_index(ql) is not None:
        return True

    choice_tokens = (
        "first one",
        "second one",
        "third one",
        "option",
        "use ",
        "go with",
        "choose",
        "chosen",
        "pick",
        "selected",
        "former",
        "latter",
        "both",
        "all",
        "either",
    )
    if any(tok in ql for tok in choice_tokens):
        return True

    words = [w for w in re.split(r"\s+", ql) if w]
    analytic = {"what", "which", "how", "why", "when", "where", "show", "list", "give", "find", "report"}
    # Comma / "and" lists of agent or source names.
    if ("," in q or " and " in ql) and len(q) <= 200 and not any(w in analytic for w in words):
        agentish = (
            "agent",
            "database",
            "source",
            "system",
            "both",
            "all",
            "each",
        )
        if any(x in ql for x in agentish):
            return True
    # Short noun-like replies (e.g. "order management", "sap sales") are usually selections.
    if len(words) <= 4 and not any(w in analytic for w in words):
        return True
    # Slightly longer agent picks without analytic verbs (e.g. "Use Order Management and AI agent").
    if len(words) <= 14 and not any(w in analytic for w in words):
        if any(x in ql for x in ("agent", "source", "system", "both", "all")):
            return True
    return False


def _safe_str(v: Any, max_len: int = 240) -> str:
    try:
        s = str(v)
    except Exception:
        s = repr(v)
    s = s.replace("\n", "\\n")
    return s if len(s) <= max_len else (s[: max_len - 3] + "...")


def _extract_talkback_options(text: str) -> List[str]:
    """
    Extract numbered suggestions from a talkback continuation string.
    Expected format: "If you want, I can also:\n1. Suggestion\n2. Suggestion"
    """
    if not text or not isinstance(text, str):
        return []
    # Match lines starting with a number and period, e.g. "1. "
    lines = re.findall(r"^\s*\d+\.\s*(.+)$", text, re.MULTILINE)
    return [line.strip() for line in lines if line.strip()]


def _parse_df_input_light(df_input: Any) -> Optional[List[Any]]:
    """Mirror AskmeService._parse_df_input for bypass path (no service instance)."""
    if df_input is None:
        return None
    if isinstance(df_input, dict):
        data = df_input.get("data")
        if isinstance(data, list):
            return data
        return None
    if isinstance(df_input, list):
        if len(df_input) == 0:
            return []
        first = df_input[0]
        if (
            isinstance(first, dict)
            and "data" in first
            and isinstance(first.get("data"), list)
            and ("headers" in first or "agent_id" in first)
        ):
            rows_out: List[Any] = []
            for block in df_input:
                if not isinstance(block, dict):
                    continue
                part = block.get("data")
                if isinstance(part, list):
                    rows_out.extend(part)
            return rows_out or None
        return df_input
    if isinstance(df_input, str):
        try:
            parsed = json.loads(df_input)
            return _parse_df_input_light(parsed)
        except json.JSONDecodeError:
            return None
    return None


def _clone_ask_request(request: Any, updates: Dict[str, Any]) -> Any:
    if hasattr(request, "model_copy"):
        return request.model_copy(update=updates)
    import copy

    r = copy.copy(request)
    for k, v in updates.items():
        setattr(r, k, v)
    return r


def _ensure_web_prefix_for_search_web_request(request: Any, state: Any) -> None:
    """
    Previously forced ``@web`` onto the question when ``search_web`` was true.
    Clients must include ``@web`` or ``@askme`` in the question text when needed; this is a no-op.
    """
    _ = (request, state)
    return
