diff --git a/README.md b/README.md index 874cb65..b4235c6 100644 --- a/README.md +++ b/README.md @@ -108,6 +108,31 @@ https://vk.com/wall-239548476_123 - админ/редактор группы может открыть и отредактировать пост; - вложенное фото видно в браузере. +## Site Parser + +Сайты обрабатывает отдельный stateless-воркер из `site_parser_worker`. Основная +коробка хранит источники, расписание, RuCaptcha-ключ и браузерное состояние, а +воркер возвращает нормализованные материалы в существующий пайплайн. + +Сейчас поддерживается RSS/Atom, включая Cloudflare: + +```json +{ + "format": "rss", + "access": "auto", + "max_items": 20 +} +``` + +В глобальном разделе `Site Parser` задаются URL воркера, его токен, ключ +RuCaptcha и таймаут. Внешний модуль запускается командой: + +```bash +docker build -t site-parser-worker site_parser_worker +docker run -d --restart unless-stopped -p 8080:8080 \ + -e WORKER_TOKEN=replace-me site-parser-worker +``` + ## Запуск локально Админка: diff --git a/db/migrations/046_site_parser.sql b/db/migrations/046_site_parser.sql new file mode 100644 index 0000000..72e1511 --- /dev/null +++ b/db/migrations/046_site_parser.sql @@ -0,0 +1,10 @@ +ALTER TABLE sources +ADD COLUMN IF NOT EXISTS runtime_state_json JSONB NOT NULL DEFAULT '{}'::jsonb; + +INSERT INTO app_settings(key, value_json, value_type, title, description, category) +VALUES + ('site_parser_url', '"http://192.168.1.113:8080"'::jsonb, 'str', 'URL воркера', 'Адрес внешнего Site Parser.', 'Site Parser'), + ('site_parser_token', '""'::jsonb, 'secret', 'Токен воркера', 'Должен совпадать с WORKER_TOKEN внешнего модуля.', 'Site Parser'), + ('site_parser_rucaptcha_token', '""'::jsonb, 'secret', 'Ключ RuCaptcha', 'Используется воркером только при встрече Cloudflare.', 'Site Parser'), + ('site_parser_timeout_sec', '180'::jsonb, 'int', 'Таймаут, сек', 'Максимальное время одного запуска источника.', 'Site Parser') +ON CONFLICT (key) DO NOTHING; diff --git a/site_parser_worker/Dockerfile b/site_parser_worker/Dockerfile new file mode 100644 index 0000000..074445f --- /dev/null +++ b/site_parser_worker/Dockerfile @@ -0,0 +1,11 @@ +FROM mcr.microsoft.com/playwright/python:v1.62.0-noble + +WORKDIR /app +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt +RUN playwright install chrome +COPY app.py . + +ENV PYTHONUNBUFFERED=1 +EXPOSE 8080 +CMD ["bash", "-lc", "Xvfb :99 -screen 0 1280x1024x24 -nolisten tcp & export DISPLAY=:99; sleep 1; exec uvicorn app:app --host 0.0.0.0 --port 8080"] diff --git a/site_parser_worker/app.py b/site_parser_worker/app.py new file mode 100644 index 0000000..3986f2b --- /dev/null +++ b/site_parser_worker/app.py @@ -0,0 +1,192 @@ +from __future__ import annotations + +import asyncio +import os +import secrets +from datetime import datetime, timezone +from time import struct_time +from typing import Annotated, Any + +import feedparser +import httpx +from bs4 import BeautifulSoup +from fastapi import FastAPI, Header, HTTPException +from playwright.async_api import 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") +browser_lock = asyncio.Lock() + + +class ParseRequest(BaseModel): + url: HttpUrl + config: dict[str, Any] + rucaptcha_token: SecretStr | None = None + browser_state: dict[str, Any] | None = None + + +class ParsedItem(BaseModel): + external_id: str + url: str + title: str + text: str + html: str + published_at: str | None + author: str | None + media: list[dict[str, str]] = Field(default_factory=list) + + +def require_token(worker_token: str | None) -> None: + expected = os.getenv("WORKER_TOKEN", "") + if not expected: + raise HTTPException(status_code=503, detail="WORKER_TOKEN is not configured") + if not worker_token or not secrets.compare_digest(worker_token, expected): + raise HTTPException(status_code=401, detail="Invalid worker token") + + +def is_cloudflare_challenge(status: int, body: str) -> bool: + sample = body[:20_000].lower() + return status in {403, 429, 503} and ( + "just a moment" in sample or "cf-chl" in sample or "challenge-platform" in sample + ) + + +def iso_date(value: struct_time | None) -> str | None: + if not value: + return None + return datetime(*value[:6], tzinfo=timezone.utc).isoformat() + + +def clean_html(value: Any) -> tuple[str, str, list[dict[str, str]]]: + html = str(value or "").strip() + soup = BeautifulSoup(html, "html.parser") + text = soup.get_text("\n", strip=True) + media = [{"type": "photo", "url": str(image["src"])} for image in soup.find_all("img", src=True)] + return html, text, media + + +def parse_rss(xml: str, max_items: int) -> list[ParsedItem]: + feed = feedparser.parse(xml) + 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) + html, text, media = clean_html(entry.get("description") or entry.get("summary") or "") + for enclosure in entry.get("enclosures") or []: + url = str(enclosure.get("href") or enclosure.get("url") or "").strip() + media_type = str(enclosure.get("type") or "") + if url and (not media_type or media_type.startswith("image/")): + media.append({"type": "photo", "url": url}) + url = str(entry.get("link") or "").strip() + 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()), + ) + ) + return items + + +async def fetch_in_browser( + url: str, + rucaptcha_token: str, + browser_state: dict[str, Any] | None, +) -> tuple[str, dict[str, Any]]: + # ponytail: one browser at a time; use a queue only when parallel source parsing is needed. + 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) if browser_state else await browser.new_context() + 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: + response = await page.goto(url, wait_until="domcontentloaded", timeout=60_000) + if response and response.status in {403, 429, 503} and "Just a moment" in await page.title(): + 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() + state = await context.storage_state() + await browser.close() + return xml, state + + +@app.get("/health") +async def health() -> dict[str, bool]: + return {"ok": True} + + +@app.post("/v1/parse") +async def parse_source( + request: ParseRequest, + 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") + url = str(request.url) + xml = "" + state = request.browser_state + 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 = await fetch_in_browser(url, request.rucaptcha_token.get_secret_value(), state) + except Exception as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + try: + items = parse_rss(xml, max_items) + except ValueError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + return { + "source_url": url, + "fetched_via": fetched_via, + "items": [item.model_dump() for item in items], + "browser_state": state, + } diff --git a/site_parser_worker/requirements.txt b/site_parser_worker/requirements.txt new file mode 100644 index 0000000..7438ef2 --- /dev/null +++ b/site_parser_worker/requirements.txt @@ -0,0 +1,8 @@ +2captcha-python-async==1.5.1 +beautifulsoup4==4.15.0 +fastapi==0.141.1 +feedparser==6.0.14 +httpx==0.28.1 +playwright==1.62.0 +playwright-captcha==0.1.5 +uvicorn==0.52.1 diff --git a/site_parser_worker/test_parser.py b/site_parser_worker/test_parser.py new file mode 100644 index 0000000..46d6143 --- /dev/null +++ b/site_parser_worker/test_parser.py @@ -0,0 +1,16 @@ +from app import parse_rss + + +def test_parse_rss() -> None: + xml = """1Title + https://example.test/1 + <p>Body</p><img src="https://example.test/1.jpg"> + Mon, 10 Aug 2026 10:00:00 +0000""" + item = parse_rss(xml, 20)[0] + assert item.external_id == "1" + assert item.text == "Body" + assert item.media[0]["url"] == "https://example.test/1.jpg" + + +if __name__ == "__main__": + test_parse_rss() diff --git a/src/vk_parser_app/admin.py b/src/vk_parser_app/admin.py index b0163f7..83c2be5 100644 --- a/src/vk_parser_app/admin.py +++ b/src/vk_parser_app/admin.py @@ -15,16 +15,17 @@ from urllib.parse import urlencode from zoneinfo import ZoneInfo import aiohttp -from fastapi import FastAPI, File, Form, Request, UploadFile, status +from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile, status from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from loguru import logger from .config import settings -from .constants import PLATFORM_VK +from .constants import PLATFORM_SITE, PLATFORM_VK from .db import fetch_int_setting, fetch_setting, get_pool from .security import hash_password, new_token, token_hash, verify_password +from .source_adapters import validate_source_config from .text_utils import build_publication_text, normalize_hash_tag, parse_categories from .vk_api import VKAPIClient, normalize_vk_source from .workers.ai_qualifier import AIQualifierWorker, normalize_model, response_usage @@ -196,6 +197,7 @@ CATEGORY_TITLES = { "Publishing": "Публикации", "Daily Report": "Ежедневный отчет", "Parser": "Парсер", + "Site Parser": "Site Parser", "Uploader": "Аплоадер", "VK": "VK API", "General": "Общие", @@ -212,6 +214,7 @@ CATEGORY_ORDER = { "Publishing": 53, "Daily Report": 55, "Parser": 60, + "Site Parser": 65, "VK": 70, "Uploader": 80, "General": 100, @@ -2361,12 +2364,22 @@ async def source_create( url: str = Form(...), active: str = Form("off"), priority: int = Form(100), + settings_json: str = Form("{}"), ): user = await get_current_user(request) if not user: return redirect("/login") require_csrf(user, csrf_token) platform = platform.strip().lower() or PLATFORM_VK + if platform not in {PLATFORM_VK, PLATFORM_SITE}: + raise HTTPException(status_code=422, detail="Неподдерживаемая площадка") + try: + source_settings = json.loads(settings_json or "{}") + if not isinstance(source_settings, dict): + raise ValueError("Настройки должны быть JSON-объектом") + validate_source_config(platform, source_settings) + except (json.JSONDecodeError, ValueError) as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc external_id = normalize_vk_source(url) if platform == PLATFORM_VK else "" external_owner_id = None resolved_name = name.strip() @@ -2389,8 +2402,8 @@ async def source_create( pool = await get_pool() row = await pool.fetchrow( """ - INSERT INTO sources(platform, name, tag, url, external_id, external_owner_id, active, priority, status, status_msg, created_by) - VALUES($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) + INSERT INTO sources(platform, name, tag, url, external_id, external_owner_id, active, priority, status, status_msg, settings_json, created_by) + VALUES($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11::jsonb, $12) ON CONFLICT DO NOTHING RETURNING id """, @@ -2404,6 +2417,7 @@ async def source_create( priority, status_value, status_msg, + json.dumps(source_settings, ensure_ascii=False), user["id"], ) if row: @@ -2443,12 +2457,22 @@ async def source_update( url: str = Form(...), active: str = Form("off"), priority: int = Form(100), + settings_json: str = Form("{}"), ): user = await get_current_user(request) if not user: return redirect("/login") require_csrf(user, csrf_token) platform = platform.strip().lower() or PLATFORM_VK + if platform not in {PLATFORM_VK, PLATFORM_SITE}: + raise HTTPException(status_code=422, detail="Неподдерживаемая площадка") + try: + source_settings = json.loads(settings_json or "{}") + if not isinstance(source_settings, dict): + raise ValueError("Настройки должны быть JSON-объектом") + validate_source_config(platform, source_settings) + except (json.JSONDecodeError, ValueError) as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc external_id = normalize_vk_source(url) if platform == PLATFORM_VK else "" pool = await get_pool() await pool.execute( @@ -2459,8 +2483,14 @@ async def source_update( tag=$4, url=$5, external_id=$6, + external_owner_id=CASE WHEN platform=$2 AND url=$5 THEN external_owner_id ELSE NULL END, active=$7, priority=$8, + settings_json=$9::jsonb, + runtime_state_json=CASE + WHEN platform=$2 AND url=$5 AND settings_json=$9::jsonb THEN runtime_state_json + ELSE '{}'::jsonb + END, updated_at=NOW() WHERE id=$1 """, @@ -2472,6 +2502,7 @@ async def source_update( external_id, active == "on", priority, + json.dumps(source_settings, ensure_ascii=False), ) await audit(user["id"], "source.update", "source", source_id, {"url": url, "platform": platform}) return redirect("/sources") diff --git a/src/vk_parser_app/constants.py b/src/vk_parser_app/constants.py index 35d313f..a8b99df 100644 --- a/src/vk_parser_app/constants.py +++ b/src/vk_parser_app/constants.py @@ -1,4 +1,5 @@ PLATFORM_VK = "vk" +PLATFORM_SITE = "site" SOURCE_STATUS_NEW = "new" SOURCE_STATUS_OK = "ok" diff --git a/src/vk_parser_app/source_adapters.py b/src/vk_parser_app/source_adapters.py new file mode 100644 index 0000000..87690e6 --- /dev/null +++ b/src/vk_parser_app/source_adapters.py @@ -0,0 +1,115 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any + +import aiohttp + +from .constants import PLATFORM_SITE, PLATFORM_VK + + +@dataclass +class SourceMedia: + url: str + media_type: str = "photo" + + +@dataclass +class SourceItem: + external_id: str + url: str + text: str + posted_at: datetime + media: list[SourceMedia] = field(default_factory=list) + raw: dict[str, Any] = field(default_factory=dict) + + +def validate_source_config(platform: str, config: dict[str, Any]) -> None: + if platform == PLATFORM_VK: + return + if platform != PLATFORM_SITE: + raise ValueError("Неподдерживаемая площадка") + if not config: + raise ValueError("Для сайта нужен конфиг JSON") + if config.get("format") != "rss": + raise ValueError('Сейчас поддерживается только "format": "rss"') + if str(config.get("access") or "auto") not in {"auto", "http", "cloudflare"}: + raise ValueError('access должен быть "auto", "http" или "cloudflare"') + try: + max_items = int(config.get("max_items", 20)) + except (TypeError, ValueError) as exc: + raise ValueError("max_items должен быть целым числом") from exc + if not 1 <= max_items <= 100: + raise ValueError("max_items должен быть от 1 до 100") + + +def _posted_at(value: Any) -> datetime: + try: + parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except (TypeError, ValueError): + return datetime.now(timezone.utc) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +class SiteParserClient: + def __init__( + self, + session: aiohttp.ClientSession, + base_url: str, + token: str, + rucaptcha_token: str, + timeout_sec: int, + ) -> None: + self.session = session + self.base_url = base_url.rstrip("/") + self.token = token + 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]: + if not self.base_url or not self.token: + raise RuntimeError("Site Parser URL или токен не настроены") + config = dict(source.get("settings_json") or {}) + validate_source_config(PLATFORM_SITE, config) + runtime_state = dict(source.get("runtime_state_json") or {}) + payload = { + "url": source["url"], + "config": config, + "rucaptcha_token": self.rucaptcha_token or None, + "browser_state": runtime_state.get("browser_state"), + } + try: + async with self.session.post( + f"{self.base_url}/v1/parse", + json=payload, + headers={"X-Worker-Token": self.token}, + timeout=self.timeout, + ) as response: + data = await response.json(content_type=None) + if response.status >= 400: + raise RuntimeError(f"Site Parser HTTP {response.status}: {data.get('detail', data)}") + except TimeoutError as exc: + raise RuntimeError(f"Site Parser превысил таймаут {int(self.timeout.total)} сек") from exc + except aiohttp.ClientError as exc: + raise RuntimeError(f"Site Parser недоступен: {exc}") from exc + + items = [] + for raw in data.get("items") or []: + title = str(raw.get("title") or "").strip() + body = str(raw.get("text") or "").strip() + text = "\n\n".join(part for part in (title, body) if part) + url = str(raw.get("url") or source["url"]).strip() + external_id = str(raw.get("external_id") or url).strip() + if not external_id: + continue + media = [ + SourceMedia(str(item["url"]), str(item.get("type") or "photo")) + for item in raw.get("media") or [] + if isinstance(item, dict) and item.get("url") + ] + items.append(SourceItem(external_id, url, text, _posted_at(raw.get("published_at")), media, raw)) + state = data.get("browser_state") + return items, ({"browser_state": state} if isinstance(state, dict) else None) diff --git a/src/vk_parser_app/templates/source_form.html b/src/vk_parser_app/templates/source_form.html index 3010cdd..d6bc697 100644 --- a/src/vk_parser_app/templates/source_form.html +++ b/src/vk_parser_app/templates/source_form.html @@ -17,6 +17,7 @@ @@ -39,6 +40,20 @@ + +
+ + +
+ Пример и параметры + 
{
+  "format": "rss",
+  "access": "auto",
+  "max_items": 20
+}
+

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

+
+
@@ -54,4 +69,11 @@
+ {% endblock %} diff --git a/src/vk_parser_app/templates/sources.html b/src/vk_parser_app/templates/sources.html index 578bed5..c991e14 100644 --- a/src/vk_parser_app/templates/sources.html +++ b/src/vk_parser_app/templates/sources.html @@ -5,7 +5,7 @@ Источники -
VK-источники для парсинга. Название идёт в prompt, тэг — в будущие хэштеги.
+
Источники для парсинга. Название идёт в prompt, тэг — в будущие хэштеги.
diff --git a/src/vk_parser_app/workers/ai_alerts.py b/src/vk_parser_app/workers/ai_alerts.py index b687b01..cbd616f 100644 --- a/src/vk_parser_app/workers/ai_alerts.py +++ b/src/vk_parser_app/workers/ai_alerts.py @@ -39,7 +39,7 @@ def parse_recipients(value: Any) -> list[int]: return recipients -async def send_ai_worker_error_alert(worker_name: str, model: str, post_ids: list[int], error: str) -> None: +async def send_system_error_alert(text: str) -> None: token = ( str(await fetch_setting("daily_report_bot_token", "") or "").strip() or str(await fetch_setting("tg_poster_bot_token", "") or "").strip() @@ -47,17 +47,8 @@ async def send_ai_worker_error_alert(worker_name: str, model: str, post_ids: lis ) recipients = parse_recipients(await fetch_setting("daily_report_recipient_ids", [442509142])) if not token or not recipients: - logger.warning("AI worker alert skipped: token or recipients are empty") + logger.warning("System alert skipped: token or recipients are empty") return - - post_part = ", ".join(str(post_id) for post_id in post_ids) if post_ids else "-" - text = ( - "AI worker batch failed\n" - f"worker: {worker_name}\n" - f"model: {model or '-'}\n" - f"posts: {post_part}\n" - f"error: {error[:1000]}" - ) bot = Bot(token=token) try: for recipient_id in recipients: @@ -69,3 +60,14 @@ async def send_ai_worker_error_alert(worker_name: str, model: str, post_ids: lis await asyncio.sleep(float(exc.retry_after) + 1) finally: await bot.session.close() + + +async def send_ai_worker_error_alert(worker_name: str, model: str, post_ids: list[int], error: str) -> None: + post_part = ", ".join(str(post_id) for post_id in post_ids) if post_ids else "-" + await send_system_error_alert( + "AI worker batch failed\n" + f"worker: {worker_name}\n" + f"model: {model or '-'}\n" + f"posts: {post_part}\n" + f"error: {error[:1000]}" + ) diff --git a/src/vk_parser_app/workers/parser.py b/src/vk_parser_app/workers/parser.py index c505c13..3ef9980 100644 --- a/src/vk_parser_app/workers/parser.py +++ b/src/vk_parser_app/workers/parser.py @@ -5,11 +5,13 @@ import hashlib import json from datetime import datetime, timedelta, timezone +import aiohttp from loguru import logger from ..config import settings from ..constants import ( JOB_TYPE_VK_STORAGE_COPY, + PLATFORM_SITE, PLATFORM_VK, POST_STATUS_SKIPPED, POST_STATUS_STORAGE_PENDING, @@ -17,9 +19,10 @@ from ..constants import ( SOURCE_STATUS_OK, WORKER_PARSER, ) -from ..db import fetch_bool_setting, fetch_float_setting, fetch_int_setting, get_pool +from ..db import fetch_bool_setting, fetch_float_setting, fetch_int_setting, fetch_setting, get_pool from ..heartbeat import HeartbeatReporter from ..jobs import is_worker_enabled +from ..source_adapters import SiteParserClient, SourceItem from ..vk_api import ( VKAPIClient, VKAPIError, @@ -29,6 +32,7 @@ from ..vk_api import ( is_repost, post_vk_url, ) +from .ai_alerts import send_system_error_alert def utc_from_ts(value: int) -> datetime: @@ -56,12 +60,12 @@ class VKParserWorker: """ SELECT * FROM sources - WHERE platform=$1 + WHERE platform=ANY($1::text[]) AND active=TRUE AND archived_at IS NULL ORDER BY last_checked_at NULLS FIRST, priority ASC, id ASC """, - PLATFORM_VK, + [PLATFORM_VK, PLATFORM_SITE], ) return [dict(r) for r in rows] @@ -96,7 +100,12 @@ class VKParserWorker: message[:1000], ) - async def mark_source_ok(self, source_id: int, last_parsed_at: datetime | None) -> None: + async def mark_source_ok( + self, + source_id: int, + last_parsed_at: datetime | None, + runtime_state: dict | None = None, + ) -> None: await self.pool.execute( """ UPDATE sources @@ -104,12 +113,14 @@ class VKParserWorker: status_msg=NULL, last_checked_at=NOW(), last_parsed_at=COALESCE($3, last_parsed_at), + runtime_state_json=COALESCE($4::jsonb, runtime_state_json), updated_at=NOW() WHERE id=$1 """, source_id, SOURCE_STATUS_OK, last_parsed_at, + json.dumps(runtime_state, ensure_ascii=False) if runtime_state is not None else None, ) async def resolve_source_if_needed(self, client: VKAPIClient, source: dict) -> dict: @@ -136,6 +147,133 @@ class VKParserWorker: source["name"] = resolved_name return source + async def save_source_item( + self, + source: dict, + item: SourceItem, + *, + status: str, + skip_reason: str | None = None, + create_storage_job: bool = True, + ) -> int | None: + source_id = int(source["id"]) + platform = str(source["platform"]) + media_urls = ",".join(media.url for media in item.media) + text_hash = make_hash(item.text) + content_hash = make_hash(item.text, media_urls) + raw = {**item.raw, "url": item.url, "media": [media.url for media in item.media]} + + async with self.pool.acquire() as conn: + async with conn.transaction(): + raw_post_id = await conn.fetchval( + """ + INSERT INTO raw_posts( + source_id, platform, external_post_id, original_url, + raw_text, raw_json, text_hash, content_hash, + posted_at, status, skip_reason + ) + VALUES($1,$2,$3,$4,$5,$6::jsonb,$7,$8,$9,$10,$11) + ON CONFLICT (source_id, external_post_id) DO NOTHING + RETURNING id + """, + source_id, + platform, + item.external_id, + item.url, + item.text, + json.dumps(raw, ensure_ascii=False), + text_hash, + content_hash, + item.posted_at, + status, + skip_reason, + ) + if raw_post_id is None: + return None + for order, media in enumerate(item.media): + await conn.execute( + """ + INSERT INTO raw_post_media( + raw_post_id, platform, media_type, original_url, sort_order + ) + VALUES($1,$2,$3,$4,$5) + """, + raw_post_id, + platform, + media.media_type, + media.url, + order, + ) + if create_storage_job: + await conn.execute( + """ + INSERT INTO jobs(type, entity_type, entity_id, payload_json, status) + VALUES($1, 'raw_post', $2, '{}'::jsonb, 'pending') + ON CONFLICT DO NOTHING + """, + JOB_TYPE_VK_STORAGE_COPY, + raw_post_id, + ) + return int(raw_post_id) + + async def parse_site_source(self, client: SiteParserClient, source: dict) -> int: + source_id = int(source["id"]) + lookback_days = max(1, await fetch_int_setting("parser_new_source_lookback_days", 14)) + overlap_minutes = max(0, await fetch_int_setting("parser_reparse_overlap_minutes", 120)) + last_parsed_at = source.get("last_parsed_at") + parse_from = source.get("parse_from") + since_dt = ( + last_parsed_at - timedelta(minutes=overlap_minutes) + if last_parsed_at + else parse_from or datetime.now(timezone.utc) - timedelta(days=lookback_days) + ) + if since_dt.tzinfo is None: + since_dt = since_dt.replace(tzinfo=timezone.utc) + + fetched, runtime_state = 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] + min_text_length = max(0, await fetch_int_setting("parser_min_text_length", 0)) + skip_empty_text = await fetch_bool_setting("parser_skip_empty_text", True) + skip_no_media = await fetch_bool_setting("parser_skip_no_media", True) + skip_short_text = await fetch_bool_setting("parser_skip_text_too_short", True) + store_skipped = await fetch_bool_setting("parser_store_skipped_posts", False) + dedupe_content_hash = await fetch_bool_setting("parser_dedupe_content_hash", True) + hashes = {item.external_id: make_hash(item.text, ",".join(m.url for m in item.media)) for item in candidates} + known_hashes = await self.known_content_hashes(list(hashes.values())) if dedupe_content_hash else set() + saved = 0 + for item in candidates: + content_hash = hashes[item.external_id] + if dedupe_content_hash and content_hash in known_hashes: + continue + skip_reason = None + if skip_empty_text and not item.text: + skip_reason = "empty_text" + elif skip_no_media and not item.media: + skip_reason = "no_media" + elif skip_short_text and len(item.text) < min_text_length: + skip_reason = "text_too_short" + if skip_reason and not store_skipped: + continue + raw_id = await self.save_source_item( + source, + item, + status=POST_STATUS_SKIPPED if skip_reason else POST_STATUS_STORAGE_PENDING, + skip_reason=skip_reason, + create_storage_job=not bool(skip_reason), + ) + if raw_id: + 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) + logger.info( + "Parsed site source {}: fetched={} recent={} known={} saved={}", + source.get("name"), len(fetched), len(recent), len(known), saved, + ) + return saved + async def known_post_ids(self, source_id: int, external_post_ids: list[str]) -> set[str]: if not external_post_ids: return set() @@ -385,7 +523,7 @@ class VKParserWorker: sources = await self.active_sources() await self.heartbeat.beat(self.pool, meta={"sources": len(sources)}) if not sources: - logger.info("No active VK sources") + logger.info("No active sources") return logger.info( @@ -401,30 +539,50 @@ class VKParserWorker: source_pause, ) - async with VKAPIClient( - rps=rps, - timeout_total_sec=timeout_total, - timeout_connect_sec=timeout_connect, - rate_limit_sleep_sec=rate_limit_sleep, - retry_attempts=retry_attempts, - retry_min_delay_sec=retry_min_delay, - retry_max_delay_sec=retry_max_delay, - ) as client: - for source in sources: - try: - await self.parse_source(client, source) - except VKAPIError as e: - if is_fatal_source_error(e): - await self.deactivate_source(int(source["id"]), str(e)) - logger.warning("VK source deactivated {}: {}", source.get("name"), e) - else: + async with aiohttp.ClientSession() as web_session: + site_client = SiteParserClient( + web_session, + str(await fetch_setting("site_parser_url", "") or "").strip(), + str(await fetch_setting("site_parser_token", "") or "").strip(), + str(await fetch_setting("site_parser_rucaptcha_token", "") or "").strip(), + await fetch_int_setting("site_parser_timeout_sec", 180), + ) + async with VKAPIClient( + rps=rps, + timeout_total_sec=timeout_total, + timeout_connect_sec=timeout_connect, + rate_limit_sleep_sec=rate_limit_sleep, + retry_attempts=retry_attempts, + retry_min_delay_sec=retry_min_delay, + retry_max_delay_sec=retry_max_delay, + ) as client: + for source in sources: + try: + if source.get("platform") == PLATFORM_VK: + await self.parse_source(client, source) + else: + await self.parse_site_source(site_client, source) + except VKAPIError as e: + if is_fatal_source_error(e): + await self.deactivate_source(int(source["id"]), str(e)) + logger.warning("VK source deactivated {}: {}", source.get("name"), e) + else: + await self.mark_source_error(int(source["id"]), str(e)) + logger.warning("VK source temporary error {}: {}", source.get("name"), e) + except Exception as e: await self.mark_source_error(int(source["id"]), str(e)) - logger.warning("VK source temporary error {}: {}", source.get("name"), e) - except Exception as e: - await self.mark_source_error(int(source["id"]), str(e)) - logger.exception("Unexpected source error {}: {}", source.get("name"), e) - if source_pause: - await asyncio.sleep(source_pause) + logger.exception("Unexpected source error {}: {}", source.get("name"), e) + if source.get("platform") == PLATFORM_SITE and source.get("status") != SOURCE_STATUS_ERROR: + try: + await send_system_error_alert( + "Site Parser source failed\n" + f"source: {source.get('name') or source.get('url')}\n" + f"error: {str(e)[:1000]}" + ) + except Exception as alert_exc: + logger.warning("Site Parser alert failed: {}", alert_exc) + if source_pause: + await asyncio.sleep(source_pause) async def run_loop(self) -> None: await self.init() diff --git a/tests/test_source_adapters.py b/tests/test_source_adapters.py new file mode 100644 index 0000000..bc7114d --- /dev/null +++ b/tests/test_source_adapters.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import unittest + +from vk_parser_app.source_adapters import SiteParserClient, validate_source_config + + +class FakeResponse: + status = 200 + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return None + + async def json(self, **_kwargs): + return { + "items": [{ + "external_id": "post-1", + "url": "https://example.test/1", + "title": "Title", + "text": "Body", + "published_at": "2026-08-10T10:00:00+00:00", + "media": [{"type": "photo", "url": "https://example.test/1.jpg"}], + }], + "browser_state": {"cookies": [{"name": "cf_clearance"}]}, + } + + +class FakeSession: + def post(self, *_args, **_kwargs): + return FakeResponse() + + +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({ + "url": "https://example.test/rss.xml", + "settings_json": {"format": "rss", "access": "auto"}, + "runtime_state_json": {}, + }) + + self.assertEqual(items[0].text, "Title\n\nBody") + self.assertEqual(items[0].media[0].url, "https://example.test/1.jpg") + self.assertEqual(state["browser_state"]["cookies"][0]["name"], "cf_clearance") + + def test_site_config_is_required(self) -> None: + with self.assertRaisesRegex(ValueError, "нужен конфиг"): + validate_source_config("site", {}) + + +if __name__ == "__main__": + unittest.main()