#!/usr/bin/env python3
"""CONTINUOUS WhatsApp -> jarvis-memory CRM sync.

Runs hourly (launchd, StartInterval 3600). Picks up every NEW WhatsApp DM message
and refreshes a single per-lead CRM fact in jarvis-memory. The memory service's
own write pipeline supersedes the previous fact for that lead automatically (it
sees the similar older "WhatsApp lead +<phone>" fact), so each lead keeps ONE
current fact whose value marches forward as the conversation evolves.

WATERMARK (Ahmed's exact rule): the cursor is the message TIMESTAMP, per thread,
never read-state. Every new message is processed EVEN IF Ahmed already read/replied
from his phone. Ahmed's own outbound replies (from_me=1) are part of the arc and
are ingested. The whatsapp.db `messages` table has NO read/unread column at all, so
read-state is not merely ignored — it is structurally impossible to consult here.
A thread is "due" purely when MAX(ts) > the stored last_ts for that peer.

READ-ONLY on whatsapp.db EXCEPT this script's own `sync_state` watermark table.

Flags:
  --once           one pass then exit (the default; launchd re-invokes hourly).
  --dry            compose + print facts, no POST, no watermark advance.
  --peer <number>  process a single peer_number thread (still POSTs unless --dry).

Env (from ../.env, never printed): DEEPSEEK_API_KEY, MEMORY_API_KEY,
optionally MEMORY_API_URL (base), DEEPSEEK_URL (base).
"""
from __future__ import annotations

import argparse
import json
import logging
import os
import sqlite3
import sys
import urllib.error
import urllib.request

HERE = os.path.dirname(os.path.abspath(__file__))
DB = os.path.join(HERE, "whatsapp.db")
ENV = os.path.normpath(os.path.join(HERE, "..", ".env"))
LOG = os.path.join(HERE, "crm_sync.log")

MEM_BASE_DEFAULT = "https://jarvis-production-3e5f.up.railway.app"
DEEPSEEK_BASE_DEFAULT = "https://api.deepseek.com"
DEEPSEEK_MODEL = "deepseek-v4-flash"

MAX_THREADS_PER_RUN = 30      # a burst never blows the hour; oldest-activity first
SELF_NUMBERS = {"966565625560"}  # Ahmed's own WA number(s) — never a lead peer
CONVO_CHAR_CAP = 24000        # keep DeepSeek input sane on very long threads
LOG_MAX_BYTES = 1_000_000     # truncate crm_sync.log at ~1MB

log = logging.getLogger("crm_sync")


# --------------------------------------------------------------------------- log
def setup_log() -> None:
    # Rotating by truncation: if the log already exceeds the cap, start fresh.
    try:
        if os.path.exists(LOG) and os.path.getsize(LOG) > LOG_MAX_BYTES:
            open(LOG, "w").close()
    except OSError:
        pass
    log.setLevel(logging.INFO)
    log.handlers.clear()
    fh = logging.FileHandler(LOG, encoding="utf-8")
    fh.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
    log.addHandler(fh)
    # Also echo to stdout so --dry / manual runs are visible in the terminal.
    sh = logging.StreamHandler(sys.stdout)
    sh.setFormatter(logging.Formatter("%(message)s"))
    log.addHandler(sh)


# --------------------------------------------------------------------------- env
def load_env(path: str) -> dict:
    out = {}
    try:
        with open(path, encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                k, v = line.split("=", 1)
                out[k.strip()] = v.strip()
    except OSError as e:
        raise SystemExit(f"cannot read {path}: {e}")
    return out


# ------------------------------------------------------------------------ sqlite
def get_conn() -> sqlite3.Connection:
    con = sqlite3.connect(DB, timeout=15)
    con.row_factory = sqlite3.Row
    con.execute("PRAGMA busy_timeout=8000")
    return con


def ensure_sync_state(con: sqlite3.Connection) -> None:
    con.execute(
        "CREATE TABLE IF NOT EXISTS sync_state ("
        "peer_number TEXT PRIMARY KEY, last_ts INTEGER, fact_hint TEXT)")
    con.commit()


def has_column(con: sqlite3.Connection, table: str, col: str) -> bool:
    return any(r["name"] == col for r in con.execute(
        f"PRAGMA table_info({table})"))


# ------------------------------------------------------------------ thread select
def due_peers(con: sqlite3.Connection):
    """DM peers whose MAX(ts) > stored last_ts, that have at least one NEW row
    which is NOT a still-untranscribed media placeholder (media_status='pending'
    with empty body). Oldest-activity-first, capped so a burst can't blow the hour.
    """
    rows = con.execute(
        """
        SELECT m.peer_number                          AS peer,
               MAX(m.ts)                               AS max_ts,
               COALESCE(s.last_ts, 0)                  AS wm,
               SUM(CASE WHEN m.ts > COALESCE(s.last_ts, 0)
                         AND NOT (m.media_status = 'pending'
                                  AND (m.body IS NULL OR TRIM(m.body) = ''))
                        THEN 1 ELSE 0 END)             AS ready_new
          FROM messages m
          LEFT JOIN sync_state s ON s.peer_number = m.peer_number
         WHERE m.is_group = 0 AND m.peer_number IS NOT NULL
         GROUP BY m.peer_number
        """).fetchall()
    due = [r for r in rows
           if r["peer"] not in SELF_NUMBERS
           and r["max_ts"] is not None
           and r["max_ts"] > r["wm"]
           and (r["ready_new"] or 0) > 0]
    due.sort(key=lambda r: r["max_ts"])          # oldest activity first
    return due[:MAX_THREADS_PER_RUN]


def fetch_thread(con: sqlite3.Connection, peer: str, has_peer_phone: bool):
    cols = ("from_me, account, push_name, body, media_type, media_status, ts, "
            "peer_number" + (", peer_phone" if has_peer_phone else ""))
    return con.execute(
        f"SELECT {cols} FROM messages "
        "WHERE is_group=0 AND peer_number=? ORDER BY ts", (peer,)).fetchall()


# --------------------------------------------------------------------- formatting
def _date(ts) -> str:
    import datetime
    try:
        return datetime.datetime.fromtimestamp(
            int(ts) / 1000, datetime.timezone.utc).strftime("%Y-%m-%d")
    except Exception:
        return ""


def build_meta(rows, has_peer_phone: bool) -> dict:
    accounts = {}
    push = ""
    phone = ""
    for r in rows:
        acc = (r["account"] or "").strip()
        if acc:
            accounts[acc] = accounts.get(acc, 0) + 1
        if not push and (r["push_name"] or "").strip():
            push = r["push_name"].strip()
        if has_peer_phone and not phone and (r["peer_phone"] or "").strip():
            phone = r["peer_phone"].strip()
    if not phone:
        phone = (rows[0]["peer_number"] or "").strip()
    account = max(accounts, key=accounts.get) if accounts else "966565625560"
    max_ts = max(int(r["ts"]) for r in rows)
    human_reply = any(r["from_me"] == 0 and (r["body"] or "").strip() for r in rows)
    return {"phone": phone, "account": account, "push": push,
            "last_date": _date(max_ts), "max_ts": max_ts,
            "human_reply": human_reply}


def build_convo(rows) -> str:
    lines = []
    for r in rows:
        who = "ME" if r["from_me"] == 1 else "THEM"
        body = (r["body"] or "").strip()
        if not body:
            mt = (r["media_type"] or "").strip()
            if (r["media_status"] or "") == "pending" and not mt:
                continue
            body = f"[{mt or 'media'} sent]"
        lines.append(f"[{_date(r['ts'])}] {who}: {body}")
    convo = "\n".join(lines)
    if len(convo) > CONVO_CHAR_CAP:
        head = convo[:3000]
        tail = convo[-(CONVO_CHAR_CAP - 3000):]
        convo = head + "\n...[middle of conversation elided]...\n" + tail
    return convo


# ------------------------------------------------------------------- deepseek call
PROMPT = """You are maintaining a sales CRM. Below is the FULL WhatsApp conversation \
between Ahmed (a Saudi outreach salesperson, his messages marked ME) and ONE lead \
(their messages marked THEM). Ahmed cold-messages Saudi businesses for several \
campaigns. Read the WHOLE thread — including Ahmed's own outbound replies and any \
[voice note] transcripts — and produce ONE dense, current CRM fact for this lead.

CLASSIFY the campaign this thread belongs to, using these cues:
- "AIN (عين cameras)" — AIN / tryain / عين / كاميرات / المراقبة / CCTV analytics cameras.
- "Modern Intelligent Solution (MIS)" — MIS / Modern Intelligent Solution / Saud Altamimi (سعود التميمي).
- "Reval" — Reval / ريفال creative agency; NOTE which pitch: free-design (تصميم مجاني), \
social-media / content, corporate games / team-building (لعبة / موظفين), or dev / software.
- "Ahmed Alrajeh job search" — a PERSONAL job / CV / "prove myself" (أثبت نفسي) outreach, \
Talent Acquisition, hiring / applying for a role.
- "other" — anything else (e.g. a distinct venture); name it briefly.

OUTPUT the "text" field EXACTLY in this shape — ONE dense fact, no line breaks:
WhatsApp lead +<phone> (<name>) — <campaign + which pitch> cold outreach from Ahmed's WA \
+<account>: <dated arc of what happened, key dates as YYYY-MM-DD, BOTH sides, questions \
asked/answered, referrals>. STOPPED AT: <the current stage>. <the concrete next action / \
who to contact / any referral>.

Rules:
- ONE fact block only. Dense and factual, written in English (keep people/business names \
as written; Arabic is fine for names).
- Ahmed's OWN replies (ME lines) are part of the arc — include them.
- If only a WhatsApp Business auto-greeting replied and no human engaged, say so \
("auto-reply only, no human engagement").
- <name> = the person's real name if known, else the business name, else "unknown".
- Use phone +{phone} and account +{account} verbatim in the text.

Also return:
- "importance": 4 if the lead is interested / a meeting or demo is proposed / they were \
referred to someone / a portfolio was requested; 3 if a human replied but with no clear \
interest; 2 if there was no human reply (cold-sent only, or auto-reply only).
- "lead": the entity name for this lead — the person's name, else the business name, else "+{phone}".
- "campaign": the canonical campaign — one of "AIN (عين cameras)", \
"Modern Intelligent Solution (MIS)", "Reval", "Ahmed Alrajeh job search", or a short name for "other".

Thread metadata: lead phone = +{phone}; Ahmed's WA account = +{account}; \
lead's WhatsApp display name = "{push}"; last message date = {last_date}.

Conversation (oldest first):
{convo}

Output ONLY a JSON object and nothing else:
{{"text": "...", "importance": 4, "lead": "...", "campaign": "..."}}"""


def deepseek_summarize(convo: str, meta: dict, key: str, base: str) -> dict:
    prompt = PROMPT.format(phone=meta["phone"], account=meta["account"],
                           push=meta["push"], last_date=meta["last_date"],
                           convo=convo)
    body = json.dumps({
        "model": DEEPSEEK_MODEL,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.2,
        "max_tokens": 2500,
        "stream": False,
    }).encode("utf-8")
    req = urllib.request.Request(
        base.rstrip("/") + "/chat/completions", data=body,
        headers={"Content-Type": "application/json",
                 "Authorization": f"Bearer {key}"})
    with urllib.request.urlopen(req, timeout=120) as r:
        o = json.loads(r.read())
    content = ((o.get("choices") or [{}])[0].get("message") or {}
               ).get("content", "").strip()
    return _parse_json(content)


def _parse_json(content: str) -> dict:
    """Robustly pull the JSON object out of the model's reply."""
    s = content.strip()
    if s.startswith("```"):
        s = s.strip("`")
        if s[:4].lower() == "json":
            s = s[4:]
    i, j = s.find("{"), s.rfind("}")
    if i == -1 or j == -1 or j <= i:
        raise ValueError(f"no JSON object in model reply: {content[:200]!r}")
    return json.loads(s[i:j + 1])


# ------------------------------------------------------------------- memory POST
def post_remember(payload: dict, key: str, base: str) -> int:
    url = base.rstrip("/") + "/remember"
    req = urllib.request.Request(
        url, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        headers={"Content-Type": "application/json",
                 "Authorization": f"Bearer {key}"})
    with urllib.request.urlopen(req, timeout=45) as r:
        return r.status


def advance(con: sqlite3.Connection, peer: str, max_ts: int, hint: str) -> None:
    con.execute(
        "INSERT INTO sync_state(peer_number, last_ts, fact_hint) VALUES(?,?,?) "
        "ON CONFLICT(peer_number) DO UPDATE SET "
        "last_ts=excluded.last_ts, fact_hint=excluded.fact_hint",
        (peer, int(max_ts), (hint or "")[:200]))
    con.commit()


# ------------------------------------------------------------------- per-thread
def make_payload(res: dict, meta: dict) -> dict:
    text = (res.get("text") or "").strip()
    if not text:
        raise ValueError("model returned empty text")
    try:
        imp = int(res.get("importance"))
    except (TypeError, ValueError):
        imp = 3 if meta["human_reply"] else 2
    if imp not in (2, 3, 4):
        imp = 3 if meta["human_reply"] else 2
    lead = (res.get("lead") or "").strip() or ("+" + meta["phone"])
    campaign = (res.get("campaign") or "").strip() or "other"
    return {
        "text": text,
        "kind": "fact",
        "when": meta["last_date"],
        "importance": imp,
        "confidence": 0.8,
        "source": "wa-sync",
        "group": "ahmed",
        "entities": [{"name": lead}, {"name": campaign}],
    }


def process(con, peer, has_peer_phone, keys, dry):
    rows = fetch_thread(con, peer, has_peer_phone)
    if not rows:
        log.info("peer %s: no rows, skip", peer)
        return False
    meta = build_meta(rows, has_peer_phone)
    convo = build_convo(rows)
    if not convo.strip():
        log.info("peer %s: nothing textual yet (awaiting transcripts), skip", peer)
        return False
    try:
        res = deepseek_summarize(convo, meta, keys["ds"], keys["ds_base"])
        payload = make_payload(res, meta)
    except (urllib.error.URLError, ValueError, json.JSONDecodeError) as e:
        log.error("peer %s: DeepSeek/compose failed (%s) — not advancing", peer, e)
        return False

    if dry:
        log.info("DRY peer=%s imp=%s campaign=%s\n  %s",
                 peer, payload["importance"], payload["entities"][1]["name"],
                 payload["text"])
        return True

    try:
        st = post_remember(payload, keys["mem"], keys["mem_base"])
    except urllib.error.URLError as e:
        log.error("peer %s: POST /remember failed (%s) — not advancing", peer, e)
        return False
    if st not in (200, 201):
        log.error("peer %s: /remember HTTP %s — not advancing", peer, st)
        return False
    advance(con, peer, meta["max_ts"],
            f"{payload['entities'][1]['name']} | imp{payload['importance']} | "
            f"{meta['last_date']}")
    log.info("peer %s: OK imp=%s campaign=%s -> watermark %s",
             peer, payload["importance"], payload["entities"][1]["name"],
             meta["max_ts"])
    return True


# ------------------------------------------------------------------------- main
def main() -> int:
    ap = argparse.ArgumentParser(description="Continuous WhatsApp -> memory CRM sync")
    ap.add_argument("--once", action="store_true",
                    help="one pass then exit (default behaviour; launchd repeats hourly)")
    ap.add_argument("--dry", action="store_true",
                    help="compose + print facts, no POST, no watermark advance")
    ap.add_argument("--peer", metavar="NUMBER",
                    help="process a single peer_number thread")
    args = ap.parse_args()

    setup_log()
    env = load_env(ENV)
    keys = {
        "ds": env.get("DEEPSEEK_API_KEY", ""),
        "mem": env.get("MEMORY_API_KEY", ""),
        "ds_base": env.get("DEEPSEEK_URL", DEEPSEEK_BASE_DEFAULT),
        "mem_base": env.get("MEMORY_API_URL", MEM_BASE_DEFAULT),
    }
    if not keys["ds"]:
        log.error("DEEPSEEK_API_KEY missing from %s", ENV)
        return 2
    if not args.dry and not keys["mem"]:
        log.error("MEMORY_API_KEY missing from %s", ENV)
        return 2

    con = get_conn()
    ensure_sync_state(con)
    has_pp = has_column(con, "messages", "peer_phone")

    if args.peer:
        ok = process(con, args.peer, has_pp, keys, args.dry)
        con.close()
        return 0 if ok else 1

    due = due_peers(con)
    log.info("run start: %d thread(s) due (cap %d, oldest-first)%s",
             len(due), MAX_THREADS_PER_RUN, " [DRY]" if args.dry else "")
    done = 0
    for r in due:
        if process(con, r["peer"], has_pp, keys, args.dry):
            done += 1
    log.info("run done: %d/%d processed", done, len(due))
    con.close()
    return 0


if __name__ == "__main__":
    sys.exit(main())