rss-formatting-preserve: сохранять абзацы и разметку из RSS (html_to_text/extract_full_text normalize; safe_html bleach для предпросмотра; пересборка старых постов)

crawler/rss_crawler.py

@@ -0,0 +1,362 @@
+# VESTI — RSS/Atom-краулер (feedparser + httpx).
+# Читает источники из sources.yaml (crawler: rss), условные GET (etag/modified),
+# нормализует URL (UTM/якоря), дотягивает полный текст (trafilatura) для summary-only,
+# пишет посты в posts (sha256-дедуп) и состояние ленты в rss_state.
+#
+# Запуск:
+#   python -m crawler.rss_crawler --all            # все включённые rss-источники
+#   python -m crawler.rss_crawler --direction linux
+#   python -m crawler.rss_crawler --source lwn
+#   python -m crawler.rss_crawler --source lwn --dry-run   # печать без записи в БД
+import argparse
+import hashlib
+import html as html_mod
+import logging
+import re
+import sys
+import time
+from pathlib import Path
+from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from config import RSS_DEFAULT_TIMEOUT, RSS_USER_AGENT, ensure_dirs
+from db.db import db
+from sources.sources import (
+    get_enabled_sources, get_source, load_sources_yaml, sync_sources_to_db,
+    start_run, finish_run,
+)
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
+log = logging.getLogger("vesti.rss")
+
+MAX_ITEMS = 25          # максимум записей ленты за запуск (остальные — следующими запусками)
+RATE_LIMIT_SEC = 1.5    # пауза между источниками (вежливый опрос)
+MIN_SUMMARY_LEN = 220   # если выжимка короче — пробуем дотянуть полный текст
+ERROR_DEAD_THRESHOLD = 5  # N ошибок подряд → status='dead' (фид-здоровье)
+
+# Параметры запроса, которые режут из URL (трекинг/мусор)
+UTM_PARAMS = {
+    "utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content",
+    "fbclid", "gclid", "ref", "ref_src", "mc_cid", "mc_eid",
+}
+
+
+def normalize_url(url):
+    """Чистит URL от UTM-параметров, fbclid/gclid/ref и якоря -> canonical_url.
+
+    Возвращает None для мусорных ссылок (без схемы/хоста).
+    """
+    if not url:
+        return None
+    url = url.strip()
+    if not url.startswith(("http://", "https://")):
+        return None
+    u = urlparse(url)
+    if not u.netloc:
+        return None
+    q = [(k, v) for k, v in parse_qsl(u.query, keep_blank_values=True)
+         if k.lower() not in UTM_PARAMS]
+    return urlunparse(u._replace(query=urlencode(q), fragment=""))
+
+
+def sha256_text(text: str) -> str:
+    return hashlib.sha256((text or "").encode("utf-8", "ignore")).hexdigest()
+
+
+# Блочные теги, вокруг которых сохраняем переносы (абзацы/списки/заголовки/цитаты)
+_BLOCK_TAGS = [
+    "p", "div", "li", "h1", "h2", "h3", "h4", "h5", "h6",
+    "blockquote", "pre", "tr", "section", "article",
+]
+
+# Нормализация текста: сохраняем структуру абзацев (\n\n), убираем мусор.
+# НЕ схлопываем всё в одну строку — иначе новость в предпросмотре — сплошной текст.
+def normalize_text(text):
+    """Чистит текст: пробелы по краям строк, 3+ переноса → 2, без изменения слов."""
+    if not text:
+        return ""
+    text = text.replace("\r\n", "\n").replace("\r", "\n")
+    text = re.sub(r"[ \t]+\n", "\n", text)      # пробелы перед концом строки
+    text = re.sub(r"\n[ \t]+", "\n", text)      # пробелы после начала строки
+    text = re.sub(r"\n{3,}", "\n\n", text)      # не больше одной пустой строки
+    # пробелы по краям каждой строки
+    text = "\n".join(l.strip() for l in text.split("\n"))
+    text = re.sub(r"\n{3,}", "\n\n", text)
+    return text.strip()
+
+
+def html_to_text(html_str):
+    """Раскрывает HTML (из ленты) в читаемый text с сохранением абзацев/списков.
+
+    HTML-теги разметки (b/i/a/...) вырезаются, их содержимое остаётся текстом;
+    блочные теги (p/li/h1-6/blockquote/pre) дают переносы строк, чтобы текст
+    не превращался в сплошную кашу.
+    """
+    from selectolax.parser import HTMLParser
+    if not html_str:
+        return ""
+    try:
+        tree = HTMLParser(html_str)
+        tree.strip_tags(["script", "style", "noscript"])
+        if tree.body is None:
+            return ""
+        for node in tree.css(", ".join(_BLOCK_TAGS)):
+            node.insert_before("\n\n")
+        text = tree.body.text()
+        return normalize_text(text)
+    except Exception as e:
+        log.debug("html_to_text failed: %s", e)
+        return re.sub(r"<[^>]+>", " ", html_mod.unescape(html_str)).strip()
+
+
+def extract_full_text(url, timeout=15):
+    """Полный текст страницы через trafilatura. Возвращает str или None (не падает)."""
+    import httpx
+    try:
+        r = httpx.get(url, headers={"User-Agent": RSS_USER_AGENT},
+                      timeout=timeout, follow_redirects=True)
+        if r.status_code != 200:
+            return None
+        import trafilatura
+        txt = trafilatura.extract(
+            r.text,
+            include_comments=False,
+            include_tables=False,
+            favor_precision=True,
+            # include_formatting=True сохраняет абзацы (\\n\\n) — иначе текст
+            # схлопывается в сплошную строку и новость нечитаема
+            include_formatting=True,
+        )
+        return (txt or "").strip() or None
+    except Exception as e:
+        log.debug("full text %s failed: %s", url, e)
+        return None
+
+
+def entry_published_at(entry):
+    """published_parsed/updated_parsed -> ISO-строка; None если нет даты."""
+    for key in ("published_parsed", "updated_parsed"):
+        st = getattr(entry, key, None)
+        if st:
+            try:
+                import datetime
+                return datetime.datetime(*st[:6], tzinfo=datetime.timezone.utc).isoformat()
+            except Exception:
+                return None
+    return None
+
+
+def parse_entries(feed, source, dry_run=False):
+    """Превращает записи ленты feedparser в список dict постов.
+
+    returns (posts, last_build_date)
+    """
+    posts = []
+    for entry in feed.entries[:MAX_ITEMS]:
+        link = getattr(entry, "link", None) or ""
+        canonical = normalize_url(link)
+        if not canonical:
+            continue
+        title = getattr(entry, "title", "") or ""
+        # полный текст из ленты (content[].value), иначе summary
+        content_parts = [c.get("value", "") for c in (getattr(entry, "content", None) or [])]
+        raw_html = content_parts[0] if content_parts else ""
+        summary_html = getattr(entry, "summary", "") or ""
+        text = html_to_text(raw_html or summary_html)
+        # summary-only лента: пробуем дотянуть полный текст
+        if (not raw_html or len(text) < MIN_SUMMARY_LEN) and canonical:
+            full = extract_full_text(canonical)
+            if full:
+                text = full
+        if not text:
+            text = title  # последний рубеж: заголовок
+        text = normalize_text(text)
+        posts.append({
+            "url": canonical,
+            "title": title,
+            "text": text,
+            "published_at": entry_published_at(entry),
+            "author": getattr(entry, "author", None) or None,
+        })
+    last_build = None
+    fb = getattr(feed, "last_build_date_parsed", None) or getattr(feed, "published_parsed", None)
+    if fb:
+        try:
+            import datetime
+            last_build = datetime.datetime(*fb[:6], tzinfo=datetime.timezone.utc).isoformat()
+        except Exception:
+            last_build = None
+    return posts, last_build
+
+
+def store_rss_posts(conn, source_id, posts):
+    """Вставка RSS-постов в posts с дедупом по sha256(text) и url.
+
+    Возвращает количество НОВЫХ. Повторная вставка той же статьи (даже с другим
+    URL в ленте) не создаёт второй строки — sha256 от текста.
+    """
+    new = 0
+    for p in posts:
+        sha = sha256_text(p["text"])
+        dup = conn.execute(
+            "SELECT id FROM posts WHERE sha256=? OR url=?",
+            (sha, p["url"]),
+        ).fetchone()
+        if dup:
+            continue
+        conn.execute(
+            """INSERT INTO posts
+               (sha256, source_id, url, text, content_type, published_at, status, is_own, is_own_canonical)
+               VALUES (?,?,?,?, 'text', ?, 'new', 0, 0)""",
+            (sha, source_id, p["url"], p["text"], p["published_at"]),
+        )
+        new += 1
+    return new
+
+
+def run_source(source, dry_run=False):
+    """Краулинг одного RSS-источника: условный GET, парсинг, запись, состояние."""
+    slug = source["slug"]
+    feed_url = source.get("feed_url")
+    if not feed_url:
+        log.error("SKIP %s: нет feed_url (crawler=rss, а лента не задана)", slug)
+        return
+    run_id = start_run(source["id"], f"rss:{slug}", "cron")
+    t0 = time.monotonic()
+
+    try:
+        import httpx
+        # --- состояние ленты ---
+        with db() as conn:
+            st = conn.execute("SELECT * FROM rss_state WHERE slug=?", (slug,)).fetchone()
+        state = dict(st) if st else {}
+
+        headers = {"User-Agent": RSS_USER_AGENT, "Accept": "application/rss+xml, application/atom+xml, application/xml;q=0.9, */*;q=0.8"}
+        if state.get("etag"):
+            headers["If-None-Match"] = state["etag"]
+        if state.get("modified"):
+            headers["If-Modified-Since"] = state["modified"]
+
+        resp = httpx.get(feed_url, headers=headers, timeout=RSS_DEFAULT_TIMEOUT,
+                         follow_redirects=True)
+        # вежливый ретрай на 429/503 (LWN троттлит частые опросы)
+        for attempt in range(3):
+            if resp.status_code in (429, 503):
+                retry_after = 0
+                ra = resp.headers.get("retry-after")
+                if ra:
+                    try: retry_after = int(ra)
+                    except ValueError: retry_after = 10
+                wait = max(retry_after, 5 * (attempt + 1))
+                log.warning("%s: HTTP %d, retry %d/3 через %ds", slug, resp.status_code, attempt + 1, wait)
+                time.sleep(wait)
+                resp = httpx.get(feed_url, headers=headers, timeout=RSS_DEFAULT_TIMEOUT,
+                                 follow_redirects=True)
+            else:
+                break
+        if resp.status_code == 304:
+            # лента не менялась — экономим (0 fetched/0 new, ok)
+            duration = int((time.monotonic() - t0) * 1000)
+            if not dry_run:
+                finish_run(run_id, "ok", 0, 0, None, duration)
+            log.info("OK %s: 304 (не менялась), %dms", slug, duration)
+            return
+
+        if resp.status_code != 200:
+            raise RuntimeError(f"HTTP {resp.status_code} for {feed_url}")
+
+        import feedparser
+        feed = feedparser.parse(resp.content, response_headers=resp.headers)
+
+        if getattr(feed, "bozo", False) and not feed.entries:
+            raise RuntimeError(f"bozo feed (не парсится): {getattr(feed, 'bozo_exception', '')}")
+
+        posts, last_build = parse_entries(feed, source, dry_run)
+
+        if dry_run:
+            print(f"=== {slug} ({source['name']}) — dry_run, {len(posts)} записей ===")
+            for p in posts:
+                print(f"  - {p['published_at'] or '????'} | {p['title'][:80]}")
+                print(f"    {p['url']}")
+            return
+
+        with db() as conn:
+            new = store_rss_posts(conn, source["id"], posts)
+            new_etag = (resp.headers.get("etag") or "").strip()
+            new_modified = (resp.headers.get("last-modified") or "").strip()
+            conn.execute(
+                """INSERT INTO rss_state (slug, etag, modified, last_build_date, last_error, error_count, status, updated_at)
+                   VALUES (?,?,?,?, NULL, 0, 'alive', datetime('now'))
+                   ON CONFLICT(slug) DO UPDATE SET
+                     etag=excluded.etag, modified=excluded.modified,
+                     last_build_date=excluded.last_build_date,
+                     last_error=NULL, error_count=0, status='alive',
+                     updated_at=datetime('now')""",
+                (slug, new_etag or state.get("etag"), new_modified or state.get("modified"), last_build),
+            )
+            conn.execute(
+                "UPDATE sources SET last_fetch=datetime('now'), status='alive', last_error=NULL WHERE id=?",
+                (source["id"],),
+            )
+        duration = int((time.monotonic() - t0) * 1000)
+        finish_run(run_id, "ok", len(posts), new, None, duration)
+        log.info("OK %s: fetched=%d new=%d (%.1fs)", slug, len(posts), new, duration / 1000)
+
+    except Exception as e:
+        duration = int((time.monotonic() - t0) * 1000)
+        if not dry_run:
+            with db() as conn:
+                st = conn.execute("SELECT error_count FROM rss_state WHERE slug=?", (slug,)).fetchone()
+                err_count = (st[0] if st else 0) + 1
+                status = "dead" if err_count >= ERROR_DEAD_THRESHOLD else "alive"
+                conn.execute(
+                    """INSERT INTO rss_state (slug, last_error, error_count, status, updated_at)
+                       VALUES (?,?,?,?, datetime('now'))
+                       ON CONFLICT(slug) DO UPDATE SET
+                         last_error=excluded.last_error, error_count=excluded.error_count,
+                         status=excluded.status, updated_at=datetime('now')""",
+                    (slug, str(e)[:500], err_count, status),
+                )
+                conn.execute(
+                    "UPDATE sources SET last_error=?, status=? WHERE id=?",
+                    (str(e)[:500], status, source["id"]),
+                )
+            finish_run(run_id, "error", 0, 0, str(e)[:500], duration)
+        log.error("ERROR %s: %s", slug, e)
+
+
+def main():
+    ap = argparse.ArgumentParser(description="VESTI RSS/Atom crawler")
+    ap.add_argument("--source", help="slug источника (например lwn)")
+    ap.add_argument("--direction", help="направление (например linux)")
+    ap.add_argument("--all", action="store_true", help="все включённые rss-источники")
+    ap.add_argument("--dry-run", action="store_true", help="печать записей без записи в БД")
+    args = ap.parse_args()
+
+    ensure_dirs()
+    data = load_sources_yaml()
+    sync_sources_to_db(data)
+
+    if args.source:
+        src = get_source(args.source)
+        if not src:
+            log.error("source %s not found", args.source)
+            sys.exit(1)
+        sources = [src]
+    elif args.direction:
+        sources = get_enabled_sources(crawler="rss", direction=args.direction)
+    else:
+        sources = get_enabled_sources(crawler="rss")
+
+    if not sources:
+        log.info("no rss sources to crawl")
+        return
+
+    for src in sources:
+        run_source(src, dry_run=args.dry_run)
+        time.sleep(RATE_LIMIT_SEC)
+
+
+if __name__ == "__main__":
+    main()