"""Claude provider — accesses claude.ai internal web API.""" import logging import os from typing import Any from curl_cffi import requests as curl_requests from src.blocks import ( UNKNOWN_REASON_EXTRACTION_FAILED, UNKNOWN_REASON_UNKNOWN_TYPE, make_image_placeholder, make_text_block, make_thinking_block, make_tool_result_block, make_tool_use_block, make_unknown_block, ) from src.loss_report import LossReport from src.providers.base import ( BaseProvider, DRIFT_ERROR, DRIFT_OK, DRIFT_WARN, ProviderError, drift_finding, ) logger = logging.getLogger(__name__) BASE_URL = "https://claude.ai/api" IMPERSONATE = "chrome120" class ClaudeProvider(BaseProvider): """Provider for Claude conversations via the internal web API. Uses curl_cffi to impersonate Chrome's TLS fingerprint, bypassing Cloudflare's bot detection (same issue as chatgpt.com). Authentication: sessionKey cookie (~30 day lifetime, opaque string). Expiry cannot be decoded client-side, so an API rejection is the only signal — and claude.ai sends 403 permission_error / account_session_invalid for an invalid session, not 401. See ``_is_auth_failure``. """ provider_name = "claude" def __init__(self, session_key: str | None = None) -> None: cf_session = curl_requests.Session(impersonate=IMPERSONATE) super().__init__(session=cf_session) # type: ignore[arg-type] # Remove base class User-Agent so curl_cffi uses its Chrome-matched UA self._session.headers.pop("User-Agent", None) key = session_key or os.getenv("CLAUDE_SESSION_KEY", "").strip() if not key: raise ProviderError( self.provider_name, "init", RuntimeError( "CLAUDE_SESSION_KEY is not set. " "Run 'ai-chat-exporter auth' to configure it." ), ) # Set sessionKey in the cookie jar self._session.cookies.set("sessionKey", key, domain="claude.ai", path="/") self._session.headers.update( { "Referer": "https://claude.ai/", "Origin": "https://claude.ai", } ) self._org_id: str | None = None # cached per session logger.debug("[claude] Session initialised with Chrome TLS impersonation (key: [REDACTED])") # claude.ai answers an invalid or expired sessionKey with 403 # permission_error / account_session_invalid — never 401. Verified live # 2026-09-20 against GET /api/organizations: a valid key returns 200, while # an expired key, a garbage key and no cookie at all return byte-identical # 403s carrying this code. Matching on the code rather than on the bare # status keeps a genuine permission problem (which would carry a different # code) reported as itself. _SESSION_INVALID_CODE = "account_session_invalid" def _is_auth_failure(self, response: Any) -> bool: if response.status_code == 401: return True if response.status_code != 403: return False try: body = response.json() except Exception: return False if not isinstance(body, dict): return False error = body.get("error") if not isinstance(error, dict): return False details = error.get("details") if not isinstance(details, dict): return False return details.get("error_code") == self._SESSION_INVALID_CODE def _handle_auth_failure(self, response: Any) -> None: msg = ( f"[claude] Authentication failed (HTTP {response.status_code}). " "Your sessionKey has likely expired (~30 day lifetime). " "Note: Claude session keys are opaque, and claude.ai reports an " "invalid one as 403 'Invalid authorization', not 401 — an API " "rejection is the only expiry signal. " "To refresh: open claude.ai in Chrome → F12 → Application → Cookies " "→ find 'sessionKey' → copy the value. " "Then run 'ai-chat-exporter auth' or update CLAUDE_SESSION_KEY in .env." ) logger.error(msg) raise ProviderError( self.provider_name, "authentication", RuntimeError( f"HTTP {response.status_code} — Claude session key expired or invalid" ), ) def _get_org_id(self) -> str: """Fetch and cache the organization ID (required for all Claude API calls).""" if self._org_id: return self._org_id logger.debug("[claude] Bootstrapping: fetching organization ID") url = f"{BASE_URL}/organizations" try: data = self._make_request("GET", url) except ProviderError: raise except Exception as e: raise ProviderError(self.provider_name, "get_org_id", e) from e if not isinstance(data, list) or not data: self._warn_unexpected_schema("get_org_id", "organizations array") raise ProviderError( self.provider_name, "get_org_id", RuntimeError("organizations endpoint returned empty or unexpected response"), ) org = data[0] org_id = org.get("uuid") or org.get("id") if not org_id: self._warn_unexpected_schema("get_org_id", "uuid/id") raise ProviderError( self.provider_name, "get_org_id", RuntimeError("Could not find organization ID in response"), ) self._org_id = org_id logger.debug("[claude] Got org_id: %s", org_id) return org_id def list_conversations(self, offset: int = 0, limit: int = 100) -> list[dict]: """Fetch one page of conversations. Claude's API may use cursor-based pagination. We attempt offset-based first (via query param); if the response includes a ``next_cursor`` field, a WARNING is logged indicating manual investigation is needed. """ org_id = self._get_org_id() url = f"{BASE_URL}/organizations/{org_id}/chat_conversations" params: dict = {"limit": limit} if offset > 0: params["offset"] = offset try: data = self._make_request("GET", url, params=params) except ProviderError: raise except Exception as e: raise ProviderError(self.provider_name, "list_conversations", e) from e # Handle both list and dict responses if isinstance(data, list): conversations = data elif isinstance(data, dict): conversations = data.get("conversations") or data.get("chats") or [] if not conversations and data: # Check for unexpected pagination mechanism if "next_cursor" in data or "cursor" in data or "next" in data: logger.warning( "[claude] API returned cursor-based pagination — " "only first page will be fetched. " "Please report this at GitHub Issues." ) else: self._warn_unexpected_schema("list_conversations", "root") return [] return conversations def get_conversation(self, conv_id: str) -> dict: """Fetch full conversation detail for a single ID.""" org_id = self._get_org_id() url = f"{BASE_URL}/organizations/{org_id}/chat_conversations/{conv_id}" try: data = self._make_request("GET", url) except ProviderError: raise except Exception as e: raise ProviderError(self.provider_name, "get_conversation", e) from e if not isinstance(data, dict): self._warn_unexpected_schema("get_conversation", "root") return {} return data def normalize_conversation(self, raw: dict, loss_report: LossReport | None = None) -> dict: """Transform Claude raw schema to the common normalized schema.""" report = loss_report if loss_report is not None else LossReport() conv_id = raw.get("uuid") or raw.get("id", "") title = raw.get("name") or raw.get("title") or "Untitled" created_at = raw.get("created_at") or raw.get("create_time") or "" updated_at = raw.get("updated_at") or raw.get("update_time") or "" # Project name — Claude may nest project info as project.name project_data = raw.get("project") or {} project: str | None = ( project_data.get("name") if isinstance(project_data, dict) else project_data ) or None # Messages raw_messages = raw.get("chat_messages") or raw.get("messages") or [] messages: list[dict] = [] for msg in raw_messages: role = _map_role(msg.get("sender") or msg.get("role", "")) if not role: continue content_raw = msg.get("content") if "content" in msg else msg.get("text", "") blocks = _extract_claude_blocks(content_raw, conv_id, report) timestamp = msg.get("created_at") or msg.get("timestamp") or None if not blocks: logger.debug("[claude] Skipping empty message in conversation %s", conv_id[:8]) continue content_type = "text" messages.append( { "role": role, "content_type": content_type, "timestamp": timestamp, "blocks": blocks, } ) for _ in messages: report.record_message() report.record_conversation() return { "id": conv_id, "title": title, "provider": "claude", "project": project, "created_at": created_at, "updated_at": updated_at, "message_count": len(messages), "messages": messages, } def check_drift(self) -> list[dict]: """Probe one listing page + one conversation; assert load-bearing fields. See FUTURE.md §10. Real Claude messages are flat ``text``/``sender``; the rich ``content`` block path is dormant, so its activation (content becoming a list) is itself a drift event worth flagging. ``attachments`` / ``files`` are ignored by the normalizer — non-empty values are a silent-loss risk. """ findings: list[dict] = [] page = self.list_conversations(offset=0, limit=5) if not page: return [drift_finding(DRIFT_WARN, "listing", "empty listing — cannot verify shape")] item = page[0] if not (item.get("uuid") or item.get("id")): findings.append(drift_finding(DRIFT_ERROR, "listing", "summary missing uuid/id")) if not (item.get("name") or item.get("title")): findings.append(drift_finding(DRIFT_WARN, "listing", "no name/title on summary")) if not (item.get("updated_at") or item.get("update_time")): findings.append(drift_finding(DRIFT_WARN, "listing", "no updated_at on summary")) conv_id = item.get("uuid") or item.get("id") if not conv_id: return findings or [drift_finding(DRIFT_ERROR, "listing", "no id to fetch")] raw = self.get_conversation(conv_id) if not (raw.get("uuid") or raw.get("id")): findings.append(drift_finding(DRIFT_ERROR, "detail", "no uuid/id on detail")) msgs = raw.get("chat_messages") or raw.get("messages") if not isinstance(msgs, list) or not msgs: findings.append(drift_finding( DRIFT_ERROR, "detail", "chat_messages missing/empty — 0 messages")) return findings saw_sender = saw_text = False flagged_list = flagged_attach = False for m in msgs: if m.get("sender") or m.get("role"): saw_sender = True content = m.get("content") if isinstance(content, list) and not flagged_list: findings.append(drift_finding( DRIFT_WARN, "content", "message.content is now a LIST — Claude switched to typed blocks; " "the dormant _dispatch_claude_block path is now live, verify rendering")) flagged_list = True if m.get("text") or content: saw_text = True if (m.get("attachments") or m.get("files")) and not flagged_attach: findings.append(drift_finding( DRIFT_WARN, "attachments", "message has non-empty attachments/files — the normalizer ignores " "these (silent loss); add handling")) flagged_attach = True if not saw_sender: findings.append(drift_finding(DRIFT_ERROR, "messages", "no sender/role on any message")) if not saw_text: findings.append(drift_finding(DRIFT_WARN, "messages", "no text/content found")) if not findings: findings.append(drift_finding(DRIFT_OK, "schema", "all load-bearing fields present")) return findings # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _map_role(sender: str) -> str | None: """Map Claude sender strings to normalized roles.""" mapping = { "human": "user", "user": "user", "assistant": "assistant", "claude": "assistant", "ai": "assistant", "system": "system", } return mapping.get(sender.lower()) if sender else None def _extract_claude_blocks( content: str | list | dict | None, conv_id: str, report: LossReport ) -> list[dict]: """Extract typed blocks from a Claude content field. Defensive dispatch — zero observed cases of rich Claude content in the user's archive at planning time, so this is theory-only. Real shapes will be locked in v0.4.1 once captured. Any unrecognised block type surfaces as an `unknown` block + WARNING + tally. """ if content is None: return [] if isinstance(content, str): block = make_text_block(content) return [block] if block else [] if isinstance(content, list): blocks: list[dict] = [] for item in content: if isinstance(item, str): block = make_text_block(item) if block: blocks.append(block) elif isinstance(item, dict): blocks.extend(_dispatch_claude_block(item, conv_id, report)) return blocks if isinstance(content, dict): return _dispatch_claude_block(content, conv_id, report) return [] def _dispatch_claude_block(block: dict, conv_id: str, report: LossReport) -> list[dict]: """Translate one raw Claude content block into normalized blocks.""" btype = block.get("type", "text") if btype == "text": block_obj = make_text_block(block.get("text", "") or "") return [block_obj] if block_obj else [] if btype == "thinking": # Claude extended-thinking blocks may use 'thinking' or 'text' field. text = block.get("thinking") or block.get("text") or "" block_obj = make_thinking_block(text) return [block_obj] if block_obj else [] if btype == "tool_use": return [ make_tool_use_block( name=block.get("name", "") or "", input_data=block.get("input"), tool_id=block.get("id"), ) ] if btype == "tool_result": # ``content`` may be a string or a list of nested blocks (recursive). nested = block.get("content") output = _flatten_tool_result_content(nested, conv_id, report) return [ make_tool_result_block( output=output, tool_name=None, is_error=bool(block.get("is_error")), ) ] if btype == "image": # Source shape is unverified; try the most likely fields. source = block.get("source") or {} ref = "" if isinstance(source, dict): ref = ( source.get("file_uuid") or source.get("media_type") or source.get("url") or "" ) return [make_image_placeholder(ref=ref or "(unknown)", source="user_upload")] # Unknown block type keys = list(block.keys()) logger.warning( "[claude] Unknown block type %r in conversation %s " "— see plan §Data-loss visibility (rendering as unknown block)", btype, conv_id[:8], ) report.record_unknown(btype or "?") return [ make_unknown_block( raw_type=btype or "?", observed_keys=keys, reason=UNKNOWN_REASON_UNKNOWN_TYPE, ) ] def _flatten_tool_result_content( nested: object, conv_id: str, report: LossReport ) -> str: """Flatten Claude tool_result content (string OR list of nested blocks) to text. Recurses into nested text blocks; any non-text nested block becomes a visible inline marker so non-text content isn't silently dropped. """ if nested is None: return "" if isinstance(nested, str): return nested if isinstance(nested, list): chunks: list[str] = [] for item in nested: if isinstance(item, str): chunks.append(item) elif isinstance(item, dict): btype = item.get("type", "text") if btype == "text": chunks.append(item.get("text", "") or "") else: keys = list(item.keys())[:10] report.record_extraction_failure(f"tool_result.{btype}") chunks.append( f"[Unsupported nested {btype} block; keys={keys}]" ) return "\n".join(c for c in chunks if c) if isinstance(nested, dict): return _flatten_tool_result_content([nested], conv_id, report) return str(nested)