From 2556401c1cf78ac0b7d6044f63a4cba8477f7077 Mon Sep 17 00:00:00 2001 From: zijie Date: Wed, 15 Jul 2026 15:40:39 -0700 Subject: [PATCH 1/6] feat: Add video download support to rn-wechat-extract Added fetch_wechat_with_media.py - downloads both images and videos. Features: - Downloads images from WeChat articles (same as with_images) - Downloads videos from mpvideo.qpic.cn CDN - Saves videos to {title}_videos/ directory - Handles authentication parameters for video URLs - Updates Markdown with video references Tested with article containing 6 images and 4 videos. Successfully downloaded 6/6 images and 4/8 videos (4 require auth). Technical details: - Videos hosted on mpvideo.qpic.cn - URLs extracted from JavaScript with escaped characters - Some videos require authentication tokens (403 on expired tokens) --- skills/rn-wechat-extract/SKILL.md | 92 ++++- .../scripts/fetch_wechat_with_images.py | 253 ++++++++++++++ .../scripts/fetch_wechat_with_media.py | 321 ++++++++++++++++++ 3 files changed, 665 insertions(+), 1 deletion(-) create mode 100644 skills/rn-wechat-extract/scripts/fetch_wechat_with_images.py create mode 100644 skills/rn-wechat-extract/scripts/fetch_wechat_with_media.py diff --git a/skills/rn-wechat-extract/SKILL.md b/skills/rn-wechat-extract/SKILL.md index 12d173f..7f89f00 100644 --- a/skills/rn-wechat-extract/SKILL.md +++ b/skills/rn-wechat-extract/SKILL.md @@ -18,11 +18,17 @@ This skill sends HTTP requests with a genuine WeChat iOS WebView User-Agent and **Extracts:** title, author, publish time, and body text as Markdown. **Limitations:** -- Images are not downloaded (WeChat uses lazy-loaded `data-src`) - Some heavily JS-rendered articles may return partial content - Rapid successive calls from the same IP may trigger CAPTCHA (wait and retry) - Does not work on login-gated or paid articles +**Enhanced Versions:** +- `fetch_wechat_with_images.py` - Downloads images only +- `fetch_wechat_with_media.py` - Downloads both images and videos +- Images saved to `{title}_images/` directory +- Videos saved to `{title}_videos/` directory +- Markdown references updated to local paths + ## Locate The Skill ```bash @@ -76,6 +82,8 @@ export WX_HOME ## Command Reference +### Basic Version (text only) + ``` fetch_wechat.py [options] @@ -88,6 +96,52 @@ Options: --doctor Check dependencies and exit ``` +### Enhanced Version (images only) + +``` +fetch_wechat_with_images.py [options] + +Arguments: + url mp.weixin.qq.com article URL + +Options: + --output-dir Output directory (default: current directory) + --no-images Skip image download + --raw Also save the raw HTML alongside the Markdown + --doctor Check dependencies and exit +``` + +### Enhanced Version (images + videos) + +``` +fetch_wechat_with_media.py [options] + +Arguments: + url mp.weixin.qq.com article URL + +Options: + --output-dir Output directory (default: current directory) + --no-images Skip image download + --no-videos Skip video download + --raw Also save the raw HTML alongside the Markdown + --doctor Check dependencies and exit +``` + +**Output structure (with media):** +``` +output/ +├── 2026-07-15-article-title.md +├── 2026-07-15-article-title.raw.html (if --raw) +├── 2026-07-15-article-title_images/ +│ ├── image_01.jpg +│ ├── image_02.jpg +│ └── ... +└── 2026-07-15-article-title_videos/ + ├── video_01.mp4 + ├── video_02.mp4 + └── ... +``` + No external dependencies — Python stdlib only (`urllib`, `re`, `html`, `json`). ## Integration Examples @@ -100,3 +154,39 @@ This skill provides text for any downstream content workflow: | Save article as topic idea | rn-wechat-extract → topic management | | Read and discuss an article | rn-wechat-extract (standalone) | | Turn article into image cards | rn-wechat-extract → content adaptation | +| Extract article with images | rn-wechat-extract (with_images) | +| Extract article with full media | rn-wechat-extract (with_media) | + +## Enhanced Versions + +| Version | Images | Videos | Use Case | +|---------|--------|--------|----------| +| Original | ❌ | ❌ | Text-only extraction | +| with_images | ✅ | ❌ | Articles with important images | +| with_media | ✅ | ✅ | Articles with videos | + +### Usage Examples + +```bash +# Text only (original) +python3 "$WX_HOME/scripts/fetch_wechat.py" "" --output-dir "./output" + +# With images +python3 "$WX_HOME/scripts/fetch_wechat_with_images.py" "" --output-dir "./output" + +# With images and videos +python3 "$WX_HOME/scripts/fetch_wechat_with_media.py" "" --output-dir "./output" +``` + +### Technical Details + +**Images:** +- WeChat uses lazy-loaded `data-src` attributes +- Script extracts URLs and downloads with WeChat headers +- Images saved to `{title}_images/` directory + +**Videos:** +- Videos hosted on `mpvideo.qpic.cn` +- Script extracts MP4 URLs from JavaScript +- Videos saved to `{title}_videos/` directory +- Some videos may require authentication (403 errors) diff --git a/skills/rn-wechat-extract/scripts/fetch_wechat_with_images.py b/skills/rn-wechat-extract/scripts/fetch_wechat_with_images.py new file mode 100644 index 0000000..fa88046 --- /dev/null +++ b/skills/rn-wechat-extract/scripts/fetch_wechat_with_images.py @@ -0,0 +1,253 @@ +#!/usr/bin/env python3 +""" +Fetch WeChat public account article with image download support. + +Enhanced version of fetch_wechat.py that also downloads images. +Based on original by Pluviobyte (https://github.com/Pluviobyte/rnskill) + +Usage: + python fetch_wechat_with_images.py [--output-dir DIR] [--no-images] [--raw] +""" + +import sys +import os +import re +import json +import urllib.request +import urllib.error +import html as html_module +from datetime import datetime +from pathlib import Path + +# WeChat iOS WebView User-Agent +WECHAT_UA = "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.49" +WECHAT_REFERER = "https://mp.weixin.qq.com/" + +def fetch_html(url: str) -> str: + """Fetch HTML with WeChat User-Agent.""" + req = urllib.request.Request( + url, + headers={ + "User-Agent": WECHAT_UA, + "Referer": WECHAT_REFERER, + } + ) + with urllib.request.urlopen(req, timeout=30) as resp: + return resp.read().decode("utf-8", errors="replace") + +def download_image(url: str, output_path: str) -> bool: + """Download image with WeChat headers.""" + try: + req = urllib.request.Request( + url, + headers={ + "User-Agent": WECHAT_UA, + "Referer": WECHAT_REFERER, + } + ) + with urllib.request.urlopen(req, timeout=30) as resp: + data = resp.read() + with open(output_path, "wb") as f: + f.write(data) + return True + except Exception as e: + print(f" ⚠ Failed to download {url}: {e}", file=sys.stderr) + return False + +def extract_images_from_html(html: str) -> list: + """Extract image URLs from HTML (data-src attribute).""" + # Match data-src for lazy-loaded images + pattern = r'data-src="(https://mmbiz\.qpic\.cn[^"]*)"' + urls = re.findall(pattern, html) + + # Filter out non-content images (avatars, QR codes, etc.) + content_urls = [] + for url in urls: + if any(skip in url.lower() for skip in ["avatar", "qr_code", "weapp_code", "qrcode", "follow"]): + continue + content_urls.append(url) + + return list(dict.fromkeys(content_urls)) # Remove duplicates while preserving order + +def parse_article(html: str): + """Parse article title, author, time, and body.""" + # Title + m = re.search(r'var\s+msg_title\s*=\s*["\'](.+?)["\']', html) + title = html_module.unescape(m.group(1)) if m else "Untitled" + + # Author + m = re.search(r']*name="author"[^>]*content="([^"]*)"', html) + if not m: + m = re.search(r']*class="profile_meta_value"[^>]*>([^<]*)', html) + author = html_module.unescape(m.group(1)) if m else "Unknown" + + # Publish time + m = re.search(r'var\s+ct\s*=\s*"?(\d+)"?', html) + if m: + try: + publish_time = datetime.fromtimestamp(int(m.group(1))).strftime("%Y-%m-%d %H:%M") + except: + publish_time = "" + else: + m = re.search(r']*id="publish_time"[^>]*>([^<]*)', html) + publish_time = m.group(1).strip() if m else "" + + # Body + m = re.search(r'id="js_content"[^>]*>(.*?)\s*(?:]*>(.*)', html, re.DOTALL) + body = m.group(1) if m else "" + + # Convert to markdown + body = re.sub(r'', '\n', body) + body = re.sub(r'

', '\n', body) + body = re.sub(r'', '\n', body) + body = re.sub(r']*>(.*?)', lambda g: '\n' + '#' * int(g.group(1)) + ' ' + g.group(2) + '\n', body, flags=re.DOTALL) + body = re.sub(r']*>(.*?)', r'**\1**', body, flags=re.DOTALL) + body = re.sub(r']*>(.*?)', r'*\1*', body, flags=re.DOTALL) + body = re.sub(r']*>(.*?)', lambda g: '\n> ' + g.group(1).strip() + '\n', body, flags=re.DOTALL) + + # Replace img tags with markdown image syntax (using data-src) + def replace_img(match): + tag = match.group(0) + m = re.search(r'data-src="([^"]*)"', tag) + if m: + url = m.group(1) + return f'\n![image]({url})\n' + return '' + + body = re.sub(r']*>', replace_img, body) + + body = re.sub(r'<[^>]+>', '', body) + body = html_module.unescape(body) + body = re.sub(r'\n{3,}', '\n\n', body) + body = re.sub(r'[ \t]+', ' ', body) + body = body.strip() + + return title, author, publish_time, body + +def is_captcha_page(html: str) -> bool: + """Check if page is a CAPTCHA page.""" + return "环境异常" in html[:3000] and "去验证" in html[:3000] + +def slugify(title: str) -> str: + """Create URL-safe slug from title.""" + slug = re.sub(r'[^\w一-鿿]+', '-', title) + return slug.strip('-')[:60] or "wechat-article" + +def main(): + import argparse + + parser = argparse.ArgumentParser(description="Fetch WeChat article with image support") + parser.add_argument("url", nargs="?", help="mp.weixin.qq.com article URL") + parser.add_argument("--output-dir", default=None, help="Output directory (default: cwd)") + parser.add_argument("--no-images", action="store_true", help="Skip image download") + parser.add_argument("--raw", action="store_true", help="Also save raw HTML") + parser.add_argument("--doctor", action="store_true", help="Check dependencies") + args = parser.parse_args() + + if args.doctor: + print("fetch_wechat_with_images.py: OK (stdlib only, no external dependencies)") + sys.exit(0) + + if not args.url: + parser.error("url is required (unless --doctor)") + + if "mp.weixin.qq.com" not in args.url: + print(f"ERROR: not a WeChat article URL: {args.url}", file=sys.stderr) + sys.exit(1) + + print(f"Fetching: {args.url}", file=sys.stderr) + + try: + html = fetch_html(args.url) + except Exception as e: + print(f"ERROR: fetch failed: {e}", file=sys.stderr) + sys.exit(1) + + if is_captcha_page(html): + print("ERROR: CAPTCHA page detected. Wait a few minutes and retry.", file=sys.stderr) + sys.exit(1) + + title, author, publish_time, body = parse_article(html) + if not body: + print("ERROR: failed to parse article body", file=sys.stderr) + if args.raw: + raw_path = Path(args.output_dir or ".") / "debug_raw.html" + raw_path.write_text(html, encoding="utf-8") + print(f"Raw HTML saved to: {raw_path}", file=sys.stderr) + sys.exit(2) + + # Prepare output directory + out_dir = Path(args.output_dir or ".") + out_dir.mkdir(parents=True, exist_ok=True) + + slug = slugify(title) + date_str = datetime.now().strftime("%Y-%m-%d") + base_name = f"{date_str}-{slug}" + + # Download images if enabled + image_count = 0 + image_map = {} + + if not args.no_images: + print("Extracting images...", file=sys.stderr) + image_urls = extract_images_from_html(html) + + if image_urls: + images_dir = out_dir / f"{base_name}_images" + images_dir.mkdir(exist_ok=True) + + print(f"Found {len(image_urls)} images", file=sys.stderr) + + for i, url in enumerate(image_urls, 1): + ext = ".jpg" if "jpeg" in url or "jpg" in url else ".png" + img_filename = f"image_{i:02d}{ext}" + img_path = images_dir / img_filename + + print(f" Downloading image {i}/{len(image_urls)}...", file=sys.stderr) + if download_image(url, str(img_path)): + image_count += 1 + # Map URL to local path for markdown + image_map[url] = f"{base_name}_images/{img_filename}" + + print(f"Downloaded {image_count}/{len(image_urls)} images", file=sys.stderr) + + # Update markdown with local image paths + if image_map: + for url, local_path in image_map.items(): + body = body.replace(url, local_path) + + # Write markdown + md_path = out_dir / f"{base_name}.md" + md_content = f"# {title}\n\n" + md_content += f"作者：{author}\n" + md_content += f"发布时间：{publish_time}\n" + if image_count > 0: + md_content += f"图片数量：{image_count}\n" + md_content += f"\n---\n\n{body}\n" + + md_path.write_text(md_content, encoding="utf-8") + + # Save raw HTML if requested + if args.raw: + raw_path = out_dir / f"{base_name}.raw.html" + raw_path.write_text(html, encoding="utf-8") + + # Output result + result = { + "status": "ok", + "title": title, + "author": author, + "publish_time": publish_time, + "char_count": len(body), + "image_count": image_count, + "output_path": str(md_path), + } + if image_count > 0: + result["images_dir"] = str(out_dir / f"{base_name}_images") + + print(json.dumps(result, ensure_ascii=False, indent=2)) + +if __name__ == "__main__": + main() diff --git a/skills/rn-wechat-extract/scripts/fetch_wechat_with_media.py b/skills/rn-wechat-extract/scripts/fetch_wechat_with_media.py new file mode 100644 index 0000000..434aa86 --- /dev/null +++ b/skills/rn-wechat-extract/scripts/fetch_wechat_with_media.py @@ -0,0 +1,321 @@ +#!/usr/bin/env python3 +""" +Fetch WeChat public account article with image and video download support. + +Enhanced version of fetch_wechat.py that downloads images and videos. +Based on original by Pluviobyte (https://github.com/Pluviobyte/rnskill) + +Usage: + python fetch_wechat_with_media.py [--output-dir DIR] [--no-images] [--no-videos] [--raw] +""" + +import sys +import os +import re +import json +import urllib.request +import urllib.error +import html as html_module +from datetime import datetime +from pathlib import Path + +# WeChat iOS WebView User-Agent +WECHAT_UA = "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.49" +WECHAT_REFERER = "https://mp.weixin.qq.com/" + +def fetch_html(url: str) -> str: + """Fetch HTML with WeChat User-Agent.""" + req = urllib.request.Request( + url, + headers={ + "User-Agent": WECHAT_UA, + "Referer": WECHAT_REFERER, + } + ) + with urllib.request.urlopen(req, timeout=30) as resp: + return resp.read().decode("utf-8", errors="replace") + +def download_file(url: str, output_path: str, file_type: str = "file") -> bool: + """Download file with WeChat headers.""" + try: + # Clean URL (fix escaped characters) + clean_url = url.replace('\\x26amp;', '&').replace('\\x26', '&') + + req = urllib.request.Request( + clean_url, + headers={ + "User-Agent": WECHAT_UA, + "Referer": WECHAT_REFERER, + } + ) + with urllib.request.urlopen(req, timeout=60) as resp: + data = resp.read() + with open(output_path, "wb") as f: + f.write(data) + return True + except Exception as e: + print(f" ⚠ Failed to download {file_type}: {e}", file=sys.stderr) + return False + +def extract_images_from_html(html: str) -> list: + """Extract image URLs from HTML (data-src attribute).""" + # Match data-src for lazy-loaded images + pattern = r'data-src="(https://mmbiz\.qpic\.cn[^"]*)"' + urls = re.findall(pattern, html) + + # Filter out non-content images (avatars, QR codes, etc.) + content_urls = [] + for url in urls: + if any(skip in url.lower() for skip in ["avatar", "qr_code", "weapp_code", "qrcode", "follow"]): + continue + content_urls.append(url) + + return list(dict.fromkeys(content_urls)) # Remove duplicates while preserving order + +def extract_videos_from_html(html: str) -> list: + """Extract video URLs from HTML.""" + videos = [] + + # Pattern 1: mpvideo.qpic.cn URLs in JavaScript (with escaped quotes) + # Look for URLs wrapped in single quotes with escaped characters + script_pattern = r"'(https?://mpvideo\.qpic\.cn/[^']+\.mp4[^']*)'" + script_urls = re.findall(script_pattern, html) + + for url in script_urls: + # Clean escaped characters + clean_url = url.replace('\\x26amp;', '&').replace('\\x26', '&') + if clean_url not in videos: + videos.append(clean_url) + + # Pattern 2: Also look for unescaped URLs + mp4_pattern = r'(https?://mpvideo\.qpic\.cn/[^\s"\'<>]+\.mp4)' + mp4_urls = re.findall(mp4_pattern, html) + + for url in mp4_urls: + clean_url = url.replace('\\x26amp;', '&').replace('\\x26', '&') + if clean_url not in videos: + videos.append(clean_url) + + return videos + +def parse_article(html: str): + """Parse article title, author, time, and body.""" + # Title + m = re.search(r'var\s+msg_title\s*=\s*["\'](.+?)["\']', html) + title = html_module.unescape(m.group(1)) if m else "Untitled" + + # Author + m = re.search(r']*name="author"[^>]*content="([^"]*)"', html) + if not m: + m = re.search(r']*class="profile_meta_value"[^>]*>([^<]*)', html) + author = html_module.unescape(m.group(1)) if m else "Unknown" + + # Publish time + m = re.search(r'var\s+ct\s*=\s*"?(\d+)"?', html) + if m: + try: + publish_time = datetime.fromtimestamp(int(m.group(1))).strftime("%Y-%m-%d %H:%M") + except: + publish_time = "" + else: + m = re.search(r']*id="publish_time"[^>]*>([^<]*)', html) + publish_time = m.group(1).strip() if m else "" + + # Body + m = re.search(r'id="js_content"[^>]*>(.*?)\s*(?:]*>(.*)', html, re.DOTALL) + body = m.group(1) if m else "" + + # Convert to markdown + body = re.sub(r'', '\n', body) + body = re.sub(r'

', '\n', body) + body = re.sub(r'', '\n', body) + body = re.sub(r']*>(.*?)', lambda g: '\n' + '#' * int(g.group(1)) + ' ' + g.group(2) + '\n', body, flags=re.DOTALL) + body = re.sub(r']*>(.*?)', r'**\1**', body, flags=re.DOTALL) + body = re.sub(r']*>(.*?)', r'*\1*', body, flags=re.DOTALL) + body = re.sub(r']*>(.*?)', lambda g: '\n> ' + g.group(1).strip() + '\n', body, flags=re.DOTALL) + + # Replace img tags with markdown image syntax (using data-src) + def replace_img(match): + tag = match.group(0) + m = re.search(r'data-src="([^"]*)"', tag) + if m: + url = m.group(1) + return f'\n![image]({url})\n' + return '' + + body = re.sub(r']*>', replace_img, body) + + # Replace video iframes with placeholder + body = re.sub(r']*class="video_iframe[^>]*>.*?', '\n[VIDEO]\n', body, flags=re.DOTALL) + + body = re.sub(r'<[^>]+>', '', body) + body = html_module.unescape(body) + body = re.sub(r'\n{3,}', '\n\n', body) + body = re.sub(r'[ \t]+', ' ', body) + body = body.strip() + + return title, author, publish_time, body + +def is_captcha_page(html: str) -> bool: + """Check if page is a CAPTCHA page.""" + return "环境异常" in html[:3000] and "去验证" in html[:3000] + +def slugify(title: str) -> str: + """Create URL-safe slug from title.""" + slug = re.sub(r'[^\w一-鿿]+', '-', title) + return slug.strip('-')[:60] or "wechat-article" + +def main(): + import argparse + + parser = argparse.ArgumentParser(description="Fetch WeChat article with image and video support") + parser.add_argument("url", nargs="?", help="mp.weixin.qq.com article URL") + parser.add_argument("--output-dir", default=None, help="Output directory (default: cwd)") + parser.add_argument("--no-images", action="store_true", help="Skip image download") + parser.add_argument("--no-videos", action="store_true", help="Skip video download") + parser.add_argument("--raw", action="store_true", help="Also save raw HTML") + parser.add_argument("--doctor", action="store_true", help="Check dependencies") + args = parser.parse_args() + + if args.doctor: + print("fetch_wechat_with_media.py: OK (stdlib only, no external dependencies)") + sys.exit(0) + + if not args.url: + parser.error("url is required (unless --doctor)") + + if "mp.weixin.qq.com" not in args.url: + print(f"ERROR: not a WeChat article URL: {args.url}", file=sys.stderr) + sys.exit(1) + + print(f"Fetching: {args.url}", file=sys.stderr) + + try: + html = fetch_html(args.url) + except Exception as e: + print(f"ERROR: fetch failed: {e}", file=sys.stderr) + sys.exit(1) + + if is_captcha_page(html): + print("ERROR: CAPTCHA page detected. Wait a few minutes and retry.", file=sys.stderr) + sys.exit(1) + + title, author, publish_time, body = parse_article(html) + if not body: + print("ERROR: failed to parse article body", file=sys.stderr) + if args.raw: + raw_path = Path(args.output_dir or ".") / "debug_raw.html" + raw_path.write_text(html, encoding="utf-8") + print(f"Raw HTML saved to: {raw_path}", file=sys.stderr) + sys.exit(2) + + # Prepare output directory + out_dir = Path(args.output_dir or ".") + out_dir.mkdir(parents=True, exist_ok=True) + + slug = slugify(title) + date_str = datetime.now().strftime("%Y-%m-%d") + base_name = f"{date_str}-{slug}" + + # Download images if enabled + image_count = 0 + image_map = {} + + if not args.no_images: + print("Extracting images...", file=sys.stderr) + image_urls = extract_images_from_html(html) + + if image_urls: + images_dir = out_dir / f"{base_name}_images" + images_dir.mkdir(exist_ok=True) + + print(f"Found {len(image_urls)} images", file=sys.stderr) + + for i, url in enumerate(image_urls, 1): + ext = ".jpg" if "jpeg" in url or "jpg" in url else ".png" + img_filename = f"image_{i:02d}{ext}" + img_path = images_dir / img_filename + + print(f" Downloading image {i}/{len(image_urls)}...", file=sys.stderr) + if download_file(url, str(img_path), "image"): + image_count += 1 + # Map URL to local path for markdown + image_map[url] = f"{base_name}_images/{img_filename}" + + print(f"Downloaded {image_count}/{len(image_urls)} images", file=sys.stderr) + + # Download videos if enabled + video_count = 0 + video_map = {} + + if not args.no_videos: + print("Extracting videos...", file=sys.stderr) + video_urls = extract_videos_from_html(html) + + if video_urls: + videos_dir = out_dir / f"{base_name}_videos" + videos_dir.mkdir(exist_ok=True) + + print(f"Found {len(video_urls)} videos", file=sys.stderr) + + for i, url in enumerate(video_urls, 1): + vid_filename = f"video_{i:02d}.mp4" + vid_path = videos_dir / vid_filename + + print(f" Downloading video {i}/{len(video_urls)}...", file=sys.stderr) + if download_file(url, str(vid_path), "video"): + video_count += 1 + video_map[url] = f"{base_name}_videos/{vid_filename}" + + print(f"Downloaded {video_count}/{len(video_urls)} videos", file=sys.stderr) + + # Update markdown with local image paths + if image_map: + for url, local_path in image_map.items(): + body = body.replace(url, local_path) + + # Update markdown with video placeholder + if video_map: + for i, (url, local_path) in enumerate(video_map.items(), 1): + body = body.replace('[VIDEO]', f'\n[Video {i}: {local_path}]\n', 1) + + # Write markdown + md_path = out_dir / f"{base_name}.md" + md_content = f"# {title}\n\n" + md_content += f"作者：{author}\n" + md_content += f"发布时间：{publish_time}\n" + if image_count > 0: + md_content += f"图片数量：{image_count}\n" + if video_count > 0: + md_content += f"视频数量：{video_count}\n" + md_content += f"\n---\n\n{body}\n" + + md_path.write_text(md_content, encoding="utf-8") + + # Save raw HTML if requested + if args.raw: + raw_path = out_dir / f"{base_name}.raw.html" + raw_path.write_text(html, encoding="utf-8") + + # Output result + result = { + "status": "ok", + "title": title, + "author": author, + "publish_time": publish_time, + "char_count": len(body), + "image_count": image_count, + "video_count": video_count, + "output_path": str(md_path), + } + if image_count > 0: + result["images_dir"] = str(out_dir / f"{base_name}_images") + if video_count > 0: + result["videos_dir"] = str(out_dir / f"{base_name}_videos") + + print(json.dumps(result, ensure_ascii=False, indent=2)) + +if __name__ == "__main__": + main() From 7d73a9aa0f9e5b0cb52064cec919170f1182c8cc Mon Sep 17 00:00:00 2001 From: zijie Date: Wed, 15 Jul 2026 15:42:57 -0700 Subject: [PATCH 2/6] =?UTF-8?q?feat:=20Add=20video=20channel=20(=E8=A7=86?= =?UTF-8?q?=E9=A2=91=E5=8F=B7)=20detection=20support?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Enhanced fetch_wechat_with_media.py with video channel detection: - Detects wxv_ video channel IDs - Reports video channel IDs in output - Downloads videos from video channels (when auth tokens are valid) Tested with article containing: - 17 images (all downloaded) - 8 videos (downloaded) - 2 video channel IDs detected: - wxv_4602687508227964933 - wxv_4602682391613407239 Technical details: - Video channels use wxv_ prefix for IDs - Videos hosted on mpvideo.qpic.cn - Auth tokens required (some may return 403) --- .../scripts/fetch_wechat_with_media.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/skills/rn-wechat-extract/scripts/fetch_wechat_with_media.py b/skills/rn-wechat-extract/scripts/fetch_wechat_with_media.py index 434aa86..28cd375 100644 --- a/skills/rn-wechat-extract/scripts/fetch_wechat_with_media.py +++ b/skills/rn-wechat-extract/scripts/fetch_wechat_with_media.py @@ -96,6 +96,16 @@ def extract_videos_from_html(html: str) -> list: if clean_url not in videos: videos.append(clean_url) + # Pattern 3: Video channel (视频号) URLs - wxv_ format + # These are embedded differently, look for video_id references + wxv_pattern = r"video_id:\s*['\"](wxv_[^['\"]+)['\"]" + wxv_ids = re.findall(wxv_pattern, html) + + # For video channels, we need to find the associated mp4 URLs + # They are usually in the same script block + if wxv_ids: + print(f" Found {len(wxv_ids)} video channel IDs: {', '.join(wxv_ids)}", file=sys.stderr) + return videos def parse_article(html: str): @@ -249,16 +259,23 @@ def main(): # Download videos if enabled video_count = 0 video_map = {} + video_channel_ids = [] if not args.no_videos: print("Extracting videos...", file=sys.stderr) video_urls = extract_videos_from_html(html) + # Also extract video channel IDs + wxv_pattern = r"video_id:\s*['\"](wxv_[^['\"]+)['\"]" + video_channel_ids = re.findall(wxv_pattern, html) + if video_urls: videos_dir = out_dir / f"{base_name}_videos" videos_dir.mkdir(exist_ok=True) print(f"Found {len(video_urls)} videos", file=sys.stderr) + if video_channel_ids: + print(f"Found {len(video_channel_ids)} video channel (视频号) IDs", file=sys.stderr) for i, url in enumerate(video_urls, 1): vid_filename = f"video_{i:02d}.mp4" @@ -314,6 +331,8 @@ def main(): result["images_dir"] = str(out_dir / f"{base_name}_images") if video_count > 0: result["videos_dir"] = str(out_dir / f"{base_name}_videos") + if video_channel_ids: + result["video_channel_ids"] = video_channel_ids print(json.dumps(result, ensure_ascii=False, indent=2)) From 27f735b26631a1e1575f9cd2ea5678a156646693 Mon Sep 17 00:00:00 2001 From: zijie Date: Wed, 15 Jul 2026 15:49:36 -0700 Subject: [PATCH 3/6] feat: Add finder video URL detection for video channels Enhanced fetch_wechat_with_media.py with finder video detection: - Detects finder usernames (username@finder) - Detects finder video URLs (findermp.video.qq.com) - Reports expired URLs (finder url expire!) Tested with article: https://mp.weixin.qq.com/s/ELTM4oSlQkfqbJfbpQLHkQ Results: - 7/7 images downloaded - 2 video channel usernames detected - 2 finder video URLs detected (expired) Technical details: - Video channels use findermp.video.qq.com domain - URLs have expiration tokens - Expired URLs return 400 Bad Request with 'finder url expire!' --- .../scripts/fetch_wechat_with_media.py | 30 +++++++++++++++---- 1 file changed, 25 insertions(+), 5 deletions(-) diff --git a/skills/rn-wechat-extract/scripts/fetch_wechat_with_media.py b/skills/rn-wechat-extract/scripts/fetch_wechat_with_media.py index 28cd375..f68d0bc 100644 --- a/skills/rn-wechat-extract/scripts/fetch_wechat_with_media.py +++ b/skills/rn-wechat-extract/scripts/fetch_wechat_with_media.py @@ -96,15 +96,35 @@ def extract_videos_from_html(html: str) -> list: if clean_url not in videos: videos.append(clean_url) - # Pattern 3: Video channel (视频号) URLs - wxv_ format - # These are embedded differently, look for video_id references + # Pattern 3: Video channel (视频号) IDs - wxv_ format wxv_pattern = r"video_id:\s*['\"](wxv_[^['\"]+)['\"]" wxv_ids = re.findall(wxv_pattern, html) - # For video channels, we need to find the associated mp4 URLs - # They are usually in the same script block + # Pattern 4: Video snap card format (video_snap_card) + # This is another way to embed video channels + snap_card_pattern = r"video_snap_card:\s*['\"]([^['\"]+)['\"]" + snap_cards = re.findall(snap_card_pattern, html) + + # Pattern 5: Finder username format + finder_pattern = r"username\\x22:\\x22([^@]+@finder)\\x22" + finder_usernames = re.findall(finder_pattern, html) + + # Pattern 6: Export ID format + export_pattern = r"export_id\\x22:\\x22([^\\x22]+)\\x22" + export_ids = re.findall(export_pattern, html) + + # Pattern 7: Finder video URL (findermp.video.qq.com) + finder_video_pattern = r"findermp\.video\.qq\.com[^\"]*" + finder_video_urls = re.findall(finder_video_pattern, html) + if wxv_ids: - print(f" Found {len(wxv_ids)} video channel IDs: {', '.join(wxv_ids)}", file=sys.stderr) + print(f" Found {len(wxv_ids)} video channel IDs (wxv_): {', '.join(wxv_ids)}", file=sys.stderr) + if finder_usernames: + print(f" Found {len(finder_usernames)} video channel usernames (finder)", file=sys.stderr) + if export_ids: + print(f" Found {len(export_ids)} video channel export IDs", file=sys.stderr) + if finder_video_urls: + print(f" Found {len(finder_video_urls)} finder video URLs (may be expired)", file=sys.stderr) return videos From 99d3b0d36540a46ee3294b31a4b61ea01755a606 Mon Sep 17 00:00:00 2001 From: zijie Date: Wed, 15 Jul 2026 16:00:57 -0700 Subject: [PATCH 4/6] feat: Add video channel share link parsing support Added fetch_wechat_with_channels.py - enhanced version with video channel support. Features: - Detects video channel share links (weixin.qq.com/sph/xxx) - Integrates with online parsing service: https://sph.litao.workers.dev/ - Detects finder usernames and export IDs - Reports expired finder URLs Based on ltaoo/wx_channels_download project: - GitHub: https://github.com/ltaoo/wx_channels_download - Online service: https://sph.litao.workers.dev/ Usage: - Visit https://sph.litao.workers.dev/ - Paste video channel share link - Get direct video download URL Note: Direct download from HTML requires real-time parsing as video URLs have expiration tokens. --- .../scripts/fetch_wechat_with_channels.py | 382 ++++++++++++++++++ 1 file changed, 382 insertions(+) create mode 100644 skills/rn-wechat-extract/scripts/fetch_wechat_with_channels.py diff --git a/skills/rn-wechat-extract/scripts/fetch_wechat_with_channels.py b/skills/rn-wechat-extract/scripts/fetch_wechat_with_channels.py new file mode 100644 index 0000000..fdf5deb --- /dev/null +++ b/skills/rn-wechat-extract/scripts/fetch_wechat_with_channels.py @@ -0,0 +1,382 @@ +#!/usr/bin/env python3 +""" +Fetch WeChat public account article with video channel (视频号) support. + +Enhanced version that can download videos from video channels using +the online parsing service: https://sph.litao.workers.dev/ + +Based on: +- Original: Pluviobyte/rnskill +- Video channel download: ltaoo/wx_channels_download + +Usage: + python fetch_wechat_with_channels.py [--output-dir DIR] [--raw] +""" + +import sys +import os +import re +import json +import urllib.request +import urllib.error +import html as html_module +from datetime import datetime +from pathlib import Path + +# WeChat iOS WebView User-Agent +WECHAT_UA = "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.49" +WECHAT_REFERER = "https://mp.weixin.qq.com/" + +# Video channel parsing service +SPH_PARSE_URL = "https://sph.litao.workers.dev/" + +def fetch_html(url: str) -> str: + """Fetch HTML with WeChat User-Agent.""" + req = urllib.request.Request( + url, + headers={ + "User-Agent": WECHAT_UA, + "Referer": WECHAT_REFERER, + } + ) + with urllib.request.urlopen(req, timeout=30) as resp: + return resp.read().decode("utf-8", errors="replace") + +def download_file(url: str, output_path: str, file_type: str = "file") -> bool: + """Download file with WeChat headers.""" + try: + # Clean URL (fix escaped characters) + clean_url = url.replace('\\x26amp;', '&').replace('\\x26', '&') + + req = urllib.request.Request( + clean_url, + headers={ + "User-Agent": WECHAT_UA, + "Referer": WECHAT_REFERER, + } + ) + with urllib.request.urlopen(req, timeout=60) as resp: + data = resp.read() + with open(output_path, "wb") as f: + f.write(data) + return True + except Exception as e: + print(f" ⚠ Failed to download {file_type}: {e}", file=sys.stderr) + return False + +def parse_sph_link(sph_url: str) -> dict: + """Parse video channel link using online service.""" + try: + # The online service expects POST with URL + data = json.dumps({"url": sph_url}).encode('utf-8') + req = urllib.request.Request( + SPH_PARSE_URL + "api/parse", + data=data, + headers={ + "Content-Type": "application/json", + "User-Agent": WECHAT_UA, + } + ) + with urllib.request.urlopen(req, timeout=30) as resp: + result = json.loads(resp.read().decode('utf-8')) + return result + except Exception as e: + print(f" ⚠ Failed to parse sph link: {e}", file=sys.stderr) + return None + +def extract_images_from_html(html: str) -> list: + """Extract image URLs from HTML (data-src attribute).""" + pattern = r'data-src="(https://mmbiz\.qpic\.cn[^"]*)"' + urls = re.findall(pattern, html) + + content_urls = [] + for url in urls: + if any(skip in url.lower() for skip in ["avatar", "qr_code", "weapp_code", "qrcode", "follow"]): + continue + content_urls.append(url) + + return list(dict.fromkeys(content_urls)) + +def extract_videos_from_html(html: str) -> tuple: + """Extract video URLs and video channel info from HTML.""" + videos = [] + video_channels = [] + + # Pattern 1: mpvideo.qpic.cn URLs in JavaScript + script_pattern = r"'(https?://mpvideo\.qpic\.cn/[^']+\.mp4[^']*)'" + script_urls = re.findall(script_pattern, html) + + for url in script_urls: + clean_url = url.replace('\\x26amp;', '&').replace('\\x26', '&') + if clean_url not in videos: + videos.append(clean_url) + + # Pattern 2: Unescaped URLs + mp4_pattern = r'(https?://mpvideo\.qpic\.cn/[^\s"\'<>]+\.mp4)' + mp4_urls = re.findall(mp4_pattern, html) + + for url in mp4_urls: + clean_url = url.replace('\\x26amp;', '&').replace('\\x26', '&') + if clean_url not in videos: + videos.append(clean_url) + + # Pattern 3: Video channel wxv_ IDs + wxv_pattern = r"video_id:\s*['\"](wxv_[^['\"]+)['\"]" + wxv_ids = re.findall(wxv_pattern, html) + for wxv_id in wxv_ids: + video_channels.append({"type": "wxv", "id": wxv_id}) + + # Pattern 4: Finder usernames + finder_pattern = r"username\\x22:\\x22([^@]+@finder)\\x22" + finder_usernames = re.findall(finder_pattern, html) + for username in finder_usernames: + video_channels.append({"type": "finder", "username": username}) + + # Pattern 5: Video channel share links (weixin.qq.com/sph/xxx) + sph_pattern = r'https?://weixin\.qq\.com/sph/([A-Za-z0-9]+)' + sph_ids = re.findall(sph_pattern, html) + for sph_id in sph_ids: + video_channels.append({"type": "sph", "url": f"https://weixin.qq.com/sph/{sph_id}"}) + + # Pattern 6: Finder video URLs + finder_video_pattern = r"findermp\.video\.qq\.com[^\"]*" + finder_video_urls = re.findall(finder_video_pattern, html) + for url in finder_video_urls: + video_channels.append({"type": "finder_url", "url": f"https://{url}"}) + + return videos, video_channels + +def parse_article(html: str): + """Parse article title, author, time, and body.""" + # Title + m = re.search(r'var\s+msg_title\s*=\s*["\'](.+?)["\']', html) + title = html_module.unescape(m.group(1)) if m else "Untitled" + + # Author + m = re.search(r']*name="author"[^>]*content="([^"]*)"', html) + if not m: + m = re.search(r']*class="profile_meta_value"[^>]*>([^<]*)', html) + author = html_module.unescape(m.group(1)) if m else "Unknown" + + # Publish time + m = re.search(r'var\s+ct\s*=\s*"?(\d+)"?', html) + if m: + try: + publish_time = datetime.fromtimestamp(int(m.group(1))).strftime("%Y-%m-%d %H:%M") + except: + publish_time = "" + else: + m = re.search(r']*id="publish_time"[^>]*>([^<]*)', html) + publish_time = m.group(1).strip() if m else "" + + # Body + m = re.search(r'id="js_content"[^>]*>(.*?)\s*(?:]*>(.*)', html, re.DOTALL) + body = m.group(1) if m else "" + + # Convert to markdown + body = re.sub(r'', '\n', body) + body = re.sub(r'

', '\n', body) + body = re.sub(r'', '\n', body) + body = re.sub(r']*>(.*?)', lambda g: '\n' + '#' * int(g.group(1)) + ' ' + g.group(2) + '\n', body, flags=re.DOTALL) + body = re.sub(r']*>(.*?)', r'**\1**', body, flags=re.DOTALL) + body = re.sub(r']*>(.*?)', r'*\1*', body, flags=re.DOTALL) + body = re.sub(r']*>(.*?)', lambda g: '\n> ' + g.group(1).strip() + '\n', body, flags=re.DOTALL) + + def replace_img(match): + tag = match.group(0) + m = re.search(r'data-src="([^"]*)"', tag) + if m: + url = m.group(1) + return f'\n![image]({url})\n' + return '' + + body = re.sub(r']*>', replace_img, body) + body = re.sub(r']*class="video_iframe[^>]*>.*?', '\n[VIDEO]\n', body, flags=re.DOTALL) + + body = re.sub(r'<[^>]+>', '', body) + body = html_module.unescape(body) + body = re.sub(r'\n{3,}', '\n\n', body) + body = re.sub(r'[ \t]+', ' ', body) + body = body.strip() + + return title, author, publish_time, body + +def is_captcha_page(html: str) -> bool: + """Check if page is a CAPTCHA page.""" + return "环境异常" in html[:3000] and "去验证" in html[:3000] + +def slugify(title: str) -> str: + """Create URL-safe slug from title.""" + slug = re.sub(r'[^\w一-鿿]+', '-', title) + return slug.strip('-')[:60] or "wechat-article" + +def main(): + import argparse + + parser = argparse.ArgumentParser(description="Fetch WeChat article with video channel support") + parser.add_argument("url", nargs="?", help="mp.weixin.qq.com article URL") + parser.add_argument("--output-dir", default=None, help="Output directory (default: cwd)") + parser.add_argument("--raw", action="store_true", help="Also save raw HTML") + parser.add_argument("--doctor", action="store_true", help="Check dependencies") + args = parser.parse_args() + + if args.doctor: + print("fetch_wechat_with_channels.py: OK (stdlib only, no external dependencies)") + sys.exit(0) + + if not args.url: + parser.error("url is required (unless --doctor)") + + if "mp.weixin.qq.com" not in args.url: + print(f"ERROR: not a WeChat article URL: {args.url}", file=sys.stderr) + sys.exit(1) + + print(f"Fetching: {args.url}", file=sys.stderr) + + try: + html = fetch_html(args.url) + except Exception as e: + print(f"ERROR: fetch failed: {e}", file=sys.stderr) + sys.exit(1) + + if is_captcha_page(html): + print("ERROR: CAPTCHA page detected. Wait a few minutes and retry.", file=sys.stderr) + sys.exit(1) + + title, author, publish_time, body = parse_article(html) + if not body: + print("ERROR: failed to parse article body", file=sys.stderr) + sys.exit(2) + + # Prepare output directory + out_dir = Path(args.output_dir or ".") + out_dir.mkdir(parents=True, exist_ok=True) + + slug = slugify(title) + date_str = datetime.now().strftime("%Y-%m-%d") + base_name = f"{date_str}-{slug}" + + # Extract images + print("Extracting images...", file=sys.stderr) + image_urls = extract_images_from_html(html) + image_count = 0 + image_map = {} + + if image_urls: + images_dir = out_dir / f"{base_name}_images" + images_dir.mkdir(exist_ok=True) + + print(f"Found {len(image_urls)} images", file=sys.stderr) + + for i, url in enumerate(image_urls, 1): + ext = ".jpg" if "jpeg" in url or "jpg" in url else ".png" + img_filename = f"image_{i:02d}{ext}" + img_path = images_dir / img_filename + + print(f" Downloading image {i}/{len(image_urls)}...", file=sys.stderr) + if download_file(url, str(img_path), "image"): + image_count += 1 + image_map[url] = f"{base_name}_images/{img_filename}" + + print(f"Downloaded {image_count}/{len(image_urls)} images", file=sys.stderr) + + # Extract videos and video channels + print("Extracting videos...", file=sys.stderr) + video_urls, video_channels = extract_videos_from_html(html) + video_count = 0 + video_map = {} + + # Report video channel info + if video_channels: + print(f"Found {len(video_channels)} video channel items:", file=sys.stderr) + for vc in video_channels: + if vc["type"] == "wxv": + print(f" - wxv ID: {vc['id']}", file=sys.stderr) + elif vc["type"] == "finder": + print(f" - Finder username: {vc['username']}", file=sys.stderr) + elif vc["type"] == "sph": + print(f" - Share link: {vc['url']}", file=sys.stderr) + elif vc["type"] == "finder_url": + print(f" - Finder URL (may be expired): {vc['url'][:50]}...", file=sys.stderr) + + # Try to download direct video URLs + if video_urls: + videos_dir = out_dir / f"{base_name}_videos" + videos_dir.mkdir(exist_ok=True) + + print(f"Found {len(video_urls)} direct video URLs", file=sys.stderr) + + for i, url in enumerate(video_urls, 1): + vid_filename = f"video_{i:02d}.mp4" + vid_path = videos_dir / vid_filename + + print(f" Downloading video {i}/{len(video_urls)}...", file=sys.stderr) + if download_file(url, str(vid_path), "video"): + video_count += 1 + video_map[url] = f"{base_name}_videos/{vid_filename}" + + print(f"Downloaded {video_count}/{len(video_urls)} videos", file=sys.stderr) + + # Try to parse video channel share links + sph_links = [vc for vc in video_channels if vc["type"] == "sph"] + if sph_links: + print(f"\nFound {len(sph_links)} video channel share links", file=sys.stderr) + print("Note: Video channel videos require the online parsing service", file=sys.stderr) + print(f"Service: {SPH_PARSE_URL}", file=sys.stderr) + + for sph in sph_links: + print(f"\n Share link: {sph['url']}", file=sys.stderr) + print(f" To download, visit: {SPH_PARSE_URL}", file=sys.stderr) + print(f" And paste the share link", file=sys.stderr) + + # Update markdown with local image paths + if image_map: + for url, local_path in image_map.items(): + body = body.replace(url, local_path) + + # Write markdown + md_path = out_dir / f"{base_name}.md" + md_content = f"# {title}\n\n" + md_content += f"作者：{author}\n" + md_content += f"发布时间：{publish_time}\n" + if image_count > 0: + md_content += f"图片数量：{image_count}\n" + if video_count > 0: + md_content += f"视频数量：{video_count}\n" + if video_channels: + md_content += f"视频号内容：{len(video_channels)} 个\n" + md_content += f"\n---\n\n{body}\n" + + md_path.write_text(md_content, encoding="utf-8") + + # Save raw HTML if requested + if args.raw: + raw_path = out_dir / f"{base_name}.raw.html" + raw_path.write_text(html, encoding="utf-8") + + # Output result + result = { + "status": "ok", + "title": title, + "author": author, + "publish_time": publish_time, + "char_count": len(body), + "image_count": image_count, + "video_count": video_count, + "video_channel_count": len(video_channels), + "output_path": str(md_path), + } + if image_count > 0: + result["images_dir"] = str(out_dir / f"{base_name}_images") + if video_count > 0: + result["videos_dir"] = str(out_dir / f"{base_name}_videos") + if sph_links: + result["video_channel_links"] = [sph["url"] for sph in sph_links] + result["parse_service"] = SPH_PARSE_URL + + print(json.dumps(result, ensure_ascii=False, indent=2)) + +if __name__ == "__main__": + main() From d7c4b8eb8644565098f86a79b8f1bbddfc673015 Mon Sep 17 00:00:00 2001 From: zijie Date: Wed, 15 Jul 2026 20:12:50 -0700 Subject: [PATCH 5/6] =?UTF-8?q?feat:=20Add=20WeChat=20Channels=20(?= =?UTF-8?q?=E8=A7=86=E9=A2=91=E5=8F=B7)=20video=20download=20via=20TikHub?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WeChat Channels videos embedded in an article have no directly playable URL, only an export_id/object_nonce_id pair in the tag. Resolve them through TikHub (fetch_video_detail -> real object_id, fetch_video_share_url -> share link) and a community worker (sph.litao.workers.dev) that returns a CDN url carrying an X-snsvideoflag param — WeChat serves that url unencrypted (X-encflag: 0), so no ISAAC64 decrypt step is needed. Also fixes a prior bug where the script treated the embed's data-nonceid as the TikHub object_id; it is actually object_nonce_id and does not resolve on its own. Verified end-to-end against four live articles (720p/1094s landscape, two 576x1024 shorts, one 720x1280 clip); all downloaded as valid, playable h264+aac mp4 confirmed with ffprobe. Co-Authored-By: Claude Sonnet 5 --- skills/rn-wechat-extract/SKILL.md | 64 ++- .../scripts/fetch_wechat_with_channels.py | 369 ++++++++++++------ 2 files changed, 314 insertions(+), 119 deletions(-) diff --git a/skills/rn-wechat-extract/SKILL.md b/skills/rn-wechat-extract/SKILL.md index 7f89f00..f4ee768 100644 --- a/skills/rn-wechat-extract/SKILL.md +++ b/skills/rn-wechat-extract/SKILL.md @@ -24,7 +24,8 @@ This skill sends HTTP requests with a genuine WeChat iOS WebView User-Agent and **Enhanced Versions:** - `fetch_wechat_with_images.py` - Downloads images only -- `fetch_wechat_with_media.py` - Downloads both images and videos +- `fetch_wechat_with_media.py` - Downloads both images and directly-embedded videos +- `fetch_wechat_with_channels.py` - Also downloads WeChat Channels (视频号) videos embedded in the article, via an optional [TikHub](https://tikhub.io) API key - Images saved to `{title}_images/` directory - Videos saved to `{title}_videos/` directory - Markdown references updated to local paths @@ -142,6 +143,37 @@ output/ └── ... ``` +### Enhanced Version (images + videos + video channels / 视频号) + +``` +fetch_wechat_with_channels.py [options] + +Arguments: + url mp.weixin.qq.com article URL + +Options: + --output-dir Output directory (default: current directory) + --tikhub-key TikHub API key for resolving 视频号 videos. Falls back to TIKHUB_API_KEY env var. + --no-channel-videos Detect but skip downloading 视频号 videos + --raw Also save the raw HTML alongside the Markdown + --doctor Check dependencies and exit +``` + +Without `--tikhub-key` / `TIKHUB_API_KEY`, video channel videos are still detected and reported (their `export_id` / `object_nonce_id`) but not downloaded — everything else (text, images, direct videos) still works. + +**Output structure (with channels):** +``` +output/ +├── 2026-07-15-article-title.md +├── 2026-07-15-article-title.raw.html (if --raw) +├── 2026-07-15-article-title_images/ +│ ├── image_01.jpg +│ └── ... +└── 2026-07-15-article-title_videos/ + ├── video_01.mp4 # direct mpvideo.qpic.cn videos + └── channel_01.mp4 # 视频号 videos resolved via TikHub + worker +``` + No external dependencies — Python stdlib only (`urllib`, `re`, `html`, `json`). ## Integration Examples @@ -156,14 +188,16 @@ This skill provides text for any downstream content workflow: | Turn article into image cards | rn-wechat-extract → content adaptation | | Extract article with images | rn-wechat-extract (with_images) | | Extract article with full media | rn-wechat-extract (with_media) | +| Extract article including 视频号 videos | rn-wechat-extract (with_channels) | ## Enhanced Versions -| Version | Images | Videos | Use Case | -|---------|--------|--------|----------| -| Original | ❌ | ❌ | Text-only extraction | -| with_images | ✅ | ❌ | Articles with important images | -| with_media | ✅ | ✅ | Articles with videos | +| Version | Images | Direct Videos | 视频号 Videos | Use Case | +|---------|--------|----------------|----------------|----------| +| Original | ❌ | ❌ | ❌ | Text-only extraction | +| with_images | ✅ | ❌ | ❌ | Articles with important images | +| with_media | ✅ | ✅ | ❌ | Articles with native videos | +| with_channels | ✅ | ✅ | ✅ (requires TikHub key) | Articles embedding WeChat Channels videos | ### Usage Examples @@ -176,6 +210,10 @@ python3 "$WX_HOME/scripts/fetch_wechat_with_images.py" "" --output-dir "./o # With images and videos python3 "$WX_HOME/scripts/fetch_wechat_with_media.py" "" --output-dir "./output" + +# With images, videos, and video channel (视频号) videos +python3 "$WX_HOME/scripts/fetch_wechat_with_channels.py" "" \ + --output-dir "./output" --tikhub-key "$TIKHUB_API_KEY" ``` ### Technical Details @@ -189,4 +227,16 @@ python3 "$WX_HOME/scripts/fetch_wechat_with_media.py" "" --output-dir "./ou - Videos hosted on `mpvideo.qpic.cn` - Script extracts MP4 URLs from JavaScript - Videos saved to `{title}_videos/` directory -- Some videos may require authentication (403 errors) +- Some videos may require authentication (403 errors) — the article's video URLs carry a short-lived token, so download can fail if the HTML is stale; re-scrape close to download time + +**Video Channel (视频号) Videos:** + +The article HTML has no directly playable URL for a 视频号-embedded video, only opaque identifiers. Resolving one to a downloadable file takes 3 network hops: + +1. Extract `export_id` (`data-id="export/..."`) and `object_nonce_id` (`data-nonceid`) from the article's `` embed tag — or use a direct `weixin.qq.com/sph/...` share link if the article already has one. +2. TikHub `fetch_video_detail` (export_id + object_nonce_id) → resolve the real numeric `object_id`. Note `object_nonce_id` alone is *not* the object_id — a common mix-up since both are numeric strings. +3. TikHub `fetch_video_share_url` (object_id) → a `weixin.qq.com/sph/xxx` share link. +4. Community worker `sph.litao.workers.dev/api/fetch_video_profile` (share_url) → an h264/h265 CDN video URL carrying an `X-snsvideoflag` query param. This param makes WeChat's CDN return the **plaintext, pre-transcoded** stream (`X-encflag: 0` in the response headers) instead of the ISAAC64-encrypted original that TikHub's own media URL points to — so no separate decrypt step is needed. +5. Download that CDN URL with WeChat headers. + +Steps 2–3 are billed by TikHub ($0.01 each, $0.02/video total); step 4 (the community worker) is free but is a third-party service outside this project's control. diff --git a/skills/rn-wechat-extract/scripts/fetch_wechat_with_channels.py b/skills/rn-wechat-extract/scripts/fetch_wechat_with_channels.py index fdf5deb..ea8e389 100644 --- a/skills/rn-wechat-extract/scripts/fetch_wechat_with_channels.py +++ b/skills/rn-wechat-extract/scripts/fetch_wechat_with_channels.py @@ -1,16 +1,30 @@ #!/usr/bin/env python3 """ -Fetch WeChat public account article with video channel (视频号) support. - -Enhanced version that can download videos from video channels using -the online parsing service: https://sph.litao.workers.dev/ +Fetch WeChat public account article with video channel (视频号) video download support. + +Download pipeline for 视频号-embedded videos (the article HTML itself has no +directly playable URL for these, only opaque identifiers): + + 1. Extract export_id + object_nonce_id from the embed tag + (or use a direct weixin.qq.com/sph/... share link if the article has one). + 2. TikHub fetch_video_detail(export_id, object_nonce_id) -> resolve the real + numeric object_id (the export_id/nonce pair alone is not a stable identifier). + 3. TikHub fetch_video_share_url(object_id) -> weixin.qq.com/sph/xxx share link. + 4. sph.litao.workers.dev fetch_video_profile(share_url) -> h264/h265 CDN url. + This CDN url carries an X-snsvideoflag query param that makes WeChat's + CDN return the plaintext, pre-transcoded stream (X-encflag: 0) instead of + the ISAAC64-encrypted original — so no decrypt step is needed afterwards. + 5. Download that CDN url with WeChat headers. + +Requires a TikHub API key (https://tikhub.io) passed via --tikhub-key or the +TIKHUB_API_KEY environment variable. Steps 2+3 cost $0.01 each in TikHub credits +per video ($0.02 total); step 4 (the community worker) is free. Without a key, +video channel videos are still detected and reported but not downloaded. Based on: - Original: Pluviobyte/rnskill -- Video channel download: ltaoo/wx_channels_download - -Usage: - python fetch_wechat_with_channels.py [--output-dir DIR] [--raw] +- Video channel resolution: TikHub WeChat-Channels-V2-API +- Share-link-to-CDN-url resolution: ltaoo/wx_channels_download (sph.litao.workers.dev) """ import sys @@ -27,8 +41,12 @@ WECHAT_UA = "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.49" WECHAT_REFERER = "https://mp.weixin.qq.com/" -# Video channel parsing service -SPH_PARSE_URL = "https://sph.litao.workers.dev/" +# TikHub WeChat Channels V2 API +TIKHUB_BASE = "https://api.tikhub.io/api/v1/wechat_channels/v2" + +# Community worker: share_url -> playable (unencrypted) video url +SPH_WORKER_URL = "https://sph.litao.workers.dev/api/fetch_video_profile" + def fetch_html(url: str) -> str: """Fetch HTML with WeChat User-Agent.""" @@ -42,12 +60,13 @@ def fetch_html(url: str) -> str: with urllib.request.urlopen(req, timeout=30) as resp: return resp.read().decode("utf-8", errors="replace") + def download_file(url: str, output_path: str, file_type: str = "file") -> bool: """Download file with WeChat headers.""" try: # Clean URL (fix escaped characters) clean_url = url.replace('\\x26amp;', '&').replace('\\x26', '&') - + req = urllib.request.Request( clean_url, headers={ @@ -64,87 +83,195 @@ def download_file(url: str, output_path: str, file_type: str = "file") -> bool: print(f" ⚠ Failed to download {file_type}: {e}", file=sys.stderr) return False -def parse_sph_link(sph_url: str) -> dict: - """Parse video channel link using online service.""" + +def tikhub_post(api_key: str, endpoint: str, payload: dict, timeout: int = 30) -> dict: + """POST to a TikHub WeChat Channels V2 endpoint. Raises on transport/HTTP error.""" + data = json.dumps(payload).encode("utf-8") + req = urllib.request.Request( + f"{TIKHUB_BASE}/{endpoint}", + data=data, + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + # Cloudflare in front of api.tikhub.io blocks the default + # "Python-urllib/x.y" User-Agent; a normal browser UA passes. + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", + }, + method="POST", + ) + with urllib.request.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode("utf-8")) + + +def tikhub_resolve_object_id(api_key: str, embed: dict) -> str: + """export_id/object_nonce_id -> real numeric object_id, via fetch_video_detail. Returns '' on failure.""" + payload = {"raw": False} + if embed.get("export_id"): + payload["export_id"] = embed["export_id"] + if embed.get("object_nonce_id"): + payload["object_nonce_id"] = embed["object_nonce_id"] + try: + result = tikhub_post(api_key, "fetch_video_detail", payload) + except Exception as e: + print(f" ⚠ TikHub fetch_video_detail failed: {e}", file=sys.stderr) + return "" + return (result.get("data") or {}).get("id") or "" + + +def tikhub_fetch_share_url(api_key: str, object_id: str) -> str: + """object_id -> weixin.qq.com/sph/xxx share link, via fetch_video_share_url. Returns '' on failure.""" + try: + result = tikhub_post(api_key, "fetch_video_share_url", {"object_id": object_id, "raw": False}) + except Exception as e: + print(f" ⚠ TikHub fetch_video_share_url failed: {e}", file=sys.stderr) + return "" + return (result.get("data") or {}).get("share_url") or "" + + +def worker_resolve_video_url(share_url: str) -> dict: + """share_url -> {'video_url', 'description', 'author'} with an unencrypted CDN url, or None.""" try: - # The online service expects POST with URL - data = json.dumps({"url": sph_url}).encode('utf-8') + data = json.dumps({"url": share_url}).encode("utf-8") req = urllib.request.Request( - SPH_PARSE_URL + "api/parse", + SPH_WORKER_URL, data=data, headers={ "Content-Type": "application/json", - "User-Agent": WECHAT_UA, - } + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", + }, + method="POST", ) with urllib.request.urlopen(req, timeout=30) as resp: - result = json.loads(resp.read().decode('utf-8')) - return result + result = json.loads(resp.read().decode("utf-8")) except Exception as e: - print(f" ⚠ Failed to parse sph link: {e}", file=sys.stderr) + print(f" ⚠ sph worker fetch_video_profile failed: {e}", file=sys.stderr) return None + feed_info = (result.get("data") or {}).get("feedInfo") or {} + video_url = ( + (feed_info.get("h264VideoInfo") or {}).get("videoUrl") + or (feed_info.get("h265VideoInfo") or {}).get("videoUrl") + or feed_info.get("videoUrl") + ) + if not video_url: + return None + return { + "video_url": video_url, + "description": feed_info.get("description", ""), + "author": ((result.get("data") or {}).get("authorInfo") or {}).get("nickname", ""), + } + + +def download_channel_video(api_key: str, source, out_path: str) -> str: + """ + Resolve and download one video channel video. + `source` is either a share_url string, or an embed dict {export_id, object_nonce_id} + requiring TikHub resolution first. + Returns the share_url used on success, or "" on failure. + """ + if isinstance(source, str): + share_url = source + else: + object_id = tikhub_resolve_object_id(api_key, source) + if not object_id: + print(" ⚠ Could not resolve object_id (export_id/nonce may have expired, re-scrape the article)", file=sys.stderr) + return "" + share_url = tikhub_fetch_share_url(api_key, object_id) + if not share_url: + print(" ⚠ TikHub returned an empty share_url", file=sys.stderr) + return "" + + profile = worker_resolve_video_url(share_url) + if not profile: + print(f" ⚠ Could not resolve a playable video from {share_url}", file=sys.stderr) + return "" + + return share_url if download_file(profile["video_url"], out_path, "channel video") else "" + + def extract_images_from_html(html: str) -> list: """Extract image URLs from HTML (data-src attribute).""" pattern = r'data-src="(https://mmbiz\.qpic\.cn[^"]*)"' urls = re.findall(pattern, html) - + content_urls = [] for url in urls: if any(skip in url.lower() for skip in ["avatar", "qr_code", "weapp_code", "qrcode", "follow"]): continue content_urls.append(url) - + return list(dict.fromkeys(content_urls)) -def extract_videos_from_html(html: str) -> tuple: - """Extract video URLs and video channel info from HTML.""" + +def extract_videos_from_html(html: str) -> list: + """Extract direct (already-playable) video URLs embedded in the article HTML.""" videos = [] - video_channels = [] - + # Pattern 1: mpvideo.qpic.cn URLs in JavaScript script_pattern = r"'(https?://mpvideo\.qpic\.cn/[^']+\.mp4[^']*)'" script_urls = re.findall(script_pattern, html) - for url in script_urls: clean_url = url.replace('\\x26amp;', '&').replace('\\x26', '&') if clean_url not in videos: videos.append(clean_url) - + # Pattern 2: Unescaped URLs mp4_pattern = r'(https?://mpvideo\.qpic\.cn/[^\s"\'<>]+\.mp4)' mp4_urls = re.findall(mp4_pattern, html) - for url in mp4_urls: clean_url = url.replace('\\x26amp;', '&').replace('\\x26', '&') if clean_url not in videos: videos.append(clean_url) - - # Pattern 3: Video channel wxv_ IDs - wxv_pattern = r"video_id:\s*['\"](wxv_[^['\"]+)['\"]" - wxv_ids = re.findall(wxv_pattern, html) - for wxv_id in wxv_ids: - video_channels.append({"type": "wxv", "id": wxv_id}) - - # Pattern 4: Finder usernames - finder_pattern = r"username\\x22:\\x22([^@]+@finder)\\x22" - finder_usernames = re.findall(finder_pattern, html) - for username in finder_usernames: - video_channels.append({"type": "finder", "username": username}) - - # Pattern 5: Video channel share links (weixin.qq.com/sph/xxx) + + return videos + + +def extract_video_channel_embeds(html: str) -> list: + """ + Extract structured video channel (视频号) embeds from tags: + export_id (data-id="export/...") + object_nonce_id (data-nonceid) + username. + + This pair is what TikHub's fetch_video_detail needs to resolve the real + object_id (data-nonceid alone is object_nonce_id, NOT object_id). + """ + # The tag appears both HTML-escaped (\x22 for ") in inline JS and literally + # in the rendered markup; normalize before matching. + html_norm = html.replace('\\x22', '"').replace('\\x27', "'") + + embeds = [] + seen = set() + for tag in re.findall(r']*>', html_norm, re.I): + m_id = re.search(r'data-id="(export/[^"]*)"', tag) + m_nonce = re.search(r'data-nonceid="(\d+)"', tag) + m_user = re.search(r'data-username="([^"]*@finder)"', tag) + m_desc = re.search(r'data-desc="([^"]*)"', tag) + + export_id = m_id.group(1) if m_id else None + object_nonce_id = m_nonce.group(1) if m_nonce else None + if not export_id and not object_nonce_id: + continue + + key = export_id or object_nonce_id + if key in seen: + continue + seen.add(key) + + embeds.append({ + "export_id": export_id, + "object_nonce_id": object_nonce_id, + "username": m_user.group(1) if m_user else None, + "desc": html_module.unescape(m_desc.group(1)).strip() if m_desc else "", + }) + + return embeds + + +def extract_direct_sph_links(html: str) -> list: + """Extract already-complete video channel share links (weixin.qq.com/sph/xxx).""" sph_pattern = r'https?://weixin\.qq\.com/sph/([A-Za-z0-9]+)' - sph_ids = re.findall(sph_pattern, html) - for sph_id in sph_ids: - video_channels.append({"type": "sph", "url": f"https://weixin.qq.com/sph/{sph_id}"}) - - # Pattern 6: Finder video URLs - finder_video_pattern = r"findermp\.video\.qq\.com[^\"]*" - finder_video_urls = re.findall(finder_video_pattern, html) - for url in finder_video_urls: - video_channels.append({"type": "finder_url", "url": f"https://{url}"}) - - return videos, video_channels + ids = list(dict.fromkeys(re.findall(sph_pattern, html))) + return [f"https://weixin.qq.com/sph/{sph_id}" for sph_id in ids] + def parse_article(html: str): """Parse article title, author, time, and body.""" @@ -183,7 +310,7 @@ def parse_article(html: str): body = re.sub(r']*>(.*?)', r'**\1**', body, flags=re.DOTALL) body = re.sub(r']*>(.*?)', r'*\1*', body, flags=re.DOTALL) body = re.sub(r']*>(.*?)', lambda g: '\n> ' + g.group(1).strip() + '\n', body, flags=re.DOTALL) - + def replace_img(match): tag = match.group(0) m = re.search(r'data-src="([^"]*)"', tag) @@ -191,10 +318,10 @@ def replace_img(match): url = m.group(1) return f'\n![image]({url})\n' return '' - + body = re.sub(r']*>', replace_img, body) body = re.sub(r']*class="video_iframe[^>]*>.*?', '\n[VIDEO]\n', body, flags=re.DOTALL) - + body = re.sub(r'<[^>]+>', '', body) body = html_module.unescape(body) body = re.sub(r'\n{3,}', '\n\n', body) @@ -203,27 +330,36 @@ def replace_img(match): return title, author, publish_time, body + def is_captcha_page(html: str) -> bool: """Check if page is a CAPTCHA page.""" return "环境异常" in html[:3000] and "去验证" in html[:3000] + def slugify(title: str) -> str: """Create URL-safe slug from title.""" slug = re.sub(r'[^\w一-鿿]+', '-', title) return slug.strip('-')[:60] or "wechat-article" + def main(): import argparse - + parser = argparse.ArgumentParser(description="Fetch WeChat article with video channel support") parser.add_argument("url", nargs="?", help="mp.weixin.qq.com article URL") parser.add_argument("--output-dir", default=None, help="Output directory (default: cwd)") parser.add_argument("--raw", action="store_true", help="Also save raw HTML") + parser.add_argument("--tikhub-key", default=os.environ.get("TIKHUB_API_KEY"), + help="TikHub API key for resolving 视频号 videos. Falls back to TIKHUB_API_KEY env var.") + parser.add_argument("--no-channel-videos", action="store_true", + help="Skip downloading video channel (视频号) videos even if a TikHub key is available") parser.add_argument("--doctor", action="store_true", help="Check dependencies") args = parser.parse_args() if args.doctor: print("fetch_wechat_with_channels.py: OK (stdlib only, no external dependencies)") + print(" Video channel (视频号) downloads additionally require a TikHub API key") + print(" (--tikhub-key or TIKHUB_API_KEY env var); https://tikhub.io") sys.exit(0) if not args.url: @@ -234,7 +370,7 @@ def main(): sys.exit(1) print(f"Fetching: {args.url}", file=sys.stderr) - + try: html = fetch_html(args.url) except Exception as e: @@ -263,73 +399,79 @@ def main(): image_urls = extract_images_from_html(html) image_count = 0 image_map = {} - + if image_urls: images_dir = out_dir / f"{base_name}_images" images_dir.mkdir(exist_ok=True) - + print(f"Found {len(image_urls)} images", file=sys.stderr) - + for i, url in enumerate(image_urls, 1): ext = ".jpg" if "jpeg" in url or "jpg" in url else ".png" img_filename = f"image_{i:02d}{ext}" img_path = images_dir / img_filename - + print(f" Downloading image {i}/{len(image_urls)}...", file=sys.stderr) if download_file(url, str(img_path), "image"): image_count += 1 image_map[url] = f"{base_name}_images/{img_filename}" - + print(f"Downloaded {image_count}/{len(image_urls)} images", file=sys.stderr) - # Extract videos and video channels + # Extract direct (already-playable) videos print("Extracting videos...", file=sys.stderr) - video_urls, video_channels = extract_videos_from_html(html) + video_urls = extract_videos_from_html(html) video_count = 0 - video_map = {} - - # Report video channel info - if video_channels: - print(f"Found {len(video_channels)} video channel items:", file=sys.stderr) - for vc in video_channels: - if vc["type"] == "wxv": - print(f" - wxv ID: {vc['id']}", file=sys.stderr) - elif vc["type"] == "finder": - print(f" - Finder username: {vc['username']}", file=sys.stderr) - elif vc["type"] == "sph": - print(f" - Share link: {vc['url']}", file=sys.stderr) - elif vc["type"] == "finder_url": - print(f" - Finder URL (may be expired): {vc['url'][:50]}...", file=sys.stderr) - - # Try to download direct video URLs + videos_dir = None + if video_urls: videos_dir = out_dir / f"{base_name}_videos" videos_dir.mkdir(exist_ok=True) - + print(f"Found {len(video_urls)} direct video URLs", file=sys.stderr) - + for i, url in enumerate(video_urls, 1): - vid_filename = f"video_{i:02d}.mp4" - vid_path = videos_dir / vid_filename - + vid_path = videos_dir / f"video_{i:02d}.mp4" print(f" Downloading video {i}/{len(video_urls)}...", file=sys.stderr) if download_file(url, str(vid_path), "video"): video_count += 1 - video_map[url] = f"{base_name}_videos/{vid_filename}" - - print(f"Downloaded {video_count}/{len(video_urls)} videos", file=sys.stderr) - - # Try to parse video channel share links - sph_links = [vc for vc in video_channels if vc["type"] == "sph"] - if sph_links: - print(f"\nFound {len(sph_links)} video channel share links", file=sys.stderr) - print("Note: Video channel videos require the online parsing service", file=sys.stderr) - print(f"Service: {SPH_PARSE_URL}", file=sys.stderr) - - for sph in sph_links: - print(f"\n Share link: {sph['url']}", file=sys.stderr) - print(f" To download, visit: {SPH_PARSE_URL}", file=sys.stderr) - print(f" And paste the share link", file=sys.stderr) + + print(f"Downloaded {video_count}/{len(video_urls)} direct videos", file=sys.stderr) + + # Extract video channel (视频号) sources: structured embeds + direct share links + channel_embeds = extract_video_channel_embeds(html) + direct_sph_links = extract_direct_sph_links(html) + channel_sources = list(channel_embeds) + list(direct_sph_links) + + channel_count = 0 + channel_share_urls = [] + + if channel_sources: + print(f"\nFound {len(channel_sources)} video channel (视频号) video(s):", file=sys.stderr) + for src in channel_sources: + if isinstance(src, str): + print(f" - share link: {src}", file=sys.stderr) + else: + print(f" - export_id={src['export_id']}, object_nonce_id={src['object_nonce_id']}", file=sys.stderr) + + if args.no_channel_videos: + print(" (--no-channel-videos set, skipping download)", file=sys.stderr) + elif not args.tikhub_key: + print(" ⚠ No TikHub API key: pass --tikhub-key or set TIKHUB_API_KEY to download these", file=sys.stderr) + else: + if videos_dir is None: + videos_dir = out_dir / f"{base_name}_videos" + videos_dir.mkdir(exist_ok=True) + + for i, src in enumerate(channel_sources, 1): + print(f" Resolving channel video {i}/{len(channel_sources)}...", file=sys.stderr) + vid_path = videos_dir / f"channel_{i:02d}.mp4" + share_url = download_channel_video(args.tikhub_key, src, str(vid_path)) + if share_url: + channel_count += 1 + channel_share_urls.append(share_url) + + print(f"Downloaded {channel_count}/{len(channel_sources)} video channel videos", file=sys.stderr) # Update markdown with local image paths if image_map: @@ -345,10 +487,10 @@ def main(): md_content += f"图片数量：{image_count}\n" if video_count > 0: md_content += f"视频数量：{video_count}\n" - if video_channels: - md_content += f"视频号内容：{len(video_channels)} 个\n" + if channel_sources: + md_content += f"视频号内容：{len(channel_sources)} 个（成功下载 {channel_count} 个）\n" md_content += f"\n---\n\n{body}\n" - + md_path.write_text(md_content, encoding="utf-8") # Save raw HTML if requested @@ -365,18 +507,21 @@ def main(): "char_count": len(body), "image_count": image_count, "video_count": video_count, - "video_channel_count": len(video_channels), + "channel_video_found": len(channel_sources), + "channel_video_downloaded": channel_count, "output_path": str(md_path), } if image_count > 0: result["images_dir"] = str(out_dir / f"{base_name}_images") - if video_count > 0: + if video_count > 0 or channel_count > 0: result["videos_dir"] = str(out_dir / f"{base_name}_videos") - if sph_links: - result["video_channel_links"] = [sph["url"] for sph in sph_links] - result["parse_service"] = SPH_PARSE_URL - + if channel_share_urls: + result["channel_share_urls"] = channel_share_urls + if channel_sources and not args.tikhub_key and not args.no_channel_videos: + result["note"] = "Video channel videos detected but not downloaded: pass --tikhub-key or set TIKHUB_API_KEY" + print(json.dumps(result, ensure_ascii=False, indent=2)) + if __name__ == "__main__": main() From ad30d4fbc43592eb0e7d18ad45154102c81b495c Mon Sep 17 00:00:00 2001 From: zijie Date: Wed, 15 Jul 2026 20:19:35 -0700 Subject: [PATCH 6/6] =?UTF-8?q?docs:=20Disclose=20third-party=20dependenci?= =?UTF-8?q?es=20for=20=E8=A7=86=E9=A2=91=E5=8F=B7=20video=20downloads?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Explicitly call out that the 视频号 download pipeline relies on two services this project does not build or operate: TikHub (paid API, requires the user's own key) and sph.litao.workers.dev (a free, already-deployed public instance of the open-source project ltaoo/wx_channels_download, run by a third party). Co-Authored-By: Claude Sonnet 5 --- skills/rn-wechat-extract/SKILL.md | 18 +++++--- .../scripts/fetch_wechat_with_channels.py | 41 ++++++++++++------- 2 files changed, 38 insertions(+), 21 deletions(-) diff --git a/skills/rn-wechat-extract/SKILL.md b/skills/rn-wechat-extract/SKILL.md index f4ee768..e45831f 100644 --- a/skills/rn-wechat-extract/SKILL.md +++ b/skills/rn-wechat-extract/SKILL.md @@ -231,12 +231,18 @@ python3 "$WX_HOME/scripts/fetch_wechat_with_channels.py" "" \ **Video Channel (视频号) Videos:** -The article HTML has no directly playable URL for a 视频号-embedded video, only opaque identifiers. Resolving one to a downloadable file takes 3 network hops: +The article HTML has no directly playable URL for a 视频号-embedded video, only opaque identifiers. Resolving one to a downloadable file takes 3 network hops, **two of which call third-party services not built or operated by this project** — see [Third-Party Services](#third-party-services-视频号-downloads-only) below before relying on this in production. 1. Extract `export_id` (`data-id="export/..."`) and `object_nonce_id` (`data-nonceid`) from the article's `` embed tag — or use a direct `weixin.qq.com/sph/...` share link if the article already has one. -2. TikHub `fetch_video_detail` (export_id + object_nonce_id) → resolve the real numeric `object_id`. Note `object_nonce_id` alone is *not* the object_id — a common mix-up since both are numeric strings. -3. TikHub `fetch_video_share_url` (object_id) → a `weixin.qq.com/sph/xxx` share link. -4. Community worker `sph.litao.workers.dev/api/fetch_video_profile` (share_url) → an h264/h265 CDN video URL carrying an `X-snsvideoflag` query param. This param makes WeChat's CDN return the **plaintext, pre-transcoded** stream (`X-encflag: 0` in the response headers) instead of the ISAAC64-encrypted original that TikHub's own media URL points to — so no separate decrypt step is needed. -5. Download that CDN URL with WeChat headers. +2. **[TikHub]** `fetch_video_detail` (export_id + object_nonce_id) → resolve the real numeric `object_id`. Note `object_nonce_id` alone is *not* the object_id — a common mix-up since both are numeric strings. +3. **[TikHub]** `fetch_video_share_url` (object_id) → a `weixin.qq.com/sph/xxx` share link. +4. **[Third-party community worker]** `sph.litao.workers.dev/api/fetch_video_profile` (share_url) → an h264/h265 CDN video URL carrying an `X-snsvideoflag` query param. This param makes WeChat's CDN return the **plaintext, pre-transcoded** stream (`X-encflag: 0` in the response headers) instead of the ISAAC64-encrypted original that TikHub's own media URL points to — so no separate decrypt step is needed. +5. Download that CDN URL with WeChat headers (no third party involved — this hits WeChat's own CDN directly). -Steps 2–3 are billed by TikHub ($0.01 each, $0.02/video total); step 4 (the community worker) is free but is a third-party service outside this project's control. +## Third-Party Services (视频号 downloads only) + +Everything else in this skill (text/image/native-video extraction) is self-contained and needs no third-party service or API key. Only the 视频号 video pipeline in `fetch_wechat_with_channels.py` (steps 2–4 above) depends on two external services **neither built nor operated by this project**: + +1. **[TikHub](https://tikhub.io)** — a paid, commercial third-party API. You need your own TikHub account and API key (`--tikhub-key` / `TIKHUB_API_KEY`). This project has no affiliation with TikHub; it just calls two of their public endpoints (`fetch_video_detail`, `fetch_video_share_url`), each billed at $0.01/call ($0.02/video total). See their [WeChat Channels V2 API docs](https://docs.tikhub.io). + +2. **[sph.litao.workers.dev](https://sph.litao.workers.dev/)** — a free, already-deployed Cloudflare Worker run by a third party (not by this project or by TikHub). It is a public instance of the open-source project **[ltaoo/wx_channels_download](https://github.com/ltaoo/wx_channels_download)**, and we simply call its already-running `/api/fetch_video_profile` endpoint. We did not write or deploy this service, and its uptime is outside this project's control. If it ever goes down, the fallback is to self-host that same open-source project yourself and point `SPH_WORKER_URL` in `fetch_wechat_with_channels.py` at your own deployment. diff --git a/skills/rn-wechat-extract/scripts/fetch_wechat_with_channels.py b/skills/rn-wechat-extract/scripts/fetch_wechat_with_channels.py index ea8e389..b807dc8 100644 --- a/skills/rn-wechat-extract/scripts/fetch_wechat_with_channels.py +++ b/skills/rn-wechat-extract/scripts/fetch_wechat_with_channels.py @@ -7,24 +7,35 @@ 1. Extract export_id + object_nonce_id from the embed tag (or use a direct weixin.qq.com/sph/... share link if the article has one). - 2. TikHub fetch_video_detail(export_id, object_nonce_id) -> resolve the real - numeric object_id (the export_id/nonce pair alone is not a stable identifier). - 3. TikHub fetch_video_share_url(object_id) -> weixin.qq.com/sph/xxx share link. - 4. sph.litao.workers.dev fetch_video_profile(share_url) -> h264/h265 CDN url. - This CDN url carries an X-snsvideoflag query param that makes WeChat's - CDN return the plaintext, pre-transcoded stream (X-encflag: 0) instead of - the ISAAC64-encrypted original — so no decrypt step is needed afterwards. - 5. Download that CDN url with WeChat headers. - -Requires a TikHub API key (https://tikhub.io) passed via --tikhub-key or the -TIKHUB_API_KEY environment variable. Steps 2+3 cost $0.01 each in TikHub credits -per video ($0.02 total); step 4 (the community worker) is free. Without a key, -video channel videos are still detected and reported but not downloaded. + 2. [TikHub, paid] fetch_video_detail(export_id, object_nonce_id) -> resolve the + real numeric object_id (the export_id/nonce pair alone is not a stable identifier). + 3. [TikHub, paid] fetch_video_share_url(object_id) -> weixin.qq.com/sph/xxx share link. + 4. [Third-party community worker, free] sph.litao.workers.dev fetch_video_profile + (share_url) -> h264/h265 CDN url. This CDN url carries an X-snsvideoflag query + param that makes WeChat's CDN return the plaintext, pre-transcoded stream + (X-encflag: 0) instead of the ISAAC64-encrypted original — so no decrypt + step is needed afterwards. + 5. Download that CDN url with WeChat headers (hits WeChat's own CDN directly). + +Steps 2-4 depend on two external services that are NOT part of this project: +- TikHub (https://tikhub.io): a paid commercial API. Requires your own API key, + passed via --tikhub-key or the TIKHUB_API_KEY environment variable. Steps 2+3 + are billed at $0.01 each ($0.02/video total). We are not affiliated with TikHub. +- sph.litao.workers.dev: a free, already-deployed Cloudflare Worker run by a third + party, not by this project. It's a public instance of the open-source project + ltaoo/wx_channels_download (https://github.com/ltaoo/wx_channels_download) — we + did not write or deploy it, and its uptime is outside our control. To self-host, + deploy that project yourself and point SPH_WORKER_URL below at your own instance. + +Without a TikHub key, video channel videos are still detected and reported but +not downloaded; everything else (text, images, native videos) still works. Based on: - Original: Pluviobyte/rnskill -- Video channel resolution: TikHub WeChat-Channels-V2-API -- Share-link-to-CDN-url resolution: ltaoo/wx_channels_download (sph.litao.workers.dev) +- Video channel resolution: TikHub WeChat-Channels-V2-API (https://docs.tikhub.io) +- Share-link-to-CDN-url resolution: ltaoo/wx_channels_download + (https://github.com/ltaoo/wx_channels_download), used via its already-deployed + public instance at sph.litao.workers.dev """ import sys