"""Cross-source person / company lookup for the jarvis-tools service (Railway).

Ahmed's leads live in three scattered places and he does NOT want a Supabase
restructure — so this is the bridge that searches all of them at once and merges
the answer. One tool, `person_lookup(query)`, takes a phone number (any Saudi
format) or a name/company and asks EVERY source Ahmed keeps people in:

  A. the memory graph  (MEMORY_API_URL /entity + /search — reuses the same env
     the whole fleet already uses; no new creds)
  B. the Ultron master DB  (leads table, via ultron_db.query() — SQL built
     server-side, phone matched digits-only so raw input never reaches SQL)
  C. the Jarvis Drive / shared-storage sheets  (CSV + XLSX under 5MB, via the
     drive.py + storage.py primitives; parsed sheets cached in-process so repeat
     lookups are instant, whole scan capped at ~8s)

Everything degrades gracefully: a missing credential or an unreachable source is
reported ("checked X: not configured"), never raised, so Ahmed can trust a miss.
Output is readable prose because the consumers are LLMs (Jarvis voice + Claude).
"""
from __future__ import annotations

import csv
import io
import json
import os
import re
import time
import urllib.error
import urllib.parse
import urllib.request

import drive
import storage
import ultron_db

_SCAN_BUDGET_S = 8.0            # hard cap on the whole Source-C sheet scan
_MAX_SHEET_BYTES = 5 * 1024 * 1024
_MAX_SHEET_ROWS = 200_000      # guard a pathological sheet
_HITS_PER_FILE = 3             # sample rows to show per matching file

# path/etag -> (version, rows) so a second lookup never re-downloads/re-parses.
_SHEET_CACHE: dict[str, tuple[str, list[list[str]]]] = {}


# ---------------------------------------------------------------------------
# env (read at call time)
# ---------------------------------------------------------------------------
def _mem_url() -> str:
    return os.environ.get("MEMORY_API_URL", "").rstrip("/")


def _mem_key() -> str:
    return os.environ.get("MEMORY_API_KEY", "")


def _group() -> str:
    return os.environ.get("MEMORY_GROUP", "ahmed")


# ---------------------------------------------------------------------------
# query classification + phone normalisation
# ---------------------------------------------------------------------------
def _digits(s) -> str:
    return re.sub(r"\D", "", str(s or ""))


def _norm_phone(s) -> str:
    """Any Saudi format → the trailing 9 digits (5XXXXXXXX), so 0501234567,
    +966501234567, 966501234567 and '050 123 4567' all normalise to the same
    key and match each other."""
    d = _digits(s)
    return d[-9:] if len(d) >= 9 else d


def _looks_like_phone(q: str) -> bool:
    """True when the query is only digits + phone punctuation (space/-/+/()/.),
    with at least 7 digits — otherwise it's a name/company."""
    q = str(q or "").strip()
    rest = re.sub(r"[\d\s\-\+\(\).]", "", q)
    return rest == "" and len(_digits(q)) >= 7


# ---------------------------------------------------------------------------
# Source A — memory graph (HTTP, stdlib urllib like storage.py/smartlead.py)
# ---------------------------------------------------------------------------
def _mem_get(path: str, params: dict, timeout: float = 6.0) -> dict:
    url = _mem_url() + path + "?" + urllib.parse.urlencode(params)
    req = urllib.request.Request(url, method="GET")
    key = _mem_key()
    if key:
        req.add_header("Authorization", f"Bearer {key}")
    req.add_header("Accept", "application/json")
    with urllib.request.urlopen(req, timeout=timeout) as r:
        raw = r.read()
    return json.loads(raw) if raw else {}


def _source_memory(q: str, is_phone: bool, norm: str) -> dict:
    res = {"name": "MEMORY GRAPH", "found": False, "lines": [],
           "status": "", "candidate": None, "tag": ""}
    if not _mem_url():
        res["status"] = "not configured"
        return res
    try:
        if is_phone:
            # phones live inside fact text — semantic+keyword /search finds them.
            facts = _mem_get("/search", {"q": q, "k": 5,
                                         "group": _group()}).get("facts", [])
            if not facts and norm:
                facts = _mem_get("/search", {"q": norm, "k": 5,
                                             "group": _group()}).get("facts", [])
            for f in facts[:6]:
                t = (f.get("fact") or "").strip()
                if t:
                    res["lines"].append("- " + t)
            if res["lines"]:
                # facts aren't clean names — let a leads/sheet row name the
                # best-guess; memory just contributes the "seen in" tag.
                res["tag"] = "memory graph"
        else:
            ents = _mem_get("/entity", {"name": q,
                                        "group": _group()}).get("entities", [])
            for e in ents[:2]:
                nm = e.get("name") or e.get("key")
                res["lines"].append(f"- {nm} ({e.get('type') or 'thing'})")
                gist = (e.get("summary") or e.get("read") or "").strip()
                if gist:
                    res["lines"].append("  " + gist[:220])
                for fct in (e.get("facts") or [])[:3]:
                    t = (fct.get("fact") or "").strip()
                    if t:
                        res["lines"].append("  • " + t)
                rels = e.get("relations") or []
                if rels:
                    res["lines"].append("  ~ " + ", ".join(
                        f"{r.get('rtype') or 'related'} {r.get('other')}"
                        for r in rels[:4]))
            if ents:
                res["candidate"] = ents[0].get("name") or ents[0].get("key")
                res["tag"] = "memory graph"
            else:  # no entity — fall back to a free-text fact search
                facts = _mem_get("/search", {"q": q, "k": 5,
                                             "group": _group()}).get("facts", [])
                for f in facts[:6]:
                    t = (f.get("fact") or "").strip()
                    if t:
                        res["lines"].append("- " + t)
                if res["lines"]:
                    res["tag"] = "memory graph"
        res["found"] = bool(res["lines"])
        res["status"] = f"ok ({len(res['lines'])} hit(s))" if res["found"] \
            else "ok (no hits)"
    except Exception as e:  # noqa: BLE001 — a source must never crash the lookup
        res["status"] = f"error: {str(e)[:100]}"
    return res


# ---------------------------------------------------------------------------
# Source B — Ultron leads DB (SQL built server-side; input never interpolated
# raw — phone is digits-only, name is escaped for the string literal + LIKE)
# ---------------------------------------------------------------------------
def _esc_like(s: str) -> str:
    """Escape for an ILIKE pattern literal: neutralise LIKE wildcards, then
    double single-quotes so the value can't break out of the string literal."""
    s = str(s).replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
    return s.replace("'", "''")


def _parse_query_rows(text: str) -> list[dict]:
    """ultron_db.query() prints 'col | col\\n---\\nval | val'. We control the
    SELECT column order, so this round-trips cleanly back into dicts."""
    lines = [ln for ln in text.strip("\n").split("\n") if ln.strip()]
    if len(lines) < 3:
        return []
    cols = [c.strip() for c in lines[0].split(" | ")]
    rows = []
    for ln in lines[2:]:
        if ln.startswith(("…", "-")):
            continue
        vals = [v.strip() for v in ln.split(" | ")]
        rows.append(dict(zip(cols, vals)))
    return rows


def _source_ultron(q: str, is_phone: bool, norm: str) -> dict:
    res = {"name": "ULTRON LEADS DB", "found": False, "lines": [],
           "status": "", "candidate": None, "tag": ""}
    if not ultron_db.configured():
        res["status"] = "not configured"
        return res
    if is_phone:
        n = len(norm)
        # norm is digits-only (from _norm_phone) → safe to inline. Match the
        # trailing N digits of phone OR whatsapp after stripping non-digits.
        cond = (f"right(regexp_replace(coalesce(phone,''),'\\D','','g'),{n})='{norm}'"
                f" or right(regexp_replace(coalesce(whatsapp,''),'\\D','','g'),{n})"
                f"='{norm}'")
    else:
        pat = _esc_like(q)
        cond = (f"company ilike '%{pat}%' or email ilike '%{pat}%' "
                f"or coalesce(instagram,'') ilike '%{pat}%'")
    sql = ("select company, phone, email, city, category, rating, maps_url "
           f"from leads where ({cond}) order by last_seen desc nulls last limit 5")
    try:
        rows = _parse_query_rows(ultron_db.query(sql))
    except Exception as e:  # noqa: BLE001
        res["status"] = f"error: {str(e)[:100]}"
        return res
    if not rows:
        res["status"] = "ok (0 rows)"
        return res
    for r in rows:
        parts = [r.get("company") or "(unnamed)"]
        for key in ("category", "city", "phone", "email"):
            if r.get(key):
                parts.append(r[key])
        if r.get("rating"):
            parts.append(f"rating {r['rating']}")
        res["lines"].append("- " + " · ".join(parts))
        if r.get("maps_url"):
            res["lines"].append("  " + r["maps_url"])
    first = rows[0]
    where = first.get("city") or first.get("category") or ""
    res["candidate"] = (first.get("company") or "").strip() or None
    res["found"] = True
    res["status"] = f"ok ({len(rows)} row(s))"
    res["tag"] = f"Ultron leads DB{f' ({where})' if where else ''}"
    return res


# ---------------------------------------------------------------------------
# Source C — Jarvis Drive + shared-storage sheets (CSV/XLSX, cached, time-capped)
# ---------------------------------------------------------------------------
def _ext_of(name: str):
    low = name.lower()
    if low.endswith(".csv"):
        return "csv"
    if low.endswith(".xlsx"):
        return "xlsx"
    if low.endswith(".xls"):
        return "xls"
    return None


def _list_drive_sheets() -> list[dict] | None:
    """Sheet-like files in the Google 'Jarvis Drive' folder, or None if the
    Drive service account isn't configured on this deploy."""
    try:
        svc = drive._service()
        fid = drive._folder_id()
    except Exception:  # noqa: BLE001 — DRIVE_SA_JSON / DRIVE_FOLDER_ID missing
        return None
    files = svc.files().list(
        q=f"'{fid}' in parents and trashed=false", spaces="drive",
        orderBy="modifiedTime desc",
        fields="files(id,name,mimeType,size,modifiedTime)"
    ).execute().get("files", [])
    out = []
    for f in files:
        name = f.get("name", "")
        mime = f.get("mimeType", "")
        gsheet = mime == "application/vnd.google-apps.spreadsheet"
        ext = "csv" if gsheet else _ext_of(name)
        if not ext:
            continue
        size = int(f["size"]) if f.get("size") else None
        fid_, ver = f["id"], (f.get("modifiedTime") or "") + str(size)

        def _fetch(fid_=fid_, gsheet=gsheet):
            if gsheet:
                return svc.files().export(fileId=fid_,
                                          mimeType="text/csv").execute()
            return svc.files().get_media(fileId=fid_).execute()

        out.append({"store": "drive", "name": name, "key": "drive:" + fid_,
                    "version": ver, "size": size, "ext": ext, "fetch": _fetch})
    return out


def _list_storage_sheets() -> list[dict] | None:
    """Sheet-like files in the writable shared Supabase-storage bucket, or None
    if storage isn't configured."""
    if not storage.configured():
        return None
    try:
        data = storage._do(
            "POST", f"/object/list/{storage.BUCKET}",
            json.dumps({"prefix": "", "limit": 1000,
                        "sortBy": {"column": "name", "order": "asc"}}).encode(),
            {"Content-Type": "application/json"})
        items = json.loads(data)
    except Exception:  # noqa: BLE001
        return None
    out = []
    for i in items:
        key = i.get("name")
        if not key:
            continue
        name = storage._dec_name(key)
        ext = _ext_of(name)
        if not ext:
            continue
        meta = i.get("metadata") or {}
        size = meta.get("size")
        size = int(size) if str(size or "").isdigit() else None
        ver = (str(meta.get("lastModified") or meta.get("updated_at") or "")
               + str(meta.get("eTag") or meta.get("etag") or "") + str(size))

        def _fetch(key=key):
            return storage._do(
                "GET", f"/object/{storage.BUCKET}/"
                       f"{urllib.parse.quote(key, safe='/')}")

        out.append({"store": "storage", "name": name, "key": "storage:" + key,
                    "version": ver, "size": size, "ext": ext, "fetch": _fetch})
    return out


def _rows_from_bytes(data: bytes, ext: str) -> list[list[str]]:
    if ext == "csv":
        txt = data.decode("utf-8-sig", "ignore")
        rows = []
        for r in csv.reader(io.StringIO(txt)):
            rows.append([str(c) for c in r])
            if len(rows) >= _MAX_SHEET_ROWS:
                break
        return rows
    if ext == "xlsx":
        import openpyxl  # lazy — may be absent; caller turns that into a skip
        wb = openpyxl.load_workbook(io.BytesIO(data), read_only=True,
                                    data_only=True)
        rows = []
        try:
            for ws in wb.worksheets:
                for r in ws.iter_rows(values_only=True):
                    rows.append(["" if c is None else str(c) for c in r])
                    if len(rows) >= _MAX_SHEET_ROWS:
                        break
                if len(rows) >= _MAX_SHEET_ROWS:
                    break
        finally:
            wb.close()
        return rows
    raise ValueError(f"unsupported ext {ext!r}")


def _get_rows(f: dict) -> list[list[str]]:
    cached = _SHEET_CACHE.get(f["key"])
    if cached and cached[0] == f["version"]:
        return cached[1]
    rows = _rows_from_bytes(f["fetch"](), f["ext"])
    _SHEET_CACHE[f["key"]] = (f["version"], rows)
    return rows


def _match_rows(rows: list[list[str]], is_phone: bool, norm: str,
                q_lower: str) -> tuple[list[str], list[list[str]]]:
    """Return (header, matched_rows[:_HITS_PER_FILE]). Assumes row 0 is a header
    when there's more than one row (true for lead exports)."""
    if not rows:
        return [], []
    header = rows[0]
    body = rows[1:] if len(rows) > 1 else rows
    hits = []
    for r in body:
        if is_phone:
            ok = any(len(_digits(c)) >= len(norm) and _digits(c).endswith(norm)
                     for c in r)
        else:
            ok = q_lower in " ".join(r).lower()
        if ok:
            hits.append(r)
            if len(hits) >= _HITS_PER_FILE:
                break
    return header, hits


def _fmt_hit(header: list[str], row: list[str]) -> str:
    """Pair header labels with values (non-empty only) for a readable line."""
    pairs = []
    for i, val in enumerate(row):
        val = (val or "").strip()
        if not val:
            continue
        label = header[i].strip() if i < len(header) and header[i].strip() else f"c{i}"
        pairs.append(f"{label}={val}")
        if len(pairs) >= 8:
            break
    return "  • " + " · ".join(pairs) if pairs else "  • " + " · ".join(row[:8])


def _source_sheets(q: str, is_phone: bool, norm: str, deadline: float) -> dict:
    res = {"name": "JARVIS DRIVE SHEETS", "found": False, "lines": [],
           "status": "", "candidate": None, "tag": ""}
    files: list[dict] = []
    stores_off = []
    for lister, label in ((_list_drive_sheets, "Google Drive"),
                          (_list_storage_sheets, "shared storage")):
        try:
            got = lister()
        except Exception:  # noqa: BLE001
            got = None
        if got is None:
            stores_off.append(label)
        else:
            files.extend(got)

    if not files:
        res["status"] = ("no sheet files found"
                         + (f" ({', '.join(stores_off)} not configured)"
                            if stores_off else ""))
        return res

    q_lower = q.lower()
    scanned, skipped = [], []
    for f in files:
        if time.monotonic() >= deadline:
            skipped.append(f["name"] + " (time budget)")
            continue
        if f["size"] is not None and f["size"] > _MAX_SHEET_BYTES:
            skipped.append(f"{f['name']} (>5MB)")
            continue
        if f["ext"] == "xls":
            skipped.append(f"{f['name']} (.xls not supported)")
            continue
        try:
            rows = _get_rows(f)
        except ImportError:
            skipped.append(f"{f['name']} (xlsx: openpyxl not installed)")
            continue
        except Exception as e:  # noqa: BLE001
            skipped.append(f"{f['name']} (read error: {str(e)[:40]})")
            continue
        scanned.append(f["name"])
        header, hits = _match_rows(rows, is_phone, norm, q_lower)
        if hits:
            res["found"] = True
            res["lines"].append(f"FILE {f['name']} ({f['store']}): "
                                f"{len(hits)} match(es)")
            for h in hits:
                res["lines"].append(_fmt_hit(header, h))
            if res["candidate"] is None:
                # best name-ish cell = first mostly-text cell in the first hit
                textish = [c.strip() for c in hits[0]
                           if c.strip() and len(_digits(c)) < len(c.strip())]
                res["candidate"] = (textish[0][:80] if textish else f["name"])
                res["tag"] = f"sheet {f['name']}"

    parts = [f"scanned {len(scanned)}"]
    if scanned:
        parts[0] += " (" + ", ".join(scanned[:6]) + (
            "…" if len(scanned) > 6 else "") + ")"
    if skipped:
        parts.append(f"skipped {len(skipped)} ({'; '.join(skipped[:6])})")
    if stores_off:
        parts.append(f"{', '.join(stores_off)} not configured")
    res["status"] = "; ".join(parts)
    return res


# ---------------------------------------------------------------------------
# merge → readable text
# ---------------------------------------------------------------------------
def person_lookup(query: str) -> str:
    q = str(query or "").strip()
    if not q:
        return ("person_lookup: give me a name, a company, or a phone number "
                "to look up.")
    is_phone = _looks_like_phone(q)
    norm = _norm_phone(q) if is_phone else ""
    subject = (f"phone …{norm[-4:]}" if is_phone and len(norm) >= 4
               else (f"phone {norm}" if is_phone else f"'{q}'"))
    deadline = time.monotonic() + _SCAN_BUDGET_S

    results = [_source_memory(q, is_phone, norm),
               _source_ultron(q, is_phone, norm),
               _source_sheets(q, is_phone, norm, deadline)]
    hits = [r for r in results if r["found"]]

    checked = "Checked — " + "; ".join(
        f"{r['name'].lower()}: {r['status']}" for r in results) + "."

    if not hits:
        return f"No match for {subject} in any source.\n{checked}"

    out = [f"FOUND for {subject} in {len(hits)} source(s):", ""]
    for r in hits:
        out.append(r["name"])
        out.extend(r["lines"])
        out.append("")

    # one-line synthesis: prefer a real name (memory entity / lead company)
    best = None
    for r in results:
        if r["found"] and r.get("candidate"):
            best = r["candidate"]
            break
    tags = [r["tag"] for r in hits if r.get("tag")]
    synthesis = f"Best guess: {best or q}"
    if tags:
        synthesis += " — seen in " + ", ".join(tags)
    synthesis += "."
    out.append(synthesis)
    out.append(checked)
    return "\n".join(out)