rssKeeper/backend/rss_fetcher.py

"""RSS 抓取核心逻辑"""
import time
import re
import html
from datetime import datetime, timezone
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urljoin
import requests
import feedparser
from bs4 import BeautifulSoup
from sqlalchemy.orm import Session
from models import Feed, Article, FetchLog
from database import SessionLocal
import config


def fetch_feed(url: str, timeout: int = config.FETCH_TIMEOUT) -> dict:
    """抓取单个 RSS 源
    返回 {"success": bool, "feed_data": parsed, "error": str, "response_time_ms": int}
    """
    start_time = time.time()
    try:
        headers = {
            "User-Agent": "rssKeeper/1.0 (+https://github.com/rssKeeper)",
            "Accept": "application/rss+xml, application/atom+xml, application/xml, text/xml, */*",
        }
        response = requests.get(url, headers=headers, timeout=timeout, allow_redirects=True)
        response.raise_for_status()

        # 解析 RSS
        parsed = feedparser.parse(response.content)

        response_time_ms = int((time.time() - start_time) * 1000)

        if parsed.bozo and hasattr(parsed, 'bozo_exception'):
            # 有解析警告但可能仍然可用
            pass

        return {
            "success": True,
            "feed_data": parsed,
            "error": None,
            "response_time_ms": response_time_ms,
        }
    except requests.exceptions.RequestException as e:
        return {"success": False, "feed_data": None, "error": str(e), "response_time_ms": None}
    except Exception as e:
        return {"success": False, "feed_data": None, "error": str(e), "response_time_ms": None}


def discover_feed_url(url: str, timeout: int = 15) -> list:
    """从任意网页自动发现 RSS/Atom feed URL
    返回找到的 feed URL 列表
    """
    try:
        headers = {
            "User-Agent": "rssKeeper/1.0 (+https://github.com/rssKeeper)",
        }
        response = requests.get(url, headers=headers, timeout=timeout, allow_redirects=True)
        response.raise_for_status()

        soup = BeautifulSoup(response.content, "html.parser")
        feed_urls = []

        # 查找 <link rel="alternate"> 标签
        for link in soup.find_all("link", rel="alternate"):
            link_type = link.get("type", "").lower()
            href = link.get("href", "")
            if href and any(t in link_type for t in ["rss", "atom", "xml"]):
                full_url = urljoin(response.url, href)
                feed_urls.append(full_url)

        # 也查找常见的 RSS 链接
        common_patterns = [
            "/rss", "/feed", "/feeds", "/atom.xml", "/rss.xml",
            "/index.xml", "/feed.xml", "/?feed=rss2",
        ]
        for pattern in common_patterns:
            candidate = urljoin(response.url, pattern)
            if candidate not in feed_urls:
                # 验证是否是有效的 feed
                try:
                    resp = requests.head(candidate, headers=headers, timeout=5, allow_redirects=True)
                    content_type = resp.headers.get("Content-Type", "").lower()
                    if any(t in content_type for t in ["rss", "atom", "xml"]):
                        feed_urls.append(candidate)
                except Exception:
                    pass

        return list(dict.fromkeys(feed_urls))  # 去重保持顺序
    except Exception:
        return []


def parse_article(entry, feed_id: int) -> dict:
    """从 feedparser entry 解析文章数据"""
    title = entry.get("title", "")
    link = entry.get("link", "")
    author = entry.get("author", "")

    # 发布时间 — 统一存为 UTC aware datetime
    published_at = None
    if hasattr(entry, "published_parsed") and entry.published_parsed:
        try:
            published_at = datetime(*entry.published_parsed[:6], tzinfo=timezone.utc)
        except (ValueError, TypeError):
            pass
    if not published_at and hasattr(entry, "updated_parsed") and entry.updated_parsed:
        try:
            published_at = datetime(*entry.updated_parsed[:6], tzinfo=timezone.utc)
        except (ValueError, TypeError):
            pass

    # 内容：优先 summary，其次 content
    content = ""
    if hasattr(entry, "content") and entry.content:
        content = entry.content[0].value
    elif hasattr(entry, "summary"):
        content = entry.summary

    # 清洗 HTML
    content = clean_html(content)

    # 生成摘要
    summary = generate_summary(content)

    return {
        "feed_id": feed_id,
        "title": title[:1024],
        "link": link[:2048],
        "author": author[:256],
        "published_at": published_at,
        "content": content[:config.MAX_ARTICLE_CONTENT_LENGTH],
        "summary": summary[:config.MAX_SUMMARY_LENGTH],
    }


def clean_html(html_text: str) -> str:
    """清洗 HTML，去除 script/style 标签，转为安全文本"""
    if not html_text:
        return ""

    # 先解码 HTML 实体
    text = html.unescape(html_text)

    # 用 BeautifulSoup 清理
    soup = BeautifulSoup(text, "html.parser")

    # 移除 script 和 style
    for tag in soup(["script", "style", "iframe", "object", "embed"]):
        tag.decompose()

    # 获取纯文本
    cleaned = soup.get_text(separator="\n")

    # 压缩空白行
    cleaned = re.sub(r"\n\s*\n+", "\n\n", cleaned)
    cleaned = cleaned.strip()

    return cleaned


def generate_summary(content: str, max_length: int = 300) -> str:
    """从内容生成摘要"""
    if not content:
        return ""

    # 去掉多余空白
    text = re.sub(r"\s+", " ", content).strip()

    if len(text) <= max_length:
        return text

    # 在句子边界截断（支持中英文标点）
    truncated = text[:max_length]
    last_period = max(
        truncated.rfind("。"), truncated.rfind(". "),
        truncated.rfind("! "), truncated.rfind("? "),
        truncated.rfind("？"), truncated.rfind("！"),
        truncated.rfind("；"),
    )
    if last_period > max_length * 0.5:
        return truncated[:last_period + 1]

    return truncated + "..."


def fetch_and_store_feed(feed_id: int) -> dict:
    """抓取指定 RSS 源并存储文章
    返回抓取结果统计
    """
    db = SessionLocal()
    try:
        feed = db.query(Feed).filter(Feed.id == feed_id).first()
        if not feed:
            return {"success": False, "error": "Feed not found", "articles_count": 0}

        result = fetch_feed(feed.url)

        if not result["success"]:
            # 记录失败
            feed.last_fetch_at = datetime.now(timezone.utc)
            feed.last_fetch_status = "fail"
            feed.last_error = result["error"]
            feed.fail_count += 1

            log = FetchLog(
                feed_id=feed_id,
                status="fail",
                error_message=result["error"],
                response_time_ms=result.get("response_time_ms"),
            )
            db.add(log)
            db.commit()
            return {"success": False, "error": result["error"], "articles_count": 0}

        parsed = result["feed_data"]

        # 更新 feed 元信息
        if hasattr(parsed.feed, "title"):
            feed.title = parsed.feed.title[:512]
        if hasattr(parsed.feed, "description"):
            feed.description = parsed.feed.description[:1000]

        # 存储文章 — 先收集所有文章，内存去重后批量入库
        seen_links = set()
        articles_to_add = []
        articles_to_update = []

        for entry in parsed.entries:
            article_data = parse_article(entry, feed_id)
            link = article_data.get("link", "")
            if not link or link in seen_links:
                continue
            seen_links.add(link)
            articles_to_add.append(article_data)

        # 批量查询已有文章
        if articles_to_add:
            existing_links = {
                row[0] for row in db.query(Article.link).filter(
                    Article.link.in_([a["link"] for a in articles_to_add])
                ).all()
            }

            new_count = 0
            for article_data in articles_to_add:
                if article_data["link"] in existing_links:
                    articles_to_update.append(article_data)
                else:
                    article = Article(**article_data)
                    db.add(article)
                    new_count += 1

            # 更新已有文章
            for article_data in articles_to_update:
                existing = db.query(Article).filter(Article.link == article_data["link"]).first()
                if existing:
                    existing.title = article_data["title"] or existing.title
                    existing.content = article_data["content"] or existing.content
                    existing.summary = article_data["summary"] or existing.summary
                    existing.author = article_data["author"] or existing.author
                    if article_data["published_at"]:
                        existing.published_at = article_data["published_at"]

        # 更新 feed 统计
        feed.last_fetch_at = datetime.now(timezone.utc)
        feed.last_fetch_status = "success"
        feed.last_error = ""
        feed.success_count += 1
        feed.article_count += new_count

        log = FetchLog(
            feed_id=feed_id,
            status="success",
            articles_fetched=new_count,
            response_time_ms=result.get("response_time_ms"),
        )
        db.add(log)
        db.commit()

        return {
            "success": True,
            "articles_count": new_count,
            "feed_title": feed.title,
        }
    except Exception as e:
        db.rollback()
        return {"success": False, "error": str(e), "articles_count": 0}
    finally:
        db.close()


def fetch_all_feeds(feed_ids: list = None) -> list:
    """并发抓取多个 RSS 源
    返回每个源的抓取结果列表
    """
    db = SessionLocal()
    try:
        query = db.query(Feed).filter(Feed.is_active == True)
        if feed_ids:
            query = query.filter(Feed.id.in_(feed_ids))
        feeds = query.all()
    finally:
        db.close()

    results = []
    with ThreadPoolExecutor(max_workers=config.FETCH_CONCURRENCY) as executor:
        future_to_feed = {
            executor.submit(fetch_and_store_feed, feed.id): feed
            for feed in feeds
        }
        for future in as_completed(future_to_feed):
            feed = future_to_feed[future]
            try:
                result = future.result()
                results.append({"feed_id": feed.id, **result})
            except Exception as e:
                results.append({"feed_id": feed.id, "success": False, "error": str(e), "articles_count": 0})

    return results