diff --git a/PROJECT.md b/PROJECT.md index 37b6d34..e229f3f 100644 --- a/PROJECT.md +++ b/PROJECT.md @@ -586,7 +586,13 @@ Retention: записи старше 30 дней удаляются при ст Записи без достаточного текста, включая video-only, не сохраняются и не попадают в media uploader. -`access='cloudflare'` включает Playwright и RuCaptcha для конкретного +Конфиг v1 описывает `discovery` (`rss` или `html`), точные `detail.fields`, +`detail.media`, транспорт detail-страниц и ограниченные ретраи. Каждый field +может иметь ordered `candidates`: CSS/attribute/JSON/date fallback. Ошибка +одного detail возвращает `partial`, сохраняет успешных соседей и показывает +диагностику в статусе источника. + +`access.type='cloudflare'` включает Playwright и RuCaptcha для конкретного источника; обычные RSS и сайты идут без браузера. Полученные `cf_clearance` и точный User-Agent браузера сохраняются в runtime state источника и повторно используются media uploader при скачивании защищённых картинок. @@ -1136,11 +1142,9 @@ Worker учитывает: - active - обязательный JSON-конфиг конкретного источника -Пример рабочего PopularAirsoft: - -```json -{"format":"rss","access":"cloudflare","max_items":10,"min_text_length":50} -``` +Полный проверенный конфиг PopularAirsoft находится в +`site_parser_worker/README.md`. Для него используется HTML discovery главной +страницы, потому что RSS смешивает короткие video pages и полные новости. Ошибки конфигурации, внешнего worker или сайта показываются в статусе источника. После загрузки защищённые изображения отображаются в raw/editor по diff --git a/RAA.md b/RAA.md index 84c3079..c13a304 100644 --- a/RAA.md +++ b/RAA.md @@ -1,6 +1,6 @@ # RAA deployment -Last updated: 2026-08-10. +Last updated: 2026-08-11. RAA is not a separate code project anymore. The source of truth for code and documentation is: @@ -96,20 +96,15 @@ All source records, JSON configs, schedules, errors, raw posts, and media state remain in the RAA database. The worker is managed by Docker Compose from `/opt/site-parser` and should be healthy as `site-parser-worker`. -PopularAirsoft source id `72`: +PopularAirsoft source id `72` uses `https://popularairsoft.com/` and the full +v1 config from `site_parser_worker/README.md`. Its RSS is not used for +discovery because it mixes short video pages with full articles. HTML discovery +selects the `The Latest News` cards and the worker then opens every `/news/...` +detail. Verified live 2026-08-11: nine details parsed, zero errors. -- URL: `https://popularairsoft.com/rss.xml` -- Config: `{"format":"rss","access":"cloudflare","max_items":10,"min_text_length":50}` -- `access='cloudflare'` deliberately starts the browser/RuCaptcha flow before - fetching the RSS. -- Keep `follow_links` disabled for this feed. Nine current entries have no - body, only title/video, and are intentionally filtered before media upload. -- Cloudflare cookies and browser User-Agent are reused by `media-uploader` for - protected images. -- Admin image previews use authenticated Telegram-backed - `/raw/media/{media_id}` URLs. -- Verified 2026-08-10: source `ok`; raw post `102` is `storage_ready`; four - images are `uploaded`. The AI later rejected the content as non-target. +Cloudflare session state is returned to the main app and is also reused by +`media-uploader` for protected images. Admin previews remain authenticated +Telegram-backed `/raw/media/{media_id}` URLs. ## RAA Categories diff --git a/STATE.md b/STATE.md index 3781d2c..da2ade4 100644 --- a/STATE.md +++ b/STATE.md @@ -1,6 +1,6 @@ # N8 Parser: current state -Last updated: 2026-08-10 +Last updated: 2026-08-11 This file is the short handoff state for future Codex threads. Read this first before touching the project, so the whole chat history does not need to be carried forward. @@ -107,20 +107,18 @@ and returns normalized JSON. `/raw/media/{media_id}` URLs backed by Telegram storage, not by the original Cloudflare-protected URL. -Current RAA test source: +Site configs now use the versioned declarative v1 contract. The worker supports +RSS or HTML discovery, exact per-field selectors with ordered fallbacks, +detail-page extraction, media normalization, bounded retries, and per-item +errors. A failed detail does not discard successful siblings; the source is +marked with a visible partial diagnostic. -```json -{"format":"rss","access":"cloudflare","max_items":10,"min_text_length":50} -``` - -- Source: `Popularairsoft`, `https://popularairsoft.com/rss.xml`, id `72`. -- Do not add `follow_links` for the current feed: one of ten entries has a real - body and four photos; the other nine are title-plus-video entries and are - intentionally skipped by the body-length threshold. -- Verified on 2026-08-10: source status `ok`, raw post `102` reached - `storage_ready`, and all four photos were uploaded. Its later rejection is - an AI non-target decision, not a parser/media failure. -- Verified RAA deploy: commit `f866a13a8a83598e26a7a6a3406b8bc1c01c9a80`. +PopularAirsoft source id `72` should use `https://popularairsoft.com/`, not its +mixed-content RSS feed. The complete config is documented in +`site_parser_worker/README.md`. It discovers the `The Latest News` cards and +then extracts each `article.news.full` page. Verified live on 2026-08-11: nine +items, nine full details, zero errors. Twitch embeds return as +`external_video`; YouTube embeds are normalized for the existing downloader. LXC 108 operations: diff --git a/site_parser_worker/Dockerfile b/site_parser_worker/Dockerfile index 074445f..5f85d87 100644 --- a/site_parser_worker/Dockerfile +++ b/site_parser_worker/Dockerfile @@ -4,7 +4,7 @@ WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt RUN playwright install chrome -COPY app.py . +COPY app.py extractor.py ./ ENV PYTHONUNBUFFERED=1 EXPOSE 8080 diff --git a/site_parser_worker/README.md b/site_parser_worker/README.md new file mode 100644 index 0000000..df66871 --- /dev/null +++ b/site_parser_worker/README.md @@ -0,0 +1,146 @@ +# Site Parser Worker + +The worker is a stateless executor. The main application owns source configs, +schedules, database state, and queues. Each request gives the worker a source +URL, a versioned JSON instruction, and optional Cloudflare session state. The +worker returns normalized items plus per-item diagnostics. + +## Config v1 + +- `discovery.type`: `rss` or `html`. +- `discovery.fields`: where list/RSS values are located. +- `detail.enabled`: open every discovered item when enabled. +- `detail.root_selector`: limits extraction to the actual article. +- `detail.fields`: exact field rules. A rule may contain ordered `candidates`. +- `detail.media`: exact selectors and fallback attributes for media. +- `detail.transport`: `http`, `fetch`, or `browser`. +- `access.type`: `auto`, `http`, `browser`, or `cloudflare`. +- `retry`: bounded attempts, delay, and per-attempt timeout. + +Field extraction modes are `text`, `html`, `attr`, and `json`. Date rules can +provide `formats` and `timezone`. Relative links are resolved against the page +URL. Unknown iframe/video providers are returned as `external_video`; they do +not fail the item. + +Errors are isolated to one item. Successful items are returned with +`status=partial`; the source receives a visible diagnostic in the main app. +Only discovery/network failure prevents the whole request from returning a +batch. + +## PopularAirsoft + +Use `https://popularairsoft.com/` as the source URL. Its RSS mixes full news +articles with short video pages, so the reliable discovery surface is the +`The Latest News` HTML block. The worker opens the main page through +Cloudflare, collects `/news/...` links, and downloads each article using the +same clearance and an impersonated browser TLS fingerprint. + +```json +{ + "version": 1, + "discovery": { + "type": "html", + "item_selector": "#block-views-block-latest-news-list-block-2 .feature-contents, #block-views-block-latest-news-list-block-1 .lt-teasure", + "limit": 10, + "fields": { + "url": { + "selector": "a.link-title", + "extract": "attr", + "attribute": "href", + "required": true + }, + "external_id": { + "selector": "a.link-title", + "extract": "attr", + "attribute": "href", + "required": true + }, + "title": { + "selector": "a.link-title", + "extract": "text", + "required": true + }, + "published_at": { + "selector": "time[datetime]", + "extract": "attr", + "attribute": "datetime", + "required": true + } + }, + "media": [ + { + "type": "photo", + "selector": ".site-image img", + "attributes": ["src", "data-src", "srcset"] + } + ] + }, + "detail": { + "enabled": true, + "always": true, + "transport": "http", + "root_selector": "article.news.full", + "fields": { + "title": { + "candidates": [ + { + "selector": ".feature-contents > .news-story-texts:first-child h2", + "extract": "text" + }, + {"source": "list.title"} + ], + "required": true + }, + "published_at": { + "candidates": [ + { + "selector": ".feature-contents > .news-story-texts:first-child .news-story-date", + "extract": "text", + "formats": ["%d %b %Y"], + "timezone": "UTC" + }, + {"source": "list.published_at"} + ], + "required": true + }, + "author": { + "selector": ".feature-contents > .news-story-texts:first-child h4", + "extract": "text" + }, + "text": { + "selector": ".feature-contents > .news-story-texts:last-child .field--name-body", + "extract": "text", + "required": true, + "min_length": 50 + } + }, + "media": [ + { + "type": "photo", + "selector": ".feature-contents > .site-image .field--name-field-image img, .field--name-body img", + "attributes": ["src", "data-src", "srcset"] + }, + { + "type": "video", + "selector": ".field--name-body iframe, .field--name-body video, .field--name-body source", + "attributes": ["src", "data-src"] + } + ] + }, + "access": { + "type": "cloudflare", + "wait_for": "#block-views-block-latest-news-list-block-1" + }, + "retry": { + "attempts": 2, + "delay_seconds": 2, + "timeout_seconds": 90 + }, + "min_text_length": 50 +} +``` + +Verified against the live site on 2026-08-11: nine items, nine successful +details, zero errors. The VFC sample returned 5,041 text characters, one photo, +and two Twitch links. The Double Bell sample returned 1,293 characters, one +photo, and one normalized YouTube URL. diff --git a/site_parser_worker/app.py b/site_parser_worker/app.py index 7ccf6b0..a704970 100644 --- a/site_parser_worker/app.py +++ b/site_parser_worker/app.py @@ -1,25 +1,30 @@ from __future__ import annotations import asyncio +import logging import os -import re import secrets +from contextlib import asynccontextmanager from datetime import datetime, timezone from time import struct_time -from typing import Annotated, Any -from urllib.parse import parse_qs, urljoin, urlparse +from typing import Annotated, Any, AsyncIterator +from urllib.parse import urljoin import feedparser import httpx from bs4 import BeautifulSoup +from curl_cffi.requests import AsyncSession as CurlAsyncSession from fastapi import FastAPI, Header, HTTPException -from playwright.async_api import async_playwright +from playwright.async_api import BrowserContext, Page, TimeoutError as PlaywrightTimeoutError, async_playwright from playwright_captcha import CaptchaType, FrameworkType, TwoCaptchaSolver from pydantic import BaseModel, Field, HttpUrl, SecretStr from twocaptcha import AsyncTwoCaptcha -app = FastAPI(title="Site Parser Worker", version="0.1.0") +from extractor import ConfigError, detail_from_html, extract_fields, extract_media, normalize_config + +app = FastAPI(title="Site Parser Worker", version="1.0.0") browser_lock = asyncio.Lock() +logger = logging.getLogger("site_parser") class ParseRequest(BaseModel): @@ -41,6 +46,14 @@ class ParsedItem(BaseModel): media: list[dict[str, str]] = Field(default_factory=list) +class CloudflareRequired(RuntimeError): + pass + + +class PermanentPageError(RuntimeError): + pass + + def require_token(worker_token: str | None) -> None: expected = os.getenv("WORKER_TOKEN", "") if not expected: @@ -62,159 +75,354 @@ def iso_date(value: struct_time | None) -> str | None: return datetime(*value[:6], tzinfo=timezone.utc).isoformat() -def youtube_url(value: str, base_url: str = "") -> str | None: - url = urljoin(base_url, str(value or "").strip()) - parsed = urlparse(url) - host = (parsed.hostname or "").lower().removeprefix("www.").removeprefix("m.") - video_id = "" - if host == "youtu.be": - video_id = parsed.path.strip("/").split("/", 1)[0] - elif host in {"youtube.com", "youtube-nocookie.com"}: - if parsed.path == "/watch": - video_id = (parse_qs(parsed.query).get("v") or [""])[0] - elif parsed.path.startswith(("/embed/", "/shorts/", "/live/")): - video_id = parsed.path.strip("/").split("/", 1)[1] - if not re.fullmatch(r"[A-Za-z0-9_-]{6,20}", video_id): - return None - return f"https://www.youtube.com/watch?v={video_id}" - - -def clean_html(value: Any, base_url: str = "") -> tuple[str, str, list[dict[str, str]]]: +def clean_feed_content(value: Any, base_url: str) -> tuple[str, str, list[dict[str, str]]]: html = str(value or "").strip() soup = BeautifulSoup(html, "html.parser") for element in soup.find_all(["script", "style", "noscript"]): element.decompose() text = soup.get_text("\n", strip=True) - media = [ - {"type": "photo", "url": urljoin(base_url, str(image["src"]))} - for image in soup.find_all("img", src=True) - ] - for element in soup.find_all(["video", "source"], src=True): - media.append({"type": "video", "url": urljoin(base_url, str(element["src"]))}) - for element in soup.find_all(["iframe", "a"]): - url = youtube_url(str(element.get("src") or element.get("href") or ""), base_url) - if url: - media.append({"type": "video", "url": url}) + media = extract_media( + soup, + [ + {"type": "photo", "selector": "img", "attributes": ["src", "data-src", "srcset"]}, + {"type": "video", "selector": "iframe, video, source", "attributes": ["src", "data-src"]}, + ], + base_url, + ) return html, text, media -def parse_rss(xml: str, max_items: int) -> list[ParsedItem]: +def default_rss_fields() -> dict[str, Any]: + return { + "url": {"candidates": [{"source": "rss.link"}], "required": True}, + "external_id": { + "candidates": [{"source": "rss.id"}, {"source": "rss.guid"}, {"source": "rss.link"}], + "required": True, + }, + "title": {"candidates": [{"source": "rss.title"}]}, + "published_at": { + "candidates": [ + {"source": "rss.published"}, + {"source": "rss.updated"}, + ] + }, + "author": {"candidates": [{"source": "rss.author"}]}, + } + + +def discover_rss(xml: str, config: dict[str, Any]) -> list[dict[str, Any]]: feed = feedparser.parse(xml, sanitize_html=False) if feed.bozo and not feed.entries: raise ValueError(f"Invalid RSS: {feed.bozo_exception}") - items = [] - for entry in feed.entries[:max_items]: - title = BeautifulSoup(str(entry.get("title") or ""), "html.parser").get_text(" ", strip=True) - url = str(entry.get("link") or "").strip() - content = (entry.get("content") or [{}])[0].get("value") or entry.get("description") or entry.get("summary") or "" - html, text, media = clean_html(content, url) - for enclosure in entry.get("enclosures") or []: + discovery = config["discovery"] + fields = discovery.get("fields") or default_rss_fields() + items: list[dict[str, Any]] = [] + for entry in feed.entries[: discovery["limit"]]: + raw = dict(entry) + values, errors = extract_fields(None, fields, {"rss": raw}) + url = str(values.get("url") or raw.get("link") or "").strip() + external_id = str(values.get("external_id") or raw.get("id") or raw.get("guid") or url).strip() + if not url or not external_id: + items.append({"url": url, "discovery_errors": errors or [{"field": "url", "error": "URL not found"}]}) + continue + content = (raw.get("content") or [{}])[0].get("value") or raw.get("description") or raw.get("summary") or "" + html, text, media = clean_feed_content(content, url) + for enclosure in raw.get("enclosures") or []: enclosure_url = urljoin(url, str(enclosure.get("href") or enclosure.get("url") or "").strip()) media_type = str(enclosure.get("type") or "") - if enclosure_url and (not media_type or media_type.startswith("image/")): - media.append({"type": "photo", "url": enclosure_url}) - elif enclosure_url and media_type.startswith("video/"): + if enclosure_url and media_type.startswith("video/"): media.append({"type": "video", "url": enclosure_url}) + elif enclosure_url: + media.append({"type": "photo", "url": enclosure_url}) items.append( - ParsedItem( - external_id=str(entry.get("id") or entry.get("guid") or url).strip(), - url=url, - title=title, - text=text, - html=html, - published_at=iso_date(entry.get("published_parsed") or entry.get("updated_parsed")), - author=str(entry.get("author") or "").strip() or None, - media=list({item["url"]: item for item in media}.values()), - ) + { + "external_id": external_id, + "url": url, + "title": str(values.get("title") or BeautifulSoup(str(raw.get("title") or ""), "html.parser").get_text(" ", strip=True)), + "text": text, + "html": html, + "published_at": values.get("published_at") or iso_date(raw.get("published_parsed") or raw.get("updated_parsed")), + "author": values.get("author") or str(raw.get("author") or "").strip() or None, + "media": list({entry["url"]: entry for entry in media}.values()), + "source_context": {"rss": raw}, + "discovery_errors": errors, + } ) return items -async def fetch_in_browser( +def discover_html(html: str, source_url: str, config: dict[str, Any]) -> list[dict[str, Any]]: + soup = BeautifulSoup(html, "html.parser") + discovery = config["discovery"] + fields = discovery.get("fields") or {} + items: list[dict[str, Any]] = [] + for element in soup.select(str(discovery["item_selector"]))[: discovery["limit"]]: + values, errors = extract_fields(element, fields, {"page": {"url": source_url}}) + url = urljoin(source_url, str(values.get("url") or "").strip()) + external_id = str(values.get("external_id") or url).strip() + if not url or not external_id: + items.append({"url": url, "discovery_errors": errors or [{"field": "url", "error": "URL not found"}]}) + continue + items.append( + { + "external_id": external_id, + "url": url, + "title": str(values.get("title") or ""), + "text": str(values.get("text") or ""), + "html": "", + "published_at": values.get("published_at"), + "author": values.get("author"), + "media": extract_media(element, discovery.get("media") or [], source_url), + "source_context": {"list": values}, + "discovery_errors": errors, + } + ) + return items + + +def apply_detail(item: dict[str, Any], html: str, config: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, str]]]: + values, errors = detail_from_html(html, item["url"], config, item.get("source_context")) + if errors: + return item, errors + for key in ("title", "text", "published_at", "author", "html"): + if values.get(key) not in {None, ""}: + item[key] = values[key] + item["media"] = list( + {(entry["type"], entry["url"]): entry for entry in [*item.get("media", []), *values.get("media", [])]}.values() + ) + return item, [] + + +def public_item(item: dict[str, Any]) -> ParsedItem: + return ParsedItem( + external_id=str(item["external_id"]), + url=str(item["url"]), + title=str(item.get("title") or ""), + text=str(item.get("text") or ""), + html=str(item.get("html") or ""), + published_at=str(item["published_at"]) if item.get("published_at") else None, + author=str(item["author"]) if item.get("author") else None, + media=item.get("media") or [], + ) + + +async def retry(operation, attempts: int, delay: float, timeout: float): + last_error: Exception | None = None + for attempt in range(1, attempts + 1): + try: + return await asyncio.wait_for(operation(), timeout=timeout), attempt + except PermanentPageError: + raise + except Exception as exc: + last_error = exc + if attempt < attempts and delay: + await asyncio.sleep(delay) + raise RuntimeError(str(last_error or "operation failed")) from last_error + + +async def http_document(client: httpx.AsyncClient, url: str) -> str: + response = await client.get(url) + if is_cloudflare_challenge(response.status_code, response.text): + raise CloudflareRequired("Cloudflare challenge received") + response.raise_for_status() + return response.text + + +async def curl_document(client: CurlAsyncSession, url: str) -> str: + response = await client.get(url, allow_redirects=True, timeout=30) + body = response.text + if is_cloudflare_challenge(response.status_code, body) or "Attention Required" in body[:10_000]: + raise CloudflareRequired("Cloudflare blocked impersonated request") + if response.status_code < 200 or response.status_code >= 300: + raise RuntimeError(f"detail returned HTTP {response.status_code}") + return body + + +async def enrich_items( + items: list[dict[str, Any]], + config: dict[str, Any], + document_loader, +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + attempts = config["retry"]["attempts"] + delay = config["retry"]["delay_seconds"] + timeout = config["retry"]["timeout_seconds"] + errors: list[dict[str, Any]] = [] + enriched = [] + for item in items: + if item.get("discovery_errors"): + errors.append({"url": item.get("url"), "stage": "discovery", "attempts": 1, "errors": item["discovery_errors"]}) + continue + if not config["detail"]["always"] and item.get("text") and item.get("media"): + enriched.append(item) + continue + try: + html, used_attempts = await retry(lambda item=item: document_loader(item["url"]), attempts, delay, timeout) + item, field_errors = apply_detail(item, html, config) + if field_errors: + errors.append({"url": item["url"], "stage": "detail", "attempts": used_attempts, "errors": field_errors}) + continue + enriched.append(item) + logger.info("Parsed detail %s", item["url"]) + except CloudflareRequired: + raise + except Exception as exc: + errors.append({"url": item.get("url"), "stage": "detail", "attempts": attempts, "error": str(exc)}) + logger.warning("Detail failed %s: %s", item.get("url"), exc) + return enriched, errors + + +async def parse_over_http(source_url: str, config: dict[str, Any]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + attempts = config["retry"]["attempts"] + delay = config["retry"]["delay_seconds"] + timeout = config["retry"]["timeout_seconds"] + async with httpx.AsyncClient( + follow_redirects=True, + timeout=30, + headers={"User-Agent": "Mozilla/5.0 SiteParser/1.0"}, + ) as client: + document, _ = await retry(lambda: http_document(client, source_url), attempts, delay, timeout) + items = discover_rss(document, config) if config["discovery"]["type"] == "rss" else discover_html(document, source_url, config) + if config["detail"]["enabled"]: + return await enrich_items(items, config, lambda url: http_document(client, url)) + return items, [] + + +@asynccontextmanager +async def captcha_solver(page: Page, token: str | None) -> AsyncIterator[TwoCaptchaSolver | None]: + if not token: + yield None + return + async with TwoCaptchaSolver( + framework=FrameworkType.PLAYWRIGHT, + page=page, + async_two_captcha_client=AsyncTwoCaptcha(token), + max_attempts=1, + ) as solver: + yield solver + + +async def browser_document( + page: Page, url: str, - rucaptcha_token: str, + solver: TwoCaptchaSolver | None, + wait_for: str | None = None, +) -> str: + await page.goto(url, wait_until="domcontentloaded", timeout=60_000) + title = (await page.title()).lower() + if "just a moment" in title or "attention required" in title: + if solver is None: + raise RuntimeError("Cloudflare challenge received but RuCaptcha is not configured") + await solver.solve_captcha( + captcha_container=page, + captcha_type=CaptchaType.CLOUDFLARE_INTERSTITIAL, + ) + if wait_for: + try: + await page.locator(wait_for).first.wait_for(state="attached", timeout=20_000) + except PlaywrightTimeoutError as exc: + raise PermanentPageError( + f"selector not found: {wait_for}; url={page.url}; title={await page.title()}" + ) from exc + return await page.content() + + +async def browser_discovery_document(page: Page, source_url: str, config: dict[str, Any], solver) -> str: + wait_for = config["access"].get("wait_for") if config["discovery"]["type"] == "html" else None + html = await browser_document(page, source_url, solver, wait_for) + if config["discovery"]["type"] != "rss": + return html + pre = page.locator("pre").first + await pre.wait_for(state="attached", timeout=60_000) + return await pre.inner_text() + + +async def browser_fetch_document(page: Page, url: str) -> str: + result = await page.evaluate( + """ + async (url) => { + const response = await fetch(url, {credentials: 'include'}); + return {status: response.status, text: await response.text()}; + } + """, + url, + ) + status = int(result.get("status") or 0) + body = str(result.get("text") or "") + if is_cloudflare_challenge(status, body) or "Attention Required" in body[:10_000]: + raise CloudflareRequired("Cloudflare blocked browser fetch") + if status < 200 or status >= 300: + raise RuntimeError(f"detail returned HTTP {status}") + return body + + +async def parse_in_browser( + source_url: str, + config: dict[str, Any], + rucaptcha_token: str | None, browser_state: dict[str, Any] | None, browser_user_agent: str | None, -) -> tuple[str, dict[str, Any], str]: - # ponytail: one browser at a time; use a queue only when parallel source parsing is needed. +) -> tuple[list[dict[str, Any]], list[dict[str, Any]], dict[str, Any], str]: + attempts = config["retry"]["attempts"] + delay = config["retry"]["delay_seconds"] + timeout = config["retry"]["timeout_seconds"] + errors: list[dict[str, Any]] = [] async with browser_lock: async with async_playwright() as playwright: browser = await playwright.chromium.launch(channel="chrome", headless=False) - context = await browser.new_context( - storage_state=browser_state, - user_agent=browser_user_agent, - ) + context: BrowserContext = await browser.new_context(storage_state=browser_state, user_agent=browser_user_agent) page = await context.new_page() - async with TwoCaptchaSolver( - framework=FrameworkType.PLAYWRIGHT, - page=page, - async_two_captcha_client=AsyncTwoCaptcha(rucaptcha_token), - max_attempts=1, - ) as solver: - await page.goto(url, wait_until="domcontentloaded", timeout=60_000) - if "just a moment" in (await page.title()).lower(): - await solver.solve_captcha( - captcha_container=page, - captcha_type=CaptchaType.CLOUDFLARE_INTERSTITIAL, - ) - try: - await page.locator("pre").wait_for(state="visible", timeout=60_000) - except Exception as exc: - raise RuntimeError(f"RSS did not load after Cloudflare challenge: {await page.title()}") from exc - xml = await page.locator("pre").inner_text() + async with captcha_solver(page, rucaptcha_token) as solver: + document, _ = await retry( + lambda: browser_discovery_document(page, source_url, config, solver), attempts, delay, timeout + ) + items = discover_rss(document, config) if config["discovery"]["type"] == "rss" else discover_html(document, source_url, config) + logger.info("Discovered %s items from %s", len(items), source_url) + if config["detail"]["enabled"]: + if config["detail"]["transport"] == "http": + original_state = browser_state if isinstance(browser_state, dict) else {} + original_cookies = original_state.get("cookies") or [] + cookies = { + str(cookie["name"]): str(cookie["value"]) + for cookie in (original_cookies or await context.cookies()) + if cookie.get("name") and cookie.get("value") + } + user_agent = browser_user_agent or await page.evaluate("navigator.userAgent") + async with CurlAsyncSession( + impersonate="chrome", + cookies=cookies, + headers={"User-Agent": str(user_agent), "Referer": source_url}, + ) as client: + items, errors = await enrich_items(items, config, lambda url: curl_document(client, url)) + elif config["detail"]["transport"] == "fetch": + items, errors = await enrich_items(items, config, lambda url: browser_fetch_document(page, url)) + else: + enriched = [] + wait_for = str(config["detail"].get("wait_for") or config["detail"]["root_selector"]) + for item in items: + if item.get("discovery_errors"): + errors.append({"url": item.get("url"), "stage": "discovery", "attempts": 1, "errors": item["discovery_errors"]}) + continue + if not config["detail"]["always"] and item.get("text") and item.get("media"): + enriched.append(item) + continue + try: + html, used_attempts = await retry( + lambda item=item: browser_document(page, item["url"], solver, wait_for), attempts, delay, timeout + ) + item, field_errors = apply_detail(item, html, config) + if field_errors: + errors.append({"url": item["url"], "stage": "detail", "attempts": used_attempts, "errors": field_errors}) + continue + enriched.append(item) + logger.info("Parsed detail %s", item["url"]) + except Exception as exc: + logger.warning("Detail failed %s: %s", item.get("url"), exc) + errors.append({"url": item.get("url"), "stage": "detail", "attempts": attempts, "error": str(exc)}) + items = enriched state = await context.storage_state() user_agent = await page.evaluate("navigator.userAgent") await browser.close() - return xml, state, user_agent - - -async def enrich_items_in_browser( - items: list[ParsedItem], - rucaptcha_token: str, - browser_state: dict[str, Any] | None, - browser_user_agent: str | None, - content_selector: str, - text_selector: str, -) -> tuple[list[ParsedItem], dict[str, Any], str]: - async with browser_lock: - async with async_playwright() as playwright: - browser = await playwright.chromium.launch(channel="chrome", headless=False) - context = await browser.new_context( - storage_state=browser_state, - user_agent=browser_user_agent, - ) - page = await context.new_page() - async with TwoCaptchaSolver( - framework=FrameworkType.PLAYWRIGHT, - page=page, - async_two_captcha_client=AsyncTwoCaptcha(rucaptcha_token), - max_attempts=1, - ) as solver: - for item in items: - if item.text and item.media: - continue - await page.goto(item.url, wait_until="domcontentloaded", timeout=60_000) - if "just a moment" in (await page.title()).lower(): - await solver.solve_captcha( - captcha_container=page, - captcha_type=CaptchaType.CLOUDFLARE_INTERSTITIAL, - ) - try: - content = page.locator(content_selector).first - await content.wait_for(state="visible", timeout=60_000) - except Exception as exc: - raise RuntimeError(f"Item content not found: {item.url} ({content_selector})") from exc - html, text, media = clean_html(await content.inner_html(), item.url) - if text_selector: - text_content = content.locator(text_selector).first - text = clean_html(await text_content.inner_html())[1] if await text_content.count() else "" - item.html = html - item.text = text - item.media = list({entry["url"]: entry for entry in [*item.media, *media]}.values()) - state = await context.storage_state() - user_agent = await page.evaluate("navigator.userAgent") - await browser.close() - return items, state, user_agent + return items, errors, state, user_agent @app.get("/health") @@ -228,82 +436,53 @@ async def parse_source( worker_token: Annotated[str | None, Header(alias="X-Worker-Token")] = None, ) -> dict[str, Any]: require_token(worker_token) - config = request.config - if not config: - raise HTTPException(status_code=422, detail="config is required and cannot be empty") - if config.get("format") != "rss": - raise HTTPException(status_code=422, detail="Only config.format=rss is supported") - access = str(config.get("access") or "auto") - if access not in {"auto", "http", "cloudflare"}: - raise HTTPException(status_code=422, detail="config.access must be auto, http or cloudflare") try: - max_items = int(config.get("max_items", 20)) - except (TypeError, ValueError) as exc: - raise HTTPException(status_code=422, detail="config.max_items must be an integer") from exc - if not 1 <= max_items <= 100: - raise HTTPException(status_code=422, detail="config.max_items must be between 1 and 100") - follow_links = config.get("follow_links", False) - if not isinstance(follow_links, bool): - raise HTTPException(status_code=422, detail="config.follow_links must be true or false") - content_selector = str(config.get("content_selector") or "").strip() - text_selector = str(config.get("text_selector") or "").strip() - if follow_links and not content_selector: - raise HTTPException(status_code=422, detail="config.content_selector is required when follow_links=true") - url = str(request.url) - xml = "" + config = normalize_config(request.config) + except ConfigError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + + source_url = str(request.url) + access = config["access"]["type"] state = request.browser_state browser_user_agent = request.browser_user_agent fetched_via = "http" - if access != "cloudflare": - try: - async with httpx.AsyncClient(follow_redirects=True, timeout=30) as client: - response = await client.get(url, headers={"User-Agent": "Mozilla/5.0 SiteParser/0.1"}) - except httpx.HTTPError as exc: - raise HTTPException(status_code=502, detail=f"Source request failed: {exc}") from exc - if not is_cloudflare_challenge(response.status_code, response.text): - try: - response.raise_for_status() - except httpx.HTTPStatusError as exc: - raise HTTPException(status_code=502, detail=f"Source returned HTTP {response.status_code}") from exc - xml = response.text - elif access == "http": - raise HTTPException(status_code=502, detail="Cloudflare challenge received in http-only mode") - if not xml: - if not request.rucaptcha_token: - raise HTTPException(status_code=422, detail="rucaptcha_token is required for Cloudflare") - fetched_via = "cloudflare" - try: - xml, state, browser_user_agent = await fetch_in_browser( - url, - request.rucaptcha_token.get_secret_value(), - state, - browser_user_agent, - ) - except Exception as exc: - raise HTTPException(status_code=502, detail=str(exc)) from exc + logger.info("Parse request source=%s discovery=%s detail=%s", source_url, config["discovery"]["type"], config["detail"]["enabled"]) try: - items = parse_rss(xml, max_items) - except ValueError as exc: - raise HTTPException(status_code=502, detail=str(exc)) from exc - if follow_links: - if not request.rucaptcha_token: - raise HTTPException(status_code=422, detail="rucaptcha_token is required when follow_links=true") - try: - items, state, browser_user_agent = await enrich_items_in_browser( - items, - request.rucaptcha_token.get_secret_value(), + if access in {"browser", "cloudflare"}: + items, errors, state, browser_user_agent = await parse_in_browser( + source_url, + config, + request.rucaptcha_token.get_secret_value() if request.rucaptcha_token else None, state, browser_user_agent, - content_selector, - text_selector, ) - except Exception as exc: - raise HTTPException(status_code=502, detail=str(exc)) from exc - fetched_via += "+item-pages" + fetched_via = "browser" + else: + try: + items, errors = await parse_over_http(source_url, config) + except CloudflareRequired: + if access == "http": + raise + items, errors, state, browser_user_agent = await parse_in_browser( + source_url, + config, + request.rucaptcha_token.get_secret_value() if request.rucaptcha_token else None, + state, + browser_user_agent, + ) + fetched_via = "browser" + except Exception as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + + if not items and not errors: + errors = [{"url": source_url, "stage": "discovery", "attempts": 1, "error": "no items found"}] + status = "partial" if errors and items else "failed" if errors else "ok" return { - "source_url": url, + "status": status, + "source_url": source_url, "fetched_via": fetched_via, - "items": [item.model_dump() for item in items], + "items": [public_item(item).model_dump() for item in items], + "errors": errors, "browser_state": state, "browser_user_agent": browser_user_agent, } diff --git a/site_parser_worker/extractor.py b/site_parser_worker/extractor.py new file mode 100644 index 0000000..b0b466b --- /dev/null +++ b/site_parser_worker/extractor.py @@ -0,0 +1,357 @@ +from __future__ import annotations + +import json +import re +from datetime import datetime, timezone +from email.utils import parsedate_to_datetime +from typing import Any +from urllib.parse import parse_qs, urljoin, urlparse +from zoneinfo import ZoneInfo + +from bs4 import BeautifulSoup, Tag + + +class ConfigError(ValueError): + pass + + +def normalize_config(value: dict[str, Any]) -> dict[str, Any]: + if not isinstance(value, dict) or not value: + raise ConfigError("config is required and cannot be empty") + + if "discovery" not in value: + follow_links = value.get("follow_links", False) + config = { + "version": 1, + "discovery": {"type": "rss", "limit": value.get("max_items", 20)}, + "detail": { + "enabled": follow_links, + "always": follow_links, + "root_selector": value.get("content_selector", ""), + "fields": { + "text": { + "selector": value.get("text_selector") or ":root", + "extract": "text", + "required": True, + } + }, + "media": [ + {"type": "photo", "selector": "img", "attributes": ["src", "data-src"]}, + {"type": "video", "selector": "iframe, video, source", "attributes": ["src", "data-src"]}, + ], + }, + "access": {"type": value.get("access", "auto")}, + "retry": {"attempts": 1, "delay_seconds": 0}, + "min_text_length": value.get("min_text_length", 0), + } + else: + config = dict(value) + + if config.get("version", 1) != 1: + raise ConfigError("only config.version=1 is supported") + discovery = config.get("discovery") + if not isinstance(discovery, dict) or discovery.get("type") not in {"rss", "html"}: + raise ConfigError("discovery.type must be rss or html") + try: + limit = int(discovery.get("limit", 20)) + except (TypeError, ValueError) as exc: + raise ConfigError("discovery.limit must be an integer") from exc + if not 1 <= limit <= 100: + raise ConfigError("discovery.limit must be between 1 and 100") + discovery["limit"] = limit + if discovery["type"] == "html" and not str(discovery.get("item_selector") or "").strip(): + raise ConfigError("discovery.item_selector is required for html discovery") + + access = config.get("access", {"type": "auto"}) + if isinstance(access, str): + access = {"type": access} + if not isinstance(access, dict) or access.get("type", "auto") not in {"auto", "http", "browser", "cloudflare"}: + raise ConfigError("access.type must be auto, http, browser or cloudflare") + config["access"] = access + + detail = config.get("detail") or {"enabled": False} + if not isinstance(detail, dict): + raise ConfigError("detail must be an object") + detail["enabled"] = bool(detail.get("enabled", False)) + detail["always"] = bool(detail.get("always", detail["enabled"])) + if detail["enabled"]: + if not str(detail.get("root_selector") or "").strip(): + raise ConfigError("detail.root_selector is required when detail is enabled") + if not isinstance(detail.get("fields") or {}, dict): + raise ConfigError("detail.fields must be an object") + if not isinstance(detail.get("media") or [], list): + raise ConfigError("detail.media must be an array") + if detail.get("transport", "browser") not in {"http", "fetch", "browser"}: + raise ConfigError("detail.transport must be http, fetch or browser") + detail["transport"] = detail.get("transport", "browser") + config["detail"] = detail + + retry = config.get("retry") or {} + try: + attempts = int(retry.get("attempts", 2)) + delay = float(retry.get("delay_seconds", 2)) + timeout = float(retry.get("timeout_seconds", 90)) + except (TypeError, ValueError) as exc: + raise ConfigError("retry values must be numeric") from exc + if not 1 <= attempts <= 5 or not 0 <= delay <= 60 or not 10 <= timeout <= 300: + raise ConfigError("retry attempts must be 1..5, delay_seconds 0..60 and timeout_seconds 10..300") + config["retry"] = {"attempts": attempts, "delay_seconds": delay, "timeout_seconds": timeout} + + try: + min_length = int(config.get("min_text_length", 0)) + except (TypeError, ValueError) as exc: + raise ConfigError("min_text_length must be an integer") from exc + if not 0 <= min_length <= 100_000: + raise ConfigError("min_text_length must be between 0 and 100000") + config["min_text_length"] = min_length + return config + + +def nested_value(value: Any, path: str) -> Any: + current = value + for part in path.split("."): + if isinstance(current, dict): + current = current.get(part) + else: + return None + return current + + +def json_path_values(value: Any, path: str) -> list[Any]: + parts = path.split(".") + found: list[Any] = [] + + def visit(node: Any, index: int) -> None: + if index == len(parts): + found.append(node) + return + if isinstance(node, list): + for entry in node: + visit(entry, index) + elif isinstance(node, dict): + if parts[index] in node: + visit(node[parts[index]], index + 1) + for entry in node.values(): + if isinstance(entry, (dict, list)): + visit(entry, index) + + visit(value, 0) + return found + + +def candidate_elements(root: Tag | BeautifulSoup, candidate: dict[str, Any]) -> list[Tag]: + selectors = candidate.get("selectors") or candidate.get("selector") or [] + if isinstance(selectors, str): + selectors = [selectors] + elements: list[Tag] = [] + for selector in selectors: + if selector == ":root": + elements.append(root) + elif str(selector).strip(): + elements.extend(root.select(str(selector))) + return elements + + +def element_value(element: Tag, candidate: dict[str, Any]) -> Any: + mode = str(candidate.get("extract") or "text") + if mode == "text": + return element.get_text("\n", strip=True) + if mode == "html": + return element.decode_contents() + if mode == "attr": + attributes = candidate.get("attributes") or candidate.get("attribute") or [] + if isinstance(attributes, str): + attributes = [attributes] + for attribute in attributes: + value = element.get(str(attribute)) + if value: + return value + return None + if mode == "json": + try: + payload = json.loads(element.string or element.get_text("", strip=True)) + except (TypeError, json.JSONDecodeError): + return None + values = json_path_values(payload, str(candidate.get("path") or "")) + return next((entry for entry in values if entry is not None and entry != ""), None) + raise ConfigError(f"unsupported extract mode: {mode}") + + +def parse_date(value: Any, candidate: dict[str, Any]) -> str | None: + if not value: + return None + raw = str(value).strip() + try: + parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) + except ValueError: + try: + parsed = parsedate_to_datetime(raw) + except (TypeError, ValueError, OverflowError): + parsed = None + if parsed is None: + formats = candidate.get("formats") or candidate.get("date_format") or [] + if isinstance(formats, str): + formats = [formats] + for date_format in formats: + try: + parsed = datetime.strptime(raw, str(date_format)) + break + except ValueError: + continue + if parsed is None: + return None + if parsed.tzinfo is None: + try: + parsed = parsed.replace(tzinfo=ZoneInfo(str(candidate.get("timezone") or "UTC"))) + except Exception: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc).isoformat() + + +def apply_regex(value: Any, candidate: dict[str, Any]) -> Any: + pattern = candidate.get("regex") + if not pattern or value is None: + return value + match = re.search(str(pattern), str(value), flags=re.DOTALL) + if not match: + return None + group = candidate.get("group", 1 if match.lastindex else 0) + try: + return match.group(group) + except (IndexError, KeyError): + return None + + +def extract_field( + root: Tag | BeautifulSoup | None, + rule: dict[str, Any], + source: dict[str, Any] | None = None, + *, + field_name: str = "", +) -> Any: + candidates = rule.get("candidates") or [rule] + for candidate in candidates: + if not isinstance(candidate, dict): + continue + values: list[Any] = [] + if candidate.get("source"): + values.append(nested_value(source or {}, str(candidate["source"]))) + elif root is not None: + values.extend(element_value(element, candidate) for element in candidate_elements(root, candidate)) + for value in values: + value = apply_regex(value, candidate) + if value is None or (isinstance(value, str) and not value.strip()): + continue + if field_name == "published_at" or candidate.get("type") == "date": + value = parse_date(value, candidate) + if not value: + continue + if isinstance(value, str): + value = value.strip() + if len(str(value)) < int(rule.get("min_length", 0)): + continue + return value + return None + + +def extract_fields( + root: Tag | BeautifulSoup | None, + fields: dict[str, Any], + source: dict[str, Any] | None = None, +) -> tuple[dict[str, Any], list[dict[str, str]]]: + values: dict[str, Any] = {} + errors: list[dict[str, str]] = [] + for name, rule in fields.items(): + if not isinstance(rule, dict): + errors.append({"field": name, "error": "field rule must be an object"}) + continue + try: + value = extract_field(root, rule, source, field_name=name) + except Exception as exc: + errors.append({"field": name, "error": str(exc)}) + continue + if value is None and rule.get("required"): + errors.append({"field": name, "error": "required field not found"}) + elif value is not None: + values[name] = value + return values, errors + + +def youtube_url(value: str, base_url: str = "") -> str | None: + url = urljoin(base_url, str(value or "").strip()) + parsed = urlparse(url) + host = (parsed.hostname or "").lower().removeprefix("www.").removeprefix("m.") + video_id = "" + if host == "youtu.be": + video_id = parsed.path.strip("/").split("/", 1)[0] + elif host in {"youtube.com", "youtube-nocookie.com"}: + if parsed.path == "/watch": + video_id = (parse_qs(parsed.query).get("v") or [""])[0] + elif parsed.path.startswith(("/embed/", "/shorts/", "/live/")): + video_id = parsed.path.strip("/").split("/", 1)[1] + if not re.fullmatch(r"[A-Za-z0-9_-]{6,20}", video_id): + return None + return f"https://www.youtube.com/watch?v={video_id}" + + +def srcset_urls(value: str) -> list[str]: + return [entry.strip().split()[0] for entry in value.split(",") if entry.strip()] + + +def extract_media(root: Tag | BeautifulSoup, specs: list[dict[str, Any]], base_url: str) -> list[dict[str, str]]: + media: list[dict[str, str]] = [] + for spec in specs: + if not isinstance(spec, dict): + continue + media_type = str(spec.get("type") or "photo") + attributes = spec.get("attributes") or spec.get("attribute") or ["src"] + if isinstance(attributes, str): + attributes = [attributes] + for element in candidate_elements(root, spec): + raw_values: list[str] = [] + for attribute in attributes: + raw = element.get(str(attribute)) + if not raw: + continue + raw_values.extend(srcset_urls(str(raw)) if attribute == "srcset" else [str(raw)]) + break + for raw in raw_values: + url = urljoin(base_url, raw.strip()) + if not url.startswith(("http://", "https://")): + continue + item_type = media_type + provider = "" + if media_type == "video": + normalized = youtube_url(url) + if normalized: + url, provider = normalized, "youtube" + else: + host = (urlparse(url).hostname or "").lower() + provider = "twitch" if "twitch.tv" in host else host + if element.name in {"iframe", "a"} or provider == "twitch": + item_type = "external_video" + item = {"type": item_type, "url": url} + if provider: + item["provider"] = provider + media.append(item) + return list({(item["type"], item["url"]): item for item in media}.values()) + + +def detail_from_html( + html: str, + url: str, + config: dict[str, Any], + discovery_values: dict[str, Any] | None = None, +) -> tuple[dict[str, Any], list[dict[str, str]]]: + soup = BeautifulSoup(html, "html.parser") + detail = config["detail"] + root = soup.select_one(str(detail["root_selector"])) + if root is None: + return {}, [{"field": "detail", "error": f"root selector not found: {detail['root_selector']}"}] + for selector in detail.get("remove_selectors") or []: + for element in root.select(str(selector)): + element.decompose() + values, errors = extract_fields(root, detail.get("fields") or {}, discovery_values) + values["media"] = extract_media(root, detail.get("media") or [], url) + values["html"] = str(root) + return values, errors diff --git a/site_parser_worker/requirements.txt b/site_parser_worker/requirements.txt index 7438ef2..01893e1 100644 --- a/site_parser_worker/requirements.txt +++ b/site_parser_worker/requirements.txt @@ -1,5 +1,6 @@ 2captcha-python-async==1.5.1 beautifulsoup4==4.15.0 +curl_cffi==0.16.0 fastapi==0.141.1 feedparser==6.0.14 httpx==0.28.1 diff --git a/site_parser_worker/test_extractor.py b/site_parser_worker/test_extractor.py new file mode 100644 index 0000000..2edd076 --- /dev/null +++ b/site_parser_worker/test_extractor.py @@ -0,0 +1,171 @@ +from __future__ import annotations + +import unittest + +from bs4 import BeautifulSoup + +from extractor import detail_from_html, extract_fields, normalize_config + + +POPULARAIRSOFT_CONFIG = { + "version": 1, + "discovery": { + "type": "html", + "item_selector": "#block-views-block-latest-news-list-block-2 .feature-contents, #block-views-block-latest-news-list-block-1 .lt-teasure", + "limit": 10, + "fields": { + "url": {"selector": "a.link-title", "extract": "attr", "attribute": "href", "required": True}, + "external_id": {"selector": "a.link-title", "extract": "attr", "attribute": "href", "required": True}, + "title": {"selector": "a.link-title", "extract": "text", "required": True}, + "published_at": { + "selector": "time[datetime]", + "extract": "attr", + "attribute": "datetime", + "required": True, + }, + }, + "media": [ + {"type": "photo", "selector": ".site-image img", "attributes": ["src", "data-src", "srcset"]} + ], + }, + "detail": { + "enabled": True, + "always": True, + "transport": "http", + "root_selector": "article.news.full", + "fields": { + "title": { + "selector": ".feature-contents > .news-story-texts:first-child h2", + "extract": "text", + "required": True, + }, + "published_at": { + "candidates": [ + { + "selector": ".feature-contents > .news-story-texts:first-child .news-story-date", + "extract": "text", + "formats": ["%d %b %Y"], + "timezone": "UTC", + }, + {"source": "list.published_at"}, + ], + "required": True, + }, + "text": { + "selector": ".feature-contents > .news-story-texts:last-child .field--name-body", + "extract": "text", + "required": True, + "min_length": 50, + }, + }, + "media": [ + { + "type": "photo", + "selector": ".feature-contents > .site-image .field--name-field-image img", + "attributes": ["src", "data-src", "srcset"], + }, + { + "type": "video", + "selector": ".field--name-body iframe, .field--name-body video, .field--name-body source", + "attributes": ["src", "data-src"], + }, + ], + }, + "access": {"type": "cloudflare", "wait_for": "#block-views-block-latest-news-list-block-1"}, + "retry": {"attempts": 3, "delay_seconds": 5, "timeout_seconds": 90}, + "min_text_length": 50, +} + + +PAGE = """ + + +
+
+
+

Double Bell M16A2

+

OptimusPrime

+

10 Aug 2026

+
+
+
+
+
+
+

This is the complete article body with enough useful text to pass validation safely.

+ +
+
+
+
+
+ +""" + +LIST = """ +
+
+ VFC Vityaz + +
+
+
+
+ Double Bell + +
+
+""" + + +class ExtractorTests(unittest.TestCase): + def test_popularairsoft_discovery_selects_news_links(self) -> None: + config = normalize_config(POPULARAIRSOFT_CONFIG) + soup = BeautifulSoup(LIST, "html.parser") + cards = soup.select(config["discovery"]["item_selector"]) + values = [extract_fields(card, config["discovery"]["fields"])[0] for card in cards] + self.assertEqual([item["url"] for item in values], ["/news/vfc-vityaz", "/news/double-bell"]) + self.assertEqual(values[1]["published_at"], "2026-08-10T06:05:49+00:00") + + def test_popularairsoft_extracts_only_article_fields(self) -> None: + config = normalize_config(POPULARAIRSOFT_CONFIG) + item, errors = detail_from_html( + PAGE, + "https://popularairsoft.com/news/example", + config, + {"list": {"published_at": "2026-08-10T06:05:49+00:00"}}, + ) + self.assertEqual(errors, []) + self.assertEqual(item["title"], "Double Bell M16A2") + self.assertEqual(item["published_at"], "2026-08-10T00:00:00+00:00") + self.assertNotIn("garbage", item["text"]) + self.assertEqual(item["media"][0]["url"], "https://popularairsoft.com/cover.jpg") + self.assertEqual(item["media"][1], { + "type": "video", + "url": "https://www.youtube.com/watch?v=t6mvlySpXNk", + "provider": "youtube", + }) + + def test_date_falls_back_to_rss(self) -> None: + fields = { + "published_at": { + "candidates": [ + {"selector": "time", "extract": "attr", "attribute": "datetime"}, + {"source": "rss.published"}, + ], + "required": True, + } + } + values, errors = extract_fields(None, fields, {"rss": {"published": "Mon, 10 Aug 2026 06:05:49 +0000"}}) + self.assertEqual(errors, []) + self.assertEqual(values["published_at"], "2026-08-10T06:05:49+00:00") + + def test_missing_required_field_is_an_item_error(self) -> None: + config = normalize_config(POPULARAIRSOFT_CONFIG) + item, errors = detail_from_html("
", "https://example.test/1", config) + self.assertEqual(item["media"], []) + self.assertTrue(any(error["field"] == "text" for error in errors)) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/vk_parser_app/source_adapters.py b/src/vk_parser_app/source_adapters.py index 705e505..2d77eaa 100644 --- a/src/vk_parser_app/source_adapters.py +++ b/src/vk_parser_app/source_adapters.py @@ -49,6 +49,50 @@ def validate_source_config(platform: str, config: dict[str, Any]) -> None: raise ValueError("Неподдерживаемая площадка") if not config: raise ValueError("Для сайта нужен конфиг JSON") + if "discovery" in config: + if config.get("version", 1) != 1: + raise ValueError("Поддерживается только version=1") + discovery = config.get("discovery") + if not isinstance(discovery, dict) or discovery.get("type") not in {"rss", "html"}: + raise ValueError('discovery.type должен быть "rss" или "html"') + try: + limit = int(discovery.get("limit", 20)) + except (TypeError, ValueError) as exc: + raise ValueError("discovery.limit должен быть целым числом") from exc + if not 1 <= limit <= 100: + raise ValueError("discovery.limit должен быть от 1 до 100") + if discovery.get("type") == "html" and not str(discovery.get("item_selector") or "").strip(): + raise ValueError("Для HTML discovery нужен item_selector") + detail = config.get("detail") or {"enabled": False} + if not isinstance(detail, dict): + raise ValueError("detail должен быть JSON-объектом") + if detail.get("enabled"): + if not str(detail.get("root_selector") or "").strip(): + raise ValueError("Для detail нужен root_selector") + if not isinstance(detail.get("fields") or {}, dict): + raise ValueError("detail.fields должен быть JSON-объектом") + if not isinstance(detail.get("media") or [], list): + raise ValueError("detail.media должен быть массивом") + if detail.get("transport", "browser") not in {"http", "fetch", "browser"}: + raise ValueError('detail.transport должен быть "http", "fetch" или "browser"') + access = config.get("access") or {"type": "auto"} + if isinstance(access, str): + access = {"type": access} + if not isinstance(access, dict) or access.get("type", "auto") not in {"auto", "http", "browser", "cloudflare"}: + raise ValueError('access.type должен быть "auto", "http", "browser" или "cloudflare"') + retry = config.get("retry") or {} + try: + attempts = int(retry.get("attempts", 2)) + delay = float(retry.get("delay_seconds", 2)) + timeout = float(retry.get("timeout_seconds", 90)) + min_text_length = int(config.get("min_text_length", 0)) + except (TypeError, ValueError) as exc: + raise ValueError("retry и min_text_length должны быть числовыми") from exc + if not 1 <= attempts <= 5 or not 0 <= delay <= 60 or not 10 <= timeout <= 300: + raise ValueError("retry: attempts 1..5, delay_seconds 0..60, timeout_seconds 10..300") + if not 0 <= min_text_length <= 100_000: + raise ValueError("min_text_length должен быть от 0 до 100000") + return if config.get("format") != "rss": raise ValueError('Сейчас поддерживается только "format": "rss"') if str(config.get("access") or "auto") not in {"auto", "http", "cloudflare"}: @@ -97,7 +141,7 @@ class SiteParserClient: self.rucaptcha_token = rucaptcha_token self.timeout = aiohttp.ClientTimeout(total=max(10, timeout_sec)) - async def fetch(self, source: dict) -> tuple[list[SourceItem], dict[str, Any] | None]: + async def fetch(self, source: dict) -> tuple[list[SourceItem], dict[str, Any] | None, str | None]: if not self.base_url or not self.token: raise RuntimeError("Site Parser URL или токен не настроены") config = json_object(source.get("settings_json")) @@ -125,6 +169,8 @@ class SiteParserClient: except aiohttp.ClientError as exc: raise RuntimeError(f"Site Parser недоступен: {exc}") from exc + result_status = str(data.get("status") or "ok") + result_errors = data.get("errors") or [] items = [] for raw in data.get("items") or []: title = str(raw.get("title") or "").strip() @@ -145,11 +191,19 @@ class SiteParserClient: ] items.append(SourceItem(external_id, url, text, _posted_at(raw.get("published_at")), media, raw)) state = data.get("browser_state") - return items, ( + runtime = ( { "browser_state": state, "browser_user_agent": str(data.get("browser_user_agent") or "") or None, + "last_result_status": result_status, + "last_result_errors": result_errors[:20], } if isinstance(state, dict) else None ) + warning = ( + f"Site Parser {result_status}: {len(result_errors)} ошибок; {str(result_errors)[:700]}" + if result_status in {"partial", "failed"} + else None + ) + return items, runtime, warning diff --git a/src/vk_parser_app/templates/source_form.html b/src/vk_parser_app/templates/source_form.html index 154911e..131b2e5 100644 --- a/src/vk_parser_app/templates/source_form.html +++ b/src/vk_parser_app/templates/source_form.html @@ -43,16 +43,31 @@
- +
Пример и параметры 
{
-  "format": "rss",
-  "access": "auto",
-  "max_items": 20
+  "version": 1,
+  "discovery": {
+    "type": "rss",
+    "limit": 20
+  },
+  "detail": {
+    "enabled": true,
+    "always": true,
+    "root_selector": "article",
+    "fields": {
+      "title": {"selector": "h1", "extract": "text", "required": true},
+      "text": {"selector": ".article-body", "extract": "text", "required": true}
+    }
+  },
+  "access": {"type": "auto"},
+  "retry": {"attempts": 2, "delay_seconds": 2, "timeout_seconds": 90},
+  "min_text_length": 50
 }
-

access: auto сначала пробует обычный запрос и при Cloudflare использует RuCaptcha; http запрещает браузер; cloudflare сразу запускает браузер.

-

Если RSS не содержит текст или медиа, добавьте "follow_links": true, селектор материала "content_selector": "article.full" и при необходимости отдельный селектор текста "text_selector": ".field--name-body".

+

discovery.type: rss читает ссылки из ленты; html собирает карточки по item_selector. Секция detail описывает точные селекторы полей на странице материала.

+

access.type: auto сначала пробует обычный запрос; http запрещает браузер; browser выполняет страницу в браузере; cloudflare разрешает RuCaptcha.

+

Для fallback используйте candidates. Извлечение поддерживает text, html, attr и json. Ошибка одной detail-страницы не прерывает готовую пачку.

min_text_length переопределяет общий порог Site Parser только для этого источника.

@@ -78,7 +93,7 @@ const syncConfig = () => { siteConfig.hidden = platform.value !== 'site'; if (platform.value === 'site' && configInput.value.trim() === '{}') { - configInput.value = '{\n "format": "rss",\n "access": "auto",\n "max_items": 20\n}'; + configInput.value = '{\n "version": 1,\n "discovery": {\n "type": "rss",\n "limit": 20\n },\n "detail": {\n "enabled": false\n },\n "access": {\n "type": "auto"\n },\n "retry": {\n "attempts": 2,\n "delay_seconds": 2,\n "timeout_seconds": 90\n },\n "min_text_length": 50\n}'; } }; platform.addEventListener('change', syncConfig); diff --git a/src/vk_parser_app/workers/parser.py b/src/vk_parser_app/workers/parser.py index b07c3eb..5982157 100644 --- a/src/vk_parser_app/workers/parser.py +++ b/src/vk_parser_app/workers/parser.py @@ -242,7 +242,7 @@ class VKParserWorker: if since_dt.tzinfo is None: since_dt = since_dt.replace(tzinfo=timezone.utc) - fetched, runtime_state = await client.fetch(source) + fetched, runtime_state, partial_warning = await client.fetch(source) recent = [item for item in fetched if item.posted_at > since_dt] known = await self.known_post_ids(source_id, [item.external_id for item in recent]) candidates = [item for item in recent if item.external_id not in known] @@ -281,7 +281,23 @@ class VKParserWorker: saved += 1 known_hashes.add(content_hash) max_seen = max((item.posted_at for item in fetched), default=last_parsed_at) - await self.mark_source_ok(source_id, max_seen, runtime_state) + if partial_warning: + await self.pool.execute( + """ + UPDATE sources + SET status=$2, status_msg=$3, last_checked_at=NOW(), + last_parsed_at=COALESCE($4, last_parsed_at), + runtime_state_json=COALESCE($5::jsonb, runtime_state_json), updated_at=NOW() + WHERE id=$1 + """, + source_id, + SOURCE_STATUS_ERROR, + partial_warning[:1000], + max_seen, + json.dumps(runtime_state, ensure_ascii=False) if runtime_state is not None else None, + ) + else: + await self.mark_source_ok(source_id, max_seen, runtime_state) logger.info( "Parsed site source {}: fetched={} recent={} known={} saved={}", source.get("name"), len(fetched), len(recent), len(known), saved, diff --git a/tests/test_source_adapters.py b/tests/test_source_adapters.py index 5de33a2..e0370ff 100644 --- a/tests/test_source_adapters.py +++ b/tests/test_source_adapters.py @@ -61,7 +61,7 @@ class EmptyBodySession(FakeSession): class SourceAdapterTests(unittest.IsolatedAsyncioTestCase): async def test_worker_response_is_normalized(self) -> None: client = SiteParserClient(FakeSession(), "http://worker", "token", "captcha", 30) - items, state = await client.fetch({ + items, state, warning = await client.fetch({ "url": "https://example.test/rss.xml", "settings_json": '{"format":"rss","access":"auto"}', "runtime_state_json": '{}', @@ -71,10 +71,11 @@ class SourceAdapterTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(items[0].media[0].url, "https://example.test/1.jpg") self.assertEqual(state["browser_state"]["cookies"][0]["name"], "cf_clearance") self.assertEqual(state["browser_user_agent"], "Test Browser") + self.assertIsNone(warning) async def test_followed_page_does_not_repeat_title(self) -> None: client = SiteParserClient(FollowedPageSession(), "http://worker", "token", "captcha", 30) - items, _ = await client.fetch({ + items, _, _ = await client.fetch({ "url": "https://example.test/rss.xml", "settings_json": '{"format":"rss","follow_links":true,"content_selector":"article.full"}', "runtime_state_json": "{}", @@ -84,7 +85,7 @@ class SourceAdapterTests(unittest.IsolatedAsyncioTestCase): async def test_title_is_not_counted_as_site_body(self) -> None: client = SiteParserClient(EmptyBodySession(), "http://worker", "token", "captcha", 30) - items, _ = await client.fetch({ + items, _, _ = await client.fetch({ "url": "https://example.test/rss.xml", "settings_json": '{"format":"rss"}', "runtime_state_json": "{}", @@ -104,6 +105,27 @@ class SourceAdapterTests(unittest.IsolatedAsyncioTestCase): with self.assertRaisesRegex(ValueError, "min_text_length"): validate_source_config("site", {"format": "rss", "min_text_length": "many"}) + def test_versioned_site_config_is_validated(self) -> None: + validate_source_config("site", { + "version": 1, + "discovery": {"type": "rss", "limit": 10}, + "detail": { + "enabled": True, + "root_selector": "article", + "fields": {"text": {"selector": ".body", "extract": "text"}}, + }, + "access": {"type": "cloudflare"}, + "retry": {"attempts": 3, "delay_seconds": 5, "timeout_seconds": 90}, + "min_text_length": 50, + }) + + def test_html_discovery_requires_item_selector(self) -> None: + with self.assertRaisesRegex(ValueError, "item_selector"): + validate_source_config("site", { + "version": 1, + "discovery": {"type": "html"}, + }) + if __name__ == "__main__": unittest.main()