284 lines
14 KiB
Python
284 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""下载微信/普通网页文章为 Markdown。Python 接口：save_article(url, output_root='.')。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
from dataclasses import asdict, dataclass
|
||
import json
|
||
from pathlib import Path
|
||
import re
|
||
import shutil
|
||
import sys
|
||
import tempfile
|
||
from urllib.parse import unquote, urljoin, urlsplit
|
||
|
||
import requests
|
||
from bs4 import BeautifulSoup
|
||
from markdownify import markdownify
|
||
from requests.adapters import HTTPAdapter
|
||
from urllib3.util.retry import Retry
|
||
|
||
|
||
class ArticleDownloadError(RuntimeError):
|
||
"""页面不是文章、图片下载失败等可向调用方报告的错误。"""
|
||
|
||
|
||
@dataclass
|
||
class ArticleResult:
|
||
title: str
|
||
source_url: str
|
||
directory: str
|
||
markdown: str
|
||
images: list[str]
|
||
warnings: list[str]
|
||
|
||
def to_dict(self) -> dict:
|
||
return asdict(self)
|
||
|
||
|
||
def safe_name(title: str) -> str:
|
||
"""跨平台目录/文件名；目录名与其中 Markdown 的文件名保持一致。"""
|
||
name = re.sub(r'[<>:"/\\|?*\x00-\x1f\x7f]', '_', title)
|
||
name = re.sub(r'\s+', ' ', name).strip(' .')
|
||
# 同时限制 UTF-8 字节数，给 .md 和重名后缀保留空间。
|
||
name = name.encode('utf-8')[:180].decode('utf-8', errors='ignore').rstrip(' .')
|
||
if not name:
|
||
name = '未命名文章'
|
||
if re.fullmatch(r'(?i:CON|PRN|AUX|NUL|COM[1-9]|LPT[1-9])(?:\..*)?', name):
|
||
name = '_' + name
|
||
return name
|
||
|
||
|
||
def _session() -> requests.Session:
|
||
session = requests.Session()
|
||
session.headers.update({
|
||
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) '
|
||
'AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36',
|
||
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
|
||
})
|
||
retry = Retry(total=3, backoff_factor=0.6,
|
||
status_forcelist=[429, 500, 502, 503, 504], allowed_methods=['GET'])
|
||
session.mount('https://', HTTPAdapter(max_retries=retry))
|
||
session.mount('http://', HTTPAdapter(max_retries=retry))
|
||
return session
|
||
|
||
|
||
def _get(session: requests.Session, url: str, timeout: float, limit: int,
|
||
referer: str | None = None) -> tuple[bytes, str, str]:
|
||
if urlsplit(url).scheme not in ('http', 'https'):
|
||
raise ArticleDownloadError(f'不支持的下载地址：{url[:200]}')
|
||
headers = {'Referer': referer} if referer else {}
|
||
with session.get(url, timeout=(min(timeout, 15), timeout),
|
||
headers=headers, stream=True) as response:
|
||
response.raise_for_status()
|
||
chunks, size = [], 0
|
||
for chunk in response.iter_content(65536):
|
||
size += len(chunk)
|
||
if size > limit:
|
||
raise ArticleDownloadError(f'文件超过大小限制（{limit // 1024 // 1024} MB）：{url[:200]}')
|
||
chunks.append(chunk)
|
||
return b''.join(chunks), response.headers.get('Content-Type', ''), response.url
|
||
|
||
|
||
def _image_ext(data: bytes, mime: str) -> str:
|
||
if data.startswith(b'\xff\xd8\xff'):
|
||
return '.jpg'
|
||
if data.startswith(b'\x89PNG\r\n\x1a\n'):
|
||
return '.png'
|
||
if data.startswith((b'GIF87a', b'GIF89a')):
|
||
return '.gif'
|
||
if data[:4] == b'RIFF' and data[8:12] == b'WEBP':
|
||
return '.webp'
|
||
if len(data) >= 12 and data[4:8] == b'ftyp' and data[8:12] in (b'avif', b'avis'):
|
||
return '.avif'
|
||
if b'<svg' in data[:1024].lower() and mime.split(';')[0] == 'image/svg+xml':
|
||
return '.svg'
|
||
raise ArticleDownloadError(f'返回内容不是支持的图片（Content-Type: {mime or "未知"}）')
|
||
|
||
|
||
def _image_source(img) -> str:
|
||
for attr in ('data-src', 'data-original', 'data-actualsrc', 'data-lazy-src', 'src'):
|
||
value = img.get(attr)
|
||
if value and str(value).strip():
|
||
return str(value).strip()
|
||
srcset = img.get('data-srcset') or img.get('srcset')
|
||
if srcset:
|
||
return str(srcset).split(',')[-1].strip().split()[0]
|
||
return ''
|
||
|
||
|
||
def save_article(url: str, output_root: str | Path = '.', *, timeout: float = 30,
|
||
allow_missing_images: bool = False) -> ArticleResult:
|
||
"""返回绝对路径。默认任意图片失败即报错，不留下半成品；已有目录不会被覆盖。
|
||
|
||
allow_missing_images=True 时，失败图片被替换为文字占位，并记入 warnings。
|
||
网络/HTTP 错误可能抛出 requests.RequestException，文章错误抛出 ArticleDownloadError。
|
||
调用方应只传入可信的公开网页 URL（本函数允许访问调用机器能访问的 HTTP 地址）。
|
||
"""
|
||
if not isinstance(url, str) or urlsplit(url).scheme not in ('http', 'https') or not urlsplit(url).hostname:
|
||
raise ValueError('url 必须是完整的 http:// 或 https:// 地址')
|
||
if timeout <= 0:
|
||
raise ValueError('timeout 必须大于 0')
|
||
if not isinstance(allow_missing_images, bool):
|
||
raise ValueError('allow_missing_images 必须是布尔值')
|
||
with _session() as session:
|
||
raw, content_type, final_url = _get(session, url, timeout, 20 * 1024 * 1024)
|
||
if content_type and not any(t in content_type.lower() for t in ('html', 'xhtml', 'text/plain')):
|
||
raise ArticleDownloadError('网址返回的不是 HTML 网页')
|
||
# 微信为 UTF-8；普通网页由 HTML 的 charset 标记/BeautifulSoup 自动识别。
|
||
soup = BeautifulSoup(raw, 'html.parser')
|
||
is_wechat = urlsplit(url).hostname == 'mp.weixin.qq.com'
|
||
content = soup.select_one('#js_content')
|
||
if is_wechat and content is None:
|
||
raise ArticleDownloadError('未找到微信文章正文，可能需要验证、登录，或文章已失效。请在浏览器中确认链接可访问。')
|
||
if content is None:
|
||
content = soup.select_one('article') or soup.select_one('main') or soup.body
|
||
if content is None:
|
||
raise ArticleDownloadError('网页没有可保存的正文')
|
||
title_tag = soup.select_one('#activity-name') or soup.select_one('h1')
|
||
title_meta = soup.select_one('meta[property="og:title"]')
|
||
title = (title_tag.get_text(' ', strip=True) if title_tag else '')
|
||
title = title or (title_meta.get('content', '').strip() if title_meta else '')
|
||
title = title or (soup.title.get_text(' ', strip=True) if soup.title else '') or '未命名文章'
|
||
for tag in content.select('script, style, noscript, iframe, form, button'):
|
||
tag.decompose()
|
||
# 部分微信排版把插图放在行内 CSS 背景中，也转为普通图片参与下载。
|
||
for tag in list(content.find_all(style=True)):
|
||
for match in re.finditer(r'url\(\s*[\'\"]?(.*?)[\'\"]?\s*\)', str(tag['style'])):
|
||
source = match.group(1).strip()
|
||
if source and not source.startswith('#'):
|
||
background = soup.new_tag('img', src=source, alt='')
|
||
tag.insert(0, background)
|
||
if not content.get_text(strip=True) and not content.find('img'):
|
||
raise ArticleDownloadError('文章正文为空，可能需要浏览器执行 JavaScript')
|
||
for link in content.find_all('a', href=True):
|
||
href = str(link['href']).strip()
|
||
if href and not href.startswith('#'):
|
||
absolute = urljoin(final_url, href)
|
||
if urlsplit(absolute).scheme in ('http', 'https', 'mailto', 'tel'):
|
||
link['href'] = absolute
|
||
else:
|
||
del link['href']
|
||
|
||
root = Path(output_root).expanduser().resolve()
|
||
root.mkdir(parents=True, exist_ok=True)
|
||
stage = Path(tempfile.mkdtemp(prefix='.article-', dir=root))
|
||
images, warnings, downloaded = [], [], {}
|
||
try:
|
||
for img in list(content.find_all('img')):
|
||
source = _image_source(img)
|
||
if not source:
|
||
img.replace_with(soup.new_string('[图片缺少下载地址]'))
|
||
if not allow_missing_images:
|
||
raise ArticleDownloadError('正文中存在没有下载地址的图片')
|
||
warnings.append('正文中存在没有下载地址的图片')
|
||
continue
|
||
image_url = urljoin(final_url, source)
|
||
try:
|
||
if image_url not in downloaded:
|
||
if source.startswith('data:image/'):
|
||
header, payload = source.split(',', 1)
|
||
data = base64.b64decode(payload, validate=True) if ';base64' in header else unquote(payload).encode()
|
||
mime = header[5:].split(';')[0]
|
||
if len(data) > 50 * 1024 * 1024:
|
||
raise ArticleDownloadError('内嵌图片超过 50 MB')
|
||
else:
|
||
data, mime, _ = _get(session, image_url, timeout, 50 * 1024 * 1024, final_url)
|
||
ext = _image_ext(data, mime)
|
||
filename = f'image{len(images) + 1:03d}{ext}'
|
||
(stage / filename).write_bytes(data)
|
||
images.append(filename)
|
||
downloaded[image_url] = filename
|
||
# 清除懒加载/响应式属性，使转换器只引用本地文件。
|
||
alt = str(img.get('alt') or '')
|
||
img.attrs = {'src': downloaded[image_url], 'alt': alt}
|
||
except (requests.RequestException, ArticleDownloadError, ValueError) as exc:
|
||
if not allow_missing_images:
|
||
raise ArticleDownloadError(f'图片下载失败：{image_url[:200]}；{exc}') from exc
|
||
warnings.append(f'图片下载失败：{image_url[:200]}；{exc}')
|
||
img.replace_with(soup.new_string(f'[图片下载失败：{img.get("alt") or "未命名图片"}]'))
|
||
|
||
body = markdownify(str(content), heading_style='ATX', bullets='-',
|
||
strip=['picture', 'source']).strip()
|
||
# 微信富文本中各段常没有标准 p 标签；markdownify 仍会保留文字和图片顺序。
|
||
author_tag = soup.select_one('#js_name')
|
||
date_tag = soup.select_one('#publish_time')
|
||
author = author_tag.get_text(' ', strip=True) if author_tag else ''
|
||
author = author or '未知作者'
|
||
metadata = []
|
||
if date_tag and date_tag.get_text(strip=True):
|
||
metadata.append(f'发布时间：{date_tag.get_text(" ", strip=True)}')
|
||
text = f'# {title}\n\n'
|
||
if metadata:
|
||
text += '\n\n'.join(metadata) + '\n\n---\n\n'
|
||
text += body + f'\n\n来源：{author}，如有侵权，请联系删除。\n'
|
||
basename = safe_name(title)
|
||
# 原子认领新目录，支持多个进程并发保存同名文章。
|
||
index = 1
|
||
while True:
|
||
name = basename if index == 1 else f'{basename} ({index})'
|
||
destination = root / name
|
||
try:
|
||
destination.mkdir()
|
||
break
|
||
except FileExistsError:
|
||
index += 1
|
||
try:
|
||
(stage / f'{name}.md').write_text(text, encoding='utf-8')
|
||
for item in stage.iterdir():
|
||
shutil.move(str(item), destination / item.name)
|
||
except BaseException:
|
||
shutil.rmtree(destination, ignore_errors=True)
|
||
raise
|
||
return ArticleResult(title, url, str(destination), str(destination / f'{name}.md'),
|
||
[str(destination / x) for x in images], warnings)
|
||
finally:
|
||
shutil.rmtree(stage, ignore_errors=True)
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument('url', nargs='?', help='文章网址')
|
||
parser.add_argument('-o', '--output-root', default='.', help='保存到此目录下，默认当前目录')
|
||
parser.add_argument('--timeout', type=float, default=30, help='每次网络请求超时秒数')
|
||
parser.add_argument('--allow-missing-images', action='store_true', help='允许图片失败，保留文字与失败提示')
|
||
parser.add_argument('--json', action='store_true', help='向标准输出写入 JSON，便于其他任务调用')
|
||
parser.add_argument('--stdin-json', action='store_true', help='从标准输入读取一个 JSON 请求，并返回 JSON')
|
||
args = parser.parse_args()
|
||
json_mode = args.json or args.stdin_json
|
||
try:
|
||
if args.stdin_json:
|
||
payload = json.load(sys.stdin)
|
||
if not isinstance(payload, dict):
|
||
raise ValueError('JSON 请求必须是对象')
|
||
unknown = set(payload) - {'url', 'output_root', 'timeout', 'allow_missing_images'}
|
||
if unknown:
|
||
raise ValueError(f'未知参数：{", ".join(sorted(unknown))}')
|
||
if 'url' not in payload:
|
||
raise ValueError('JSON 请求缺少 url')
|
||
result = save_article(**payload)
|
||
else:
|
||
if not args.url:
|
||
raise ValueError('请提供网址，或使用 --stdin-json')
|
||
result = save_article(args.url, args.output_root, timeout=args.timeout,
|
||
allow_missing_images=args.allow_missing_images)
|
||
if json_mode:
|
||
print(json.dumps({'ok': True, **result.to_dict()}, ensure_ascii=False))
|
||
else:
|
||
print(result.directory)
|
||
for warning in result.warnings:
|
||
print(f'警告：{warning}', file=sys.stderr)
|
||
return 0
|
||
except Exception as exc:
|
||
if json_mode:
|
||
print(json.dumps({'ok': False, 'error': str(exc), 'error_type': type(exc).__name__}, ensure_ascii=False))
|
||
else:
|
||
print(f'下载失败：{exc}', file=sys.stderr)
|
||
return 1
|
||
|
||
|
||
if __name__ == '__main__':
|
||
sys.exit(main())
|