diff --git a/app/tool/page_extract.py b/app/tool/page_extract.py index 6a21512..21c0efc 100644 --- a/app/tool/page_extract.py +++ b/app/tool/page_extract.py @@ -2,16 +2,19 @@ 流程: 1. 打开公告页,取 `#page-content` 的正文文字与图片地址; -2. 图片地址下载成字节流; -3. 每张图片 OCR 提取文字; -4. 每张图片检测二维码,有则解出内容; -5. 正文 + 图片文字 + 二维码内容合并为一段字符串。 +2. 通过 `.wx_follow_avatar` 单独提取 Logo 地址; +3. 正文图片下载成字节流; +4. 每张图片 OCR 提取文字; +5. 每张图片检测二维码,有则解出内容; +6. 返回合并后的公告文本与 Logo 地址。 图片下载、OCR、二维码识别都允许失败,单张出错只记日志并跳过。 """ from __future__ import annotations +from dataclasses import dataclass + from app.core.logger import log from app.tool.browser import close_page, open_page, query from app.tool.cv import has_qr @@ -19,38 +22,62 @@ from app.tool.image_download import download_image from app.tool.ocr import ocr from app.tool.qr_decode import decode_qr + +@dataclass +class PageExtractResult: + """公告页提取结果。""" + + content: str + """正文 + 图片 OCR + 二维码内容合并后的文本""" + + logo_url: str | None + """公众号 Logo 图片地址""" + # 正文容器选择器 _SELECTOR = "#page-content" +# 底部公众号信息区域中的 Logo 容器选择器 +_LOGO_SELECTOR = "#js_article_bottom_bar .wx_follow_avatar" + # 打开页面后的等待时间(毫秒),等懒加载图片就位 _WAIT_MS = 3000 -def extract_page(url: str) -> str: - """提取公告页的全部文字信息。 +def extract_page(url: str) -> PageExtractResult: + """提取公告页的全部文字信息和 Logo 地址。 Args: url: 公告页地址。 Returns: - 正文文字、图片 OCR 文字、二维码内容合并后的字符串; - 页面打开失败时返回空字符串。 + PageExtractResult 对象,包含 content 和 logo_url。 """ if not url or not url.startswith("http"): - return "" + return PageExtractResult(content="", logo_url=None) page_id: str | None = None try: page_id = open_page(url, _WAIT_MS) nodes = query(page_id, _SELECTOR) + logo_nodes = query(page_id, _LOGO_SELECTOR) except Exception as exc: log.error(f"公告页打开失败: {url} | {exc}") - return "" + return PageExtractResult(content="", logo_url=None) finally: # 页面数据已取出,尽早释放浏览器上下文,不占着资源等后续 OCR if page_id is not None: close_page(page_id) + logo_url = next( + ( + image_url + for node in logo_nodes + for image_url in node.image_urls + if image_url + ), + None, + ) + texts: list[str] = [] image_urls: list[str] = [] @@ -81,5 +108,8 @@ def extract_page(url: str) -> str: log.warning(f"二维码识别失败: {image_url} | {exc}") content = "\n".join(text for text in texts if text) - log.info(f"公告页提取完成(图片 {len(image_urls)} 张,文本 {len(content)} 字): {url}") - return content + log.info( + f"公告页提取完成(图片 {len(image_urls)} 张,文本 {len(content)} 字," + f"Logo {'已提取' if logo_url else '未提取'}): {url}" + ) + return PageExtractResult(content=content, logo_url=logo_url)