refactor: dedupe service/collector helper twins (#1135)

src/services/channel_analytics_service.py

@@ -310,12 +310,19 @@ async def _get_subscriber_count(self, channel_id: int) -> int | None:
             return stats[0].subscriber_count
         return None
 
-    async def _calc_err(self, channel_id: int, last_n: int = 20) -> float | None:
+    async def _engagement_rate(
+        self,
+        channel_id: int,
+        rows: list[dict],
+    ) -> float | None:
+        """ERR (Engagement Rate by Reach), in percent, for a set of message rows.
+
+        Shared body of `_calc_err` / `_calc_err24`: ``sum(engagement) /
+        (rows * subscribers) * 100``. ``None`` when the channel has no
+        subscribers or the message window is empty.
+        """
         sub_count = await self._get_subscriber_count(channel_id)
-        if not sub_count:
-            return None
-        rows = await self._db.repos.messages.get_err_data(channel_id, last_n)
-        if not rows:
+        if not sub_count or not rows:
             return None
         total_engagement = sum(
             (r.get("views") or 0)
@@ -326,18 +333,10 @@ async def _calc_err(self, channel_id: int, last_n: int = 20) -> float | None:
         )
         return round(total_engagement / (len(rows) * sub_count) * 100, 2)
 
+    async def _calc_err(self, channel_id: int, last_n: int = 20) -> float | None:
+        rows = await self._db.repos.messages.get_err_data(channel_id, last_n)
+        return await self._engagement_rate(channel_id, rows)
+
     async def _calc_err24(self, channel_id: int) -> float | None:
-        sub_count = await self._get_subscriber_count(channel_id)
-        if not sub_count:
-            return None
         rows = await self._db.repos.messages.get_err24_data(channel_id)
-        if not rows:
-            return None
-        total_engagement = sum(
-            (r.get("views") or 0)
-            + (r.get("forwards") or 0)
-            + (r.get("reply_count") or 0)
-            + r.get("total_reactions", 0)
-            for r in rows
-        )
-        return round(total_engagement / (len(rows) * sub_count) * 100, 2)
+        return await self._engagement_rate(channel_id, rows)

src/services/photo_task_service.py

@@ -50,17 +50,23 @@ def normalize_mode(mode: str | PhotoSendMode, files_count: int) -> PhotoSendMode
             return PhotoSendMode.SEPARATE
         return send_mode
 
-    async def send_now(
+    async def _create_send_batch_item(
         self,
         *,
         phone: str,
         target: PhotoTarget,
-        file_paths: list[str],
-        mode: str | PhotoSendMode,
-        caption: str | None = None,
-    ) -> PhotoBatchItem:
-        files = self.validate_files(file_paths)
-        send_mode = self.normalize_mode(mode, len(files))
+        files: list[str],
+        send_mode: PhotoSendMode,
+        caption: str | None,
+        status: PhotoBatchStatus,
+        schedule_at: datetime | None = None,
+    ) -> tuple[int, int]:
+        """Create the PhotoBatch + its first PhotoBatchItem for an ad-hoc send.
+
+        Shared by :meth:`send_now` (status=RUNNING, stamps ``started_at``) and
+        :meth:`schedule_send` (status=SCHEDULED, stamps ``schedule_at``). Returns
+        ``(batch_id, item_id)``.
+        """
         batch_id = await self._bundle.create_batch(
             PhotoBatch(
                 phone=phone,
@@ -69,22 +75,45 @@ async def send_now(
                 target_type=target.target_type,
                 send_mode=send_mode,
                 caption=caption,
-                status=PhotoBatchStatus.RUNNING,
+                status=status,
             )
         )
-        item_id = await self._bundle.create_item(
-            PhotoBatchItem(
-                batch_id=batch_id,
-                phone=phone,
-                target_dialog_id=target.dialog_id,
-                target_title=target.title,
-                target_type=target.target_type,
-                file_paths=files,
-                send_mode=send_mode,
-                caption=caption,
-                status=PhotoBatchStatus.RUNNING,
-                started_at=datetime.now(timezone.utc),
-            )
+        item_kwargs: dict = {
+            "batch_id": batch_id,
+            "phone": phone,
+            "target_dialog_id": target.dialog_id,
+            "target_title": target.title,
+            "target_type": target.target_type,
+            "file_paths": files,
+            "send_mode": send_mode,
+            "caption": caption,
+            "status": status,
+        }
+        if status == PhotoBatchStatus.RUNNING:
+            item_kwargs["started_at"] = datetime.now(timezone.utc)
+        elif schedule_at is not None:
+            item_kwargs["schedule_at"] = schedule_at
+        item_id = await self._bundle.create_item(PhotoBatchItem(**item_kwargs))
+        return batch_id, item_id
+
+    async def send_now(
+        self,
+        *,
+        phone: str,
+        target: PhotoTarget,
+        file_paths: list[str],
+        mode: str | PhotoSendMode,
+        caption: str | None = None,
+    ) -> PhotoBatchItem:
+        files = self.validate_files(file_paths)
+        send_mode = self.normalize_mode(mode, len(files))
+        batch_id, item_id = await self._create_send_batch_item(
+            phone=phone,
+            target=target,
+            files=files,
+            send_mode=send_mode,
+            caption=caption,
+            status=PhotoBatchStatus.RUNNING,
         )
         # Accumulate ids of files already published live on Telegram. In SEPARATE
         # mode send_now publishes file-by-file immediately, so if a later file
@@ -181,30 +210,14 @@ async def schedule_send(
     ) -> PhotoBatchItem:
         files = self.validate_files(file_paths)
         send_mode = self.normalize_mode(mode, len(files))
-        batch_id = await self._bundle.create_batch(
-            PhotoBatch(
-                phone=phone,
-                target_dialog_id=target.dialog_id,
-                target_title=target.title,
-                target_type=target.target_type,
-                send_mode=send_mode,
-                caption=caption,
-                status=PhotoBatchStatus.SCHEDULED,
-            )
-        )
-        item_id = await self._bundle.create_item(
-            PhotoBatchItem(
-                batch_id=batch_id,
-                phone=phone,
-                target_dialog_id=target.dialog_id,
-                target_title=target.title,
-                target_type=target.target_type,
-                file_paths=files,
-                send_mode=send_mode,
-                caption=caption,
-                schedule_at=schedule_at,
-                status=PhotoBatchStatus.SCHEDULED,
-            )
+        batch_id, item_id = await self._create_send_batch_item(
+            phone=phone,
+            target=target,
+            files=files,
+            send_mode=send_mode,
+            caption=caption,
+            status=PhotoBatchStatus.SCHEDULED,
+            schedule_at=schedule_at,
         )
         # Accumulate server-scheduled message ids progressively. In SEPARATE mode
         # send_now schedules files one-by-one; if a later file fails, the earlier

src/services/pipeline_nodes/handlers.py

@@ -32,6 +32,26 @@ class ReactionInvalidError(Exception):  # type: ignore[no-redef]
 logger = logging.getLogger(__name__)
 
 
+def _build_source_messages(messages: list) -> str:
+    """Render the ``context_messages`` list into the ``[header] text`` block fed to LLM/agent nodes.
+
+    Shared by the LLM-generate and agent-loop handlers — identical rendering of
+    each message (``[channel_title|username] stripped_text (id:.. date:..)``),
+    blank-separated. Empty/whitespace texts are dropped.
+    """
+    from datetime import datetime
+
+    source_parts: list[str] = []
+    for m in messages:
+        text = (m.text or "").strip()
+        if not text:
+            continue
+        header = m.channel_title or m.channel_username or ""
+        when = m.date.isoformat() if isinstance(m.date, datetime) else str(m.date)
+        source_parts.append(f"[{header}] {text} (id:{m.message_id} date:{when})")
+    return "\n\n".join(source_parts)
+
+
 def _current_node_id(services: dict, default: str = "?") -> str:
     """Return the node id for the currently-executing handler.
 
@@ -186,8 +206,6 @@ async def execute(self, node_config: dict, context: NodeContext, services: dict)
         if provider_callable is None:
             raise RuntimeError("LlmGenerateHandler: no provider_callable in services")
 
-        from datetime import datetime
-
         from src.agent.prompt_template import render_prompt_template
 
         prompt_template = node_config.get("prompt_template") or context.get_global("prompt_template", "")
@@ -197,15 +215,7 @@ async def execute(self, node_config: dict, context: NodeContext, services: dict)
 
         # Build source messages string from context
         messages = context.get_global("context_messages", [])
-        source_parts = []
-        for m in messages:
-            text = (m.text or "").strip()
-            if not text:
-                continue
-            header = m.channel_title or m.channel_username or ""
-            when = m.date.isoformat() if isinstance(m.date, datetime) else str(m.date)
-            source_parts.append(f"[{header}] {text} (id:{m.message_id} date:{when})")
-        source_messages = "\n\n".join(source_parts)
+        source_messages = _build_source_messages(messages)
 
         rendered = render_prompt_template(
             prompt_template,
@@ -842,8 +852,6 @@ async def execute(self, node_config: dict, context: NodeContext, services: dict)
         if provider_callable is None:
             raise RuntimeError("AgentLoopHandler: no provider_callable in services")
 
-        from datetime import datetime
-
         system_prompt = node_config.get("system_prompt", "Ты полезный ассистент.")
         model = node_config.get("model") or services.get("default_model") or ""
         max_tokens = int(node_config.get("max_tokens", 2000))
@@ -870,18 +878,11 @@ async def execute(self, node_config: dict, context: NodeContext, services: dict)
         full_system = system_prompt + tool_desc + self._REACT_SUFFIX
 
         # Build source messages string from context
-        messages = context.get_global("context_messages", [])
-        source_parts = []
-        for m in messages:
-            text = (m.text or "").strip()
-            if not text:
-                continue
-            header = m.channel_title or m.channel_username or ""
-            when = m.date.isoformat() if isinstance(m.date, datetime) else str(m.date)
-            source_parts.append(f"[{header}] {text} (id:{m.message_id} date:{when})")
-        source_messages = "\n\n".join(source_parts)
+        source_messages = _build_source_messages(context.get_global("context_messages", []))
 
-        user_message = f"Сообщения для анализа:\n\n{source_messages}" if source_parts else "Нет сообщений для анализа."
+        user_message = (
+            f"Сообщения для анализа:\n\n{source_messages}" if source_messages else "Нет сообщений для анализа."
+        )
 
         conversation = [
             {"role": "system", "content": full_system},

src/services/telegram_actions.py

@@ -619,6 +619,40 @@ async def kick_participant(
             await client.kick_participant(entity, user)
             return TelegramActionResult(phone=acquired_phone)
 
+    async def _fetch_message_with_flood_wait(
+        self,
+        client,
+        acquired_phone: str,
+        entity,
+        message_id: int,
+        *,
+        operation_prefix: str,
+    ):
+        """Resolve a single message by id under ``run_with_flood_wait``, or raise.
+
+        Shared body of :meth:`download_media` / :meth:`download_media_sized`:
+        fetch the one message via ``iter_messages(ids=...)`` under the flood-wait
+        wrapper, then raise :class:`TelegramActionMessageNotFoundError` when the
+        id no longer exists. Returns the resolved message object.
+        """
+        message = None
+
+        async def _lookup_message() -> None:
+            nonlocal message
+            async for current_message in client.iter_messages(entity, ids=int(message_id)):
+                message = current_message
+                break
+
+        await run_with_flood_wait(
+            _lookup_message(),
+            operation=f"{operation_prefix}_lookup",
+            phone=acquired_phone,
+            pool=self._pool,
+        )
+        if message is None:
+            raise TelegramActionMessageNotFoundError("message_not_found")
+        return message
+
     async def download_media(
         self,
         *,
@@ -631,22 +665,9 @@ async def download_media(
         output_resolved = await asyncio.to_thread(_ensure_resolved_dir, output_dir)
         async with self._client(phone=phone, native=True) as (client, acquired_phone):
             entity = await self._resolve_entity(client, phone=acquired_phone, identifier=chat_id)
-            message = None
-
-            async def _lookup_message() -> None:
-                nonlocal message
-                async for current_message in client.iter_messages(entity, ids=int(message_id)):
-                    message = current_message
-                    break
-
-            await run_with_flood_wait(
-                _lookup_message(),
-                operation=f"{operation_prefix}_lookup",
-                phone=acquired_phone,
-                pool=self._pool,
+            message = await self._fetch_message_with_flood_wait(
+                client, acquired_phone, entity, message_id, operation_prefix=operation_prefix
             )
-            if message is None:
-                raise TelegramActionMessageNotFoundError("message_not_found")
             path = await run_with_flood_wait(
                 client.download_media(message, file=str(output_resolved)),
                 operation=operation_prefix,
@@ -679,22 +700,9 @@ async def download_media_sized(
         root_resolved = await asyncio.to_thread(_ensure_resolved_dir, output_dir)
         async with self._client(phone=phone, native=True) as (client, acquired_phone):
             entity = await self._resolve_entity(client, phone=acquired_phone, identifier=chat_id)
-            message = None
-
-            async def _lookup_message() -> None:
-                nonlocal message
-                async for current_message in client.iter_messages(entity, ids=int(message_id)):
-                    message = current_message
-                    break
-
-            await run_with_flood_wait(
-                _lookup_message(),
-                operation=f"{operation_prefix}_lookup",
-                phone=acquired_phone,
-                pool=self._pool,
+            message = await self._fetch_message_with_flood_wait(
+                client, acquired_phone, entity, message_id, operation_prefix=operation_prefix
             )
-            if message is None:
-                raise TelegramActionMessageNotFoundError("message_not_found")
             if getattr(message, "media", None) is None:
                 raise TelegramActionNoMediaError("no_media")