Update: Генерация отклика через threading

SergNik38 · SergNik38 · commit 3649a9d1ec51 · 2026-05-27T13:19:50.000+08:00
diff --git a/.gitignore b/.gitignore
@@ -1,4 +1,35 @@
+# Python
 __pycache__/
 *.pyc
+*.pyo
+*.pyd
+*.egg-info/
+dist/
+build/
+.venv/
+venv/
+
+# Env & secrets
 .env
+.env.*
+!.env.example
+
+# Project data
 data/
+*.db
+*.sqlite
+*.sqlite3
+
+# RuFlo / claude-flow runtime
+.claude-flow/
+.swarm/
+.claude/memory.db
+
+# Logs & temp
+*.log
+tmp/
+temp/
+
+# OS
+.DS_Store
+Thumbs.db
diff --git a/kwork_parser/app.py b/kwork_parser/app.py
@@ -2,6 +2,7 @@
 
 import html
 import logging
+import threading
 import time
 from datetime import datetime
 
@@ -22,6 +23,9 @@
 logger = logging.getLogger(__name__)
 
 
+_MAX_CONCURRENT_DRAFTS = 3
+
+
 class Application:
     def __init__(self, settings: Settings) -> None:
         self.settings = settings
@@ -31,6 +35,7 @@ def __init__(self, settings: Settings) -> None:
         self.ai_scorer = OpenRouterScorer(settings) if settings.ai_enabled else None
         self.response_draft_generator = ResponseDraftService(settings) if settings.response_draft_enabled else None
         self.notifier = TelegramNotifier(settings)
+        self._draft_semaphore = threading.Semaphore(_MAX_CONCURRENT_DRAFTS)
 
     def run_once(self) -> int:
         self._sync_telegram_feedback()
@@ -295,26 +300,41 @@ def _handle_draft_action(self, action: TelegramDraftAction) -> None:
                     self.notifier.answer_feedback(action.callback_query_id, "Для этого заказа пока не хватает данных на демо")
                     return
                 self.notifier.answer_feedback(action.callback_query_id, "Готовлю демо...")
-                generated, error_text = self._send_demo_project(
-                    candidate.project,
-                    candidate.rule_result,
-                    candidate.ai_result,
-                    demo_summary=draft.demo_summary,
-                    chat_id=action.chat_id,
-                )
-                if not generated:
-                    try:
-                        self.notifier.send_demo_status(candidate.project, error_text, chat_id=action.chat_id)
-                    except Exception:
-                        logger.warning(
-                            "Failed to send demo error status for project %s",
-                            action.project_id,
-                            exc_info=True,
-                        )
+                threading.Thread(
+                    target=self._run_demo_in_background,
+                    args=(action, candidate, draft),
+                    daemon=True,
+                ).start()
                 return
 
             self.notifier.answer_feedback(action.callback_query_id, "Генерирую отклик...")
             variant = "default" if action.action in {"generate", "regenerate"} else action.action
+            threading.Thread(
+                target=self._run_draft_in_background,
+                args=(action, candidate, variant),
+                daemon=True,
+            ).start()
+        except Exception:
+            logger.warning("Draft action handling failed for project %s", action.project_id, exc_info=True)
+
+    def _run_draft_in_background(
+        self,
+        action: TelegramDraftAction,
+        candidate: object,
+        variant: str,
+    ) -> None:
+        if not self._draft_semaphore.acquire(blocking=False):
+            logger.warning("Draft semaphore full, skipping generation for project %s", action.project_id)
+            try:
+                self.notifier.send_demo_status(
+                    candidate.project,
+                    "Слишком много одновременных запросов, попробуй чуть позже",
+                    chat_id=action.chat_id,
+                )
+            except Exception:
+                pass
+            return
+        try:
             self._send_response_draft(
                 candidate.project,
                 candidate.rule_result,
@@ -323,7 +343,48 @@ def _handle_draft_action(self, action: TelegramDraftAction) -> None:
                 chat_id=action.chat_id,
             )
         except Exception:
-            logger.warning("Draft action handling failed for project %s", action.project_id, exc_info=True)
+            logger.warning("Background draft generation failed for project %s", action.project_id, exc_info=True)
+        finally:
+            self._draft_semaphore.release()
+
+    def _run_demo_in_background(
+        self,
+        action: TelegramDraftAction,
+        candidate: object,
+        draft: object,
+    ) -> None:
+        if not self._draft_semaphore.acquire(blocking=False):
+            logger.warning("Draft semaphore full, skipping demo for project %s", action.project_id)
+            try:
+                self.notifier.send_demo_status(
+                    candidate.project,
+                    "Слишком много одновременных запросов, попробуй чуть позже",
+                    chat_id=action.chat_id,
+                )
+            except Exception:
+                pass
+            return
+        try:
+            generated, error_text = self._send_demo_project(
+                candidate.project,
+                candidate.rule_result,
+                candidate.ai_result,
+                demo_summary=draft.demo_summary,
+                chat_id=action.chat_id,
+            )
+            if not generated:
+                try:
+                    self.notifier.send_demo_status(candidate.project, error_text, chat_id=action.chat_id)
+                except Exception:
+                    logger.warning(
+                        "Failed to send demo error status for project %s",
+                        action.project_id,
+                        exc_info=True,
+                    )
+        except Exception:
+            logger.warning("Background demo generation failed for project %s", action.project_id, exc_info=True)
+        finally:
+            self._draft_semaphore.release()
 
     def _format_health_message(self) -> str:
         snapshot = self.storage.get_health_snapshot()
diff --git a/kwork_parser/storage.py b/kwork_parser/storage.py
@@ -2,6 +2,7 @@
 
 import json
 import sqlite3
+import threading
 from dataclasses import dataclass
 from pathlib import Path
 
@@ -51,8 +52,9 @@ class Storage:
     def __init__(self, path: Path) -> None:
         self.path = path
         self.path.parent.mkdir(parents=True, exist_ok=True)
-        self.connection = sqlite3.connect(self.path)
+        self.connection = sqlite3.connect(self.path, check_same_thread=False)
         self.connection.row_factory = sqlite3.Row
+        self._lock = threading.Lock()
         self._init_schema()
 
     def _init_schema(self) -> None:
@@ -372,32 +374,33 @@ def get_health_snapshot(self) -> HealthSnapshot:
         )
 
     def save_response_draft(self, draft: ResponseDraft) -> None:
-        self.connection.execute(
-            """
-            INSERT INTO response_drafts (
-                project_id, text, variant, demo_available, demo_summary,
-                demo_path, demo_archive_path, status, updated_at
+        with self._lock:
+            self.connection.execute(
+                """
+                INSERT INTO response_drafts (
+                    project_id, text, variant, demo_available, demo_summary,
+                    demo_path, demo_archive_path, status, updated_at
+                )
+                VALUES (?, ?, ?, ?, ?, NULL, NULL, 'draft', CURRENT_TIMESTAMP)
+                ON CONFLICT(project_id) DO UPDATE SET
+                    text = excluded.text,
+                    variant = excluded.variant,
+                    demo_available = excluded.demo_available,
+                    demo_summary = excluded.demo_summary,
+                    demo_path = NULL,
+                    demo_archive_path = NULL,
+                    status = 'draft',
+                    updated_at = CURRENT_TIMESTAMP
+                """,
+                (
+                    draft.project_id,
+                    draft.text,
+                    draft.variant,
+                    1 if draft.demo_available else 0,
+                    draft.demo_summary or None,
+                ),
             )
-            VALUES (?, ?, ?, ?, ?, NULL, NULL, 'draft', CURRENT_TIMESTAMP)
-            ON CONFLICT(project_id) DO UPDATE SET
-                text = excluded.text,
-                variant = excluded.variant,
-                demo_available = excluded.demo_available,
-                demo_summary = excluded.demo_summary,
-                demo_path = NULL,
-                demo_archive_path = NULL,
-                status = 'draft',
-                updated_at = CURRENT_TIMESTAMP
-            """,
-            (
-                draft.project_id,
-                draft.text,
-                draft.variant,
-                1 if draft.demo_available else 0,
-                draft.demo_summary or None,
-            ),
-        )
-        self.connection.commit()
+            self.connection.commit()
 
     def get_response_draft(self, project_id: int) -> ResponseDraft | None:
         row = self.connection.execute(
@@ -421,17 +424,18 @@ def get_response_draft(self, project_id: int) -> ResponseDraft | None:
         )
 
     def save_demo_project_artifacts(self, project_id: int, demo_path: str, demo_archive_path: str) -> None:
-        self.connection.execute(
-            """
-            UPDATE response_drafts
-            SET demo_path = ?,
-                demo_archive_path = ?,
-                updated_at = CURRENT_TIMESTAMP
-            WHERE project_id = ?
-            """,
-            (demo_path, demo_archive_path, project_id),
-        )
-        self.connection.commit()
+        with self._lock:
+            self.connection.execute(
+                """
+                UPDATE response_drafts
+                SET demo_path = ?,
+                    demo_archive_path = ?,
+                    updated_at = CURRENT_TIMESTAMP
+                WHERE project_id = ?
+                """,
+                (demo_path, demo_archive_path, project_id),
+            )
+            self.connection.commit()
 
     def mark_response_draft_sent_manually(self, project_id: int) -> None:
         self.connection.execute(
