116 lines
3.3 KiB
Python
116 lines
3.3 KiB
Python
"""公告页内容提取工具。
|
||
|
||
流程：
|
||
1. 打开公告页，取 `#page-content` 的正文文字与图片地址；
|
||
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
|
||
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) -> PageExtractResult:
|
||
"""提取公告页的全部文字信息和 Logo 地址。
|
||
|
||
Args:
|
||
url: 公告页地址。
|
||
|
||
Returns:
|
||
PageExtractResult 对象，包含 content 和 logo_url。
|
||
"""
|
||
if not url or not url.startswith("http"):
|
||
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 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] = []
|
||
|
||
for node in nodes:
|
||
if node.text:
|
||
texts.append(node.text)
|
||
image_urls.extend(node.image_urls)
|
||
|
||
for image_url in image_urls:
|
||
image = download_image(image_url)
|
||
if image is None:
|
||
continue
|
||
|
||
# OCR 提取图片文字
|
||
try:
|
||
image_text = ocr(image)
|
||
except Exception as exc:
|
||
log.warning(f"图片 OCR 失败: {image_url} | {exc}")
|
||
else:
|
||
if image_text:
|
||
texts.append(image_text)
|
||
|
||
# 有二维码则解出内容
|
||
try:
|
||
if has_qr(image):
|
||
texts.extend(decode_qr(image))
|
||
except Exception as exc:
|
||
log.warning(f"二维码识别失败: {image_url} | {exc}")
|
||
|
||
content = "\n".join(text for text in texts if text)
|
||
log.info(
|
||
f"公告页提取完成（图片 {len(image_urls)} 张，文本 {len(content)} 字，"
|
||
f"Logo {'已提取' if logo_url else '未提取'}）: {url}"
|
||
)
|
||
return PageExtractResult(content=content, logo_url=logo_url)
|