fix: complete site parser source workflow

2026-08-10 21:11:43 +05:00
parent 3dea6f43f3
commit 5417737c08
9 changed files with 163 additions and 23 deletions
@@ -0,0 +1,10 @@
+INSERT INTO app_settings(key, value_json, value_type, title, description, category)
+VALUES (
+  'site_parser_default_interval_minutes',
+  '30'::jsonb,
+  'int',
+  'Интервал по умолчанию, мин',
+  'Как часто проверять сайт, если у источника не задан свой интервал.',
+  'Site Parser'
+)
+ON CONFLICT (key) DO NOTHING;
@@ -25,7 +25,7 @@ from .config import settings
 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 .source_adapters import json_object, 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
@@ -366,6 +366,13 @@ SETTING_ORDER = {
        "parser_dedupe_content_hash",
        "parser_source_pause_sec",
    ],
+    "Site Parser": [
+        "site_parser_url",
+        "site_parser_token",
+        "site_parser_rucaptcha_token",
+        "site_parser_timeout_sec",
+        "site_parser_default_interval_minutes",
+    ],
    "VK": [
        "vk_requests_per_second",
        "vk_wall_page_size",
@@ -2196,6 +2203,7 @@ async def sources_list(request: Request, q: str = "", status_filter: str = "", a
    sources = []
    for row in rows:
        item = dict(row)
+        item["settings_json"] = json_object(item.get("settings_json"))
        item["last_parsed_at_fmt"] = format_dt(item.get("last_parsed_at"))
        item["last_checked_at_fmt"] = format_dt(item.get("last_checked_at"))
        item["created_at_fmt"] = format_dt(item.get("created_at"))
@@ -2274,6 +2282,7 @@ async def sources_preview(
    sources = []
    for row in rows:
        item = dict(row)
+        item["settings_json"] = json_object(item.get("settings_json"))
        item["last_parsed_at_fmt"] = format_dt(item.get("last_parsed_at"))
        item["last_checked_at_fmt"] = format_dt(item.get("last_checked_at"))
        item["created_at_fmt"] = format_dt(item.get("created_at"))
@@ -2350,7 +2359,14 @@ async def source_new(request: Request):
        return redirect("/login")
    return templates.TemplateResponse(
        "source_form.html",
-        base_context(request, user, source=None, action="/sources/new", title="Новый источник"),
+        base_context(
+            request,
+            user,
+            source=None,
+            action="/sources/new",
+            title="Новый источник",
+            site_default_interval=await fetch_int_setting("site_parser_default_interval_minutes", 30),
+        ),
    )


@@ -2365,6 +2381,7 @@ async def source_create(
    active: str = Form("off"),
    priority: int = Form(100),
    settings_json: str = Form("{}"),
+    interval_minutes: int = Form(30),
 ):
    user = await get_current_user(request)
    if not user:
@@ -2377,6 +2394,8 @@ async def source_create(
        source_settings = json.loads(settings_json or "{}")
        if not isinstance(source_settings, dict):
            raise ValueError("Настройки должны быть JSON-объектом")
+        if platform == PLATFORM_SITE:
+            source_settings["interval_minutes"] = interval_minutes
        validate_source_config(platform, source_settings)
    except (json.JSONDecodeError, ValueError) as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
@@ -2439,9 +2458,10 @@ async def source_edit(request: Request, source_id: int):
        base_context(
            request,
            user,
-            source=dict(source),
+            source={**dict(source), "settings_json": json_object(source["settings_json"])},
            action=f"/sources/{source_id}/edit",
            title=f"Источник #{source_id}",
+            site_default_interval=await fetch_int_setting("site_parser_default_interval_minutes", 30),
        ),
    )

@@ -2458,6 +2478,7 @@ async def source_update(
    active: str = Form("off"),
    priority: int = Form(100),
    settings_json: str = Form("{}"),
+    interval_minutes: int = Form(30),
 ):
    user = await get_current_user(request)
    if not user:
@@ -2470,6 +2491,8 @@ async def source_update(
        source_settings = json.loads(settings_json or "{}")
        if not isinstance(source_settings, dict):
            raise ValueError("Настройки должны быть JSON-объектом")
+        if platform == PLATFORM_SITE:
+            source_settings["interval_minutes"] = interval_minutes
        validate_source_config(platform, source_settings)
    except (json.JSONDecodeError, ValueError) as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
@@ -2532,6 +2555,7 @@ async def source_toggle(request: Request, source_id: int, csrf_token: str = Form
        source_row = await pool.fetchrow("SELECT * FROM sources WHERE id=$1", source_id)
        if source_row:
            s = dict(source_row)
+            s["settings_json"] = json_object(s.get("settings_json"))
            # Fetch stats just for this source to render correctly
            stats = await pool.fetchrow(
                """
@@ -2549,6 +2573,25 @@ async def source_toggle(request: Request, source_id: int, csrf_token: str = Form
    return redirect("/sources")


+@app.post("/sources/{source_id}/parse-now")
+async def source_parse_now(request: Request, source_id: int, 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(
+        """
+        UPDATE sources
+        SET last_checked_at=NULL, status='new', status_msg=NULL, updated_at=NOW()
+        WHERE id=$1 AND archived_at IS NULL
+        """,
+        source_id,
+    )
+    await audit(user["id"], "source.parse_now", "source", source_id)
+    return redirect("/sources")
+
+
@app.post("/sources/{source_id}/delete")
 async def source_delete(request: Request, source_id: int, csrf_token: str = Form(...)):
    user = await get_current_user(request)
@@ -3546,6 +3589,18 @@ async def workers(request: Request):
            continue
        settings.append(item)
    settings.sort(key=setting_sort_key)
+    site_parser_url = str(setting_values.get("site_parser_url") or "").strip().rstrip("/")
+    site_parser_health = {"ok": False, "message": "URL не настроен"}
+    if site_parser_url:
+        try:
+            async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=3)) as session:
+                async with session.get(f"{site_parser_url}/health") as response:
+                    site_parser_health = {
+                        "ok": response.status == 200,
+                        "message": "Доступен" if response.status == 200 else f"HTTP {response.status}",
+                    }
+        except Exception as exc:
+            site_parser_health = {"ok": False, "message": str(exc)[:200] or "Недоступен"}
    ai_provider = str(setting_value(settings, "ai_qualifier_provider", "openrouter") or "openrouter")
    ai_model = str(setting_value(settings, "ai_qualifier_model", "") or "")
    ai_api_key = str(setting_value(settings, "ai_qualifier_api_key", "") or "")
@@ -3580,6 +3635,8 @@ async def workers(request: Request):
                for r in rows
            ],
            settings=settings,
+            site_parser_url=site_parser_url,
+            site_parser_health=site_parser_health,
            provider_options=PROVIDER_OPTIONS,
            ai_provider=ai_provider,
            ai_model=ai_model,
@@ -1,5 +1,6 @@
 from __future__ import annotations

+import json
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from typing import Any
@@ -9,6 +10,18 @@ import aiohttp
 from .constants import PLATFORM_SITE, PLATFORM_VK


+def json_object(value: Any) -> dict[str, Any]:
+    if isinstance(value, dict):
+        return dict(value)
+    if isinstance(value, str):
+        try:
+            parsed = json.loads(value)
+        except json.JSONDecodeError:
+            return {}
+        return parsed if isinstance(parsed, dict) else {}
+    return {}
+
+
@dataclass
 class SourceMedia:
    url: str
@@ -42,6 +55,12 @@ def validate_source_config(platform: str, config: dict[str, Any]) -> None:
        raise ValueError("max_items должен быть целым числом") from exc
    if not 1 <= max_items <= 100:
        raise ValueError("max_items должен быть от 1 до 100")
+    try:
+        interval_minutes = int(config.get("interval_minutes", 30))
+    except (TypeError, ValueError) as exc:
+        raise ValueError("interval_minutes должен быть целым числом") from exc
+    if not 1 <= interval_minutes <= 10080:
+        raise ValueError("interval_minutes должен быть от 1 до 10080")


 def _posted_at(value: Any) -> datetime:
@@ -72,9 +91,9 @@ class SiteParserClient:
    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 {})
+        config = json_object(source.get("settings_json"))
        validate_source_config(PLATFORM_SITE, config)
-        runtime_state = dict(source.get("runtime_state_json") or {})
+        runtime_state = json_object(source.get("runtime_state_json"))
        payload = {
            "url": source["url"],
            "config": config,
@@ -42,6 +42,8 @@
        </div>

        <div id="site-config" class="form-control md:col-span-2">
+          <label class="label"><span class="label-text font-bold">Проверять каждые, минут</span></label>
+          <input name="interval_minutes" type="number" min="1" max="10080" value="{{ source.settings_json.get('interval_minutes', site_default_interval) if source and source.settings_json else site_default_interval }}" class="input input-bordered w-full max-w-xs mb-4">
          <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">
@@ -49,7 +51,8 @@
            <pre class="mt-2 p-3 bg-base-200 overflow-x-auto">{
  "format": "rss",
  "access": "auto",
-  "max_items": 20
+  "max_items": 20,
+  "interval_minutes": 30
 }</pre>
            <p class="mt-2"><code>access</code>: <code>auto</code> сначала пробует обычный запрос и при Cloudflare использует RuCaptcha; <code>http</code> запрещает браузер; <code>cloudflare</code> сразу запускает браузер.</p>
          </details>
@@ -72,7 +75,13 @@
 <script>
  const platform = document.querySelector('[name="platform"]');
  const siteConfig = document.getElementById('site-config');
-  const syncConfig = () => siteConfig.hidden = platform.value !== 'site';
+  const configInput = document.querySelector('[name="settings_json"]');
+  const syncConfig = () => {
+    siteConfig.hidden = platform.value !== 'site';
+    if (platform.value === 'site' && configInput.value.trim() === '{}') {
+      configInput.value = '{\n  "format": "rss",\n  "access": "auto",\n  "max_items": 20\n}';
+    }
+  };
  platform.addEventListener('change', syncConfig);
  syncConfig();
 </script>
@@ -7,6 +7,7 @@
    </div>
    <div class="sm:hidden text-[10px] font-mono text-app-textMuted mt-1">#{{ s.id }}</div>
    <div class="text-[10px] font-mono text-app-textMuted mt-1 truncate" title="External ID">{{ s.external_id or "" }}</div>
+    {% if s.platform == "site" %}<div class="text-[10px] text-app-textMuted mt-1">Каждые {{ s.settings_json.get('interval_minutes', 30) }} мин.</div>{% endif %}
    <div class="sm:hidden text-[10px] font-mono text-blue-400 mt-1 truncate">#{{ s.tag or "—" }}</div>
    <a href="{{ s.url }}" target="_blank" class="lg:hidden text-[10px] text-app-textMuted hover:text-white hover:underline truncate block mt-1" title="{{ s.url }}">{{ s.url }}</a>
  </td>
@@ -39,6 +40,12 @@
  <td class="hidden xl:table-cell px-4 py-3 text-right text-[10px] text-app-textMuted">{{ s.last_parsed_at_fmt or "Никогда" }}</td>
  <td class="px-3 sm:px-4 py-3 text-center">
    <div class="flex justify-center items-center gap-1">
+      <form method="post" action="/sources/{{ s.id }}/parse-now" class="m-0">
+        <input type="hidden" name="csrf_token" value="{{ user.csrf_token }}">
+        <button class="btn btn-ghost btn-icon w-8 h-8 rounded-md hover:bg-app-surface text-app-textMuted hover:text-white" type="submit" title="Парсить сейчас">
+          <i data-lucide="refresh-cw" class="w-4 h-4"></i>
+        </button>
+      </form>
      <a href="/sources/{{ s.id }}/edit" class="btn btn-ghost btn-icon w-8 h-8 rounded-md hover:bg-app-surface text-app-textMuted hover:text-white" title="Править">
        <i data-lucide="edit-2" class="w-4 h-4"></i>
      </a>
@@ -1,11 +1,14 @@
 {% extends "base.html" %}
 {% block body %}
-<div class="mb-8">
-  <h1 class="text-3xl font-bold tracking-tight text-white flex items-center gap-3 mb-2">
-    <i data-lucide="database" class="text-app-primary w-8 h-8"></i>
-    Источники
-  </h1>
-  <div class="text-app-textMuted text-sm">Источники для парсинга. Название идёт в prompt, тэг — в будущие хэштеги.</div>
+<div class="mb-8 flex flex-wrap items-start justify-between gap-4">
+  <div>
+    <h1 class="text-3xl font-bold tracking-tight text-white flex items-center gap-3 mb-2">
+      <i data-lucide="database" class="text-app-primary w-8 h-8"></i>
+      Источники
+    </h1>
+    <div class="text-app-textMuted text-sm">Источники для парсинга. Название идёт в prompt, тэг — в будущие хэштеги.</div>
+  </div>
+  <a class="btn btn-primary" href="/sources/new"><i data-lucide="plus" class="w-4 h-4"></i> Добавить источник</a>
 </div>

 <details class="card mb-8 group/details" {% if source_preview %}open{% endif %}>
@@ -14,7 +17,7 @@
      <div class="w-8 h-8 rounded-lg bg-app-primary/20 text-app-primary flex items-center justify-center">
        <i data-lucide="plus" class="w-5 h-5"></i>
      </div>
-      <span class="text-lg font-bold text-white">Добавить источники</span>
+      <span class="text-lg font-bold text-white">Добавить VK списком</span>
    </div>
    <i data-lucide="chevron-down" class="w-5 h-5 text-app-textMuted transition-transform group-open/details:rotate-180"></i>
  </summary>
@@ -8,6 +8,18 @@
  <div class="text-app-textMuted text-sm">Управление фоновыми процессами, категориями, расписанием и настройками AI.</div>
 </div>

+<div class="card mb-8 p-4 flex flex-wrap items-center justify-between gap-4">
+  <div class="flex items-center gap-3 min-w-0">
+    <i data-lucide="rss" class="w-5 h-5 text-app-primary flex-shrink-0"></i>
+    <div class="min-w-0">
+      <div class="font-bold text-white">Site Parser</div>
+      <div class="text-xs text-app-textMuted truncate">{{ site_parser_url or "URL не настроен" }}</div>
+    </div>
+    <span class="badge {% if site_parser_health.ok %}badge-success{% else %}badge-error{% endif %}">{{ site_parser_health.message }}</span>
+  </div>
+  <a href="#site-parser-settings" class="btn btn-primary-outline btn-sm"><i data-lucide="settings" class="w-4 h-4"></i> Настройки</a>
+</div>
+
 <!-- Workers Panel -->
 <div class="card overflow-hidden mb-8">
  <div class="overflow-x-auto">
@@ -25,7 +37,7 @@
      <tbody class="divide-y divide-app-border text-xs">
        {% for w in workers %}
        <tr class="hover:bg-app-surfaceHover transition-colors">
-          <td class="px-4 py-3 font-mono font-bold text-white">{{ w.name }}</td>
+          <td class="px-4 py-3 font-mono font-bold text-white">{{ "Парсер источников (VK + сайты)" if w.name == "vk-parser" else w.name }}</td>
          <td class="px-4 py-3">
            {% if w.effective_enabled %}
            <span class="badge badge-success px-2 py-0.5 flex items-center gap-1.5 w-max">
@@ -294,7 +306,7 @@

 <div class="flex flex-col gap-6">
  {% for category, rows in settings|groupby("category") %}
-  <details class="card group/details" data-workers-details="settings:{{ category }}">
+  <details id="{% if category == 'Site Parser' %}site-parser-settings{% else %}settings-{{ category|lower|replace(' ', '-') }}{% endif %}" class="card group/details" data-workers-details="settings:{{ category }}" {% if category == "Site Parser" %}open{% endif %}>
    <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>
@@ -404,6 +416,10 @@
  const scrollKey = "workers-scroll-y";
  const detailsKey = "workers-open-details";
  const detailItems = Array.from(document.querySelectorAll("details[data-workers-details]"));
+  const siteParserDetails = document.getElementById("site-parser-settings");
+  document.querySelector('a[href="#site-parser-settings"]')?.addEventListener("click", () => {
+    if (siteParserDetails) siteParserDetails.open = true;
+  });
  const savedDetails = sessionStorage.getItem(detailsKey);
  if (savedDetails) {
    try {
@@ -55,7 +55,7 @@ class VKParserWorker:
    async def init(self) -> None:
        self.pool = await get_pool()

-    async def active_sources(self) -> list[dict]:
+    async def active_sources(self, vk_interval_sec: int, site_default_interval_minutes: int) -> list[dict]:
        rows = await self.pool.fetch(
            """
            SELECT *
@@ -63,9 +63,27 @@ class VKParserWorker:
            WHERE platform=ANY($1::text[])
              AND active=TRUE
              AND archived_at IS NULL
+              AND (
+                last_checked_at IS NULL
+                OR (platform=$2 AND last_checked_at <= NOW() - $3::double precision * INTERVAL '1 second')
+                OR (
+                  platform=$4
+                  AND last_checked_at <= NOW() - (
+                    CASE
+                      WHEN settings_json->>'interval_minutes' ~ '^[0-9]+$'
+                      THEN (settings_json->>'interval_minutes')::double precision
+                      ELSE $5::double precision
+                    END
+                  ) * INTERVAL '1 minute'
+                )
+              )
            ORDER BY last_checked_at NULLS FIRST, priority ASC, id ASC
            """,
            [PLATFORM_VK, PLATFORM_SITE],
+            PLATFORM_VK,
+            vk_interval_sec,
+            PLATFORM_SITE,
+            site_default_interval_minutes,
        )
        return [dict(r) for r in rows]

@@ -512,6 +530,8 @@ class VKParserWorker:
            await self.heartbeat.beat(self.pool, status="disabled", force=True)
            return

+        parser_interval_sec = max(10, await fetch_int_setting("parser_interval_sec", 300))
+        site_default_interval = max(1, await fetch_int_setting("site_parser_default_interval_minutes", 30))
        rps = await fetch_int_setting("vk_requests_per_second", 3)
        timeout_total = await fetch_int_setting("vk_api_timeout_total_sec", 15)
        timeout_connect = await fetch_int_setting("vk_api_timeout_connect_sec", 5)
@@ -520,10 +540,10 @@ class VKParserWorker:
        retry_min_delay = await fetch_float_setting("vk_api_retry_min_delay_sec", 2.0)
        retry_max_delay = await fetch_float_setting("vk_api_retry_max_delay_sec", 10.0)
        source_pause = max(0.0, await fetch_float_setting("parser_source_pause_sec", 0.0))
-        sources = await self.active_sources()
+        sources = await self.active_sources(parser_interval_sec, site_default_interval)
        await self.heartbeat.beat(self.pool, meta={"sources": len(sources)})
        if not sources:
-            logger.info("No active sources")
+            logger.debug("No sources due for parsing")
            return

        logger.info(
@@ -592,8 +612,7 @@ class VKParserWorker:
                await self.run_once()
            except Exception as e:
                logger.exception("Parser loop error: {}", e)
-            interval = max(10, await fetch_int_setting("parser_interval_sec", 300))
-            await asyncio.sleep(interval)
+            await asyncio.sleep(10)


 async def main() -> None:
@@ -38,8 +38,8 @@ class SourceAdapterTests(unittest.IsolatedAsyncioTestCase):
        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": {},
+            "settings_json": '{"format":"rss","access":"auto","interval_minutes":30}',
+            "runtime_state_json": '{}',
        })

        self.assertEqual(items[0].text, "Title\n\nBody")