feat: add external site parser worker

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
+```
+
 ## Запуск локально
 
 Админка:

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;

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"]

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,
+    }

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

site_parser_worker/test_parser.py

@@ -0,0 +1,16 @@
+from app import parse_rss
+
+
+def test_parse_rss() -> None:
+    xml = """<rss><channel><item><guid>1</guid><title>Title</title>
+    <link>https://example.test/1</link>
+    <description>&lt;p&gt;Body&lt;/p&gt;&lt;img src="https://example.test/1.jpg"&gt;</description>
+    <pubDate>Mon, 10 Aug 2026 10:00:00 +0000</pubDate></item></channel></rss>"""
+    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()

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)
+        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)
+        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")

src/vk_parser_app/constants.py

@@ -1,4 +1,5 @@
 PLATFORM_VK = "vk"
+PLATFORM_SITE = "site"
 
 SOURCE_STATUS_NEW = "new"
 SOURCE_STATUS_OK = "ok"

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)

src/vk_parser_app/templates/source_form.html

@@ -17,6 +17,7 @@
           <label class="label"><span class="label-text font-bold">Площадка</span></label>
           <select name="platform" class="select select-bordered w-full">
             <option value="vk" {% if not source or source.platform == "vk" %}selected{% endif %}>VK</option>
+            <option value="site" {% if source and source.platform == "site" %}selected{% endif %}>Сайт</option>
           </select>
         </div>
         
@@ -39,6 +40,20 @@
           <label class="label"><span class="label-text font-bold">Ссылка</span></label>
           <input name="url" value="{{ source.url if source else '' }}" required class="input input-bordered w-full text-primary">
         </div>
+
+        <div id="site-config" class="form-control md:col-span-2">
+          <label class="label"><span class="label-text font-bold">Конфигурация JSON</span></label>
+          <textarea name="settings_json" rows="8" class="textarea textarea-bordered w-full font-mono" placeholder='{"format":"rss","access":"auto","max_items":20}'>{{ source.settings_json | tojson(indent=2) if source else '{}' }}</textarea>
+          <details class="mt-2 text-sm text-base-content/70">
+            <summary class="cursor-pointer">Пример и параметры</summary>
+            <pre class="mt-2 p-3 bg-base-200 overflow-x-auto">{
+  "format": "rss",
+  "access": "auto",
+  "max_items": 20
+}</pre>
+            <p class="mt-2"><code>access</code>: <code>auto</code> сначала пробует обычный запрос и при Cloudflare использует RuCaptcha; <code>http</code> запрещает браузер; <code>cloudflare</code> сразу запускает браузер.</p>
+          </details>
+        </div>
       </div>
       
       <div class="form-control mt-4">
@@ -54,4 +69,11 @@
     </form>
   </div>
 </div>
+<script>
+  const platform = document.querySelector('[name="platform"]');
+  const siteConfig = document.getElementById('site-config');
+  const syncConfig = () => siteConfig.hidden = platform.value !== 'site';
+  platform.addEventListener('change', syncConfig);
+  syncConfig();
+</script>
 {% endblock %}

src/vk_parser_app/templates/sources.html

@@ -5,7 +5,7 @@
     <i data-lucide="database" class="text-app-primary w-8 h-8"></i>
     Источники
   </h1>
-  <div class="text-app-textMuted text-sm">VK-источники для парсинга. Название идёт в prompt, тэг — в будущие хэштеги.</div>
+  <div class="text-app-textMuted text-sm">Источники для парсинга. Название идёт в prompt, тэг — в будущие хэштеги.</div>
 </div>
 
 <details class="card mb-8 group/details" {% if source_preview %}open{% endif %}>

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]}"
+    )

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,6 +539,14 @@ class VKParserWorker:
             source_pause,
         )
 
+        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,
@@ -412,7 +558,10 @@ class VKParserWorker:
             ) 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))
@@ -423,6 +572,15 @@ class VKParserWorker:
                     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.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)
 

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()