From 5417737c08a9edb7fe5217fd9798467203a392e3 Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 10 Aug 2026 21:11:43 +0500 Subject: [PATCH] fix: complete site parser source workflow --- db/migrations/047_site_parser_schedule.sql | 10 ++++ src/vk_parser_app/admin.py | 63 +++++++++++++++++++- src/vk_parser_app/source_adapters.py | 23 ++++++- src/vk_parser_app/templates/source_form.html | 13 +++- src/vk_parser_app/templates/source_row.html | 7 +++ src/vk_parser_app/templates/sources.html | 17 +++--- src/vk_parser_app/templates/workers.html | 20 ++++++- src/vk_parser_app/workers/parser.py | 29 +++++++-- tests/test_source_adapters.py | 4 +- 9 files changed, 163 insertions(+), 23 deletions(-) create mode 100644 db/migrations/047_site_parser_schedule.sql diff --git a/db/migrations/047_site_parser_schedule.sql b/db/migrations/047_site_parser_schedule.sql new file mode 100644 index 0000000..72af16d --- /dev/null +++ b/db/migrations/047_site_parser_schedule.sql @@ -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; diff --git a/src/vk_parser_app/admin.py b/src/vk_parser_app/admin.py index 83c2be5..33bac20 100644 --- a/src/vk_parser_app/admin.py +++ b/src/vk_parser_app/admin.py @@ -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, diff --git a/src/vk_parser_app/source_adapters.py b/src/vk_parser_app/source_adapters.py index 87690e6..09c132d 100644 --- a/src/vk_parser_app/source_adapters.py +++ b/src/vk_parser_app/source_adapters.py @@ -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, diff --git a/src/vk_parser_app/templates/source_form.html b/src/vk_parser_app/templates/source_form.html index d6bc697..f35698a 100644 --- a/src/vk_parser_app/templates/source_form.html +++ b/src/vk_parser_app/templates/source_form.html @@ -42,6 +42,8 @@
+ +
@@ -49,7 +51,8 @@ 
{
   "format": "rss",
   "access": "auto",
-  "max_items": 20
+  "max_items": 20,
+  "interval_minutes": 30
 }

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

@@ -72,7 +75,13 @@ diff --git a/src/vk_parser_app/templates/source_row.html b/src/vk_parser_app/templates/source_row.html index 16597dc..8d16a62 100644 --- a/src/vk_parser_app/templates/source_row.html +++ b/src/vk_parser_app/templates/source_row.html @@ -7,6 +7,7 @@
#{{ s.id }}
{{ s.external_id or "" }}
+ {% if s.platform == "site" %}
Каждые {{ s.settings_json.get('interval_minutes', 30) }} мин.
{% endif %}
#{{ s.tag or "—" }}
{{ s.url }} @@ -39,6 +40,12 @@ {{ s.last_parsed_at_fmt or "Никогда" }}
+
+ + +
diff --git a/src/vk_parser_app/templates/sources.html b/src/vk_parser_app/templates/sources.html index c991e14..0c1fb53 100644 --- a/src/vk_parser_app/templates/sources.html +++ b/src/vk_parser_app/templates/sources.html @@ -1,11 +1,14 @@ {% extends "base.html" %} {% block body %} -
-

- - Источники -

-
Источники для парсинга. Название идёт в prompt, тэг — в будущие хэштеги.
+
+
+

+ + Источники +

+
Источники для парсинга. Название идёт в prompt, тэг — в будущие хэштеги.
+
+ Добавить источник
@@ -14,7 +17,7 @@
- Добавить источники + Добавить VK списком
diff --git a/src/vk_parser_app/templates/workers.html b/src/vk_parser_app/templates/workers.html index 93c0952..b240dbb 100644 --- a/src/vk_parser_app/templates/workers.html +++ b/src/vk_parser_app/templates/workers.html @@ -8,6 +8,18 @@
Управление фоновыми процессами, категориями, расписанием и настройками AI.
+
+
+ +
+
Site Parser
+
{{ site_parser_url or "URL не настроен" }}
+
+ {{ site_parser_health.message }} +
+ Настройки +
+
@@ -25,7 +37,7 @@ {% for w in workers %} - {{ w.name }} + {{ "Парсер источников (VK + сайты)" if w.name == "vk-parser" else w.name }} {% if w.effective_enabled %} @@ -294,7 +306,7 @@
{% for category, rows in settings|groupby("category") %} -
+
{{ category_titles.get(category, category) }}
@@ -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 { diff --git a/src/vk_parser_app/workers/parser.py b/src/vk_parser_app/workers/parser.py index 3ef9980..53831fd 100644 --- a/src/vk_parser_app/workers/parser.py +++ b/src/vk_parser_app/workers/parser.py @@ -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: diff --git a/tests/test_source_adapters.py b/tests/test_source_adapters.py index bc7114d..122df24 100644 --- a/tests/test_source_adapters.py +++ b/tests/test_source_adapters.py @@ -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")