mirror of
https://github.com/rachelos/we-mp-rss.git
synced 2026-11-05 18:55:26 +08:00
125 lines
3.6 KiB
Python
125 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, Tuple
|
|
|
|
from core.config import cfg
|
|
from core.models.base import DATA_STATUS
|
|
from core.print import print_info, print_warning
|
|
|
|
|
|
def normalize_content_mode(mode: str | None = None) -> str:
|
|
normalized = (mode or cfg.get("gather.content_mode", "web") or "web").strip().lower()
|
|
if normalized not in {"web", "api"}:
|
|
return "web"
|
|
return normalized
|
|
|
|
|
|
def extract_origin_article_id(article_id: str, mp_id: str | None = None) -> str:
|
|
if not article_id:
|
|
return ""
|
|
|
|
mp_prefix = (mp_id or "").replace("MP_WXS_", "").strip()
|
|
if mp_prefix:
|
|
prefixed = f"{mp_prefix}-"
|
|
if article_id.startswith(prefixed):
|
|
return article_id[len(prefixed):]
|
|
|
|
return article_id
|
|
|
|
|
|
def build_article_url(article: Any) -> str:
|
|
article_url = (getattr(article, "url", "") or "").strip()
|
|
if article_url:
|
|
return article_url
|
|
|
|
origin_id = extract_origin_article_id(
|
|
getattr(article, "id", ""),
|
|
getattr(article, "mp_id", ""),
|
|
)
|
|
if not origin_id:
|
|
return ""
|
|
|
|
return f"https://mp.weixin.qq.com/s/{origin_id}"
|
|
|
|
|
|
def _fetch_with_web(url: str) -> str:
|
|
from driver.wxarticle import Web
|
|
|
|
result = Web.get_article_content(url) or {}
|
|
return (result.get("content") or "").strip()
|
|
|
|
|
|
def _fetch_with_api(url: str) -> str:
|
|
from core.wx.model.api import MpsApi
|
|
|
|
fetcher = MpsApi()
|
|
return (fetcher.content_extract(url) or "").strip()
|
|
|
|
|
|
def fetch_article_content(url: str, preferred_mode: str | None = None) -> Tuple[str, str]:
|
|
mode = normalize_content_mode(preferred_mode)
|
|
modes = [mode] + [item for item in ("web", "api") if item != mode]
|
|
|
|
for current_mode in modes:
|
|
try:
|
|
if current_mode == "api":
|
|
content = _fetch_with_api(url)
|
|
else:
|
|
content = _fetch_with_web(url)
|
|
except Exception as exc:
|
|
print_warning(f"fetch article content failed in {current_mode} mode: {exc}")
|
|
continue
|
|
|
|
if content == "DELETED":
|
|
return content, current_mode
|
|
if content:
|
|
return content, current_mode
|
|
|
|
return "", mode
|
|
|
|
|
|
def sync_article_content(
|
|
session,
|
|
article: Any,
|
|
preferred_mode: str | None = None,
|
|
force: bool = False,
|
|
) -> Tuple[bool, str]:
|
|
existing_content = (getattr(article, "content", "") or "").strip()
|
|
if existing_content and not force:
|
|
return False, "cached"
|
|
|
|
article_url = build_article_url(article)
|
|
if not article_url:
|
|
print_warning(f"article {getattr(article, 'id', '')} has no valid url")
|
|
return False, "missing_url"
|
|
|
|
content, mode = fetch_article_content(article_url, preferred_mode)
|
|
if not content:
|
|
return False, mode
|
|
|
|
try:
|
|
if content == "DELETED":
|
|
article.content = ""
|
|
article.content_html = ""
|
|
article.status = DATA_STATUS.DELETED
|
|
session.commit()
|
|
session.refresh(article)
|
|
print_info(f"article {article.id} marked as deleted via {mode}")
|
|
return True, mode
|
|
|
|
from driver.wxarticle import Web
|
|
from tools.fix import fix_html
|
|
|
|
article.content = content
|
|
article.content_html = fix_html(content)
|
|
article.status = DATA_STATUS.ACTIVE
|
|
if not (getattr(article, "description", "") or "").strip():
|
|
article.description = Web.get_description(content)
|
|
session.commit()
|
|
session.refresh(article)
|
|
print_info(f"article {article.id} content synced via {mode}")
|
|
return True, mode
|
|
except Exception:
|
|
session.rollback()
|
|
raise
|