diff --git a/packages/markitdown/src/markitdown/_markitdown.py b/packages/markitdown/src/markitdown/_markitdown.py index f342a61..654ffd5 100644 --- a/packages/markitdown/src/markitdown/_markitdown.py +++ b/packages/markitdown/src/markitdown/_markitdown.py @@ -39,6 +39,8 @@ EpubConverter, DocumentIntelligenceConverter, CsvConverter, + WeChatMPConverter, + XTwitterConverter, ) from ._base_converter import DocumentConverter, DocumentConverterResult @@ -187,6 +189,8 @@ def enable_builtins(self, **kwargs) -> None: self.register_converter( HtmlConverter(), priority=PRIORITY_GENERIC_FILE_FORMAT ) + self.register_converter(WeChatMPConverter()) + self.register_converter(XTwitterConverter()) self.register_converter(RssConverter()) self.register_converter(WikipediaConverter()) self.register_converter(YouTubeConverter()) diff --git a/packages/markitdown/src/markitdown/converters/__init__.py b/packages/markitdown/src/markitdown/converters/__init__.py index e4437a5..8ae40ce 100644 --- a/packages/markitdown/src/markitdown/converters/__init__.py +++ b/packages/markitdown/src/markitdown/converters/__init__.py @@ -23,6 +23,8 @@ ) from ._epub_converter import EpubConverter from ._csv_converter import CsvConverter +from ._wechat_mp_converter import WeChatMPConverter +from ._xtwitter_converter import XTwitterConverter __all__ = [ "PlainTextConverter", @@ -45,4 +47,6 @@ "DocumentIntelligenceFileType", "EpubConverter", "CsvConverter", + "WeChatMPConverter", + "XTwitterConverter", ] diff --git a/packages/markitdown/src/markitdown/converters/_wechat_mp_converter.py b/packages/markitdown/src/markitdown/converters/_wechat_mp_converter.py new file mode 100644 index 0000000..dc11af6 --- /dev/null +++ b/packages/markitdown/src/markitdown/converters/_wechat_mp_converter.py @@ -0,0 +1,368 @@ +""" +微信公众号文章转换器 + +将 mp.weixin.qq.com 的公众号文章转为 Markdown,并下载图片到本地。 +""" + +import hashlib +import io +import os +import re +from typing import Any, BinaryIO, Dict, Optional +from urllib.parse import urljoin + +import requests +from bs4 import BeautifulSoup, Tag + +from .._base_converter import DocumentConverter, DocumentConverterResult +from .._stream_info import StreamInfo +from ._markdownify import _CustomMarkdownify + +ACCEPTED_MIME_TYPE_PREFIXES = [ + "text/html", + "application/xhtml", +] + +ACCEPTED_FILE_EXTENSIONS = [ + ".html", + ".htm", +] + +# 公众号反爬验证页面的关键词 +_CAPTCHA_INDICATORS = ["环境异常", "captcha", "appmsgcaptcha", "验证后即可继续"] + + +class WeChatMPConverter(DocumentConverter): + """微信公众号文章转换器,提取标题/作者/正文/图片,图片下载到本地。""" + + def accepts( + self, + file_stream: BinaryIO, + stream_info: StreamInfo, + **kwargs: Any, + ) -> bool: + url = stream_info.url or "" + if not re.search(r"^https?://mp\.weixin\.qq\.com/", url): + return False + + mimetype = (stream_info.mimetype or "").lower() + extension = (stream_info.extension or "").lower() + + if extension in ACCEPTED_FILE_EXTENSIONS: + return True + + for prefix in ACCEPTED_MIME_TYPE_PREFIXES: + if mimetype.startswith(prefix): + return True + + return False + + def convert( + self, + file_stream: BinaryIO, + stream_info: StreamInfo, + **kwargs: Any, + ) -> DocumentConverterResult: + encoding = "utf-8" if stream_info.charset is None else stream_info.charset + + # 先读取内容,检测是否被反爬拦截 + cur_pos = file_stream.tell() + raw_bytes = file_stream.read() + file_stream.seek(cur_pos) + + # 如果检测到验证页面,尝试用更好的 headers 重新请求 + html_text = raw_bytes.decode(encoding, errors="replace") + original_url = stream_info.url + if self._is_captcha_page(html_text) and stream_info.url: + raw_bytes = self._refetch(stream_info.url) + if raw_bytes is not None: + html_text = raw_bytes.decode("utf-8", errors="replace") + + soup = BeautifulSoup(html_text, "html.parser") + + # 提取元信息 + title = self._extract_title(soup) + account = self._extract_account(soup) + author = self._extract_author(soup) + date = self._extract_date(soup) + + # 提取正文 + content = soup.find("div", id="js_content") + if not content: + # 回退:尝试 class 匹配 + content = soup.find("div", class_="rich_media_content") + if not content: + # 最终回退:用整个 body + content = soup.find("body") or soup + + # 下载图片并替换路径 + image_dir = kwargs.get("wechat_image_dir", None) + if image_dir is None: + # 默认在当前工作目录下创建 images 文件夹 + image_dir = os.path.join(os.getcwd(), "images") + self._download_images(content, image_dir, stream_info.url) + + # 转换为 Markdown + md_body = _CustomMarkdownify(**kwargs).convert_soup(content) + md_body = md_body.strip() + + # 拼接头部元信息 + header = f"# {title}\n\n" + meta_lines = [] + if account: + meta_lines.append(f"**公众号**: {account}") + if author and author != account: + meta_lines.append(f"**作者**: {author}") + if date: + meta_lines.append(f"**发布时间**: {date}") + # 原文链接:优先用原始 URL,如果被重定向到验证页则从中提取 + display_url = original_url or "" + if "appmsgcaptcha" in display_url and "target_url=" in display_url: + from urllib.parse import parse_qs, urlparse + params = parse_qs(urlparse(display_url).query) + if "target_url" in params: + display_url = params["target_url"][0] + if display_url: + meta_lines.append(f"**原文链接**: {display_url}") + if meta_lines: + header += "\n".join(meta_lines) + "\n\n---\n\n" + + markdown = header + md_body + + return DocumentConverterResult( + markdown=markdown, + title=title, + ) + + # ------------------------------------------------------------------ + # 反爬处理 + # ------------------------------------------------------------------ + + @staticmethod + def _is_captcha_page(html: str) -> bool: + """检测 HTML 是否是验证页面。""" + html_lower = html.lower() + return any(kw in html_lower for kw in _CAPTCHA_INDICATORS) + + @staticmethod + def _refetch(url: str) -> Optional[bytes]: + """用移动端 User-Agent 重新请求,尝试绕过反爬。""" + # 如果当前 URL 是验证页,从中提取原始 URL + original_url = url + if "appmsgcaptcha" in url and "target_url=" in url: + from urllib.parse import parse_qs, urlparse + parsed = urlparse(url) + params = parse_qs(parsed.query) + if "target_url" in params: + original_url = params["target_url"][0] + + headers_list = [ + { + "User-Agent": ( + "Mozilla/5.0 (Linux; Android 13; Pixel 7) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/120.0.0.0 Mobile Safari/537.36" + ), + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", + "Referer": "https://mp.weixin.qq.com/", + }, + { + "User-Agent": ( + "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) " + "AppleWebKit/605.1.15 (KHTML, like Gecko) " + "Version/17.0 Mobile/15E148 Safari/604.1" + ), + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", + "Referer": "https://mp.weixin.qq.com/", + }, + { + "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" + ), + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", + "Referer": "https://www.weixin.qq.com/", + }, + ] + for hdrs in headers_list: + try: + session = requests.Session() + resp = session.get(original_url, headers=hdrs, timeout=30) + resp.raise_for_status() + content = resp.content + # 验证新页面不是验证页 + text = content.decode("utf-8", errors="replace").lower() + if not any(kw in text for kw in _CAPTCHA_INDICATORS): + return content + except Exception: + continue + return None + + # ------------------------------------------------------------------ + # 元信息提取 + # ------------------------------------------------------------------ + + @staticmethod + def _extract_title(soup: BeautifulSoup) -> str: + el = soup.find("h1", id="activity-name") + if el: + return el.get_text(strip=True) + el = soup.find("h1", class_="rich_media_title") + if el: + return el.get_text(strip=True) + if soup.title and soup.title.string: + return soup.title.string.strip() + return "Untitled" + + @staticmethod + def _extract_account(soup: BeautifulSoup) -> str: + el = soup.find("a", id="js_name") + if el: + return el.get_text(strip=True) + el = soup.find("span", class_="rich_media_meta_nickname") + if el: + return el.get_text(strip=True) + return "" + + @staticmethod + def _extract_author(soup: BeautifulSoup) -> str: + el = soup.find("span", id="js_author_name") + if el: + return el.get_text(strip=True) + el = soup.find("span", class_="rich_media_meta_text") + if el: + return el.get_text(strip=True) + return "" + + @staticmethod + def _extract_date(soup: BeautifulSoup) -> str: + el = soup.find("em", id="publish_time") + if el: + text = el.get_text(strip=True) + if text: + return text + # 回退:从 meta 标签提取 + meta = soup.find("meta", {"property": "article:published_time"}) + if meta and meta.get("content"): + return meta["content"][:10] + return "" + + # ------------------------------------------------------------------ + # 图片处理 + # ------------------------------------------------------------------ + + def _download_images( + self, + content: Tag, + image_dir: str, + page_url: Optional[str], + ) -> None: + """ + 找到正文中的所有 ,下载到 image_dir,并将 src/data-src + 替换为本地相对路径。 + """ + images = content.find_all("img") + if not images: + return + + os.makedirs(image_dir, exist_ok=True) + + # 图片文件夹的名称,用于生成相对路径 + image_folder_name = os.path.basename(image_dir) + + headers = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/120.0.0.0 Safari/537.36" + ), + "Referer": page_url or "https://mp.weixin.qq.com/", + } + + for idx, img in enumerate(images, start=1): + # 优先用 data-src(公众号懒加载),其次 src + img_url = img.get("data-src") or img.get("src") or "" + if not img_url or img_url.startswith("data:"): + continue + + # 补全相对路径 + if img_url.startswith("//"): + img_url = "https:" + img_url + elif not img_url.startswith("http"): + img_url = urljoin(page_url or "", img_url) + + # 下载图片 + local_path = self._download_one( + img_url, image_dir, idx, headers + ) + if local_path: + # 替换为相对路径 + relative_path = os.path.join(image_folder_name, local_path) + # 统一用正斜杠,兼容 Markdown 渲染 + relative_path = relative_path.replace("\\", "/") + img["src"] = relative_path + # 清掉 data-src,避免 markdownify 二次处理 + if img.get("data-src"): + del img["data-src"] + + @staticmethod + def _download_one( + url: str, + image_dir: str, + index: int, + headers: Dict[str, str], + ) -> Optional[str]: + """下载单张图片,返回文件名;失败返回 None。""" + try: + resp = requests.get(url, headers=headers, timeout=30, stream=True) + resp.raise_for_status() + + # 从 Content-Type 或 URL 推断扩展名 + ext = _guess_extension(resp.headers.get("Content-Type", ""), url) + + # 用 URL 的 md5 做文件名,避免重复下载 + url_hash = hashlib.md5(url.encode()).hexdigest()[:12] + filename = f"img_{index:03d}_{url_hash}{ext}" + filepath = os.path.join(image_dir, filename) + + # 如果文件已存在,跳过下载 + if os.path.exists(filepath): + return filename + + with open(filepath, "wb") as f: + for chunk in resp.iter_content(chunk_size=8192): + f.write(chunk) + + return filename + + except Exception: + return None + + +def _guess_extension(content_type: str, url: str) -> str: + """从 Content-Type 或 URL 路径推断图片扩展名。""" + ct_map = { + "image/jpeg": ".jpg", + "image/png": ".png", + "image/gif": ".gif", + "image/webp": ".webp", + "image/svg+xml": ".svg", + "image/bmp": ".bmp", + } + ct_lower = content_type.lower().split(";")[0].strip() + if ct_lower in ct_map: + return ct_map[ct_lower] + + # 从 URL 路径猜测 + from urllib.parse import urlparse + path = urlparse(url).path + for ext in (".jpg", ".jpeg", ".png", ".gif", ".webp", ".svg", ".bmp"): + if path.lower().endswith(ext): + return ext + + # 公众号默认大多是 jpg + return ".jpg" diff --git a/packages/markitdown/src/markitdown/converters/_xtwitter_converter.py b/packages/markitdown/src/markitdown/converters/_xtwitter_converter.py new file mode 100644 index 0000000..686ad4e --- /dev/null +++ b/packages/markitdown/src/markitdown/converters/_xtwitter_converter.py @@ -0,0 +1,446 @@ +""" +X (Twitter) 推文转换器 + +支持类型: +- 普通推文(纯文字 / 带图片) +- 长文 (X Article) +- 视频推文 + +数据来源:FXTwitter API (https://api.fxtwitter.com) +""" + +import hashlib +import os +import re +from typing import Any, BinaryIO, Dict, List, Optional + +import requests +from bs4 import BeautifulSoup + +from .._base_converter import DocumentConverter, DocumentConverterResult +from .._stream_info import StreamInfo + +# X / Twitter URL 匹配模式 +_TWITTER_URL_PATTERN = re.compile( + r"^https?://(www\.)?(twitter\.com|x\.com)/\w+/status/(\d+)" +) + +# 请求头 +_HEADERS = { + "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" + ), +} + + +class XTwitterConverter(DocumentConverter): + """X (Twitter) 推文转换器,通过 FXTwitter API 获取结构化数据。""" + + def accepts( + self, + file_stream: BinaryIO, + stream_info: StreamInfo, + **kwargs: Any, + ) -> bool: + url = stream_info.url or "" + return bool(_TWITTER_URL_PATTERN.match(url)) + + def convert( + self, + file_stream: BinaryIO, + stream_info: StreamInfo, + **kwargs: Any, + ) -> DocumentConverterResult: + url = stream_info.url or "" + + # 从 URL 提取 screen_name 和 tweet_id + match = _TWITTER_URL_PATTERN.match(url) + if not match: + return DocumentConverterResult(markdown="") + + screen_name = url.split("/")[-3] + tweet_id = match.group(3) + + # 通过 FXTwitter API 获取推文数据 + tweet_data = self._fetch_tweet(screen_name, tweet_id) + if not tweet_data: + return DocumentConverterResult( + markdown=f"Failed to fetch tweet: {url}" + ) + + # 根据类型分发 + if tweet_data.get("article"): + markdown = self._convert_article(tweet_data, url) + else: + markdown = self._convert_tweet(tweet_data, url) + + title = self._extract_title(tweet_data) + + return DocumentConverterResult( + markdown=markdown, + title=title, + ) + + # ------------------------------------------------------------------ + # 数据获取 + # ------------------------------------------------------------------ + + @staticmethod + def _fetch_tweet(screen_name: str, tweet_id: str) -> Optional[Dict]: + """通过 FXTwitter API 获取推文数据。""" + api_url = f"https://api.fxtwitter.com/{screen_name}/status/{tweet_id}" + try: + resp = requests.get(api_url, headers=_HEADERS, timeout=15) + resp.raise_for_status() + data = resp.json() + return data.get("tweet") + except Exception: + return None + + # ------------------------------------------------------------------ + # 普通推文 / 图片推文 / 视频推文 + # ------------------------------------------------------------------ + + def _convert_tweet( + self, tweet: Dict, url: str + ) -> str: + """转换普通推文、图片推文、视频推文。""" + author = tweet.get("author", {}) + author_name = author.get("name", "") + author_handle = author.get("screen_name", "") + text = tweet.get("text", "") + created_at = tweet.get("created_at", "") + likes = tweet.get("likes", 0) + retweets = tweet.get("retweets", 0) + replies = tweet.get("replies", 0) + views = tweet.get("views", 0) + bookmarks = tweet.get("bookmarks", 0) + + # 拼接头部 + header = f"# {author_name} 的推文\n\n" + header += f"**作者**: {author_name} (@{author_handle})\n" + if created_at: + header += f"**时间**: {created_at}\n" + header += f"**原文链接**: {url}\n" + header += f"\n---\n\n" + + # 正文 + body = text + "\n\n" + + # 处理图片 + media = tweet.get("media", {}) + image_dir = os.path.join(os.getcwd(), "images") + if image_dir is None: + image_dir = os.path.join(os.getcwd(), "images") + + images = media.get("all", []) + for idx, img in enumerate(images, start=1): + if img.get("type") == "photo": + img_url = img.get("url", "") + if img_url: + local_name = self._download_image( + img_url, image_dir, idx + ) + if local_name: + folder = os.path.basename(image_dir) + path = f"{folder}/{local_name}".replace("\\", "/") + body += f"![图片{idx}]({path})\n\n" + + # 处理视频 + videos = media.get("videos", []) + if videos: + for idx, vid in enumerate(videos, start=1): + video_url = vid.get("url", "") + duration = vid.get("duration", 0) + thumb_url = vid.get("thumbnail_url", "") + minutes = int(duration // 60) + seconds = int(duration % 60) + + body += f"**[视频{idx}]** 时长 {minutes:02d}:{seconds:02d}\n\n" + + # 下载缩略图 + if thumb_url: + thumb_name = self._download_image( + thumb_url, image_dir, idx, prefix="thumb" + ) + if thumb_name: + folder = os.path.basename(image_dir) + path = f"{folder}/{thumb_name}".replace("\\", "/") + body += f"![视频缩略图{idx}]({path})\n\n" + + if video_url: + body += f"视频下载链接: {video_url}\n\n" + + # 互动数据 + stats = ( + f"---\n\n" + f"| 浏览 | 点赞 | 转发 | 收藏 | 评论 |\n" + f"|------|------|------|------|------|\n" + f"| {self._fmt_num(views)} | {self._fmt_num(likes)} " + f"| {self._fmt_num(retweets)} | {self._fmt_num(bookmarks)} " + f"| {self._fmt_num(replies)} |\n" + ) + body += stats + + return header + body + + # ------------------------------------------------------------------ + # 长文 (X Article) + # ------------------------------------------------------------------ + + def _convert_article( + self, tweet: Dict, url: str + ) -> str: + """转换长文推文 (X Article)。""" + author = tweet.get("author", {}) + author_name = author.get("name", "") + author_handle = author.get("screen_name", "") + article = tweet.get("article", {}) + title = article.get("title", "") + preview = article.get("preview_text", "") + cover = article.get("cover_media", {}) + created_at = tweet.get("created_at", "") + + # 拼接头部 + header = f"# {title}\n\n" + header += f"**作者**: {author_name} (@{author_handle})\n" + if created_at: + header += f"**时间**: {created_at}\n" + header += f"**原文链接**: {url}\n" + header += f"\n---\n\n" + + # 下载封面图 + image_dir = os.path.join(os.getcwd(), "images") + if image_dir is None: + image_dir = os.path.join(os.getcwd(), "images") + + body = "" + cover_url = "" + if cover: + media_info = cover.get("media_info", {}) + cover_url = media_info.get("original_img_url", "") + if cover_url: + cover_name = self._download_image( + cover_url, image_dir, 0, prefix="cover" + ) + if cover_name: + folder = os.path.basename(image_dir) + path = f"{folder}/{cover_name}".replace("\\", "/") + body += f"![封面]({path})\n\n" + + # 解析文章内容(Draft.js 块格式) + content = article.get("content", {}) + blocks = content.get("blocks", []) + entity_map = content.get("entityMap", []) + media_entities = article.get("media_entities", []) + + # 建立 entityMap 索引 + entity_lookup = {} + if isinstance(entity_map, list): + for entry in entity_map: + key = str(entry.get("key", "")) + entity_lookup[key] = entry.get("value", entry) + elif isinstance(entity_map, dict): + entity_lookup = entity_map + + # 建立 media_entities 索引 (media_id -> img_url) + media_lookup = {} + if isinstance(media_entities, list): + for me in media_entities: + mid = me.get("media_id", "") + mi = me.get("media_info", {}) + img_url = mi.get("original_img_url", "") + if mid and img_url: + media_lookup[mid] = img_url + elif isinstance(media_entities, dict): + for mid, me in media_entities.items(): + mi = me.get("media_info", {}) + img_url = mi.get("original_img_url", "") + if img_url: + media_lookup[mid] = img_url + + img_idx = 0 + for block in blocks: + block_type = block.get("type", "unstyled") + text = block.get("text", "") + entity_ranges = block.get("entityRanges", []) + inline_styles = block.get("inlineStyleRanges", []) + + # 处理图片 / 媒体块 + if block_type == "atomic" and entity_ranges: + for er in entity_ranges: + entity_key = str(er.get("key", "")) + entity = entity_lookup.get(entity_key, {}) + etype = entity.get("type", "") + edata = entity.get("data", {}) + + if etype == "MEDIA": + # 从 mediaItems 找到 media_id -> img_url + for mi in edata.get("mediaItems", []): + media_id = mi.get("mediaId", "") + img_url = media_lookup.get(media_id, "") + if img_url: + img_idx += 1 + local_name = self._download_image( + img_url, image_dir, img_idx + ) + if local_name: + folder = os.path.basename(image_dir) + path = f"{folder}/{local_name}".replace( + "\\", "/" + ) + body += f"\n![图片{img_idx}]({path})\n\n" + + elif etype == "MARKDOWN": + md_text = edata.get("markdown", "") + if md_text: + body += f"\n{md_text}\n\n" + continue + + # 应用内联样式(加粗) + text = self._apply_inline_styles(text, inline_styles) + + # 处理链接实体 + if entity_ranges: + for er in entity_ranges: + entity_key = str(er.get("key", "")) + entity = entity_lookup.get(entity_key, {}) + etype = entity.get("type", "") + if etype == "LINK": + link_url = entity.get("data", {}).get("url", "") + if link_url: + offset = er.get("offset", 0) + length = er.get("length", 0) + link_text = text[offset : offset + length] + if link_text and link_text != link_url: + replacement = f"[{link_text}]({link_url})" + text = text[:offset] + replacement + text[offset + length :] + + # 根据块类型输出 + if block_type == "header-two": + body += f"\n## {text}\n\n" + elif block_type == "header-three": + body += f"\n### {text}\n\n" + elif block_type == "blockquote": + lines = text.split("\n") + for line in lines: + body += f"> {line}\n" + body += "\n" + elif block_type == "unordered-list-item": + body += f"- {text.rstrip()}\n" + elif block_type == "ordered-list-item": + # Draft.js 的有序列表块没有自带序号,简单处理 + body += f"1. {text.rstrip()}\n" + elif block_type == "code-block": + body += f"\n```\n{text}\n```\n\n" + else: + # unstyled 或其他 + if text.strip(): + body += f"{text}\n\n" + + # 互动数据 + likes = tweet.get("likes", 0) + retweets = tweet.get("retweets", 0) + replies = tweet.get("replies", 0) + views = tweet.get("views", 0) + bookmarks = tweet.get("bookmarks", 0) + + stats = ( + f"---\n\n" + f"| 浏览 | 点赞 | 转发 | 收藏 | 评论 |\n" + f"|------|------|------|------|------|\n" + f"| {self._fmt_num(views)} | {self._fmt_num(likes)} " + f"| {self._fmt_num(retweets)} | {self._fmt_num(bookmarks)} " + f"| {self._fmt_num(replies)} |\n" + ) + body += stats + + return header + body + + # ------------------------------------------------------------------ + # 辅助方法 + # ------------------------------------------------------------------ + + @staticmethod + def _extract_title(tweet: Dict) -> str: + """提取标题:优先用文章标题,否则用作者名。""" + article = tweet.get("article", {}) + if article: + title = article.get("title", "") + if title: + return title + author = tweet.get("author", {}) + name = author.get("name", "") + if name: + return f"{name} 的推文" + return "Tweet" + + @staticmethod + def _apply_inline_styles(text: str, styles: List[Dict]) -> str: + """应用内联样式(加粗等)到文本。""" + if not styles: + return text + + # 按偏移量倒序排列,避免位置偏移 + sorted_styles = sorted(styles, key=lambda s: s.get("offset", 0), reverse=True) + + for style in sorted_styles: + offset = style.get("offset", 0) + length = style.get("length", 0) + style_name = style.get("style", "") + + if style_name == "Bold" and offset + length <= len(text): + text = text[:offset] + "**" + text[offset : offset + length] + "**" + text[offset + length :] + + return text + + @staticmethod + def _download_image( + url: str, image_dir: str, index: int, prefix: str = "img" + ) -> Optional[str]: + """下载图片到本地,返回文件名。""" + try: + os.makedirs(image_dir, exist_ok=True) + url_hash = hashlib.md5(url.encode()).hexdigest()[:12] + ext = _guess_image_ext(url) + filename = f"tw_{prefix}_{index:03d}_{url_hash}{ext}" + filepath = os.path.join(image_dir, filename) + + if os.path.exists(filepath): + return filename + + resp = requests.get(url, headers=_HEADERS, timeout=30, stream=True) + resp.raise_for_status() + + with open(filepath, "wb") as f: + for chunk in resp.iter_content(chunk_size=8192): + f.write(chunk) + + return filename + except Exception: + return None + + @staticmethod + def _fmt_num(n: Any) -> str: + """格式化数字:1200 -> 1.2K。""" + try: + n = int(n) + except (ValueError, TypeError): + return str(n) + if n >= 1_000_000: + return f"{n / 1_000_000:.1f}M" + if n >= 1_000: + return f"{n / 1_000:.1f}K" + return str(n) + + +def _guess_image_ext(url: str) -> str: + """从 URL 路径推断图片扩展名。""" + from urllib.parse import urlparse + path = urlparse(url).path + for ext in (".jpg", ".jpeg", ".png", ".gif", ".webp", ".svg"): + if path.lower().endswith(ext): + return ext + return ".jpg"