ui/parsing/share_fetch.py

"""Fetch a ChatGPT/Claude *share link* and parse its embedded conversation JSON.

A pasted share URL should "just work": we GET the public share page, dig the conversation
out of the JSON the page ships to the browser (no headless browser, no API key), and map it
to the same `ParsedExport` the paste path returns (source="paste" → single-conversation flow).

Where the JSON lives (web-verified shapes; both have several historical variants we tolerate):
  - ChatGPT: `<script id="__NEXT_DATA__">{…}</script>` (Next.js) — or newer Remix/react-router
    builds embedding `window.__remixContext = {…}` / `__reactRouterContext`. We grep all of them
    for the message mapping (or `linear_conversation`) and reuse the parser's ChatGPT mapping
    walk where possible.
  - Claude: a JSON blob carrying `chat_messages: [{sender, text|content}]`, embedded in a
    `<script>` (often the same Remix payload). sender human→user, assistant→assistant.

Robust by design: any fetch/parse failure raises `ParseError` with a copy-paste-instead hint.
The fetched chat is held in memory only — never written to disk or logged.
"""
from __future__ import annotations

import json
import os
import re

from .parser import (
    Conversation,
    Turn,
    ParsedExport,
    ParseError,
    _norm_role,
    _to_epoch,
    _text_from_parts,
    _parse_chatgpt,
)

# Host/path patterns for the two providers' share links.
SHARE_HOSTS = ("chatgpt.com", "chat.openai.com", "claude.ai")
_CHATGPT_RE = re.compile(r"^https?://(?:www\.)?(?:chatgpt\.com|chat\.openai\.com)/share/[\w-]+", re.I)
_CLAUDE_RE = re.compile(r"^https?://(?:www\.)?claude\.ai/share/[\w-]+", re.I)
_SHARE_RE = re.compile(r"^https?://\S+$", re.I)

_UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
       "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36")

_COPY_HINT = ("Couldn't read that share link automatically — open it, copy the conversation "
              "text, and paste it here instead.")


def is_share_url(text: str) -> bool:
    """True only when the *whole* paste is a single ChatGPT/Claude share URL.

    A bare URL line counts; anything carrying conversation prose (extra non-blank lines) does
    not, so a transcript that merely mentions a URL still goes down the normal paste path.
    """
    if not isinstance(text, str):
        return False
    lines = [ln.strip() for ln in text.strip().splitlines() if ln.strip()]
    if len(lines) != 1:
        return False
    url = lines[0]
    return bool(_CHATGPT_RE.match(url) or _CLAUDE_RE.match(url))


def fetch_share(url: str, *, get=None) -> ParsedExport:
    """Fetch a share URL and return its conversation as a ParsedExport (source="paste").

    `get` is injectable (defaults to requests.get) so tests need no network — mirrors how
    MiniCPMClient takes `post=`. Raises ParseError on any fetch/parse failure.
    """
    url = (url or "").strip()
    if not (_CHATGPT_RE.match(url) or _CLAUDE_RE.match(url)):
        raise ParseError(_COPY_HINT)

    html = _get_html(url, get)
    is_claude = bool(_CLAUDE_RE.match(url))
    convs = _extract_claude(html) if is_claude else _extract_chatgpt(html)
    if not convs or not any(c.turns for c in convs):
        raise ParseError(_COPY_HINT)
    return ParsedExport(source="paste", conversations=convs)


def _get_html(url: str, get) -> str:
    timeout = _timeout()
    get = get or _default_get()
    try:
        resp = get(url, headers={"User-Agent": _UA, "Accept": "text/html"}, timeout=timeout)
    except Exception as e:  # network errors, DNS, Cloudflare resets, …
        raise ParseError(_COPY_HINT) from e
    status = getattr(resp, "status_code", 200)
    if status and status >= 400:
        raise ParseError(_COPY_HINT)
    text = getattr(resp, "text", None)
    if not isinstance(text, str) or not text.strip():
        raise ParseError(_COPY_HINT)
    return text


def _default_get():
    import requests  # lazy: UI must run without this import path being exercised
    return requests.get


def _timeout() -> int:
    try:
        return int(os.environ.get("SHARE_FETCH_TIMEOUT", "20"))
    except ValueError:
        return 20


# --------------------------------------------------------------------------------------
# JSON extraction from the embedded page payloads
# --------------------------------------------------------------------------------------

# <script id="__NEXT_DATA__" ...>{…}</script>
_NEXT_DATA_RE = re.compile(
    r'<script[^>]*id=["\']__NEXT_DATA__["\'][^>]*>(.*?)</script>', re.S | re.I)
# window.__remixContext = {…}; / __reactRouterContext = {…};
_REMIX_RE = re.compile(
    r'(?:window\.)?(?:__remixContext|__reactRouterContext)\s*=\s*(\{.*?\})\s*;', re.S)
# generic fallback: any <script>{…json…}</script>
_ANY_SCRIPT_RE = re.compile(r'<script[^>]*>(.*?)</script>', re.S | re.I)


def _json_blobs(html: str):
    """Yield candidate parsed-JSON objects embedded in the page, best matches first."""
    seen = []
    for rx in (_NEXT_DATA_RE, _REMIX_RE):
        for m in rx.finditer(html):
            obj = _try_json(m.group(1))
            if obj is not None:
                seen.append(obj)
    # Last resort: scan every <script> body that looks like a JSON object.
    for m in _ANY_SCRIPT_RE.finditer(html):
        body = m.group(1).strip()
        if body.startswith("{") and ('"mapping"' in body or '"chat_messages"' in body
                                     or '"linear_conversation"' in body):
            obj = _try_json(body)
            if obj is not None:
                seen.append(obj)
    return seen


def _try_json(text: str):
    text = (text or "").strip()
    if not text:
        return None
    try:
        return json.loads(text)
    except (json.JSONDecodeError, ValueError):
        return None


def _walk(obj):
    """Depth-first walk over nested dict/list, yielding every dict node."""
    if isinstance(obj, dict):
        yield obj
        for v in obj.values():
            yield from _walk(v)
    elif isinstance(obj, list):
        for v in obj:
            yield from _walk(v)


# ---- ChatGPT ----

def _extract_chatgpt(html: str) -> list[Conversation]:
    for blob in _json_blobs(html):
        # 1) reuse the parser's mapping walk wherever a {"mapping": {...}} conv appears
        mapping_owners = [d for d in _walk(blob) if isinstance(d.get("mapping"), dict)
                          and any(isinstance(n, dict) and n.get("message")
                                  for n in d["mapping"].values())]
        if mapping_owners:
            convs = _parse_chatgpt(mapping_owners)
            if any(c.turns for c in convs):
                return convs
        # 2) linear_conversation: a flat ordered list of message dicts
        for d in _walk(blob):
            lin = d.get("linear_conversation")
            if isinstance(lin, list):
                turns = _turns_from_linear(lin)
                if turns:
                    return [Conversation(turns=turns)]
    return []


def _turns_from_linear(items: list) -> list[Turn]:
    turns: list[Turn] = []
    for it in items:
        msg = it.get("message") if isinstance(it, dict) else None
        if not isinstance(msg, dict):
            continue
        role = _norm_role((msg.get("author") or {}).get("role", ""))
        text = _text_from_parts(msg.get("content") or {})
        if role and text:
            turns.append(Turn(role=role, text=text, epoch=_to_epoch(msg.get("create_time"))))
    return turns


# ---- Claude ----

def _extract_claude(html: str) -> list[Conversation]:
    for blob in _json_blobs(html):
        for d in _walk(blob):
            msgs = d.get("chat_messages")
            if isinstance(msgs, list) and msgs:
                turns = _turns_from_claude(msgs)
                if turns:
                    return [Conversation(turns=turns)]
    return []


def _turns_from_claude(msgs: list) -> list[Turn]:
    turns: list[Turn] = []
    for m in msgs:
        if not isinstance(m, dict):
            continue
        role = _norm_role(m.get("sender", ""))
        text = (m.get("text") or "").strip() or _claude_content_text(m.get("content"))
        if role and text:
            turns.append(Turn(role=role, text=text, epoch=_to_epoch(m.get("created_at"))))
    return turns


def _claude_content_text(content) -> str:
    """Newer Claude payloads put text in `content: [{type:'text', text:'…'}]`."""
    if isinstance(content, str):
        return content.strip()
    if isinstance(content, list):
        parts = [c.get("text", "") for c in content if isinstance(c, dict)]
        return "\n".join(p for p in parts if isinstance(p, str)).strip()
    return ""