| from __future__ import annotations |
|
|
| import re |
| from urllib.parse import urlparse |
|
|
| import requests |
| from bs4 import BeautifulSoup |
|
|
| from app.article_fetchers.base import ArticleContent, ArticleFetchError |
| from app.utils.url_parser import clean_url |
|
|
|
|
| def _clean_text(value: str) -> str: |
| return re.sub(r"[ \t\r\f\v]+", " ", value or "").strip() |
|
|
|
|
| def _normalize_body(value: str) -> str: |
| lines = [_clean_text(line) for line in (value or "").splitlines()] |
| return "\n".join(line for line in lines if line) |
|
|
|
|
| def _meta_content(soup: BeautifulSoup, *selectors: tuple[str, str]) -> str: |
| for attr, value in selectors: |
| node = soup.find("meta", attrs={attr: value}) |
| if node: |
| content = _clean_text(node.get("content") or "") |
| if content: |
| return content |
| return "" |
|
|
|
|
| def _candidate_score(node) -> int: |
| text = _normalize_body(node.get_text("\n")) |
| paragraphs = node.find_all("p") |
| return len(text) + len(paragraphs) * 120 |
|
|
|
|
| def parse_generic_article_html(html: str, url: str) -> ArticleContent: |
| soup = BeautifulSoup(html, "html.parser") |
| for tag in soup(["script", "style", "noscript", "svg", "canvas", "iframe"]): |
| tag.decompose() |
| for tag in soup(["nav", "header", "footer", "aside", "form"]): |
| tag.decompose() |
|
|
| title = ( |
| _meta_content(soup, ("property", "og:title"), ("name", "twitter:title")) |
| or _clean_text(soup.title.get_text(" ")) if soup.title else "" |
| ) |
| author = _meta_content(soup, ("name", "author"), ("property", "article:author")) |
| published_at = _meta_content( |
| soup, |
| ("property", "article:published_time"), |
| ("name", "publishdate"), |
| ("name", "date"), |
| ) |
| cover = _meta_content(soup, ("property", "og:image"), ("name", "twitter:image")) |
|
|
| candidates = [] |
| for selector in ("article", "main", "[role='main']", "#content", ".content", ".article", ".post"): |
| candidates.extend(soup.select(selector)) |
| if not candidates and soup.body: |
| candidates = [soup.body] |
| best = max(candidates, key=_candidate_score, default=None) |
| body = _normalize_body(best.get_text("\n")) if best else "" |
| if len(body) < 80: |
| description = _meta_content(soup, ("name", "description"), ("property", "og:description")) |
| body = description if len(description) > len(body) else body |
| if len(body) < 40: |
| raise ValueError("网页正文为空或过短，无法生成总结") |
|
|
| parsed = urlparse(url) |
| article_id = parsed.netloc + parsed.path |
| return ArticleContent( |
| platform="generic_web", |
| url=url, |
| article_id=article_id or url, |
| title=title or parsed.netloc or "网页文章", |
| author_name=author, |
| content_text=body, |
| image_urls=[cover] if cover else [], |
| cover_url=cover, |
| published_at=published_at, |
| raw_metadata={"source": "generic_web"}, |
| ) |
|
|
|
|
| class GenericArticleFetcher: |
| platform = "generic_web" |
|
|
| def fetch(self, url: str) -> ArticleContent: |
| clean = clean_url(url) |
| try: |
| response = requests.get( |
| clean, |
| timeout=12, |
| allow_redirects=True, |
| headers={ |
| "User-Agent": ( |
| "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " |
| "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.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", |
| }, |
| ) |
| response.raise_for_status() |
| return parse_generic_article_html(response.text, response.url or clean) |
| except ValueError: |
| raise |
| except Exception as exc: |
| raise ArticleFetchError(f"网页文章抓取失败：{exc}") from exc |
|
|
| def search(self, keyword: str, limit: int = 20) -> list[ArticleContent]: |
| raise ArticleFetchError("通用网页暂不支持关键字查询，请粘贴具体文章链接") |
|
|
| def fetch_publisher(self, query: str, limit: int = 20) -> list[ArticleContent]: |
| raise ArticleFetchError("通用网页暂不支持发布者订阅，请粘贴具体文章链接") |
|
|
