_deepseek_export.py
· 10 KiB · Python
Неформатований
#!/usr/bin/env python3
"""
Export DeepSeek shared conversation to Markdown.
Usage (just run it — venv is created automatically):
python deepseek_export.py <share_url_or_id> [-o output.md] [--json] [--txt] [--token TOKEN]
Examples:
python deepseek_export.py https://chat.deepseek.com/share/nvy7v2ps1r6e2wyoyj
python deepseek_export.py nvy7v2ps1r6e2wyoyj -o my_chat.md
python deepseek_export.py nvy7v2ps1r6e2wyoyj --json -o chat.json
python deepseek_export.py nvy7v2ps1r6e2wyoyj --txt
First run will create .venv/ and install requests via uv.
"""
# ── auto-venv bootstrap ─────────────────────────────────────────────────────
import os
import subprocess
import sys
from pathlib import Path
_SCRIPT_DIR = Path(__file__).resolve().parent
_VENV_DIR = _SCRIPT_DIR / ".venv"
_VENV_PYTHON = _VENV_DIR / ("Scripts/python.exe" if os.name == "nt" else "bin/python")
def _ensure_venv():
"""Create venv + install deps if missing, then re-exec inside it."""
if sys.prefix != sys.base_prefix:
return # already inside a venv
if not _VENV_PYTHON.exists():
print("[boot] Creating .venv …", flush=True)
subprocess.check_call(["uv", "venv", str(_VENV_DIR)], cwd=_SCRIPT_DIR)
print("[boot] Installing dependencies …", flush=True)
subprocess.check_call(
["uv", "pip", "install", "--python", str(_VENV_PYTHON), "requests"],
cwd=_SCRIPT_DIR,
)
os.execv(str(_VENV_PYTHON), [str(_VENV_PYTHON), *sys.argv])
_ensure_venv()
# ── end bootstrap ────────────────────────────────────────────────────────────
import argparse
import json
import re
from datetime import datetime, timezone
import requests
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
SHARE_URL_RE = re.compile(
r"(?:https?://chat\.deepseek\.com)?/share/(?P<id>[A-Za-z0-9_-]+)"
)
def extract_share_id(raw: str) -> str:
m = SHARE_URL_RE.search(raw)
if m:
return m.group("id")
if re.fullmatch(r"[A-Za-z0-9_-]{6,}", raw):
return raw
raise ValueError(f"Cannot extract share id from: {raw}")
HEADERS = {
"accept": "application/json, text/plain, */*",
"accept-language": "en-US,en;q=0.9",
"referer": "https://chat.deepseek.com/",
"user-agent": (
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36"
),
}
# ---------------------------------------------------------------------------
# Strategy 1 — plain requests (works when WAF is not aggressive)
# ---------------------------------------------------------------------------
def fetch_via_requests(share_id: str, token: str | None = None) -> dict:
s = requests.Session()
s.headers.update(HEADERS)
s.get("https://chat.deepseek.com/", timeout=15)
if token:
s.headers["authorization"] = f"Bearer {token}"
url = f"https://chat.deepseek.com/api/v0/share/content?share_id={share_id}"
r = s.get(url, timeout=15)
r.raise_for_status()
data = r.json()
if data.get("code") != 0:
raise RuntimeError(f"API error: {data.get('msg', data)}")
return data
# ---------------------------------------------------------------------------
# Strategy 2 — Playwright headless browser (bypasses WAF reliably)
# ---------------------------------------------------------------------------
def fetch_via_playwright(share_id: str, token: str | None = None) -> dict:
try:
from playwright.sync_api import sync_playwright
except ImportError:
raise RuntimeError(
"playwright not installed. Run:\n"
" uv pip install playwright && uv run playwright install chromium"
)
captured = {}
def on_response(response):
if "/api/v0/share/content" in response.url:
try:
captured["data"] = response.json()
except Exception:
pass
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
ctx = browser.new_context(
user_agent=HEADERS["user-agent"],
viewport={"width": 1280, "height": 800},
)
page = ctx.new_page()
page.on("response", on_response)
if token:
ctx.add_cookies([{
"name": "ds_session_id",
"value": token,
"domain": "chat.deepseek.com",
"path": "/",
}])
page.goto(
f"https://chat.deepseek.com/share/{share_id}",
wait_until="networkidle",
timeout=30000,
)
page.wait_for_timeout(3000)
browser.close()
if not captured.get("data"):
raise RuntimeError("Playwright could not capture the share API response")
data = captured["data"]
if data.get("code") != 0:
raise RuntimeError(f"API error: {data.get('msg', data)}")
return data
# ---------------------------------------------------------------------------
# Main fetch (tries requests first, falls back to playwright)
# ---------------------------------------------------------------------------
def fetch_share(share_id: str, token: str | None = None) -> dict:
try:
print("[1/2] Trying direct API…", end=" ", flush=True)
data = fetch_via_requests(share_id, token)
print("OK")
return data
except Exception as e:
print(f"failed ({e})")
print("[2/2] Falling back to Playwright…", end=" ", flush=True)
try:
data = fetch_via_playwright(share_id, token)
print("OK")
return data
except Exception as e:
print(f"failed ({e})")
raise RuntimeError(
"Both strategies failed. If WAF is blocking you, install Playwright:\n"
" uv pip install playwright && uv run playwright install chromium\n"
"Or pass your token with --token."
)
# ---------------------------------------------------------------------------
# Formatting
# ---------------------------------------------------------------------------
def _ts(ts) -> str:
if not ts:
return ""
try:
return datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
except Exception:
return str(ts)
def to_markdown(data: dict) -> str:
biz = data.get("data", {}).get("biz_data", {})
messages = biz.get("messages", []) or biz.get("chat_messages", [])
title = biz.get("title") or "DeepSeek Conversation"
model = biz.get("model_type") or biz.get("model") or ""
parts = [f"# {title}\n"]
if model:
parts.append(f"**Model:** {model}")
parts.append("")
if not messages:
parts.append("_No messages found._\n")
return "\n".join(parts)
for msg in messages:
role = (msg.get("role") or "").upper()
content = msg.get("content") or msg.get("message") or ""
if not content and not role:
continue
thinking = msg.get("thinking_content") or msg.get("reasoning_content") or ""
elapsed = msg.get("thinking_elapsed_secs")
search = msg.get("search_results") or []
if role == "USER":
header = "You"
elif role in ("ASSISTANT", "AI"):
header = "DeepSeek"
elif role == "SYSTEM":
header = "System"
else:
header = role.title()
parts.append(f"## {header}\n")
if thinking:
label = f"Thinking ({elapsed}s)" if elapsed else "Thinking"
parts.append(f"<details><summary>{label}</summary>\n\n{thinking}\n\n</details>\n")
if search:
refs = []
for item in search:
t = item.get("title") or ""
u = item.get("url") or ""
if u:
refs.append(f"- [{t}]({u})" if t else f"- {u}")
if refs:
parts.append("**Sources:**\n" + "\n".join(refs) + "\n")
if content:
parts.append(f"{content}\n")
parts.append("---\n")
return "\n".join(parts)
def to_plain_text(data: dict) -> str:
biz = data.get("data", {}).get("biz_data", {})
messages = biz.get("messages", []) or biz.get("chat_messages", [])
title = biz.get("title") or "DeepSeek Conversation"
parts = [f"=== {title} ===\n"]
for msg in messages:
role = (msg.get("role") or "").upper()
content = msg.get("content") or msg.get("message") or ""
thinking = msg.get("thinking_content") or msg.get("reasoning_content") or ""
if role == "USER":
parts.append(f"[User]\n{content}\n")
elif role in ("ASSISTANT", "AI"):
if thinking:
parts.append(f"[Thinking]\n{thinking}\n")
parts.append(f"[DeepSeek]\n{content}\n")
else:
parts.append(f"[{role}]\n{content}\n")
return "\n".join(parts)
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main():
ap = argparse.ArgumentParser(description="Export a DeepSeek shared conversation")
ap.add_argument("share", help="Share URL or share ID")
ap.add_argument("-o", "--output", help="Output file path (default: auto-named)")
ap.add_argument("--json", action="store_true", help="Dump raw JSON instead of Markdown")
ap.add_argument("--txt", action="store_true", help="Dump plain text")
ap.add_argument("--token", help="Bearer token for auth (optional, needed for private shares)")
args = ap.parse_args()
share_id = extract_share_id(args.share)
print(f"Share ID: {share_id}")
data = fetch_share(share_id, args.token)
if args.output:
out = Path(args.output)
elif args.json:
out = Path(f"export_{share_id}.json")
elif args.txt:
out = Path(f"export_{share_id}.txt")
else:
out = Path(f"export_{share_id}.md")
if args.json:
content = json.dumps(data, indent=2, ensure_ascii=False)
elif args.txt:
content = to_plain_text(data)
else:
content = to_markdown(data)
out.write_text(content, encoding="utf-8")
print(f"Saved to {out} ({len(content)} bytes)")
if __name__ == "__main__":
main()
| 1 | #!/usr/bin/env python3 |
| 2 | """ |
| 3 | Export DeepSeek shared conversation to Markdown. |
| 4 | |
| 5 | Usage (just run it — venv is created automatically): |
| 6 | python deepseek_export.py <share_url_or_id> [-o output.md] [--json] [--txt] [--token TOKEN] |
| 7 | |
| 8 | Examples: |
| 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 | |
| 14 | First run will create .venv/ and install requests via uv. |
| 15 | """ |
| 16 | |
| 17 | # ── auto-venv bootstrap ───────────────────────────────────────────────────── |
| 18 | import os |
| 19 | import subprocess |
| 20 | import sys |
| 21 | from 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 | |
| 28 | def _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 | |
| 48 | import argparse |
| 49 | import json |
| 50 | import re |
| 51 | from datetime import datetime, timezone |
| 52 | |
| 53 | import requests |
| 54 | |
| 55 | # --------------------------------------------------------------------------- |
| 56 | # Helpers |
| 57 | # --------------------------------------------------------------------------- |
| 58 | |
| 59 | SHARE_URL_RE = re.compile( |
| 60 | r"(?:https?://chat\.deepseek\.com)?/share/(?P<id>[A-Za-z0-9_-]+)" |
| 61 | ) |
| 62 | |
| 63 | |
| 64 | def 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 | |
| 73 | HEADERS = { |
| 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 | |
| 88 | def 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 | |
| 110 | def 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 | |
| 165 | def 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 | |
| 192 | def _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 | |
| 201 | def 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 | |
| 259 | def 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 | |
| 284 | def 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 | |
| 318 | if __name__ == "__main__": |
| 319 | main() |
| 320 |