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_insta_parser_settings.sql

@@ -1,18 +0,0 @@
-INSERT INTO worker_controls(name, enabled, settings_json)
-VALUES ('insta-parser', FALSE, '{}'::jsonb)
-ON CONFLICT (name) DO NOTHING;
-
-INSERT INTO app_settings(key, value_json, value_type, title, description, category)
-VALUES
-  ('insta_login', '""'::jsonb, 'str', 'Instagram login', 'Username for the Instagram account used by instagrapi.', 'Instagram Parser'),
-  ('insta_password', '""'::jsonb, 'secret', 'Instagram password', 'Password for the Instagram account used by instagrapi.', 'Instagram Parser'),
-  ('insta_proxy_url', '""'::jsonb, 'str', 'Instagram proxy URL', 'Optional stable proxy, for example http://user:pass@host:port.', 'Instagram Parser'),
-  ('insta_session_path', '"insta_session.json"'::jsonb, 'str', 'Instagram session path', 'Path to the persisted instagrapi session settings file.', 'Instagram Parser'),
-  ('insta_auth_status', '""'::jsonb, 'str', 'Instagram auth status', 'Last manual Instagram login result.', 'Instagram Parser'),
-  ('insta_fetch_count', '5'::jsonb, 'int', 'Posts to inspect', 'How many latest posts to inspect per account visit.', 'Instagram Parser'),
-  ('insta_delay_base_minutes', '35'::jsonb, 'int', 'Account visit delay, minutes', 'Base pause after checking one Instagram account.', 'Instagram Parser'),
-  ('insta_delay_random_minutes', '5'::jsonb, 'int', 'Delay random spread, minutes', 'Random +/- spread added to the base account visit delay.', 'Instagram Parser'),
-  ('insta_cooldown_hours', '12'::jsonb, 'int', 'Cooldown hours', 'Sleep window after challenge, login-required or rate-limit responses.', 'Instagram Parser'),
-  ('insta_request_pause_sec', '2'::jsonb, 'float', 'Request pause, sec', 'Small pause between Instagram user lookup and media fetch.', 'Instagram Parser'),
-  ('insta_dedupe_content_hash', 'true'::jsonb, 'bool', 'Dedupe by content hash', 'Skip Instagram posts whose text and media hash already exists.', 'Instagram Parser')
-ON CONFLICT (key) DO NOTHING;

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;

db/migrations/047_insta_manual_auth.sql

@@ -1,12 +0,0 @@
-INSERT INTO app_settings(key, value_json, value_type, title, description, category)
-VALUES
-  ('insta_auth_status', '""'::jsonb, 'str', 'Instagram auth status', 'Last manual Instagram login result.', 'Instagram Parser')
-ON CONFLICT (key) DO NOTHING;
-
-UPDATE app_settings
-SET description='Used only by the manual Instagram login button. The parser reads the saved session file and does not auto-login.'
-WHERE key='insta_login';
-
-UPDATE app_settings
-SET description='Used only by the manual Instagram login button. The parser reads the saved session file and does not auto-login.'
-WHERE key='insta_password';

requirements.txt

@@ -10,4 +10,3 @@ Pillow==11.3.0
 python-multipart==0.0.20
 uvicorn[standard]==0.35.0
 yt-dlp==2026.6.9
-instagrapi==2.1.2

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_INSTAGRAM, 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
@@ -37,7 +38,6 @@ from .workers.tg_poster import TelegramPoster
 from .workers.tg_reactor import TelegramReactor
 from .workers.vk_poster import VKPoster
 from .workers.vk_storage_uploader import TelegramStorageUploader
-from .workers.insta_parser import InstaParserWorker, InstagramCodeRequired, instagram_login
 
 COOKIE_NAME = "vk_parser_admin"
 VK_OAUTH_VERIFIER_COOKIE = "vk_oauth_verifier"
@@ -196,8 +196,8 @@ CATEGORY_TITLES = {
     "MAX Poster": "MAX-постер",
     "Publishing": "Публикации",
     "Daily Report": "Ежедневный отчет",
-    "Instagram Parser": "Instagram-парсер",
     "Parser": "Парсер",
+    "Site Parser": "Site Parser",
     "Uploader": "Аплоадер",
     "VK": "VK API",
     "General": "Общие",
@@ -213,8 +213,8 @@ CATEGORY_ORDER = {
     "Site Poster": 52,
     "Publishing": 53,
     "Daily Report": 55,
-    "Instagram Parser": 56,
     "Parser": 60,
+    "Site Parser": 65,
     "VK": 70,
     "Uploader": 80,
     "General": 100,
@@ -950,7 +950,7 @@ def parse_source_line(line: str) -> dict[str, str]:
         (
             i
             for i, part in enumerate(parts)
-            if any(domain in part for domain in ("vk.com/", "vk.ru/", "m.vk.com/", "instagram.com/"))
+            if any(domain in part for domain in ("vk.com/", "vk.ru/", "m.vk.com/"))
             or part.lower().startswith(("club", "public"))
         ),
         -1,
@@ -958,9 +958,9 @@ def parse_source_line(line: str) -> dict[str, str]:
     if url_index < 0:
         if len(parts) == 1:
             value = parts[0].strip()
-            url = value if value.startswith("http") or value.startswith("@") else f"https://vk.com/{value}"
+            url = value if value.startswith("http") else f"https://vk.com/{value}"
             return {"name": "", "tag": "", "url": url, "error": ""}
-        return {"name": "", "tag": "", "url": "", "error": "Не нашёл ссылку"}
+        return {"name": "", "tag": "", "url": "", "error": "Не нашёл ссылку VK"}
     url = parts[url_index].strip()
     before_url = parts[:url_index]
     if len(before_url) >= 2:
@@ -978,47 +978,10 @@ def parse_source_line(line: str) -> dict[str, str]:
     return {"name": name, "tag": normalize_hash_tag(tag, ""), "url": url, "error": ""}
 
 
-def normalize_instagram_source(value: str) -> tuple[str, str]:
-    raw = (value or "").strip()
-    if not raw:
-        return "", ""
-    if raw.startswith("@"):
-        username = raw[1:]
-        return username, f"https://www.instagram.com/{username}/"
-    if not raw.startswith(("http://", "https://")) and "instagram.com/" not in raw:
-        return raw, f"https://www.instagram.com/{raw}/"
-    if raw.startswith(("instagram.com/", "www.instagram.com/")):
-        raw = f"https://{raw}"
-    match = re.search(r"instagram\.com/([^/?#]+)/?", raw, re.IGNORECASE)
-    username = (match.group(1) if match else "").strip()
-    if username.lower() in {"p", "reel", "stories", "explore"}:
-        username = ""
-    url = f"https://www.instagram.com/{username}/" if username else raw
-    return username, url
-
-
-def source_platform_from_url(url: str) -> str:
-    value = (url or "").strip().lower()
-    return PLATFORM_INSTAGRAM if value.startswith("@") or "instagram.com/" in value else PLATFORM_VK
-
-
 async def build_sources_preview(lines_text: str, default_active: bool = True) -> list[dict[str, Any]]:
     lines = [line for line in (lines_text or "").splitlines() if line.strip()]
     parsed = [parse_source_line(line) for line in lines]
-    normalized_values = []
+    normalized_values = [normalize_vk_source(item["url"]) for item in parsed if item.get("url")]
-    normalized_urls = []
-    for item in parsed:
-        url = item.get("url") or ""
-        if not url:
-            continue
-        if source_platform_from_url(url) == PLATFORM_INSTAGRAM:
-            external_id, normalized_url = normalize_instagram_source(url)
-        else:
-            external_id = normalize_vk_source(url)
-            normalized_url = url
-        if external_id:
-            normalized_values.append(external_id)
-        normalized_urls.append(normalized_url)
     pool = await get_pool()
     existing_rows = await pool.fetch(
         """
@@ -1028,7 +991,7 @@ async def build_sources_preview(lines_text: str, default_active: bool = True) ->
           AND (lower(external_id)=ANY($1::text[]) OR lower(url)=ANY($2::text[]) OR lower(COALESCE(tag,''))=ANY($3::text[]))
         """,
         [v.lower() for v in normalized_values if v],
-        [url.lower() for url in normalized_urls if url],
+        [str(item.get("url") or "").lower() for item in parsed],
         [normalize_hash_tag(str(item.get("tag") or ""), "").lower() for item in parsed if item.get("tag")],
     )
     existing_ids = {str(row["external_id"] or "").lower() for row in existing_rows}
@@ -1042,14 +1005,10 @@ async def build_sources_preview(lines_text: str, default_active: bool = True) ->
         preview = []
         for idx, item in enumerate(parsed, start=1):
             url = item.get("url") or ""
-            platform = source_platform_from_url(url)
-            if platform == PLATFORM_INSTAGRAM:
-                external_id, url = normalize_instagram_source(url)
-            else:
             external_id = normalize_vk_source(url) if url else ""
             row = {
                 "line_no": idx,
-                "platform": platform,
+                "platform": PLATFORM_VK,
                 "name": item.get("name") or "",
                 "tag": normalize_hash_tag(item.get("tag") or external_id, external_id or "source"),
                 "url": url,
@@ -1069,13 +1028,8 @@ async def build_sources_preview(lines_text: str, default_active: bool = True) ->
             if not row["error"] and row["tag"].lower() in existing_tags:
                 row["error"] = "Такой тэг уже есть"
             if not row["error"] and not external_id:
-                row["error"] = "Не удалось разобрать ссылку"
+                row["error"] = "Не удалось разобрать VK-ссылку"
-            if not row["error"] and platform == PLATFORM_INSTAGRAM:
+            if not row["error"]:
-                row["name"] = row["name"] or external_id
-                row["ok"] = True
-                seen.add(key)
-                seen_tags.add(row["tag"].lower())
-            elif not row["error"]:
                 try:
                     screen_name, owner_id, resolved_name = await client.resolve_group(url)
                     row["external_id"] = screen_name
@@ -1860,7 +1814,6 @@ async def startup() -> None:
         ("tg-reactor", TelegramReactor()),
         ("vk-poster", VKPoster()),
         ("vk-storage-uploader", TelegramStorageUploader()),
-        ("insta-parser", InstaParserWorker()),
     ]
     for name, worker in workers:
         asyncio.create_task(start_worker_task(worker, name))
@@ -2366,7 +2319,6 @@ async def sources_bulk_create(
         if not isinstance(item, dict) or not item.get("ok"):
             continue
         try:
-            platform = str(item.get("platform") or PLATFORM_VK).strip().lower() or PLATFORM_VK
             row = await pool.fetchrow(
                 """
                 INSERT INTO sources(platform, name, tag, url, external_id, external_owner_id, active, priority, created_by)
@@ -2374,7 +2326,7 @@ async def sources_bulk_create(
                 ON CONFLICT DO NOTHING
                 RETURNING id
                 """,
-                platform,
+                PLATFORM_VK,
                 str(item.get("name") or item.get("external_id") or "").strip(),
                 normalize_hash_tag(str(item.get("tag") or item.get("external_id") or ""), str(item.get("external_id") or "source")),
                 str(item.get("url") or "").strip(),
@@ -2385,7 +2337,7 @@ async def sources_bulk_create(
             )
             if row:
                 created += 1
-                await audit(user["id"], "source.create", "source", int(row["id"]), {"url": item.get("url"), "bulk": True, "platform": platform})
+                await audit(user["id"], "source.create", "source", int(row["id"]), {"url": item.get("url"), "bulk": True})
         except Exception:
             logger.exception("Bulk source insert failed: {}", item)
     return redirect(f"/sources?q=&status_filter=&created={created}")
@@ -2412,21 +2364,26 @@ 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
-    resolved_url = url.strip()
+    if platform not in {PLATFORM_VK, PLATFORM_SITE}:
-    if platform == PLATFORM_VK and source_platform_from_url(resolved_url) == PLATFORM_INSTAGRAM:
+        raise HTTPException(status_code=422, detail="Неподдерживаемая площадка")
-        platform = PLATFORM_INSTAGRAM
+    try:
-    if platform == PLATFORM_INSTAGRAM:
+        source_settings = json.loads(settings_json or "{}")
-        external_id, resolved_url = normalize_instagram_source(resolved_url)
+        if not isinstance(source_settings, dict):
-    else:
+            raise ValueError("Настройки должны быть JSON-объектом")
-        external_id = normalize_vk_source(resolved_url)
+        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()
+    resolved_url = url.strip()
     status_value = "new"
     status_msg = None
     if platform == PLATFORM_VK and external_id:
@@ -2445,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
         """,
@@ -2460,6 +2417,7 @@ async def source_create(
         priority,
         status_value,
         status_msg,
+        json.dumps(source_settings, ensure_ascii=False),
         user["id"],
     )
     if row:
@@ -2499,19 +2457,23 @@ 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
-    resolved_url = url.strip()
+    if platform not in {PLATFORM_VK, PLATFORM_SITE}:
-    if platform == PLATFORM_VK and source_platform_from_url(resolved_url) == PLATFORM_INSTAGRAM:
+        raise HTTPException(status_code=422, detail="Неподдерживаемая площадка")
-        platform = PLATFORM_INSTAGRAM
+    try:
-    if platform == PLATFORM_INSTAGRAM:
+        source_settings = json.loads(settings_json or "{}")
-        external_id, resolved_url = normalize_instagram_source(resolved_url)
+        if not isinstance(source_settings, dict):
-    else:
+            raise ValueError("Настройки должны быть JSON-объектом")
-        external_id = normalize_vk_source(resolved_url)
+        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(
         """
@@ -2521,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
         """,
@@ -2530,10 +2498,11 @@ async def source_update(
         platform,
         name.strip(),
         normalize_hash_tag(tag or external_id or name, external_id or "source"),
-        resolved_url,
+        url.strip(),
         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")
@@ -3623,8 +3592,6 @@ async def workers(request: Request):
             vk_schedule=await vk_poster_schedule_rows(),
             category_titles=CATEGORY_TITLES,
             prompt_hints=PROMPT_HINTS,
-            instagram_auth_status=setting_values.get("insta_auth_status") or "",
-            instagram_cooldown_until=setting_values.get("insta_cooldown_until") or "",
         ),
     )
 
@@ -3953,80 +3920,6 @@ async def worker_toggle(request: Request, worker_name: str, csrf_token: str = Fo
     return redirect("/workers")
 
 
-@app.post("/instagram-auth/login")
-async def instagram_auth_login(request: Request, csrf_token: str = Form(...), verification_code: str = Form("")):
-    user = await get_current_user(request)
-    if not user:
-        return redirect("/login")
-    require_csrf(user, csrf_token)
-    login = str(await fetch_setting("insta_login", "") or "").strip()
-    password = str(await fetch_setting("insta_password", "") or "").strip()
-    proxy = str(await fetch_setting("insta_proxy_url", "") or "").strip()
-    session_path = str(await fetch_setting("insta_session_path", "insta_session.json") or "").strip()
-    pool = await get_pool()
-    if not login or not password:
-        status_text = "missing login or password"
-    else:
-        try:
-            await asyncio.to_thread(instagram_login, login, password, proxy, session_path, verification_code.strip())
-            status_text = f"ok: session saved to {session_path or 'insta_session.json'} at {datetime.now(timezone.utc).isoformat()}"
-            await pool.execute(
-                """
-                INSERT INTO app_settings(key, value_json, value_type, title, description, category)
-                VALUES('insta_cooldown_until', '""'::jsonb, 'str', 'Cooldown until', '', 'Instagram Parser')
-                ON CONFLICT (key) DO UPDATE
-                SET value_json='""'::jsonb,
-                    description='',
-                    updated_at=NOW()
-                """
-            )
-        except InstagramCodeRequired as exc:
-            status_text = f"code_required: {exc.choice}"
-        except Exception as exc:
-            exc_name = exc.__class__.__name__
-            if exc_name in {"TwoFactorRequired", "ChallengeRequired"}:
-                status_text = f"code_required: {exc_name}"
-            else:
-                status_text = f"failed: {exc_name}: {str(exc)[:1000]}"
-    await pool.execute(
-        """
-        INSERT INTO app_settings(key, value_json, value_type, title, description, category, updated_by)
-        VALUES('insta_auth_status', $1::jsonb, 'str', 'Instagram auth status', 'Last manual Instagram login result.', 'Instagram Parser', $2)
-        ON CONFLICT (key) DO UPDATE
-        SET value_json=$1::jsonb,
-            updated_by=$2,
-            updated_at=NOW()
-        """,
-        json.dumps(status_text),
-        user["id"],
-    )
-    await audit(user["id"], "instagram.login", "setting", None, {"status": status_text})
-    return redirect("/workers")
-
-
-@app.post("/instagram-auth/clear-cooldown")
-async def instagram_auth_clear_cooldown(request: Request, csrf_token: str = Form(...)):
-    user = await get_current_user(request)
-    if not user:
-        return redirect("/login")
-    require_csrf(user, csrf_token)
-    pool = await get_pool()
-    await pool.execute(
-        """
-        INSERT INTO app_settings(key, value_json, value_type, title, description, category, updated_by)
-        VALUES('insta_cooldown_until', '""'::jsonb, 'str', 'Cooldown until', '', 'Instagram Parser', $1)
-        ON CONFLICT (key) DO UPDATE
-        SET value_json='""'::jsonb,
-            description='manual reset',
-            updated_by=$1,
-            updated_at=NOW()
-        """,
-        user["id"],
-    )
-    await audit(user["id"], "instagram.cooldown_reset", "setting", None)
-    return redirect("/workers")
-
-
 @app.post("/settings/save")
 async def settings_save(request: Request, csrf_token: str = Form(...), key: str = Form(...), value: str = Form(...)):
     user = await get_current_user(request)

src/vk_parser_app/constants.py

@@ -1,5 +1,5 @@
 PLATFORM_VK = "vk"
-PLATFORM_INSTAGRAM = "instagram"
+PLATFORM_SITE = "site"
 
 SOURCE_STATUS_NEW = "new"
 SOURCE_STATUS_OK = "ok"
@@ -39,4 +39,3 @@ WORKER_VK_POSTER = "vk-poster"
 WORKER_MAX_POSTER = "max-poster"
 WORKER_SITE_POSTER = "site-poster"
 WORKER_DAILY_REPORT = "daily-report"
-WORKER_INSTA_PARSER = "insta-parser"

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,7 +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="instagram" {% if source and source.platform == "instagram" %}selected{% endif %}>Instagram</option>
+            <option value="site" {% if source and source.platform == "site" %}selected{% endif %}>Сайт</option>
           </select>
         </div>
         
@@ -40,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">
@@ -55,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/templates/workers.html

@@ -294,50 +294,12 @@
 
 <div class="flex flex-col gap-6">
   {% for category, rows in settings|groupby("category") %}
-  <details class="card group/details" data-workers-details="settings:{{ category }}" {% if category == "Instagram Parser" %}open{% endif %}>
+  <details class="card group/details" data-workers-details="settings:{{ category }}">
     <summary class="p-4 flex items-center justify-between cursor-pointer select-none hover:bg-app-surfaceHover transition-colors border-b border-app-border list-none">
       <div class="text-lg font-bold text-white">{{ category_titles.get(category, category) }}</div>
       <i data-lucide="chevron-down" class="w-5 h-5 text-app-textMuted transition-transform group-open/details:rotate-180"></i>
     </summary>
     <div class="p-6 bg-app-bg/30">
-      {% if category == "Instagram Parser" %}
-      <div class="mb-8 p-5 bg-app-surface border border-app-border rounded-xl flex flex-col gap-4">
-        <div class="flex flex-col lg:flex-row lg:items-start lg:justify-between gap-4">
-          <div class="min-w-0">
-            <div class="font-bold text-white flex items-center gap-2">
-              <i data-lucide="instagram" class="w-5 h-5 text-pink-400"></i>
-              Авторизация Instagram
-            </div>
-            <div class="text-xs text-app-textMuted mt-2 break-words">
-              Статус: <span class="font-mono text-app-textMain">{{ instagram_auth_status or "—" }}</span>
-            </div>
-            {% if instagram_cooldown_until %}
-            <div class="text-xs text-app-warning mt-2 break-words">
-              Cooldown: <span class="font-mono">{{ instagram_cooldown_until }}</span>
-            </div>
-            {% endif %}
-          </div>
-          <form method="post" action="/instagram-auth/clear-cooldown" class="m-0">
-            <input type="hidden" name="csrf_token" value="{{ user.csrf_token }}">
-            <button class="btn btn-surface btn-sm" type="submit">
-              <i data-lucide="timer-reset" class="w-4 h-4"></i>
-              Сбросить cooldown
-            </button>
-          </form>
-        </div>
-        <form method="post" action="/instagram-auth/login" class="grid grid-cols-1 md:grid-cols-[1fr_auto] gap-3 items-end">
-          <input type="hidden" name="csrf_token" value="{{ user.csrf_token }}">
-          <div>
-            <label class="block text-[10px] uppercase font-bold text-app-textMuted mb-1">Код из SMS/email/2FA, если Instagram его просит</label>
-            <input name="verification_code" class="input w-full font-mono" autocomplete="one-time-code" placeholder="Оставь пустым для первой попытки">
-          </div>
-          <button class="btn btn-primary" type="submit">
-            <i data-lucide="key-round" class="w-4 h-4"></i>
-            Войти
-          </button>
-        </form>
-      </div>
-      {% endif %}
       
       <div class="flex flex-col gap-8">
         {% for s in rows %}
@@ -466,7 +428,7 @@
     const form = event.target;
     if (!(form instanceof HTMLFormElement)) return;
     const action = form.getAttribute("action") || "";
-    if (action.startsWith("/workers") || action.startsWith("/instagram-auth") || action.startsWith("/settings") || action.startsWith("/branding") || action.includes("-schedule/")) {
+    if (action.startsWith("/workers") || action.startsWith("/settings") || action.startsWith("/branding") || action.includes("-schedule/")) {
       saveDetails();
       sessionStorage.setItem(scrollKey, String(window.scrollY));
     }

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,58 +47,27 @@ 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
+    bot = Bot(token=token)
+    try:
+        for recipient_id in recipients:
+            while True:
+                try:
+                    await bot.send_message(recipient_id, text, disable_web_page_preview=True)
+                    break
+                except TelegramRetryAfter as exc:
+                    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 "-"
-    text = (
+    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]}"
     )
-    bot = Bot(token=token)
-    try:
-        for recipient_id in recipients:
-            while True:
-                try:
-                    await bot.send_message(recipient_id, text, disable_web_page_preview=True)
-                    break
-                except TelegramRetryAfter as exc:
-                    await asyncio.sleep(float(exc.retry_after) + 1)
-    finally:
-        await bot.session.close()
-
-
-async def send_parser_error_alert(parser_name: str, source_name: str, error: str) -> None:
-    token = (
-        str(await fetch_setting("daily_report_bot_token", "") or "").strip()
-        or str(await fetch_setting("tg_poster_bot_token", "") or "").strip()
-        or settings.tg_bot_token
-    )
-    recipients = parse_recipients(await fetch_setting("daily_report_recipient_ids", [442509142]))
-    if not token or not recipients:
-        logger.warning("Parser alert skipped: token or recipients are empty")
-        return
-
-    text = (
-        "Parser failed\n"
-        f"parser: {parser_name}\n"
-        f"source: {source_name or '-'}\n"
-        f"error: {error[:1000]}"
-    )
-    bot = Bot(token=token)
-    try:
-        for recipient_id in recipients:
-            while True:
-                try:
-                    await bot.send_message(recipient_id, text, disable_web_page_preview=True)
-                    break
-                except TelegramRetryAfter as exc:
-                    await asyncio.sleep(float(exc.retry_after) + 1)
-                except Exception as exc:
-                    logger.error("Failed to send parser alert to {}: {}", recipient_id, exc)
-                    break
-    finally:
-        await bot.session.close()

src/vk_parser_app/workers/insta_parser.py

@@ -1,521 +0,0 @@
-from __future__ import annotations
-
-import asyncio
-import hashlib
-import json
-import random
-from datetime import datetime, timedelta, timezone
-from pathlib import Path
-from typing import Any
-
-from loguru import logger
-
-from ..constants import (
-    JOB_TYPE_VK_STORAGE_COPY,
-    MEDIA_STATUS_LINK_ONLY,
-    MEDIA_STATUS_PENDING,
-    PLATFORM_INSTAGRAM,
-    POST_STATUS_SKIPPED,
-    POST_STATUS_STORAGE_PENDING,
-    SOURCE_STATUS_ERROR,
-    SOURCE_STATUS_OK,
-    WORKER_INSTA_PARSER,
-)
-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 .ai_alerts import send_parser_error_alert
-
-
-def make_hash(*parts: str) -> str:
-    h = hashlib.sha256()
-    for part in parts:
-        h.update((part or "").strip().lower().encode("utf-8"))
-        h.update(b"\0")
-    return h.hexdigest()
-
-
-def as_dict(value: Any) -> dict:
-    if isinstance(value, dict):
-        return value
-    if hasattr(value, "model_dump"):
-        return value.model_dump()
-    if hasattr(value, "dict"):
-        return value.dict()
-    return {}
-
-
-def media_value(value: Any) -> str:
-    return str(value or "").strip()
-
-
-def media_code(post: dict) -> str:
-    return str(post.get("code") or post.get("pk") or "").strip()
-
-
-def original_url(post: dict) -> str:
-    code = media_code(post)
-    return f"https://www.instagram.com/p/{code}/" if code else ""
-
-
-def media_taken_at(post: dict) -> datetime:
-    value = post.get("taken_at")
-    if isinstance(value, datetime):
-        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
-    return datetime.now(tz=timezone.utc)
-
-
-def media_user_pk(post: dict) -> int | None:
-    user = as_dict(post.get("user"))
-    try:
-        return int(user.get("pk") or 0) or None
-    except Exception:
-        return None
-
-
-def extract_instagram_media(post: dict) -> list[dict]:
-    items: list[dict] = []
-    resources = [as_dict(item) for item in post.get("resources") or []]
-    if not resources:
-        resources = [post]
-
-    for index, item in enumerate(resources):
-        media_type = int(item.get("media_type") or post.get("media_type") or 0)
-        if media_type == 1:
-            url = media_value(item.get("thumbnail_url") or post.get("thumbnail_url"))
-            kind = "photo"
-        elif media_type == 2:
-            url = media_value(item.get("video_url") or post.get("video_url"))
-            kind = "video"
-        else:
-            continue
-        if not url:
-            continue
-        items.append(
-            {
-                "media_type": kind,
-                "original_url": url,
-                "original_attachment_id": str(item.get("pk") or post.get("pk") or index),
-                "width": item.get("width") or post.get("width"),
-                "height": item.get("height") or post.get("height"),
-                "duration_sec": int(float(item.get("video_duration") or post.get("video_duration") or 0)) or None,
-                "sort_order": index,
-            }
-        )
-    return items
-
-
-class NeverRaised(Exception):
-    pass
-
-
-class InstagramAuthRequired(Exception):
-    pass
-
-
-class InstagramCodeRequired(Exception):
-    def __init__(self, choice: Any) -> None:
-        self.choice = choice
-        super().__init__(f"Instagram requested verification code: {choice}")
-
-
-def load_instagrapi():
-    from instagrapi import Client
-    import instagrapi.exceptions as exc
-
-    return Client, {
-        "challenge": getattr(exc, "ChallengeRequired", NeverRaised),
-        "login_required": getattr(exc, "LoginRequired", getattr(exc, "ClientLoginRequired", NeverRaised)),
-        "please_wait": getattr(exc, "PleaseWaitFewMinutes", NeverRaised),
-        "user_not_found": getattr(exc, "UserNotFound", NeverRaised),
-        "private_account": getattr(exc, "PrivateAccount", NeverRaised),
-    }
-
-
-def instagram_login(login: str, password: str, proxy: str, session_path: str, verification_code: str = "") -> None:
-    Client, _ = load_instagrapi()
-    client = Client()
-    if proxy:
-        client.set_proxy(proxy)
-    if verification_code:
-        client.challenge_code_handler = lambda username, choice: verification_code
-    else:
-        def challenge_code_handler(username: str, choice: Any) -> str:
-            raise InstagramCodeRequired(choice)
-
-        client.challenge_code_handler = challenge_code_handler
-    path = Path(session_path or "insta_session.json")
-    path.parent.mkdir(parents=True, exist_ok=True)
-    login_kwargs = {"verification_code": verification_code} if verification_code else {}
-    client.login(login, password, **login_kwargs)
-    client.dump_settings(path)
-
-
-class InstaParserWorker:
-    def __init__(self) -> None:
-        self.pool = None
-        self.heartbeat = HeartbeatReporter(WORKER_INSTA_PARSER, 30)
-        self.client = None
-        self.exceptions: dict[str, type[BaseException]] = {}
-
-    async def init(self) -> None:
-        self.pool = await get_pool()
-
-    async def active_source(self) -> dict | None:
-        row = await self.pool.fetchrow(
-            """
-            SELECT *
-            FROM sources
-            WHERE platform=$1
-              AND active=TRUE
-              AND archived_at IS NULL
-            ORDER BY last_checked_at NULLS FIRST, priority ASC, id ASC
-            LIMIT 1
-            """,
-            PLATFORM_INSTAGRAM,
-        )
-        return dict(row) if row else None
-
-    async def mark_source_error(self, source_id: int, message: str) -> None:
-        await self.pool.execute(
-            """
-            UPDATE sources
-            SET status=$2,
-                status_msg=$3,
-                last_checked_at=NOW(),
-                updated_at=NOW()
-            WHERE id=$1
-            """,
-            source_id,
-            SOURCE_STATUS_ERROR,
-            message[:1000],
-        )
-
-    async def deactivate_source(self, source_id: int, message: str) -> None:
-        await self.pool.execute(
-            """
-            UPDATE sources
-            SET active=FALSE,
-                status=$2,
-                status_msg=$3,
-                last_checked_at=NOW(),
-                updated_at=NOW()
-            WHERE id=$1
-            """,
-            source_id,
-            SOURCE_STATUS_ERROR,
-            message[:1000],
-        )
-
-    async def mark_source_ok(self, source_id: int, last_parsed_at: datetime | None) -> None:
-        await self.pool.execute(
-            """
-            UPDATE sources
-            SET status=$2,
-                status_msg=NULL,
-                last_checked_at=NOW(),
-                last_parsed_at=COALESCE($3, last_parsed_at),
-                updated_at=NOW()
-            WHERE id=$1
-            """,
-            source_id,
-            SOURCE_STATUS_OK,
-            last_parsed_at,
-        )
-
-    async def known_post_ids(self, source_id: int, external_post_ids: list[str]) -> set[str]:
-        if not external_post_ids:
-            return set()
-        rows = await self.pool.fetch(
-            """
-            SELECT external_post_id
-            FROM raw_posts
-            WHERE source_id=$1 AND external_post_id=ANY($2::text[])
-            """,
-            source_id,
-            external_post_ids,
-        )
-        return {str(row["external_post_id"]) for row in rows}
-
-    async def known_content_hashes(self, hashes: list[str]) -> set[str]:
-        if not hashes:
-            return set()
-        rows = await self.pool.fetch(
-            """
-            SELECT content_hash
-            FROM raw_posts
-            WHERE content_hash=ANY($1::text[])
-            """,
-            hashes,
-        )
-        return {str(row["content_hash"]) for row in rows}
-
-    async def set_cooldown(self, hours: int, reason: str) -> None:
-        until = datetime.now(tz=timezone.utc) + timedelta(hours=max(1, hours))
-        await self.pool.execute(
-            """
-            INSERT INTO app_settings(key, value_json, value_type, title, description, category)
-            VALUES('insta_cooldown_until', $1::jsonb, 'str', 'Cooldown until', $2, 'Instagram Parser')
-            ON CONFLICT (key) DO UPDATE
-            SET value_json=$1::jsonb,
-                description=$2,
-                updated_at=NOW()
-            """,
-            json.dumps(until.isoformat()),
-            reason[:1000],
-        )
-
-    async def cooldown_until(self) -> datetime | None:
-        value = await fetch_setting("insta_cooldown_until", "")
-        if not value:
-            return None
-        try:
-            dt = datetime.fromisoformat(str(value))
-            return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
-        except ValueError:
-            return None
-
-    def setup_client(self, proxy: str, session_path: str) -> None:
-        if self.client is not None:
-            return
-        Client, exceptions = load_instagrapi()
-        self.exceptions = exceptions
-        client = Client()
-        if proxy:
-            client.set_proxy(proxy)
-        path = Path(session_path or "insta_session.json")
-        if not path.exists():
-            raise InstagramAuthRequired(f"Instagram session file not found: {path}")
-        client.load_settings(path)
-        client.get_timeline_feed()
-        self.client = client
-
-    async def save_post(self, source: dict, post: dict, status: str, skip_reason: str | None, media: list[dict]) -> int | None:
-        source_id = int(source["id"])
-        external_post_id = str(post["pk"])
-        raw_text = str(post.get("caption_text") or "").strip()
-        media_ids = ",".join(item["original_attachment_id"] for item in media)
-        text_hash = make_hash(raw_text)
-        content_hash = make_hash(raw_text, media_ids)
-        has_downloadable_media = any(item["media_type"] == "photo" for item in 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, external_owner_id,
-                      original_url, raw_text, raw_json, text_hash, content_hash,
-                      posted_at, status, skip_reason
-                    )
-                    VALUES($1,$2,$3,$4,$5,$6,$7::jsonb,$8,$9,$10,$11,$12)
-                    ON CONFLICT (source_id, external_post_id) DO NOTHING
-                    RETURNING id
-                    """,
-                    source_id,
-                    PLATFORM_INSTAGRAM,
-                    external_post_id,
-                    media_user_pk(post),
-                    original_url(post),
-                    raw_text,
-                    json.dumps(post, default=str, ensure_ascii=False),
-                    text_hash,
-                    content_hash,
-                    media_taken_at(post),
-                    status,
-                    skip_reason,
-                )
-                if raw_post_id is None:
-                    return None
-
-                for item in media:
-                    await conn.execute(
-                        """
-                        INSERT INTO raw_post_media(
-                          raw_post_id, platform, media_type, original_url, original_attachment_id,
-                          width, height, duration_sec, sort_order, status, error
-                        )
-                        VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
-                        """,
-                        raw_post_id,
-                        PLATFORM_INSTAGRAM,
-                        item["media_type"],
-                        item["original_url"],
-                        item["original_attachment_id"],
-                        item.get("width"),
-                        item.get("height"),
-                        item.get("duration_sec"),
-                        item["sort_order"],
-                        MEDIA_STATUS_LINK_ONLY if item["media_type"] == "video" else MEDIA_STATUS_PENDING,
-                        "instagram video link only" if item["media_type"] == "video" else None,
-                    )
-
-                if status == POST_STATUS_STORAGE_PENDING and has_downloadable_media:
-                    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_source(self, source: dict, fetch_count: int, request_pause_sec: float) -> int:
-        username = str(source.get("external_id") or "").strip().lstrip("@")
-        if not username:
-            await self.mark_source_error(int(source["id"]), "empty instagram username")
-            return 0
-
-        user_id = await asyncio.to_thread(self.client.user_id_from_username, username)
-        if request_pause_sec:
-            await asyncio.sleep(request_pause_sec)
-        medias = await asyncio.to_thread(self.client.user_medias, user_id, fetch_count)
-        posts = [as_dict(item) for item in medias]
-        if not posts:
-            await self.mark_source_ok(int(source["id"]), None)
-            return 0
-
-        known_ids = await self.known_post_ids(int(source["id"]), [str(post.get("pk")) for post in posts if post.get("pk")])
-        candidates = [post for post in posts if str(post.get("pk")) not in known_ids]
-
-        dedupe_content_hash = await fetch_bool_setting("insta_dedupe_content_hash", True)
-        hash_by_pk: dict[str, str] = {}
-        if dedupe_content_hash:
-            hashes = []
-            for post in candidates:
-                media = extract_instagram_media(post)
-                media_ids = ",".join(item["original_attachment_id"] for item in media)
-                content_hash = make_hash(str(post.get("caption_text") or "").strip(), media_ids)
-                hash_by_pk[str(post["pk"])] = content_hash
-                hashes.append(content_hash)
-            known_hashes = await self.known_content_hashes(hashes)
-        else:
-            known_hashes = set()
-
-        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)
-        saved = 0
-        max_seen: datetime | None = None
-
-        for post in candidates:
-            posted_at = media_taken_at(post)
-            if max_seen is None or posted_at > max_seen:
-                max_seen = posted_at
-            content_hash = hash_by_pk.get(str(post.get("pk")))
-            if dedupe_content_hash and content_hash in known_hashes:
-                continue
-
-            text = str(post.get("caption_text") or "").strip()
-            media = extract_instagram_media(post)
-            skip_reason = None
-            if skip_empty_text and not text:
-                skip_reason = "empty_text"
-            elif skip_no_media and not media:
-                skip_reason = "no_media"
-            elif skip_short_text and len(text) < min_text_length:
-                skip_reason = "text_too_short"
-
-            if skip_reason and not store_skipped:
-                continue
-            status = POST_STATUS_SKIPPED if skip_reason else POST_STATUS_STORAGE_PENDING
-            raw_id = await self.save_post(source, post, status, skip_reason, media)
-            if raw_id:
-                saved += 1
-                if content_hash:
-                    known_hashes.add(content_hash)
-
-        await self.mark_source_ok(int(source["id"]), max_seen or source.get("last_parsed_at"))
-        logger.info("Parsed Instagram source {}: fetched={} known={} saved={}", username, len(posts), len(known_ids), saved)
-        return saved
-
-    async def run_once(self) -> bool:
-        enabled = await is_worker_enabled(self.pool, WORKER_INSTA_PARSER)
-        if not enabled:
-            await self.heartbeat.beat(self.pool, status="disabled", force=True)
-            return False
-
-        until = await self.cooldown_until()
-        if until and until > datetime.now(tz=timezone.utc):
-            await self.heartbeat.beat(self.pool, status="cooldown", meta={"until": until.isoformat()})
-            return False
-
-        proxy = str(await fetch_setting("insta_proxy_url", "") or "").strip()
-        session_path = str(await fetch_setting("insta_session_path", "insta_session.json") or "").strip()
-        fetch_count = max(1, min(20, await fetch_int_setting("insta_fetch_count", 5)))
-        cooldown_hours = max(1, await fetch_int_setting("insta_cooldown_hours", 12))
-        request_pause_sec = max(0.0, await fetch_float_setting("insta_request_pause_sec", 2.0))
-
-        try:
-            await asyncio.to_thread(self.setup_client, proxy, session_path)
-        except InstagramAuthRequired as exc:
-            await self.heartbeat.beat(self.pool, status="auth_required", meta={"error": str(exc)})
-            await self.set_cooldown(cooldown_hours, str(exc))
-            await send_parser_error_alert(WORKER_INSTA_PARSER, "auth", str(exc))
-            return False
-        except Exception as exc:
-            self.client = None
-            await self.heartbeat.beat(self.pool, status="auth_required", meta={"error": str(exc)})
-            await self.set_cooldown(cooldown_hours, f"session failed: {exc}")
-            await send_parser_error_alert(WORKER_INSTA_PARSER, "session", str(exc))
-            return False
-
-        source = await self.active_source()
-        if not source:
-            await self.heartbeat.beat(self.pool, status="idle")
-            return False
-
-        try:
-            await self.heartbeat.beat(self.pool, status="running", meta={"source": source.get("external_id")})
-            await self.parse_source(source, fetch_count, request_pause_sec)
-            return True
-        except (self.exceptions.get("user_not_found", NeverRaised), self.exceptions.get("private_account", NeverRaised)) as exc:
-            await self.deactivate_source(int(source["id"]), str(exc))
-            await send_parser_error_alert(WORKER_INSTA_PARSER, str(source.get("external_id") or ""), str(exc))
-            return True
-        except (
-            self.exceptions.get("challenge", NeverRaised),
-            self.exceptions.get("login_required", NeverRaised),
-            self.exceptions.get("please_wait", NeverRaised),
-        ) as exc:
-            self.client = None
-            await self.mark_source_error(int(source["id"]), str(exc))
-            await self.set_cooldown(cooldown_hours, str(exc))
-            await send_parser_error_alert(WORKER_INSTA_PARSER, str(source.get("external_id") or ""), str(exc))
-            return True
-        except Exception as exc:
-            await self.mark_source_error(int(source["id"]), str(exc))
-            logger.exception("Unexpected Instagram source error {}: {}", source.get("external_id"), exc)
-            return True
-
-    async def run_loop(self) -> None:
-        await self.init()
-        logger.info("{} started", WORKER_INSTA_PARSER)
-        while True:
-            try:
-                visited_source = await self.run_once()
-            except Exception as exc:
-                logger.exception("Instagram parser loop error: {}", exc)
-                visited_source = False
-            if not visited_source:
-                await asyncio.sleep(10)
-                continue
-            base = max(1, await fetch_int_setting("insta_delay_base_minutes", 35))
-            spread = max(0, await fetch_int_setting("insta_delay_random_minutes", 5))
-            delay = max(60.0, base * 60 + random.uniform(-spread * 60, spread * 60))
-            await asyncio.sleep(delay)
-
-
-async def main() -> None:
-    logger.remove()
-    logger.add(lambda msg: print(msg, end=""))
-    worker = InstaParserWorker()
-    await worker.run_loop()
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

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