mcncarl/yichen-skills

Repository: `https://github.com/mcncarl/agent-memory-vault`

git clone https://github.com/mcncarl/agent-memory-vault.git

[Credential-related source omitted]

只读：报告 Step、Step 国内站、豆包的脚本与凭证是否存在，以及引擎偏好与本地 Whisper 状态；不显示密钥，不调用付费接口，也不导入或运行已配置的执行器。`token_present`、`credential_present` 不代表账户有余额；豆包余额只能登录费用中心核验（账户概览 https://console.volcengine.com/finance/account-overview/ ，充值说明 https://www.volcengine.com/d

- 执行脚本：环境变量 `YICHEN_STEP_ASR_SCRIPT` 指向的兼容执行器；未设置时查找 `scripts/backends/step_asr_transcribe.py`。公开版不内置该脚本；统一入口 `scripts/transcribe.py --asr-engine stepfun` 调用它 - 接口：`https://api.stepfun.ai/v1/audio/asr/sse`（国际站）

- 接口：`https://api.stepfun.ai/v1/audio/asr/sse`（国际站）

[Credential-related source omitted]

- 执行脚本：环境变量 `YICHEN_STEP_ASR_TIMESTAMPS_SCRIPT` 指向的兼容执行器；未设置时查找 `scripts/backends/step_asr_timestamps.py`。公开版不内置该脚本 - 接口：`https://api.stepfun.com/v1/audio/asr/sse`，`enable_timestamp=true`；本地转 16 kHz PCM，base64 直传，按静音切成约 60 秒分片

- 接口：`https://api.stepfun.com/v1/audio/asr/sse`，`enable_timestamp=true`；本地转 16 kHz PCM，base64 直传，按静音切成约 60 秒分片

[Credential-related source omitted]

- 账户概览：https://console.volcengine.com/finance/account-overview/

- 充值说明：https://www.volcengine.com/docs/3019/1176800?lang=zh

- 豆包语音文档：https://www.volcengine.com/docs/6561/109885?lang=zh

- 只用 StepFun/Step ASR；用户要求“只用 step”时，不能用 Whisper、火山或其他模型补写。 - Endpoint：`https://api.stepfun.ai/v1/audio/asr/sse`；Model：`stepaudio-2.5-asr`。

- Endpoint：`https://api.stepfun.ai/v1/audio/asr/sse`；Model：`stepaudio-2.5-asr`。

[Credential-related source omitted]

- Endpoint：`https://api.stepfun.com/v1/audio/asr/sse`，默认模型 `stepaudio-2.5-asr`（0.15 元/小时）。`--model stepaudio-3-asr-max` 文字更准、2.8 元/小时，时间戳相同。

[Credential-related source omitted]

#!/usr/bin/env python3

DOUBAO_REFERENCE = SCRIPTS.parent / "references/doubao-asr.md" STEP_CN_KEY = "STEPFUN_CN_API_KEY" ACCOUNT_URL = "https://console.volcengine.com/finance/account-overview/"

[Credential-related source omitted]

ACCOUNT_URL = "https://console.volcengine.com/finance/account-overview/"

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

#!/usr/bin/env python3

[Credential-related source omitted]

[Credential-related source omitted]

#!/usr/bin/env python3

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

url = f"https://{TOS_BUCKET}.tos-{TOS_REGION}.volces.com/{filename}"

try:         resp = requests.put(url, data=file_data, headers=headers)

resp = requests.put(url, data=file_data, headers=headers)

print("上传成功!")             set_public_access(filename, access_key, secret_key)             return f"https://{TOS_BUCKET}.tos-{TOS_REGION}.volces.com/{filename}"

set_public_access(filename, access_key, secret_key)             return f"https://{TOS_BUCKET}.tos-{TOS_REGION}.volces.com/{filename}"         else:

return f"https://{TOS_BUCKET}.tos-{TOS_REGION}.volces.com/{filename}"

url = f"https://{TOS_BUCKET}.tos-{TOS_REGION}.volces.com/{filename}?acl"

<AccessControlList>         <Grant>             <Grantee xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:type="Group">

<Grant>             <Grantee xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:type="Group">                 <URI>http://acs.amazonaws.com/groups/global/AllUsers</URI>

<Grantee xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:type="Group">

<URI>http://acs.amazonaws.com/groups/global/AllUsers</URI>

try:         resp = requests.put(url, data=acl_xml, headers=headers)

resp = requests.put(url, data=acl_xml, headers=headers)

url = "https://openspeech.bytedance.com/api/v3/auc/bigmodel/submit"

[Credential-related source omitted]

resp = requests.post(url, headers=headers, json=data)

url = "https://openspeech.bytedance.com/api/v3/auc/bigmodel/query"

[Credential-related source omitted]

for i in range(max_attempts):         time.sleep(interval_seconds)         resp = requests.post(url, headers=headers, json=data)

resp = requests.post(url, headers=headers, json=data)

[Credential-related source omitted]

#!/usr/bin/env python3

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

#!/usr/bin/env python3

[Credential-related source omitted]

#!/usr/bin/env python3

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

#!/usr/bin/env python3

[Credential-related source omitted]

[Credential-related source omitted]

The X route calls [afar1/fieldtheory-cli](https://github.com/afar1/fieldtheory-cli) as an external runtime. No Field Theory source, binary, private Query ID, Cookie, credential, or bookmark data is bundled. The required `graphql-only` marke

| 小红书 | 已登录 Chrome 的个人主页 → 收藏 → 笔记 | 复用当前 Chrome 页面会话，不导出 Cookie | 保留 `xsec_token` 的完整笔记 URL | | 抖音 | 已登录 Chrome 的个人页 → 收藏 → 视频 | 复用当前 Chrome 页面会话，不导出 Cookie | 规范化 `/video/<id>` 与页面实际出现的 `/note/<id>` | | X | 本机 Field Theory 读取 Chrome 会话并调用内

| X | 本机 Field Theory 读取 Chrome 会话并调用内部 GraphQL Bookmarks | 只接受 GraphQL-only 版本 | `https://x.com/<author>/status/<id>` |

1. X 路线调用外部 [afar1/fieldtheory-cli](https://github.com/afar1/fieldtheory-cli)；上游采用 MIT License，许可证副本见 [../licenses/afar1-fieldtheory-cli-LICENSE.txt](../licenses/afar1-fieldtheory-cli-LICENSE.txt)。

https://www.douyin.com/user/self?showTab=favorite_collection

Executable file

const type = (match[1] || match[3]).toLowerCase();   const id = match[2] || match[4];   return { id, type, url: `https://www.douyin.com/${type}/${id}` };

const id = match[2] || match[4];   return { id, type, url: `https://www.douyin.com/${type}/${id}` }; }

return { id, type, url: `https://www.douyin.com/${type}/${id}` };

}         if (!id) continue;         const url = raw.startsWith("http") ? raw : `https://www.xiaohongshu.com${raw}`;

if (!id) continue;         const url = raw.startsWith("http") ? raw : `https://www.xiaohongshu.com${raw}`;         items.push({ id, url });

const url = raw.startsWith("http") ? raw : `https://www.xiaohongshu.com${raw}`;

const type = match[1].toLowerCase();         const id = match[2];         items.push({ id, type, url: `https://www.douyin.com/${type}/${id}` });

const id = match[2];         items.push({ id, type, url: `https://www.douyin.com/${type}/${id}` });       }

items.push({ id, type, url: `https://www.douyin.com/${type}/${id}` });

#!/usr/bin/env python3

def bookmark_url(item: dict) -> str | None:     url = item.get("url")     if isinstance(url, str) and url.startswith("https://x.com/") and "/status/" in url:

url = item.get("url")     if isinstance(url, str) and url.startswith("https://x.com/") and "/status/" in url:         return url.split("?", 1)[0]

if isinstance(url, str) and url.startswith("https://x.com/") and "/status/" in url:

author = str(item.get("authorHandle") or "_").strip().lstrip("@") or "_"     if tweet_id.isdigit():         return f"https://x.com/{author}/status/{tweet_id}"

if tweet_id.isdigit():         return f"https://x.com/{author}/status/{tweet_id}"     return None

return f"https://x.com/{author}/status/{tweet_id}"

#!/usr/bin/env python3

#!/usr/bin/env python3

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

- Claim the exact tab object returned by `openTabs()` with `browser.user.claimTab(...)`.    - Do not use `browser.tabs.new()` as the default first move. Opening a new tab can land in the wrong/free account.    - If no suitable ChatGPT tab i

- Do not use `browser.tabs.new()` as the default first move. Opening a new tab can land in the wrong/free account.    - If no suitable ChatGPT tab is available, ask the user to open `https://chatgpt.com/` in the already logged-in Chrome acc

- If no suitable ChatGPT tab is available, ask the user to open `https://chatgpt.com/` in the already logged-in Chrome account/profile, then continue by claiming that tab.

- If Pro status is not visible but the user explicitly requires it, stop and ask the user to make the correct Pro account/model visible. Do not treat an unconfirmed route as Pro. 4. Start a new chat inside the verified account only when nee

4. Start a new chat inside the verified account only when needed:    - Prefer the ChatGPT UI's `新聊天` control or `https://chatgpt.com/` in the already claimed/verified tab.    - Do not open a new browser window or new tab that could switch p

- Prefer the ChatGPT UI's `新聊天` control or `https://chatgpt.com/` in the already claimed/verified tab.

https://github.com/mcncarl/yichen-skills/blob/main/LICENSE

[XiaoDuoYa/codex-with-chatgpt](https://github.com/XiaoDuoYa/codex-with-chatgpt) explores a different Cloudflare/OAuth remote-MCP architecture. This Skill uses a separate Secure-Tunnel, read-only review protocol and vendors no source code fr

[XiaoDuoYa/codex-with-chatgpt](https://github.com/XiaoDuoYa/codex-with-chatgpt) 探索的是不同的 Cloudflare/OAuth 远程 MCP 架构。本 Skill 使用独立的 Secure Tunnel + 只读审查协议，没有复制或打包该仓库源码。

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

- a redacted URL reference such as `https://chatgpt.com/c/[redacted]` and capture time, never the full private conversation URL or ID;

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

- YouTube：本 Skill `youtube_known_url.py` 的 `info`、`download`、`subtitles` 和 `playlist`；只接受明确视频 URL 与用户自建播放列表（`PL`/`OLAK` 前缀），拒绝 RD（推荐 Mix）、UU（频道全部上传）等系统生成列表，不搜索或浏览频道。 - B站：`bili`（bili-cli）读取已知 BV/AV/完整 URL；下载走本 Skill `known_media_download.py

- B站：`bili`（bili-cli）读取已知 BV/AV/完整 URL；下载走本 Skill `known_media_download.py`，只接受完整 `www.bilibili.com/video/` 链接；已知播放列表用带 `--ignore-config`、`--no-cookies` 和数量上限的 `yt-dlp --flat-playlist` 平铺枚举，命令见平台路由。 - 小宇宙：`xiaoyuzhou_stepfun.py` 匿名读取/下载已知 e

- 小宇宙：`xiaoyuzhou_stepfun.py` 匿名读取/下载已知 episode，只接受 `https://www.xiaoyuzhoufm.com/episode/<id>`；默认下载后用 StepFun 转写（付费），`--download-only` 只下载不转写。首次转写前如果输出里有 `first_use_choice_available: true`，先问用户一次：StepFun（付费、快）还是本地 Whisper（免费、较慢，首次需 `uv too

# 默认只 Map，不 Crawl python3 ~/.agents/skills/yichen-content-archive/scripts/firecrawl_site.py \   --site-url "https://example.com/docs" --path-prefix "/docs" \

python3 ~/.agents/skills/yichen-content-archive/scripts/firecrawl_site.py    --site-url "https://example.com/docs" --path-prefix "/docs"

--site-url "https://example.com/docs" --path-prefix "/docs" \

# 只有用户明确要求执行，且复用完全相同的四个范围参数 python3 ~/.agents/skills/yichen-content-archive/scripts/firecrawl_site.py \   --site-url "https://example.com/docs" --path-prefix "/docs" \

python3 ~/.agents/skills/yichen-content-archive/scripts/firecrawl_site.py    --site-url "https://example.com/docs" --path-prefix "/docs"

--site-url "https://example.com/docs" --path-prefix "/docs" \

[Credential-related source omitted]

只放行笔记详情路径；xhslink 短链先匿名跟随跳转，再从最终 URL 取 note_id，跳转到验证页、404 或其他笔记即失败。`--use-cookie` 的 Cookie 只附在发往 `https://*.xiaohongshu.com` 的请求上，媒体 CDN 请求不带 Cookie；http 媒体链接升级为 https，其他协议拒绝。metadata 只保存规范笔记 URL 和去掉查询串的媒体/字幕地址，错误输出只保留主机名。媒体先写 `.part` 并核对大小

只接受微博公开单条 URL；把 `/u/<uid>?layerid=<id>` 规范为 `https://m.weibo.cn/detail/<id>` 后再调用 `yt-dlp`。固定忽略用户配置、禁用 Cookie 输入、禁止播放列表和覆盖，下载只选择同时含音视频的单文件 MP4；输出排他新 run 目录，不持久化临时签名媒体 URL。

[Credential-related source omitted]

入口只接受 `https://(www.)xiaoyuzhoufm.com/episode/<id>`，批量文件逐行同样校验并按 episode ID 去重；页面跳转必须停留在小宇宙 HTTPS 主机；音频只接受 `https://media.xyzcdn.net`。下载先写 `.part`，核对 Content-Length 一致后才改名并写 `source.json`，截断时不记录哈希。默认下载后用 StepFun 转写，`--download-only` 只下载；批量 

只接受 `https://mp.weixin.qq.com/s/<id>` 与 `/s?__biz=&mid=&idx=&sn=`（可带 `chksm`）单篇文章链接，其余参数一律丢弃，`key`、`uin`、`pass_ticket` 等不会出现在请求、输出或文件中；账号页、历史、合集、主页为 `unsupported_url`，非 HTTPS、其他主机、内嵌凭证或端口为 `invalid_url`。默认只匿名直读公开页，只跟随到文章链接的跳转；中转 down.mptext

| 平台 | 已知单项 `read` | 已知单项 `download/archive` | `known_collection` 精确枚举 | 明确禁止 | |---|---|---|---|---| | 普通网页 | Jina Reader `https://r.jina.ai/<URL>`；需图片/格式控制时用 Web Reader | 把同一已知 URL 的 Markdown、文本、HTML 或用户明确指定的原始 HTTP 响应写入新文件 | URL 文件逐行处理；不

|---|---|---|---|---| | 普通网页 | Jina Reader `https://r.jina.ai/<URL>`；需图片/格式控制时用 Web Reader | 把同一已知 URL 的 Markdown、文本、HTML 或用户明确指定的原始 HTTP 响应写入新文件 | URL 文件逐行处理；不从网页继续提取站内链接 | 无界站点爬取、搜索结果页扩展、相似链接 | | 公开 HTTPS 有界站点 | 不用于单页读取 | `firecrawl_site.

| 普通网页 | Jina Reader `https://r.jina.ai/<URL>`；需图片/格式控制时用 Web Reader | 把同一已知 URL 的 Markdown、文本、HTML 或用户明确指定的原始 HTTP 响应写入新文件 | URL 文件逐行处理；不从网页继续提取站内链接 | 无界站点爬取、搜索结果页扩展、相似链接 |

| 微信公众号 | 本 Skill `wechat_known_url.py read "<URL>"`：默认只匿名直读微信公开文章页（不带 Cookie），必须取到真实 `#js_content` 正文（图文页 item_show_type=8 例外，取页面数据里的描述文字与图片列表），验证/登录页、已删除或迁移页都算失败；中转 down.mptext.top 已于 2026-10-31 停服，`--relay-fallback` 仅保留为显式开关 | 同一脚本 `down

| YouTube | `python3 ~/.agents/skills/yichen-content-archive/scripts/youtube_known_url.py info <VIDEO_URL>` | 同一脚本的 `download`（视频或 `--audio-only`）与 `subtitles`（SRT + 保留时间戳 TXT） | `youtube_known_url.py playlist <PLAYLIST_URL> --limit N` 生成固定

| B站 | 用 `bili-cli` 的 `bili video <BV_OR_URL> --json` 读取 BV/完整 URL；AV 先规范成 `https://www.bilibili.com/video/av<ID>/` 再读取 | 本 Skill `known_media_download.py <https://www.bilibili.com/video/BV.../>`：匿名、忽略全局配置、不读 Cookie、`--no-playlist`、禁用 gener

[Credential-related source omitted]

#!/usr/bin/env python3

用法:     python3 douyin_download.py "<抖音链接>" [输出路径]     python3 douyin_download.py "https://www.douyin.com/video/7611845735025364265"

python3 douyin_download.py "<抖音链接>" [输出路径]     python3 douyin_download.py "https://www.douyin.com/video/7611845735025364265"     python3 douyin_download.py "<抖音链接>" "~/Downloads/my_video.mp4"

python3 douyin_download.py "https://www.douyin.com/video/7611845735025364265"

if not _plain_https(url) or not is_douyin_host(_host(url)):         raise ValueError("只接受 https://*.douyin.com 的已知视频链接")

raise ValueError("只接受 https://*.douyin.com 的已知视频链接")

if not video_id:         raise ValueError(DETAIL_URL_HINT)     return f"https://www.douyin.com/video/{video_id}"

return f"https://www.douyin.com/video/{video_id}"

current = url     for _ in range(MAX_REDIRECTS):         response = requests.get(

for _ in range(MAX_REDIRECTS):         response = requests.get(             current,

response = requests.get(

if not video_id:         raise ValueError(DETAIL_URL_HINT)     return f"https://www.douyin.com/video/{video_id}", video_id

return f"https://www.douyin.com/video/{video_id}", video_id

def download_video(video_url: str, output_path: str, referer: str = 'https://www.douyin.com/') -> str:

response = requests.get(video_url, headers=headers, stream=True, timeout=60)

#!/usr/bin/env python3

import hashlib import hmac import http.client

import hmac import http.client import ipaddress

import http.client

API_BASE = "https://api.firecrawl.dev/v2"

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

display_host = host     netloc = display_host if port in {None, 443} else f"{display_host}:{port}"     return host, f"https://{netloc}"

return host, f"https://{netloc}"

def redirect_request(         self,         request: urllib.request.Request,

self,         request: urllib.request.Request,         file_pointer: Any,

request: urllib.request.Request,

[Credential-related source omitted]

[Credential-related source omitted]

_validate_api_url(url)         data = None if payload is None else _canonical_json(payload)         request = urllib.request.Request(

data = None if payload is None else _canonical_json(payload)         request = urllib.request.Request(             url,

request = urllib.request.Request(

) from None             raise SiteError("firecrawl_http_error", f"Firecrawl API returned HTTP {exc.code}.") from None         except (OSError, http.client.HTTPException):

except (OSError, http.client.HTTPException):

# URLError and Python 3.9 socket.timeout are OSError subclasses; a bad             # status line or truncated body raises http.client.HTTPException.

# status line or truncated body raises http.client.HTTPException.

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

#!/usr/bin/env python3

import argparse import hashlib import http.client

import hashlib import http.client import importlib.util

import http.client

DOH_RESOLVERS = (     "https://dns.alidns.com/resolve",

"https://dns.alidns.com/resolve",

"https://cloudflare-dns.com/dns-query",

def doh_open(request: urllib.request.Request, timeout: float) -> Any:

def _doh_query(endpoint: str, host: str, record_type: int, version: int) -> list[str]:     query = urllib.parse.urlencode({"name": host, "type": record_type})     request = urllib.request.Request(

query = urllib.parse.urlencode({"name": host, "type": record_type})     request = urllib.request.Request(         f"{endpoint}?{query}",

request = urllib.request.Request(

)             body = response.read(DOH_MAX_BYTES + 1)     except (OSError, http.client.HTTPException, ValueError) as exc:

except (OSError, http.client.HTTPException, ValueError) as exc:

def download_x_file(url: str, destination: Path, *, timeout: int) -> None:     validate_x_media_url(url)     request = urllib.request.Request(

validate_x_media_url(url)     request = urllib.request.Request(         url,

request = urllib.request.Request(

# http.client returns a short body silently when the peer closes early.

except KnownMediaError:         raise     except (OSError, http.client.HTTPException) as exc:

except (OSError, http.client.HTTPException) as exc:

[Credential-related source omitted]

#!/usr/bin/env python3

[Credential-related source omitted]

import hashlib import html import http.client

import html import http.client import json

import http.client

RELAY_BACKEND = "mptext-html" RELAY_HOST = "down.mptext.top" RELAY_ENDPOINT = f"https://{RELAY_HOST}/api/public/v1/download"

RELAY_HOST = "down.mptext.top" RELAY_ENDPOINT = f"https://{RELAY_HOST}/api/public/v1/download" BROWSER_USER_AGENT = (

RELAY_ENDPOINT = f"https://{RELAY_HOST}/api/public/v1/download"

"User-Agent": BROWSER_USER_AGENT,     "Accept": "image/avif,image/webp,image/apng,image/*,*/*;q=0.8",     "Referer": f"https://{ARTICLE_HOST}/",

"Accept": "image/avif,image/webp,image/apng,image/*,*/*;q=0.8",     "Referer": f"https://{ARTICLE_HOST}/", }

"Referer": f"https://{ARTICLE_HOST}/",

) RELAY_ROOT_CANONICAL_RE = re.compile(     r"<link\s+rel=[\"']canonical[\"']\s+href=[\"']https://down\.mptext\.top/?[\"']",

RELAY_ROOT_CANONICAL_RE = re.compile(     r"<link\s+rel=[\"']canonical[\"']\s+href=[\"']https://down\.mptext\.top/?[\"']",     re.IGNORECASE,

r"<link\s+rel=[\"']canonical[\"']\s+href=[\"']https://down\.mptext\.top/?[\"']",

short_id = short.group("id")         return KnownArticle(             canonical_url=f"https://{ARTICLE_HOST}/s/{short_id}",

return KnownArticle(             canonical_url=f"https://{ARTICLE_HOST}/s/{short_id}",             form="short",

canonical_url=f"https://{ARTICLE_HOST}/s/{short_id}",

query += f"&chksm={chksm}"     return KnownArticle(         canonical_url=f"https://{ARTICLE_HOST}/s?{query}",

return KnownArticle(         canonical_url=f"https://{ARTICLE_HOST}/s?{query}",         form="long",

canonical_url=f"https://{ARTICLE_HOST}/s?{query}",

def open_request(opener: urllib.request.OpenerDirector, request: urllib.request.Request, timeout: int):

timeout: int, ) -> HttpResponse:     request = urllib.request.Request(url, headers=dict(headers), method="GET")

) -> HttpResponse:     request = urllib.request.Request(url, headers=dict(headers), method="GET")     try:

request = urllib.request.Request(url, headers=dict(headers), method="GET")

raise FetchError("timeout") from None         raise FetchError("network_error") from None     except (http.client.HTTPException, OSError):

raise FetchError("network_error") from None     except (http.client.HTTPException, OSError):         raise FetchError("network_error") from None

except (http.client.HTTPException, OSError):

request = urllib.request.Request(url, headers=dict(IMAGE_HEADERS), method="GET")

raise FetchError("image_timeout") from None         raise FetchError("image_network_error") from None     except (http.client.HTTPException, OSError):

raise FetchError("image_network_error") from None     except (http.client.HTTPException, OSError):         raise FetchError("image_network_error") from None

except (http.client.HTTPException, OSError):

#!/usr/bin/env python3

import argparse import http.client

import http.client import json import urllib.error

DEFAULT_API_BASE = "http://127.0.0.1:18901" ROOT_URL = DEFAULT_API_BASE + "/" RETIRED_COMMANDS = ("login", "search", "download")

def redirect_request(         self,         req: urllib.request.Request,

self,         req: urllib.request.Request,         fp: Any,

req: urllib.request.Request,

_NoRedirectHandler(),     )     request = urllib.request.Request(

)     request = urllib.request.Request(         ROOT_URL,

request = urllib.request.Request(

return True, exc.code, None     except (urllib.error.URLError, OSError, http.client.HTTPException) as exc:

except (urllib.error.URLError, OSError, http.client.HTTPException) as exc:

#!/usr/bin/env python3

[Credential-related source omitted]

post_id = layerids[0]         return KnownWeiboTarget(             canonical_url=f"https://m.weibo.cn/detail/{post_id}",

return KnownWeiboTarget(             canonical_url=f"https://m.weibo.cn/detail/{post_id}",             post_hint=post_id,

canonical_url=f"https://m.weibo.cn/detail/{post_id}",

post_id = mobile.group("post")         return KnownWeiboTarget(             canonical_url=f"https://m.weibo.cn/detail/{post_id}",

return KnownWeiboTarget(             canonical_url=f"https://m.weibo.cn/detail/{post_id}",             post_hint=post_id,

canonical_url=f"https://m.weibo.cn/detail/{post_id}",

return KnownWeiboTarget(             canonical_url=(                 f"https://weibo.com/{desktop.group('uid')}/{post_token}"

canonical_url=(                 f"https://weibo.com/{desktop.group('uid')}/{post_token}"             ),

f"https://weibo.com/{desktop.group('uid')}/{post_token}"

#!/usr/bin/env python3

import argparse import http.client

import http.client

FX_API = "https://api.fxtwitter.com/2"

JINA_READER = "https://r.jina.ai/http://x.com"

"handle": status_match.group("handle"),             "canonical_url": (                 f"https://x.com/{status_match.group('handle')}/status/"

"canonical_url": (                 f"https://x.com/{status_match.group('handle')}/status/"                 f"{status_match.group('id')}"

f"https://x.com/{status_match.group('handle')}/status/"

"id": web_status_match.group("id"),             "handle": None,             "canonical_url": f"https://x.com/i/status/{web_status_match.group('id')}",

"handle": None,             "canonical_url": f"https://x.com/i/status/{web_status_match.group('id')}",         }

"canonical_url": f"https://x.com/i/status/{web_status_match.group('id')}",

"id": article_match.group("id"),             "handle": None,             "canonical_url": f"https://x.com/i/article/{article_match.group('id')}",

"handle": None,             "canonical_url": f"https://x.com/i/article/{article_match.group('id')}",         }

"canonical_url": f"https://x.com/i/article/{article_match.group('id')}",

if query:         url = f"{url}?{urllib.parse.urlencode(query)}"     request = urllib.request.Request(

url = f"{url}?{urllib.parse.urlencode(query)}"     request = urllib.request.Request(         url,

request = urllib.request.Request(

)     try:         with urllib.request.urlopen(request, timeout=timeout) as response:

try:         with urllib.request.urlopen(request, timeout=timeout) as response:             body = read_limited(response, MAX_JSON_BYTES)

with urllib.request.urlopen(request, timeout=timeout) as response:

f"FxTwitter returned HTTP {exc.code}.",         ) from exc     except (OSError, http.client.HTTPException) as exc:

except (OSError, http.client.HTTPException) as exc:

)     if handle:         return f"https://x.com/{handle}/status/{status_id}"

if handle:         return f"https://x.com/{handle}/status/{status_id}"     return f"https://x.com/i/status/{status_id}"

return f"https://x.com/{handle}/status/{status_id}"

return f"https://x.com/i/status/{status_id}"

def fetch_jina(status_url: str, *, timeout: int) -> str:     request = urllib.request.Request(

request = urllib.request.Request(

)     try:         with urllib.request.urlopen(request, timeout=timeout) as response:

try:         with urllib.request.urlopen(request, timeout=timeout) as response:             body = read_limited(response, MAX_MARKDOWN_BYTES)

with urllib.request.urlopen(request, timeout=timeout) as response:

except urllib.error.HTTPError as exc:         raise KnownUrlError("jina_unavailable", "Jina Reader could not read the status.") from exc     except (OSError, http.client.HTTPException) as exc:

raise KnownUrlError("jina_unavailable", "Jina Reader could not read the status.") from exc     except (OSError, http.client.HTTPException) as exc:         raise transport_error(exc, "Jina Reader") from exc

except (OSError, http.client.HTTPException) as exc:

#!/usr/bin/env python3

默认先匿名请求网页，从 window.__INITIAL_STATE__ 中提取结构化元数据；失败时回退网页 meta。 不写入飞书；只有显式传入 --use-cookie 时才允许从 XHS_COOKIE 读取登录态，而且仍先匿名尝试。 Cookie 只附在发往 https://*.xiaohongshu.com 的请求上，媒体 CDN 请求永远匿名。

不写入飞书；只有显式传入 --use-cookie 时才允许从 XHS_COOKIE 读取登录态，而且仍先匿名尝试。 Cookie 只附在发往 https://*.xiaohongshu.com 的请求上，媒体 CDN 请求永远匿名。 只接受笔记详情链接与 xhslink 短链；短链先跟随跳转，再从最终详情 URL 提取 note_id。

Cookie 只附在发往 https://*.xiaohongshu.com 的请求上，媒体 CDN 请求永远匿名。

host = _host(url)     if not _plain_https(url) or not (is_xiaohongshu_host(host) or is_xhslink_host(host)):         raise FetchFailure("invalid_url", "只接受 https://*.xiaohongshu.com 或 https://*.xhslink.com 链接")

if not _plain_https(url) or not (is_xiaohongshu_host(host) or is_xhslink_host(host)):         raise FetchFailure("invalid_url", "只接受 https://*.xiaohongshu.com 或 https://*.xhslink.com 链接")     return url

raise FetchFailure("invalid_url", "只接受 https://*.xiaohongshu.com 或 https://*.xhslink.com 链接")

"""登录 Cookie 只能发往 https://*.xiaohongshu.com，绝不发给媒体 CDN 或其他主机。"""

[Credential-related source omitted]

def canonical_note_url(note_id: str) -> str:     return f"https://www.xiaohongshu.com/explore/{note_id}"

return f"https://www.xiaohongshu.com/explore/{note_id}"

current = url     for _ in range(MAX_REDIRECTS):         response = requests.get(

for _ in range(MAX_REDIRECTS):         response = requests.get(             current,

response = requests.get(

current = url     for _ in range(MAX_REDIRECTS + 1):         response = requests.get(

for _ in range(MAX_REDIRECTS + 1):         response = requests.get(             current,

response = requests.get(

[Credential-related source omitted]

if include_cookie:                     log(f"{label} 匿名下载失败，尝试使用已明确授权的登录态")                 response = requests.get(

log(f"{label} 匿名下载失败，尝试使用已明确授权的登录态")                 response = requests.get(                     url,

response = requests.get(

[Credential-related source omitted]

#!/usr/bin/env python3

[Credential-related source omitted]

[Credential-related source omitted]

write_text(         urls_path,         "".join(f"https://www.xiaoyuzhoufm.com/episode/{eid}\n" for eid in episode_ids),

urls_path,         "".join(f"https://www.xiaoyuzhoufm.com/episode/{eid}\n" for eid in episode_ids),         overwrite=overwrite,

"".join(f"https://www.xiaoyuzhoufm.com/episode/{eid}\n" for eid in episode_ids),

"container_kind": "xiaoyuzhou_podcast",         "container_id": podcast_id,         "container_ref": f"https://www.xiaoyuzhoufm.com/podcast/{podcast_id}",

"container_id": podcast_id,         "container_ref": f"https://www.xiaoyuzhoufm.com/podcast/{podcast_id}",         "requested_limit": limit,

"container_ref": f"https://www.xiaoyuzhoufm.com/podcast/{podcast_id}",

#!/usr/bin/env python3

[Credential-related source omitted]

[Credential-related source omitted]

if parts is None or parts[0] != "https" or parts[1] not in EPISODE_HOSTS:         raise ValueError(             "只接受 https://www.xiaoyuzhoufm.com/episode/<id> 形式的小宇宙 episode 链接"

raise ValueError(             "只接受 https://www.xiaoyuzhoufm.com/episode/<id> 形式的小宇宙 episode 链接"         )

"只接受 https://www.xiaoyuzhoufm.com/episode/<id> 形式的小宇宙 episode 链接"

raise ValueError("URL 不是小宇宙 episode 详情链接")     eid = match.group("eid")     return eid, f"https://www.xiaoyuzhoufm.com/episode/{eid}"

return eid, f"https://www.xiaoyuzhoufm.com/episode/{eid}"

parsed = urlparse(url.strip())         raise ValueError(             "音频地址不在允许范围（仅 https://media.xyzcdn.net 默认端口）: "

raise ValueError(             "音频地址不在允许范围（仅 https://media.xyzcdn.net 默认端口）: "             f"{parsed.scheme or '?'}://{parsed.hostname or '?'}"

"音频地址不在允许范围（仅 https://media.xyzcdn.net 默认端口）: "

normalized = html.unescape(page).replace("\\/", "/")     audio_patterns = (         r'https://media\.xyzcdn\.net/[^"\'<>\\ ]+?\.(?:m4a|mp3)(?:\?[^"\'<>\\ ]*)?',

audio_patterns = (         r'https://media\.xyzcdn\.net/[^"\'<>\\ ]+?\.(?:m4a|mp3)(?:\?[^"\'<>\\ ]*)?',         r'"enclosureUrl"\s*:\s*"([^"]+)"',

r'https://media\.xyzcdn\.net/[^"\'<>\\ ]+?\.(?:m4a|mp3)(?:\?[^"\'<>\\ ]*)?',

#!/usr/bin/env python3

[Credential-related source omitted]

if host == "youtu.be" and parts and VIDEO_ID_RE.fullmatch(parts[0]):         video_id = parts[0]         return "video", f"https://www.youtube.com/watch?v={video_id}", video_id

return "video", f"https://www.youtube.com/watch?v={video_id}", video_id

video_id = (query.get("v") or [None])[0]         if video_id and VIDEO_ID_RE.fullmatch(video_id):             return "video", f"https://www.youtube.com/watch?v={video_id}", video_id

return "video", f"https://www.youtube.com/watch?v={video_id}", video_id

video_id = parts[1]         if VIDEO_ID_RE.fullmatch(video_id):             return "video", f"https://www.youtube.com/watch?v={video_id}", video_id

return "video", f"https://www.youtube.com/watch?v={video_id}", video_id

[Credential-related source omitted]

"title": entry.get("title"),                 "channel": entry.get("channel") or entry.get("uploader"),                 "url": f"https://www.youtube.com/watch?v={video_id}",

"channel": entry.get("channel") or entry.get("uploader"),                 "url": f"https://www.youtube.com/watch?v={video_id}",             }

"url": f"https://www.youtube.com/watch?v={video_id}",

[Credential-related source omitted]

#!/usr/bin/env python3

import contextlib import hashlib import http.client

import hashlib import http.client import importlib.util

import http.client

[Credential-related source omitted]

def make_preflight(directory, *, links=None, limit=3, depth=1):     scope = SITE.make_scope("https://example.com/docs", "/docs", limit, depth)

scope = SITE.make_scope("https://example.com/docs", "/docs", limit, depth)

map_links=links         or [             "https://example.com/docs",

or [             "https://example.com/docs",             "https://example.com/docs/a?tracking=1",

"https://example.com/docs",

"https://example.com/docs/a?tracking=1",

[Credential-related source omitted]

class ScopeTests(unittest.TestCase):     def test_scope_is_public_https_bounded_and_seed_is_inside_prefix(self):         scope = SITE.make_scope("https://Example.com/docs/", "/docs/", 100, 3)

def test_scope_is_public_https_bounded_and_seed_is_inside_prefix(self):         scope = SITE.make_scope("https://Example.com/docs/", "/docs/", 100, 3)         self.assertEqual(scope.site_url, "https://example.com/docs")

scope = SITE.make_scope("https://Example.com/docs/", "/docs/", 100, 3)

self.assertEqual(scope.site_url, "https://example.com/docs")

self.assertEqual(scope.path_prefix, "/docs")         for url, prefix, limit, depth in (             ("http://example.com/docs", "/docs", 10, 1),

for url, prefix, limit, depth in (             ("http://example.com/docs", "/docs", 10, 1),             ("https://127.0.0.1/docs", "/docs", 10, 1),

("http://example.com/docs", "/docs", 10, 1),

("https://127.0.0.1/docs", "/docs", 10, 1),

("https://example.com/admin", "/docs", 10, 1),

("https://example.com/docs?login=1", "/docs", 10, 1),

("https://0x7f.0.0.1/docs", "/docs", 10, 1),

("https://127%2e0%2e0%2e1/docs", "/docs", 10, 1),

("https://%65xample.com/docs", "/docs", 10, 1),

("https://foo.internal/docs", "/docs", 10, 1),

("https://example.com:0/docs", "/docs", 10, 1),

("https://example.com/docs", "/docs", 101, 1),

("https://example.com/docs", "/docs", 10, 4),

with self.subTest(hostname=hostname):                 scope = SITE.make_scope(                     f"https://{hostname}/docs", "/docs", 10, 1

scope = SITE.make_scope(                     f"https://{hostname}/docs", "/docs", 10, 1                 )

f"https://{hostname}/docs", "/docs", 10, 1

self.assertEqual(scope.site_url, f"https://{expected}/docs")

with self.subTest(hostname=hostname):                 with self.assertRaises(SITE.SiteError):                     SITE.make_scope(f"https://{hostname}/docs", "/docs", 10, 1)

SITE.make_scope(f"https://{hostname}/docs", "/docs", 10, 1)

with mock.patch.object(SITE, "IDNA_UTS46", None):             ascii_scope = SITE.make_scope(                 "https://Example.com/docs", "/docs", 10, 1

ascii_scope = SITE.make_scope(                 "https://Example.com/docs", "/docs", 10, 1             )

"https://Example.com/docs", "/docs", 10, 1

)             self.assertEqual(ascii_scope.origin, "https://example.com")             with self.assertRaises(SITE.SiteError):

self.assertEqual(ascii_scope.origin, "https://example.com")

SITE.make_scope("https://Bücher.de/docs", "/docs", 10, 1)

SITE._policy_path(path, "test path")                 with self.assertRaises(SITE.SiteError):                     SITE.make_scope("https://example.com" + path, "/docs", 10, 1)

SITE.make_scope("https://example.com" + path, "/docs", 10, 1)

with self.subTest(path_prefix=path_prefix):                 with self.assertRaises(SITE.SiteError):                     SITE.make_scope("https://example.com", path_prefix, 10, 1)

with self.assertRaises(SITE.SiteError):                     SITE.make_scope("https://example.com", path_prefix, 10, 1)         scope = SITE.make_scope("https://example.com", "/", 10, 1)

SITE.make_scope("https://example.com", path_prefix, 10, 1)

scope = SITE.make_scope("https://example.com", "/", 10, 1)

self.assertEqual(scope.site_url, "https://example.com/")

def test_returned_urls_are_same_origin_and_path_and_drop_queries(self):         scope = SITE.make_scope("https://example.com/docs", "/docs", 10, 1)

scope = SITE.make_scope("https://example.com/docs", "/docs", 10, 1)

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

),             "https://example.com/docs/a",         )

"https://example.com/docs/a",

)         for value in (             "https://sub.example.com/docs/a",

for value in (             "https://sub.example.com/docs/a",             "https://example.com/admin",

"https://sub.example.com/docs/a",

"https://example.com/admin",

"http://example.com/docs/a",

"https://127.0.0.1/docs/a",

def test_matrix_parameter_dot_segments_fail_closed(self):         scope = SITE.make_scope("https://example.com/docs", "/docs", 10, 1)

scope = SITE.make_scope("https://example.com/docs", "/docs", 10, 1)

for value in (             "https://example.com/docs/..;/admin",             "https://example.com/docs/%2e%2e;/admin",

"https://example.com/docs/..;/admin",

"https://example.com/docs/%2e%2e;/admin",

"https://example.com/docs/..%3b/admin",

"https://example.com/docs/.;jsessionid=1/../admin",

"https://example.com/docs/..;jsessionid=1/admin",

"https://example.com/docs/.;x/a",

self.assertEqual(raised.exception.category, "outside_scope")         for site_url, prefix in (             ("https://example.com/docs/..;/x", "/docs"),

for site_url, prefix in (             ("https://example.com/docs/..;/x", "/docs"),             ("https://example.com/docs", "/docs/..;"),

("https://example.com/docs/..;/x", "/docs"),

("https://example.com/docs", "/docs/..;"),

self.assertEqual(             SITE.normalize_scoped_url("https://example.com/docs/a;b", scope),

SITE.normalize_scoped_url("https://example.com/docs/a;b", scope),

"https://example.com/docs/a;b",

cases = (             (                 "https://example.com/%E6%96%87%E6%A1%A3",

(                 "https://example.com/%E6%96%87%E6%A1%A3",                 "/%E6%96%87%E6%A1%A3",

"https://example.com/%E6%96%87%E6%A1%A3",

),             (                 "https://example.com/my%20docs",

(                 "https://example.com/my%20docs",                 "/my docs",

"https://example.com/my%20docs",

with self.subTest(prefix=prefix, path=path):                     self.assertFalse(re.search(pattern, path))         ascii_scope = SITE.make_scope("https://example.com/docs", "/docs", 5, 1)

self.assertFalse(re.search(pattern, path))         ascii_scope = SITE.make_scope("https://example.com/docs", "/docs", 5, 1)         self.assertEqual(

ascii_scope = SITE.make_scope("https://example.com/docs", "/docs", 5, 1)

def test_default_run_calls_map_only_filters_and_writes_signed_artifacts(self):         with tempfile.TemporaryDirectory() as directory:             scope = SITE.make_scope("https://example.com/docs", "/docs", 3, 2)

with tempfile.TemporaryDirectory() as directory:             scope = SITE.make_scope("https://example.com/docs", "/docs", 3, 2)             client = FakeClient(

scope = SITE.make_scope("https://example.com/docs", "/docs", 3, 2)

client = FakeClient(                 map_links=[                     "https://example.com/docs/a?tracking=1",

map_links=[                     "https://example.com/docs/a?tracking=1",                     "https://example.com/docs/a?tracking=2",

"https://example.com/docs/a?tracking=1",

"https://example.com/docs/a?tracking=2",

"https://sub.example.com/docs/b",

"https://example.com/outside",

{"url": "https://example.com/docs/c"},

[Credential-related source omitted]

client.map_payloads[0],                 {                     "url": "https://example.com/docs",

{                     "url": "https://example.com/docs",                     "limit": 3,

"url": "https://example.com/docs",

self.assertEqual(                 urls,                 ["https://example.com/docs/a", "https://example.com/docs/c"],

urls,                 ["https://example.com/docs/a", "https://example.com/docs/c"],             )

["https://example.com/docs/a", "https://example.com/docs/c"],

[Credential-related source omitted]

with tempfile.TemporaryDirectory() as directory:             scope = SITE.make_scope("https://example.com/docs", "/docs", 10, 1)

scope = SITE.make_scope("https://example.com/docs", "/docs", 10, 1)

client = FakeClient(                 map_links=[                     "https://example.com/docs/configuration",

map_links=[                     "https://example.com/docs/configuration",                     "https://example.com/docs/sitemap.xml",

"https://example.com/docs/configuration",

"https://example.com/docs/sitemap.xml",

"https://example.com/docs/robots.txt",

"https://example.com/docs/guide.pdf",

"https://example.com/docs/assets/logo.PNG",

"https://example.com/docs/assets/site.css",

"https://example.com/docs/assets/app.js?v=3",

"https://example.com/docs/feed.xml",

"https://example.com/docs/page.html",

"https://example.com/docs/v1.2",

"https://example.com/docs/.txt",

[Credential-related source omitted]

urls,                 [                     "https://example.com/docs/configuration",

[                     "https://example.com/docs/configuration",                     "https://example.com/docs/page.html",

"https://example.com/docs/configuration",

"https://example.com/docs/page.html",

"https://example.com/docs/v1.2",

"https://example.com/docs/.txt",

def test_depth_note_is_also_in_empty_and_failed_map_summaries(self):         with tempfile.TemporaryDirectory() as directory:             scope = SITE.make_scope("https://example.com/docs", "/docs", 5, 2)

with tempfile.TemporaryDirectory() as directory:             scope = SITE.make_scope("https://example.com/docs", "/docs", 5, 2)             result = SITE.run_preflight(

scope = SITE.make_scope("https://example.com/docs", "/docs", 5, 2)

result = SITE.run_preflight(                 scope,                 client=FakeClient(map_links=["https://example.com/docs/sitemap.xml"]),

[Credential-related source omitted]

client=FakeClient(map_links=["https://example.com/docs/sitemap.xml"]),

[Credential-related source omitted]

with tempfile.TemporaryDirectory() as directory:             scope = SITE.make_scope("https://example.com/docs", "/docs", 7, 2)

scope = SITE.make_scope("https://example.com/docs", "/docs", 7, 2)

[Credential-related source omitted]

[Credential-related source omitted]

def test_oversized_map_artifacts_fail_before_preflight_is_published(self):         with tempfile.TemporaryDirectory() as directory:             scope = SITE.make_scope("https://example.com/docs", "/docs", 1, 1)

with tempfile.TemporaryDirectory() as directory:             scope = SITE.make_scope("https://example.com/docs", "/docs", 1, 1)             client = FakeClient(

scope = SITE.make_scope("https://example.com/docs", "/docs", 1, 1)

client = FakeClient(                 map_links=["https://example.com/docs/" + ("a" * 800)]             )

map_links=["https://example.com/docs/" + ("a" * 800)]

[Credential-related source omitted]

(                 [                     "https://sub.example.com/docs/a",

[                     "https://sub.example.com/docs/a",                     "https://example.com/outside",

"https://sub.example.com/docs/a",

"https://example.com/outside",

for links, returned_count, rejected_count in cases:             with self.subTest(links=links), tempfile.TemporaryDirectory() as directory:                 scope = SITE.make_scope("https://example.com/docs", "/docs", 10, 1)

with self.subTest(links=links), tempfile.TemporaryDirectory() as directory:                 scope = SITE.make_scope("https://example.com/docs", "/docs", 10, 1)                 result = SITE.run_preflight(

scope = SITE.make_scope("https://example.com/docs", "/docs", 10, 1)

[Credential-related source omitted]

[Credential-related source omitted]

[                         "--site-url",                         "https://example.com/docs",

"--site-url",                         "https://example.com/docs",                         "--path-prefix",

"https://example.com/docs",

[Credential-related source omitted]

scope, _, result = make_preflight(directory)             pathlib.Path(result["enumerated_urls"]).write_text(                 "https://example.com/docs/changed\n", encoding="utf-8"

pathlib.Path(result["enumerated_urls"]).write_text(                 "https://example.com/docs/changed\n", encoding="utf-8"             )

"https://example.com/docs/changed\n", encoding="utf-8"

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

now=NOW + timedelta(minutes=1),                 )             mismatched = SITE.make_scope("https://example.com/docs", "/docs", 2, 1)

)             mismatched = SITE.make_scope("https://example.com/docs", "/docs", 2, 1)             with self.assertRaisesRegex(SITE.SiteError, "parameters"):

mismatched = SITE.make_scope("https://example.com/docs", "/docs", 2, 1)

[Credential-related source omitted]

{                         "markdown": "# Root\n\nSafe",                         "metadata": {"sourceURL": "https://example.com/docs"},

"markdown": "# Root\n\nSafe",                         "metadata": {"sourceURL": "https://example.com/docs"},                     },

"metadata": {"sourceURL": "https://example.com/docs"},

{                         "markdown": "# External",                         "metadata": {"sourceURL": "https://evil.example/docs"},

"markdown": "# External",                         "metadata": {"sourceURL": "https://evil.example/docs"},                     },

"metadata": {"sourceURL": "https://evil.example/docs"},

{                         "markdown": "# Not mapped",                         "metadata": {"sourceURL": "https://example.com/docs/new"},

"markdown": "# Not mapped",                         "metadata": {"sourceURL": "https://example.com/docs/new"},                     },

"metadata": {"sourceURL": "https://example.com/docs/new"},

[Credential-related source omitted]

[Credential-related source omitted]

directory,                 links=[                     "https://example.com/docs/a",

links=[                     "https://example.com/docs/a",                     "https://example.com/docs/b",

"https://example.com/docs/a",

"https://example.com/docs/b",

{                         "markdown": "",                         "metadata": {"sourceURL": "https://example.com/docs/a"},

"markdown": "",                         "metadata": {"sourceURL": "https://example.com/docs/a"},                     },

"metadata": {"sourceURL": "https://example.com/docs/a"},

{                         "markdown": "12345",                         "metadata": {"sourceURL": "https://example.com/docs/b"},

"markdown": "12345",                         "metadata": {"sourceURL": "https://example.com/docs/b"},                     },

"metadata": {"sourceURL": "https://example.com/docs/b"},

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

SITE.parse_args(                     [                         "--site-url", "https://example.com/docs",

[                         "--site-url", "https://example.com/docs",                         "--path-prefix", "/docs",

"--site-url", "https://example.com/docs",

SITE.parse_args(                     [                         "--site-url", "https://example.com/docs",

[                         "--site-url", "https://example.com/docs",                         "--path-prefix", "/docs",

"--site-url", "https://example.com/docs",

[Credential-related source omitted]

scope, _, preflight = make_preflight(                 directory,                 links=["https://example.com/docs", "https://example.com/docs/deep/page"],

directory,                 links=["https://example.com/docs", "https://example.com/docs/deep/page"],                 depth=0,

links=["https://example.com/docs", "https://example.com/docs/deep/page"],

client = FakeClient(                 crawl_documents=[                     {"markdown": "# Root", "metadata": {"sourceURL": "https://example.com/docs"}}

crawl_documents=[                     {"markdown": "# Root", "metadata": {"sourceURL": "https://example.com/docs"}}                 ]

{"markdown": "# Root", "metadata": {"sourceURL": "https://example.com/docs"}}

def test_final_url_and_status_code_must_pass_before_archiving(self):         links = [             "https://example.com/docs",

links = [             "https://example.com/docs",             "https://example.com/docs/a",

"https://example.com/docs",

"https://example.com/docs/a",

"https://example.com/docs/b",

"https://example.com/docs/c",

"https://example.com/docs/d",

"markdown": "offsite body",                 "metadata": {                     "sourceURL": "https://example.com/docs",

"metadata": {                     "sourceURL": "https://example.com/docs",                     "url": "https://other-origin.example.net/landing",

"sourceURL": "https://example.com/docs",

"url": "https://other-origin.example.net/landing",

"markdown": "not found page",                 "metadata": {                     "sourceURL": "https://example.com/docs/a",

"metadata": {                     "sourceURL": "https://example.com/docs/a",                     "url": "https://example.com/docs/a",

"sourceURL": "https://example.com/docs/a",

"url": "https://example.com/docs/a",

"markdown": "redirected inside scope",                 "metadata": {                     "sourceURL": "https://example.com/docs/b",

"metadata": {                     "sourceURL": "https://example.com/docs/b",                     "url": "https://example.com/docs/b/?utm=1",

"sourceURL": "https://example.com/docs/b",

"url": "https://example.com/docs/b/?utm=1",

},             },             {"markdown": "no final url or status", "metadata": {"sourceURL": "https://example.com/docs/c"}},

},             {"markdown": "no final url or status", "metadata": {"sourceURL": "https://example.com/docs/c"}},             {

{"markdown": "no final url or status", "metadata": {"sourceURL": "https://example.com/docs/c"}},

{                 "markdown": "odd status",                 "metadata": {"sourceURL": "https://example.com/docs/d", "statusCode": "OK"},

"markdown": "odd status",                 "metadata": {"sourceURL": "https://example.com/docs/d", "statusCode": "OK"},             },

"metadata": {"sourceURL": "https://example.com/docs/d", "statusCode": "OK"},

for path in (run_dir / "web").glob("*/metadata.json")             }             redirected = metadata["https://example.com/docs/b"]

}             redirected = metadata["https://example.com/docs/b"]             self.assertEqual(redirected["final_url"], "https://example.com/docs/b")

redirected = metadata["https://example.com/docs/b"]

self.assertEqual(redirected["final_url"], "https://example.com/docs/b")

self.assertTrue(redirected["final_url_reported"])             self.assertTrue(redirected["status_code_reported"])             unverified = metadata["https://example.com/docs/c"]

self.assertTrue(redirected["status_code_reported"])             unverified = metadata["https://example.com/docs/c"]             self.assertIsNone(unverified["final_url"])

unverified = metadata["https://example.com/docs/c"]

scope, _, preflight = make_preflight(directory)             client = FakeClient(                 crawl_documents=[{"markdown": "# Root", "metadata": {"sourceURL": "https://example.com/docs"}}]

client = FakeClient(                 crawl_documents=[{"markdown": "# Root", "metadata": {"sourceURL": "https://example.com/docs"}}]             )

crawl_documents=[{"markdown": "# Root", "metadata": {"sourceURL": "https://example.com/docs"}}]

"status": "completed",             "data": self.crawl_documents,             "next": "https://api.firecrawl.dev/v2/crawl/job-safe-1?skip=1",

"data": self.crawl_documents,             "next": "https://api.firecrawl.dev/v2/crawl/job-safe-1?skip=1",         }

"next": "https://api.firecrawl.dev/v2/crawl/job-safe-1?skip=1",

def doc(path):     return {"markdown": f"# {path}", "metadata": {"sourceURL": f"https://example.com{path}"}}

return {"markdown": f"# {path}", "metadata": {"sourceURL": f"https://example.com{path}"}}

class CrawlPaginationTests(unittest.TestCase):     def test_pagination_failure_keeps_returned_pages_as_partial(self):         links = ["https://example.com/docs", "https://example.com/docs/a", "https://example.com/docs/b"]

def test_pagination_failure_keeps_returned_pages_as_partial(self):         links = ["https://example.com/docs", "https://example.com/docs/a", "https://example.com/docs/b"]         client = PagedClient(

links = ["https://example.com/docs", "https://example.com/docs/a", "https://example.com/docs/b"]

client = PagedClient(             [                 {"data": [doc("/docs/a")], "next": "https://api.firecrawl.dev/v2/crawl/job-safe-1?skip=2"},

[                 {"data": [doc("/docs/a")], "next": "https://api.firecrawl.dev/v2/crawl/job-safe-1?skip=2"},                 SITE.SiteError("firecrawl_unavailable", "Firecrawl API could not be reached."),

{"data": [doc("/docs/a")], "next": "https://api.firecrawl.dev/v2/crawl/job-safe-1?skip=2"},

def endless(count):             return [                 {"data": [doc(f"/docs/p{index}")], "next": f"https://api.firecrawl.dev/v2/crawl/job-safe-1?skip={index + 2}"}

return [                 {"data": [doc(f"/docs/p{index}")], "next": f"https://api.firecrawl.dev/v2/crawl/job-safe-1?skip={index + 2}"}                 for index in range(count)

{"data": [doc(f"/docs/p{index}")], "next": f"https://api.firecrawl.dev/v2/crawl/job-safe-1?skip={index + 2}"}

def open(self, request, timeout):         self.requests.append(request)

self.requests.append(request)

[Credential-related source omitted]

(request,) = opener.requests         self.assertEqual(request.get_method(), "DELETE")         self.assertEqual(request.full_url, "https://api.firecrawl.dev/v2/crawl/job-safe-1")

self.assertEqual(request.get_method(), "DELETE")         self.assertEqual(request.full_url, "https://api.firecrawl.dev/v2/crawl/job-safe-1")         with self.assertRaises(SITE.SiteError):

self.assertEqual(request.full_url, "https://api.firecrawl.dev/v2/crawl/job-safe-1")

cases = (             RecordingOpener(error=socket.timeout("timed out")),             RecordingOpener(error=http.client.BadStatusLine("junk")),

RecordingOpener(error=socket.timeout("timed out")),             RecordingOpener(error=http.client.BadStatusLine("junk")),             RecordingOpener(error=urllib.error.URLError(ConnectionRefusedError())),

RecordingOpener(error=http.client.BadStatusLine("junk")),             RecordingOpener(error=urllib.error.URLError(ConnectionRefusedError())),             RecordingOpener(read_error=http.client.IncompleteRead(b"{")),

RecordingOpener(error=urllib.error.URLError(ConnectionRefusedError())),             RecordingOpener(read_error=http.client.IncompleteRead(b"{")),             RecordingOpener(read_error=socket.timeout("timed out")),

RecordingOpener(read_error=http.client.IncompleteRead(b"{")),             RecordingOpener(read_error=socket.timeout("timed out")),         )

[Credential-related source omitted]

[Credential-related source omitted]

def test_api_url_gate(self):         for url in (             "https://api.firecrawl.dev/v2/crawl/job-1",

for url in (             "https://api.firecrawl.dev/v2/crawl/job-1",             "https://api.firecrawl.dev/v2/crawl/job-1?skip=10",

"https://api.firecrawl.dev/v2/crawl/job-1",

"https://api.firecrawl.dev/v2/crawl/job-1?skip=10",

"https://api.firecrawl.dev:443/v2/map",

SITE._validate_api_url(url)         for url in (             "http://api.firecrawl.dev/v2/crawl/job-1",

for url in (             "http://api.firecrawl.dev/v2/crawl/job-1",             "https://api.firecrawl.dev.evil.example/v2/crawl/job-1",

"http://api.firecrawl.dev/v2/crawl/job-1",

"https://api.firecrawl.dev.evil.example/v2/crawl/job-1",

"https://api.firecrawl.dev:8443/v2/crawl/job-1",

"https://user:pw@api.firecrawl.dev/v2/crawl/job-1",

"https://api.firecrawl.dev/v2/crawl/job-1#frag",

"https://api.firecrawl.dev/v1/crawl/job-1",

"https://api.firecrawl.dev/v2/../v1/crawl",

"https://api.firecrawl.dev/v2/%2e%2e/v1/crawl",

"https://api.firecrawl.dev/v2/crawl/%2fjob",

"https://api.firecrawl.dev/v2/crawl\\job",

"https://api.firecrawl.dev:bad/v2/crawl",

[Credential-related source omitted]

[Credential-related source omitted]

client.next_page("https://api.firecrawl.dev/v2/map?cursor=1")

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

def test_authenticated_firecrawl_requests_refuse_redirects(self):         request = urllib.request.Request(

request = urllib.request.Request(

"Found",                 {},                 "https://attacker.invalid/capture",

{},                 "https://attacker.invalid/capture",             )

"https://attacker.invalid/capture",

302,                     "Found",                     {"Location": "https://attacker.invalid/capture"},

"Found",                     {"Location": "https://attacker.invalid/capture"},                     None,

{"Location": "https://attacker.invalid/capture"},

[Credential-related source omitted]

[Credential-related source omitted]

with self.assertRaises(SITE.SiteError) as raised:             client.map_site({"url": "https://example.com", "limit": 1})         self.assertEqual(raised.exception.category, "firecrawl_redirect_refused")

client.map_site({"url": "https://example.com", "limit": 1})

[Credential-related source omitted]

[Credential-related source omitted]

#!/usr/bin/env python3

import contextlib import email.message import http.client

import email.message import http.client import importlib.util

import http.client

self.assertEqual(             MEDIA.validate_public_https_url(                 "https://8.8.8.8/video.mp4", resolve_dns=False

MEDIA.validate_public_https_url(                 "https://8.8.8.8/video.mp4", resolve_dns=False             ),

"https://8.8.8.8/video.mp4", resolve_dns=False

),             "https://8.8.8.8/video.mp4",         )

"https://8.8.8.8/video.mp4",

with self.assertRaises(MEDIA.KnownMediaError):             MEDIA.validate_public_https_url(                 "https://127.0.0.1/video.mp4", resolve_dns=False

MEDIA.validate_public_https_url(                 "https://127.0.0.1/video.mp4", resolve_dns=False             )

"https://127.0.0.1/video.mp4", resolve_dns=False

with self.assertRaises(MEDIA.KnownMediaError):             MEDIA.validate_public_https_url(                 "https://user:password@example.com/video", resolve_dns=False

MEDIA.validate_public_https_url(                 "https://user:password@example.com/video", resolve_dns=False             )

"https://user:password@example.com/video", resolve_dns=False

{                         "id": "123",                         "url": "https://video.twimg.com/fallback.mp4",

"id": "123",                         "url": "https://video.twimg.com/fallback.mp4",                         "width": 1280,

"url": "https://video.twimg.com/fallback.mp4",

"formats": [                             {                                 "url": "https://video.twimg.com/low.mp4",

{                                 "url": "https://video.twimg.com/low.mp4",                                 "container": "mp4",

"url": "https://video.twimg.com/low.mp4",

},                             {                                 "url": "https://video.twimg.com/high.mp4",

{                                 "url": "https://video.twimg.com/high.mp4",                                 "container": "mp4",

"url": "https://video.twimg.com/high.mp4",

},                             {                                 "url": "https://video.twimg.com/master.m3u8",

{                                 "url": "https://video.twimg.com/master.m3u8",                                 "container": "m3u8",

"url": "https://video.twimg.com/master.m3u8",

selected = MEDIA.select_x_videos(content)         self.assertEqual(len(selected), 1)         self.assertEqual(selected[0]["url"], "https://video.twimg.com/high.mp4")

self.assertEqual(len(selected), 1)         self.assertEqual(selected[0]["url"], "https://video.twimg.com/high.mp4")         self.assertEqual(selected[0]["bitrate"], 2176000)

self.assertEqual(selected[0]["url"], "https://video.twimg.com/high.mp4")

"formats": [                             {                                 "url": "https://example.com/video.mp4",

{                                 "url": "https://example.com/video.mp4",                                 "container": "mp4",

"url": "https://example.com/video.mp4",

[Credential-related source omitted]

), mock.patch.object(reader, "read_known_url", return_value=no_video):             for url in (                 "https://x.com/example/status/2000000000000000001",

for url in (                 "https://x.com/example/status/2000000000000000001",                 "https://www.x.com/example/status/2000000000000000001",

"https://x.com/example/status/2000000000000000001",

"https://www.x.com/example/status/2000000000000000001",

"https://twitter.com/i/status/2000000000000000001",

"https://mobile.twitter.com/example/status/2000000000000000001",

self.assertEqual(raised.exception.category, "no_public_video")             for url, category in (                 ("http://x.com/example/status/1", "unsupported_url"),

for url, category in (                 ("http://x.com/example/status/1", "unsupported_url"),                 ("https://user:secret@x.com/example/status/1", "unsupported_url"),

("http://x.com/example/status/1", "unsupported_url"),

("https://user:secret@x.com/example/status/1", "unsupported_url"),

("https://x.com:8443/example/status/1", "unsupported_port"),

):             for url in (                 "https://www.bilibili.com/video/BV1xx411c7mD",

for url in (                 "https://www.bilibili.com/video/BV1xx411c7mD",                 "https://x.com.evil.example/example/status/1",

"https://www.bilibili.com/video/BV1xx411c7mD",

"https://x.com.evil.example/example/status/1",

):                 with self.assertRaises(MEDIA.KnownMediaError) as raised:                     MEDIA.run("https://www.bilibili.com/video/BV1Ab411c7Xy/", output_dir=None, timeout=10)

with self.assertRaises(MEDIA.KnownMediaError) as raised:                     MEDIA.run("https://www.bilibili.com/video/BV1Ab411c7Xy/", output_dir=None, timeout=10)                 self.assertEqual(raised.exception.category, "non_public_host

MEDIA.run("https://www.bilibili.com/video/BV1Ab411c7Xy/", output_dir=None, timeout=10)

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

"https://example.com/watch/1",

body=b"\x00\x00\x00\x18ftypmp42",

content_type="video/mp4",         content_length="auto",         final_url="https://video.twimg.com/ext_tw_video/1/pu/vid/a.mp4",

content_length="auto",         final_url="https://video.twimg.com/ext_tw_video/1/pu/vid/a.mp4",         read_error=None,

final_url="https://video.twimg.com/ext_tw_video/1/pu/vid/a.mp4",

def open(self, request, timeout):         self.requests.append(request)

self.requests.append(request)

X_VIDEO_URL = "https://video.twimg.com/ext_tw_video/1/pu/vid/a.mp4"

self.assertEqual(destination.read_bytes(), b"\x00\x00\x00\x18ftypmp42")

with self.assertRaises(MEDIA.KnownMediaError) as raised:             handler.redirect_request(                 None, None, 302, "Found", {}, "https://evil.example/a.mp4"

handler.redirect_request(                 None, None, 302, "Found", {}, "https://evil.example/a.mp4"             )

None, None, 302, "Found", {}, "https://evil.example/a.mp4"

self.assertEqual(raised.exception.category, "unsafe_media_url")         self.assert_download_error(             FakeOpener(FakeMediaResponse(final_url="http://video.twimg.com/a.mp4")),

self.assert_download_error(             FakeOpener(FakeMediaResponse(final_url="http://video.twimg.com/a.mp4")),             "unsafe_media_url",

FakeOpener(FakeMediaResponse(final_url="http://video.twimg.com/a.mp4")),

)         with self.assertRaises(MEDIA.KnownMediaError):             MEDIA.download_x_file("https://pbs.twimg.com/a.mp4", pathlib.Path("unused"), timeout=5)

MEDIA.download_x_file("https://pbs.twimg.com/a.mp4", pathlib.Path("unused"), timeout=5)

(FakeOpener(error=socket.timeout("timed out")), "timeout"),             (FakeOpener(error=urllib.error.URLError(socket.timeout("timed out"))), "timeout"),             (FakeOpener(error=http.client.BadStatusLine("junk")), "download_failed"),

(FakeOpener(error=urllib.error.URLError(socket.timeout("timed out"))), "timeout"),             (FakeOpener(error=http.client.BadStatusLine("junk")), "download_failed"),             (FakeOpener(error=ConnectionResetError()), "download_failed

(FakeOpener(error=http.client.BadStatusLine("junk")), "download_failed"),

(FakeOpener(error=ConnectionResetError()), "download_failed"),             (                 FakeOpener(FakeMediaResponse(read_error=http.client.IncompleteRead(b"x"))),

(                 FakeOpener(FakeMediaResponse(read_error=http.client.IncompleteRead(b"x"))),                 "download_failed",

FakeOpener(FakeMediaResponse(read_error=http.client.IncompleteRead(b"x"))),

), mock.patch.object(MEDIA.Path, "home", return_value=pathlib.Path(home)):             with self.assertRaises(MEDIA.KnownMediaError) as raised:                 MEDIA.run("https://x.com/a/status/2000000000000000001", output_dir=None, timeout

with self.assertRaises(MEDIA.KnownMediaError) as raised:                 MEDIA.run("https://x.com/a/status/2000000000000000001", output_dir=None, timeout=10)             self.assertEqual(raised.exception.category, "rate_limited")

MEDIA.run("https://x.com/a/status/2000000000000000001", output_dir=None, timeout=10)

), mock.patch.object(MEDIA, "download_x_file") as download:                 with self.assertRaises(MEDIA.KnownMediaError) as raised:                     MEDIA.run("https://x.com/i/article/2032093606551707648", output_dir=None, timeout=10)

with self.assertRaises(MEDIA.KnownMediaError) as raised:                     MEDIA.run("https://x.com/i/article/2032093606551707648", output_dir=None, timeout=10)                 self.assertEqual(raised.exception.category, "invalid_response

MEDIA.run("https://x.com/i/article/2032093606551707648", output_dir=None, timeout=10)

result = x_video_result(X_VIDEO_URL, "https://video.twimg.com/2.mp4")

), mock.patch.object(MEDIA, "download_x_file", side_effect=flaky):             output = pathlib.Path(root) / "out"             run = MEDIA.run("https://x.com/a/status/2000000000000000001", output_dir=str(output), timeout=10)

output = pathlib.Path(root) / "out"             run = MEDIA.run("https://x.com/a/status/2000000000000000001", output_dir=str(output), timeout=10)             self.assertEqual(run["status"], "partial")

run = MEDIA.run("https://x.com/a/status/2000000000000000001", output_dir=str(output), timeout=10)

stdout = io.StringIO()             with mock.patch.object(                 sys, "argv", ["known_media_download.py", "https://x.com/a/status/2000000000000000001", "--output-dir", str(output)]

with mock.patch.object(                 sys, "argv", ["known_media_download.py", "https://x.com/a/status/2000000000000000001", "--output-dir", str(output)]             ), contextlib.redirect_stdout(stdout):

sys, "argv", ["known_media_download.py", "https://x.com/a/status/2000000000000000001", "--output-dir", str(output)]

"title": "clip",         "extractor": "BiliBili",         "webpage_url": "https://www.bilibili.com/video/BV1xx411c7mD",

"extractor": "BiliBili",         "webpage_url": "https://www.bilibili.com/video/BV1xx411c7mD",         "format_id": "30080+30280",

"webpage_url": "https://www.bilibili.com/video/BV1xx411c7mD",

"format_id": "30080+30280",         "formats": [             {"format_id": "30016", "url": "http://cdn.example.com/low.m4s"},

"formats": [             {"format_id": "30016", "url": "http://cdn.example.com/low.m4s"},             {"format_id": "30080", "url": "https://upos.example.com:8082/v.m4s", "filesize": 1000},

{"format_id": "30016", "url": "http://cdn.example.com/low.m4s"},

{"format_id": "30080", "url": "https://upos.example.com:8082/v.m4s", "filesize": 1000},

{"format_id": "30280", "url": "https://upos.example.com/a.m4s", "filesize": 100},

],         "requested_formats": [             {"format_id": "30080", "url": "https://upos.example.com:8082/v.m4s", "filesize": 1000},

"requested_formats": [             {"format_id": "30080", "url": "https://upos.example.com:8082/v.m4s", "filesize": 1000},             {"format_id": "30280", "url": "https://upos.example.com/a.m4s", "filesize": 100},

{"format_id": "30080", "url": "https://upos.example.com:8082/v.m4s", "filesize": 1000},

{"format_id": "30280", "url": "https://upos.example.com/a.m4s", "filesize": 100},

private_dns = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.5", 443))]         cases = (             ({"url": "http://upos.example.com/v.m4s"}, PUBLIC_DNS),

cases = (             ({"url": "http://upos.example.com/v.m4s"}, PUBLIC_DNS),             ({"url": "https://user:pw@upos.example.com/v.m4s"}, PUBLIC_DNS),

({"url": "http://upos.example.com/v.m4s"}, PUBLIC_DNS),

({"url": "https://user:pw@upos.example.com/v.m4s"}, PUBLIC_DNS),

({"url": "https://169.254.169.254/latest"}, PUBLIC_DNS),

({"url": "https://upos.example.com/v.m4s"}, private_dns),

({"url": "https://upos.example.com/v.m4s", "manifest_url": "http://upos.example.com/m.mpd"}, PUBLIC_DNS),

({"url": "https://upos.example.com/v.m4s", "fragments": [{"url": "https://127.0.0.1/f1"}]}, PUBLIC_DNS),

too_big = generic_info(             requested_formats=[                 {"format_id": "30080", "url": "https://upos.example.com/v.m4s", "filesize": 3 * 1024**3},

requested_formats=[                 {"format_id": "30080", "url": "https://upos.example.com/v.m4s", "filesize": 3 * 1024**3},                 {"format_id": "30280", "url": "https://upos.example.com/a.m4s", "filesize_approx": 2 * 1024**3},

{"format_id": "30080", "url": "https://upos.example.com/v.m4s", "filesize": 3 * 1024**3},

{"format_id": "30280", "url": "https://upos.example.com/a.m4s", "filesize_approx": 2 * 1024**3},

unknown = generic_info(                 requested_formats=[                     {"format_id": "30080", "url": "https://upos.example.com/v.m4s"},

requested_formats=[                     {"format_id": "30080", "url": "https://upos.example.com/v.m4s"},                     {"format_id": "30280", "url": "https://upos.example.com/a.m4s", "filesize": 10},

{"format_id": "30080", "url": "https://upos.example.com/v.m4s"},

{"format_id": "30280", "url": "https://upos.example.com/a.m4s", "filesize": 10},

MEDIA, "run_yt_dlp", side_effect=fake_run         ):             self.assertEqual(MEDIA.preflight_generic("https://www.bilibili.com/video/BV1xx411c7mD", timeout=30)["id"], "BV1xx411c7mD")

):             self.assertEqual(MEDIA.preflight_generic("https://www.bilibili.com/video/BV1xx411c7mD", timeout=30)["id"], "BV1xx411c7mD")             for category in ("container_rejected", "invalid_response"):

self.assertEqual(MEDIA.preflight_generic("https://www.bilibili.com/video/BV1xx411c7mD", timeout=30)["id"], "BV1xx411c7mD")

for category in ("container_rejected", "invalid_response"):                 with self.assertRaises(MEDIA.KnownMediaError) as raised:                     MEDIA.preflight_generic("https://www.bilibili.com/video/BV1xx411c7mD", timeout=30)

with self.assertRaises(MEDIA.KnownMediaError) as raised:                     MEDIA.preflight_generic("https://www.bilibili.com/video/BV1xx411c7mD", timeout=30)                 self.assertEqual(raised.exception.category, category)

MEDIA.preflight_generic("https://www.bilibili.com/video/BV1xx411c7mD", timeout=30)

self.assertEqual(total_timeout, MEDIA.GENERIC_PREFLIGHT_SECONDS)         self.assertEqual(command[command.index("--format") + 1], MEDIA.GENERIC_FORMAT)         self.assertEqual(command[-2:], ["--", "https://www.bilibili.com/video/BV1xx411c7

self.assertEqual(command[command.index("--format") + 1], MEDIA.GENERIC_FORMAT)         self.assertEqual(command[-2:], ["--", "https://www.bilibili.com/video/BV1xx411c7mD"])         with mock.patch.object(MEDIA, "find_yt_dlp", return_value="

self.assertEqual(command[-2:], ["--", "https://www.bilibili.com/video/BV1xx411c7mD"])

):             with self.assertRaises(MEDIA.KnownMediaError) as raised:                 MEDIA.preflight_generic("https://www.bilibili.com/video/BV1xx411c7mD", timeout=30)

with self.assertRaises(MEDIA.KnownMediaError) as raised:                 MEDIA.preflight_generic("https://www.bilibili.com/video/BV1xx411c7mD", timeout=30)         self.assertEqual(raised.exception.category, "extractor_failed")

MEDIA.preflight_generic("https://www.bilibili.com/video/BV1xx411c7mD", timeout=30)

output = pathlib.Path(root) / "out"             with self.assertRaises(MEDIA.KnownMediaError) as raised:                 MEDIA.run("https://www.bilibili.com/video/BV1xx411c7mD", output_dir=str(output), timeout=30)

with self.assertRaises(MEDIA.KnownMediaError) as raised:                 MEDIA.run("https://www.bilibili.com/video/BV1xx411c7mD", output_dir=str(output), timeout=30)             self.assertEqual(raised.exception.category, "timeout")

MEDIA.run("https://www.bilibili.com/video/BV1xx411c7mD", output_dir=str(output), timeout=30)

query = urllib.parse.parse_qs(parsed.query)         key = (parsed.hostname, int(query["type"][0]))         self.requests.append((key, query["name"][0], request.get_header("Accept"), timeout))

key = (parsed.hostname, int(query["type"][0]))         self.requests.append((key, query["name"][0], request.get_header("Accept"), timeout))         reply = self.answers.get(key)

self.requests.append((key, query["name"][0], request.get_header("Accept"), timeout))

class FakeIpDohTests(unittest.TestCase):     def check(self, doh, url="https://www.bilibili.com/video/BV1Ab411c7Xy/"):

def check(self, doh, url="https://www.bilibili.com/video/BV1Ab411c7Xy/"):         checks = {}         with mock.patch.object(MEDIA.socket, "getaddrinfo", return_value=LIVE_FAKE_IP), mock.patch.object(

):                 result = MEDIA.run(                     "https://www.bilibili.com/video/BV1xx411c7mD", output_dir=str(output), timeout=30

result = MEDIA.run(                     "https://www.bilibili.com/video/BV1xx411c7mD", output_dir=str(output), timeout=30                 )

"https://www.bilibili.com/video/BV1xx411c7mD", output_dir=str(output), timeout=30

checks = {}         with mock.patch.object(MEDIA.socket, "getaddrinfo", return_value=PUBLIC_DNS):             MEDIA.validate_public_https_url("https://www.bilibili.com/video/x", dns_checks=checks)

with mock.patch.object(MEDIA.socket, "getaddrinfo", return_value=PUBLIC_DNS):             MEDIA.validate_public_https_url("https://www.bilibili.com/video/x", dns_checks=checks)         self.assertEqual(checks, {"www.bilibili.com": "local_dn

MEDIA.validate_public_https_url("https://www.bilibili.com/video/x", dns_checks=checks)

self.assertEqual(raised.exception.category, "non_public_host")                 with self.assertRaises(MEDIA.KnownMediaError):                     MEDIA.validate_public_https_url(f"https://[{value}]/v.mp4", resolve_dns=False)

with self.assertRaises(MEDIA.KnownMediaError):                     MEDIA.validate_public_https_url(f"https://[{value}]/v.mp4", resolve_dns=False)         for value in public:

MEDIA.validate_public_https_url(f"https://[{value}]/v.mp4", resolve_dns=False)

with mock.patch.object(MEDIA.socket, "getaddrinfo", return_value=answers):             with self.assertRaises(MEDIA.KnownMediaError) as raised:                 MEDIA.validate_public_https_url("https://cdn.example.com/v.mp4")

with self.assertRaises(MEDIA.KnownMediaError) as raised:                 MEDIA.validate_public_https_url("https://cdn.example.com/v.mp4")         self.assertEqual(raised.exception.category, "non_public_host")

MEDIA.validate_public_https_url("https://cdn.example.com/v.mp4")

#!/usr/bin/env python3

def test_default_repeat_uses_new_dir_and_safe_resume_reuses(self):         default_root = ARTIFACT_ROOT / "stepfun-default" / "run-fixed"         url = "https://www.xiaoyuzhoufm.com/episode/episode-safe"

url = "https://www.xiaoyuzhoufm.com/episode/episode-safe"

STEPFUN,                 "extract_episode",                 return_value=("Episode", "https://media.xyzcdn.net/audio.m4a"),

"extract_episode",                 return_value=("Episode", "https://media.xyzcdn.net/audio.m4a"),             ),

return_value=("Episode", "https://media.xyzcdn.net/audio.m4a"),             ),             mock.patch.object(STEPFUN, "download", side_effect=fake_download),

def test_resume_rejects_corrupt_source_without_rewriting_metadata(self):         default_root = ARTIFACT_ROOT / "stepfun-corrupt" / "run-fixed"         url = "https://www.xiaoyuzhoufm.com/episode/episode-corrupt"

url = "https://www.xiaoyuzhoufm.com/episode/episode-corrupt"

STEPFUN,                 "extract_episode",                 return_value=("Episode", "https://media.xyzcdn.net/audio.m4a"),

"extract_episode",                 return_value=("Episode", "https://media.xyzcdn.net/audio.m4a"),             ),

return_value=("Episode", "https://media.xyzcdn.net/audio.m4a"),             ),             mock.patch.object(STEPFUN, "download", side_effect=fake_download),

#!/usr/bin/env python3

(                 "download",                 "https://mp.weixin.qq.com/s/known",

"download",                 "https://mp.weixin.qq.com/s/known",                 "--file",

"https://mp.weixin.qq.com/s/known",

"localhost_root_reachability_only",         )         self.assertEqual(payload["root_url"], "http://127.0.0.1:18901/")

)         self.assertEqual(payload["root_url"], "http://127.0.0.1:18901/")         self.assertTrue(payload["localhost_only"])

self.assertEqual(payload["root_url"], "http://127.0.0.1:18901/")

#!/usr/bin/env python3

self.assertEqual(             douyin.normalize_url(                 "https://www.douyin.com/jingxuan?modal_id=7611845735025364265"

douyin.normalize_url(                 "https://www.douyin.com/jingxuan?modal_id=7611845735025364265"             ),

"https://www.douyin.com/jingxuan?modal_id=7611845735025364265"

),             "https://www.douyin.com/video/7611845735025364265",         )

"https://www.douyin.com/video/7611845735025364265",

)         with self.assertRaises(ValueError):             douyin.normalize_url("https://example.com/video/7611845735025364265")

douyin.normalize_url("https://example.com/video/7611845735025364265")

def test_extracts_note_id_and_rejects_other_hosts(self) -> None:         self.assertEqual(             xhs.extract_note_id("https://www.xiaohongshu.com/explore/abc123"),

self.assertEqual(             xhs.extract_note_id("https://www.xiaohongshu.com/explore/abc123"),             "abc123",

xhs.extract_note_id("https://www.xiaohongshu.com/explore/abc123"),

)         with self.assertRaises(ValueError):             xhs.extract_note_id("https://example.com/explore/abc123")

xhs.extract_note_id("https://example.com/explore/abc123")

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

try:             page = "https://www.xiaohongshu.com/explore/abc123"             self.assertNotIn("Cookie", xhs.request_headers(include_cookie=False, target_url=page))

page = "https://www.xiaohongshu.com/explore/abc123"

[Credential-related source omitted]

[Credential-related source omitted]

"og:title": "Example",                 "og:description": "Summary",                 "og:image": "https://sns-img.example/image.jpg",

"og:description": "Summary",                 "og:image": "https://sns-img.example/image.jpg",             },

"og:image": "https://sns-img.example/image.jpg",

},             "https://www.xiaohongshu.com/explore/abc123",             "abc123",

"https://www.xiaohongshu.com/explore/abc123",

class WechatRetiredStubTests(unittest.TestCase):     def local_server(self, reply):         server = socket.socket()

def local_server(self, reply):         server = socket.socket()         server.bind(("127.0.0.1", 0))

server = socket.socket()

thread = threading.Thread(target=serve, daemon=True)         thread.start()         return f"http://127.0.0.1:{server.getsockname()[1]}/"

return f"http://127.0.0.1:{server.getsockname()[1]}/"

#!/usr/bin/env python3

import contextlib import html import http.client

import html import http.client import importlib.util

import http.client

CHKSM = "84f4c5d1b3834cc7ad5a" SHORT_ID = "AbCdEfGhIjKlMnOpQrSt" SHORT_CANONICAL = f"https://mp.weixin.qq.com/s/{SHORT_ID}"

SHORT_ID = "AbCdEfGhIjKlMnOpQrSt" SHORT_CANONICAL = f"https://mp.weixin.qq.com/s/{SHORT_ID}" LONG_CANONICAL = (

SHORT_CANONICAL = f"https://mp.weixin.qq.com/s/{SHORT_ID}"

LONG_CANONICAL = (     f"https://mp.weixin.qq.com/s?__biz={BIZ}&mid={MID}&idx={IDX}&sn={SN}&chksm={CHKSM}" )

f"https://mp.weixin.qq.com/s?__biz={BIZ}&mid={MID}&idx={IDX}&sn={SN}&chksm={CHKSM}"

) LONG_WITH_SECRETS = (     f"https://mp.weixin.qq.com/s?__biz={BIZ}&mid={MID}&idx={IDX}&sn={SN}"

LONG_WITH_SECRETS = (     f"https://mp.weixin.qq.com/s?__biz={BIZ}&mid={MID}&idx={IDX}&sn={SN}"     f"&chksm={CHKSM}&key=SECRETKEY123&uin=SECRETUIN456&pass_ticket=SECRETTICKET789"

f"https://mp.weixin.qq.com/s?__biz={BIZ}&mid={MID}&idx={IDX}&sn={SN}"

<section><p>第一段<strong>加粗</strong>文字，正文足够长，用于通过最短正文门槛的检查。</p></section> <h2>小标题</h2> <p>请看<a href="https://example.com/a_b?x=1">示例链接</a>。</p>

<h2>小标题</h2> <p>请看<a href="https://example.com/a_b?x=1">示例链接</a>。</p> <p><img data-src="https://mmbiz.qpic.cn/mmbiz_png/abc/640?wx_fmt=png" src="data:image/gif;base64,AAAA" alt="图一"></p>

<p>请看<a href="https://example.com/a_b?x=1">示例链接</a>。</p>

<p><img data-src="https://mmbiz.qpic.cn/mmbiz_png/abc/640?wx_fmt=png" src="data:image/gif;base64,AAAA" alt="图一"></p>

MIGRATED_TARGET = (     "https://mp.weixin.qq.com/s?__biz=MzAwMDAwMDAwMQ==&mid=2247480001&idx=1"

"https://mp.weixin.qq.com/s?__biz=MzAwMDAwMDAwMQ==&mid=2247480001&idx=1"

"<p>该公众号已迁移至新的账号，原账号已回收。若需访问原文章链接，请点击下方按钮。</p>"     "<a href='' id='js_access_msg'>访问文章</a></div>"     "<script>var transferTargetLink = 'http://mp.weixin.qq.com/s?__biz=MzAwMDAwMDAwMQ==&amp;"

"<a href='' id='js_access_msg'>访问文章</a></div>"     "<script>var transferTargetLink = 'http://mp.weixin.qq.com/s?__biz=MzAwMDAwMDAwMQ==&amp;"     "mid=2247480001&amp;idx=1&amp;sn=fedcba9876543210fedcba9876543210#rd';</script>"

"<script>var transferTargetLink = 'http://mp.weixin.qq.com/s?__biz=MzAwMDAwMDAwMQ==&amp;"

'<!doctype html><html lang="zh-CN"><head><meta charset="utf-8" />'     "<title>公众号文章导出工具已下线</title>"     '<link rel="canonical" href="https://down.mptext.top/" /></head>'

"<title>公众号文章导出工具已下线</title>"     '<link rel="canonical" href="https://down.mptext.top/" /></head>'     "<body><main><h1>公众号文章导出工具已下线</h1>"

'<link rel="canonical" href="https://down.mptext.top/" /></head>'

meta = (             f'<meta property="og:title" content="{title}" />'             f'<meta property="og:url" content="http://mp.weixin.qq.com/s?__biz={BIZ}'

f'<meta property="og:title" content="{title}" />'             f'<meta property="og:url" content="http://mp.weixin.qq.com/s?__biz={BIZ}'             f'&amp;mid={mid}&amp;idx={IDX}&amp;sn={SN}&amp;chksm={CHKSM}#rd" />'

f'<meta property="og:url" content="http://mp.weixin.qq.com/s?__biz={BIZ}'

self.status = status         self._body = body.encode("utf-8") if isinstance(body, str) else body         self.headers = http.client.HTTPMessage()

self._body = body.encode("utf-8") if isinstance(body, str) else body         self.headers = http.client.HTTPMessage()         self.headers["Content-Type"] = content_type

self.headers = http.client.HTTPMessage()

def __call__(self, opener, request, timeout):         self.requests.append(request)

self.requests.append(request)

def http_error(code: int) -> urllib.error.HTTPError:     return urllib.error.HTTPError(         "https://mp.weixin.qq.com/s/x", code, "error", http.client.HTTPMessage(), None

return urllib.error.HTTPError(         "https://mp.weixin.qq.com/s/x", code, "error", http.client.HTTPMessage(), None     )

"https://mp.weixin.qq.com/s/x", code, "error", http.client.HTTPMessage(), None

def fetch(self, url: str, *, relay: bool = True):

def test_short_form_drops_query_and_fragment(self) -> None:         article = wechat.normalize_article_url(             f"https://MP.weixin.qq.com/s/{SHORT_ID}?scene=1&key=SECRETKEY123#rd"

article = wechat.normalize_article_url(             f"https://MP.weixin.qq.com/s/{SHORT_ID}?scene=1&key=SECRETKEY123#rd"         )

f"https://MP.weixin.qq.com/s/{SHORT_ID}?scene=1&key=SECRETKEY123#rd"

def test_html_escaped_separators_from_page_source_are_accepted(self) -> None:         article = wechat.normalize_article_url(             f"https://mp.weixin.qq.com/s?__biz={BIZ}&mid={MID}&idx={IDX}&sn={SN}"

article = wechat.normalize_article_url(             f"https://mp.weixin.qq.com/s?__biz={BIZ}&mid={MID}&idx={IDX}&sn={SN}"         )

f"https://mp.weixin.qq.com/s?__biz={BIZ}&mid={MID}&idx={IDX}&sn={SN}"

self.assertEqual(             article.canonical_url,             f"https://mp.weixin.qq.com/s?__biz={BIZ}&mid={MID}&idx={IDX}&sn={SN}",

article.canonical_url,             f"https://mp.weixin.qq.com/s?__biz={BIZ}&mid={MID}&idx={IDX}&sn={SN}",         )