Ultima attività 8 hours ago

_deepseek_export.py Raw
1#!/usr/bin/env python3
2"""
3Export DeepSeek shared conversation to Markdown.
4
5Usage (just run it — venv is created automatically):
6 python deepseek_export.py <share_url_or_id> [-o output.md] [--json] [--txt] [--token TOKEN]
7
8Examples:
9 python deepseek_export.py https://chat.deepseek.com/share/nvy7v2ps1r6e2wyoyj
10 python deepseek_export.py nvy7v2ps1r6e2wyoyj -o my_chat.md
11 python deepseek_export.py nvy7v2ps1r6e2wyoyj --json -o chat.json
12 python deepseek_export.py nvy7v2ps1r6e2wyoyj --txt
13
14First run will create .venv/ and install requests via uv.
15"""
16
17# ── auto-venv bootstrap ─────────────────────────────────────────────────────
18import os
19import subprocess
20import sys
21from pathlib import Path
22
23_SCRIPT_DIR = Path(__file__).resolve().parent
24_VENV_DIR = _SCRIPT_DIR / ".venv"
25_VENV_PYTHON = _VENV_DIR / ("Scripts/python.exe" if os.name == "nt" else "bin/python")
26
27
28def _ensure_venv():
29 """Create venv + install deps if missing, then re-exec inside it."""
30 if sys.prefix != sys.base_prefix:
31 return # already inside a venv
32
33 if not _VENV_PYTHON.exists():
34 print("[boot] Creating .venv …", flush=True)
35 subprocess.check_call(["uv", "venv", str(_VENV_DIR)], cwd=_SCRIPT_DIR)
36 print("[boot] Installing dependencies …", flush=True)
37 subprocess.check_call(
38 ["uv", "pip", "install", "--python", str(_VENV_PYTHON), "requests"],
39 cwd=_SCRIPT_DIR,
40 )
41
42 os.execv(str(_VENV_PYTHON), [str(_VENV_PYTHON), *sys.argv])
43
44
45_ensure_venv()
46# ── end bootstrap ────────────────────────────────────────────────────────────
47
48import argparse
49import json
50import re
51from datetime import datetime, timezone
52
53import requests
54
55# ---------------------------------------------------------------------------
56# Helpers
57# ---------------------------------------------------------------------------
58
59SHARE_URL_RE = re.compile(
60 r"(?:https?://chat\.deepseek\.com)?/share/(?P<id>[A-Za-z0-9_-]+)"
61)
62
63
64def extract_share_id(raw: str) -> str:
65 m = SHARE_URL_RE.search(raw)
66 if m:
67 return m.group("id")
68 if re.fullmatch(r"[A-Za-z0-9_-]{6,}", raw):
69 return raw
70 raise ValueError(f"Cannot extract share id from: {raw}")
71
72
73HEADERS = {
74 "accept": "application/json, text/plain, */*",
75 "accept-language": "en-US,en;q=0.9",
76 "referer": "https://chat.deepseek.com/",
77 "user-agent": (
78 "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
79 "(KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36"
80 ),
81}
82
83
84# ---------------------------------------------------------------------------
85# Strategy 1 — plain requests (works when WAF is not aggressive)
86# ---------------------------------------------------------------------------
87
88def fetch_via_requests(share_id: str, token: str | None = None) -> dict:
89 s = requests.Session()
90 s.headers.update(HEADERS)
91 s.get("https://chat.deepseek.com/", timeout=15)
92
93 if token:
94 s.headers["authorization"] = f"Bearer {token}"
95
96 url = f"https://chat.deepseek.com/api/v0/share/content?share_id={share_id}"
97 r = s.get(url, timeout=15)
98 r.raise_for_status()
99 data = r.json()
100
101 if data.get("code") != 0:
102 raise RuntimeError(f"API error: {data.get('msg', data)}")
103 return data
104
105
106# ---------------------------------------------------------------------------
107# Strategy 2 — Playwright headless browser (bypasses WAF reliably)
108# ---------------------------------------------------------------------------
109
110def fetch_via_playwright(share_id: str, token: str | None = None) -> dict:
111 try:
112 from playwright.sync_api import sync_playwright
113 except ImportError:
114 raise RuntimeError(
115 "playwright not installed. Run:\n"
116 " uv pip install playwright && uv run playwright install chromium"
117 )
118
119 captured = {}
120
121 def on_response(response):
122 if "/api/v0/share/content" in response.url:
123 try:
124 captured["data"] = response.json()
125 except Exception:
126 pass
127
128 with sync_playwright() as p:
129 browser = p.chromium.launch(headless=True)
130 ctx = browser.new_context(
131 user_agent=HEADERS["user-agent"],
132 viewport={"width": 1280, "height": 800},
133 )
134 page = ctx.new_page()
135 page.on("response", on_response)
136
137 if token:
138 ctx.add_cookies([{
139 "name": "ds_session_id",
140 "value": token,
141 "domain": "chat.deepseek.com",
142 "path": "/",
143 }])
144
145 page.goto(
146 f"https://chat.deepseek.com/share/{share_id}",
147 wait_until="networkidle",
148 timeout=30000,
149 )
150 page.wait_for_timeout(3000)
151 browser.close()
152
153 if not captured.get("data"):
154 raise RuntimeError("Playwright could not capture the share API response")
155 data = captured["data"]
156 if data.get("code") != 0:
157 raise RuntimeError(f"API error: {data.get('msg', data)}")
158 return data
159
160
161# ---------------------------------------------------------------------------
162# Main fetch (tries requests first, falls back to playwright)
163# ---------------------------------------------------------------------------
164
165def fetch_share(share_id: str, token: str | None = None) -> dict:
166 try:
167 print("[1/2] Trying direct API…", end=" ", flush=True)
168 data = fetch_via_requests(share_id, token)
169 print("OK")
170 return data
171 except Exception as e:
172 print(f"failed ({e})")
173
174 print("[2/2] Falling back to Playwright…", end=" ", flush=True)
175 try:
176 data = fetch_via_playwright(share_id, token)
177 print("OK")
178 return data
179 except Exception as e:
180 print(f"failed ({e})")
181 raise RuntimeError(
182 "Both strategies failed. If WAF is blocking you, install Playwright:\n"
183 " uv pip install playwright && uv run playwright install chromium\n"
184 "Or pass your token with --token."
185 )
186
187
188# ---------------------------------------------------------------------------
189# Formatting
190# ---------------------------------------------------------------------------
191
192def _ts(ts) -> str:
193 if not ts:
194 return ""
195 try:
196 return datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
197 except Exception:
198 return str(ts)
199
200
201def to_markdown(data: dict) -> str:
202 biz = data.get("data", {}).get("biz_data", {})
203 messages = biz.get("messages", []) or biz.get("chat_messages", [])
204
205 title = biz.get("title") or "DeepSeek Conversation"
206 model = biz.get("model_type") or biz.get("model") or ""
207
208 parts = [f"# {title}\n"]
209 if model:
210 parts.append(f"**Model:** {model}")
211 parts.append("")
212
213 if not messages:
214 parts.append("_No messages found._\n")
215 return "\n".join(parts)
216
217 for msg in messages:
218 role = (msg.get("role") or "").upper()
219 content = msg.get("content") or msg.get("message") or ""
220 if not content and not role:
221 continue
222
223 thinking = msg.get("thinking_content") or msg.get("reasoning_content") or ""
224 elapsed = msg.get("thinking_elapsed_secs")
225 search = msg.get("search_results") or []
226
227 if role == "USER":
228 header = "You"
229 elif role in ("ASSISTANT", "AI"):
230 header = "DeepSeek"
231 elif role == "SYSTEM":
232 header = "System"
233 else:
234 header = role.title()
235
236 parts.append(f"## {header}\n")
237
238 if thinking:
239 label = f"Thinking ({elapsed}s)" if elapsed else "Thinking"
240 parts.append(f"<details><summary>{label}</summary>\n\n{thinking}\n\n</details>\n")
241
242 if search:
243 refs = []
244 for item in search:
245 t = item.get("title") or ""
246 u = item.get("url") or ""
247 if u:
248 refs.append(f"- [{t}]({u})" if t else f"- {u}")
249 if refs:
250 parts.append("**Sources:**\n" + "\n".join(refs) + "\n")
251
252 if content:
253 parts.append(f"{content}\n")
254 parts.append("---\n")
255
256 return "\n".join(parts)
257
258
259def to_plain_text(data: dict) -> str:
260 biz = data.get("data", {}).get("biz_data", {})
261 messages = biz.get("messages", []) or biz.get("chat_messages", [])
262 title = biz.get("title") or "DeepSeek Conversation"
263
264 parts = [f"=== {title} ===\n"]
265 for msg in messages:
266 role = (msg.get("role") or "").upper()
267 content = msg.get("content") or msg.get("message") or ""
268 thinking = msg.get("thinking_content") or msg.get("reasoning_content") or ""
269 if role == "USER":
270 parts.append(f"[User]\n{content}\n")
271 elif role in ("ASSISTANT", "AI"):
272 if thinking:
273 parts.append(f"[Thinking]\n{thinking}\n")
274 parts.append(f"[DeepSeek]\n{content}\n")
275 else:
276 parts.append(f"[{role}]\n{content}\n")
277 return "\n".join(parts)
278
279
280# ---------------------------------------------------------------------------
281# CLI
282# ---------------------------------------------------------------------------
283
284def main():
285 ap = argparse.ArgumentParser(description="Export a DeepSeek shared conversation")
286 ap.add_argument("share", help="Share URL or share ID")
287 ap.add_argument("-o", "--output", help="Output file path (default: auto-named)")
288 ap.add_argument("--json", action="store_true", help="Dump raw JSON instead of Markdown")
289 ap.add_argument("--txt", action="store_true", help="Dump plain text")
290 ap.add_argument("--token", help="Bearer token for auth (optional, needed for private shares)")
291 args = ap.parse_args()
292
293 share_id = extract_share_id(args.share)
294 print(f"Share ID: {share_id}")
295
296 data = fetch_share(share_id, args.token)
297
298 if args.output:
299 out = Path(args.output)
300 elif args.json:
301 out = Path(f"export_{share_id}.json")
302 elif args.txt:
303 out = Path(f"export_{share_id}.txt")
304 else:
305 out = Path(f"export_{share_id}.md")
306
307 if args.json:
308 content = json.dumps(data, indent=2, ensure_ascii=False)
309 elif args.txt:
310 content = to_plain_text(data)
311 else:
312 content = to_markdown(data)
313
314 out.write_text(content, encoding="utf-8")
315 print(f"Saved to {out} ({len(content)} bytes)")
316
317
318if __name__ == "__main__":
319 main()
320