86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
"""公告页内容提取工具。
|
|
|
|
流程：
|
|
1. 打开公告页，取 `#page-content` 的正文文字与图片地址；
|
|
2. 图片地址下载成字节流；
|
|
3. 每张图片 OCR 提取文字；
|
|
4. 每张图片检测二维码，有则解出内容；
|
|
5. 正文 + 图片文字 + 二维码内容合并为一段字符串。
|
|
|
|
图片下载、OCR、二维码识别都允许失败，单张出错只记日志并跳过。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
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
|
|
|
|
# 正文容器选择器
|
|
_SELECTOR = "#page-content"
|
|
|
|
# 打开页面后的等待时间（毫秒），等懒加载图片就位
|
|
_WAIT_MS = 3000
|
|
|
|
|
|
def extract_page(url: str) -> str:
|
|
"""提取公告页的全部文字信息。
|
|
|
|
Args:
|
|
url: 公告页地址。
|
|
|
|
Returns:
|
|
正文文字、图片 OCR 文字、二维码内容合并后的字符串；
|
|
页面打开失败时返回空字符串。
|
|
"""
|
|
if not url or not url.startswith("http"):
|
|
return ""
|
|
|
|
page_id: str | None = None
|
|
try:
|
|
page_id = open_page(url, _WAIT_MS)
|
|
nodes = query(page_id, _SELECTOR)
|
|
except Exception as exc:
|
|
log.error(f"公告页打开失败: {url} | {exc}")
|
|
return ""
|
|
finally:
|
|
# 页面数据已取出，尽早释放浏览器上下文，不占着资源等后续 OCR
|
|
if page_id is not None:
|
|
close_page(page_id)
|
|
|
|
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)} 字）: {url}")
|
|
return content
|