diff --git a/src/services/channel_analytics_service.py b/src/services/channel_analytics_service.py index b57cc7f0..f3447ccd 100644 --- a/src/services/channel_analytics_service.py +++ b/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) diff --git a/src/services/photo_task_service.py b/src/services/photo_task_service.py index 714d2d48..deddf7a0 100644 --- a/src/services/photo_task_service.py +++ b/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 diff --git a/src/services/pipeline_nodes/handlers.py b/src/services/pipeline_nodes/handlers.py index f471daa9..6a24738e 100644 --- a/src/services/pipeline_nodes/handlers.py +++ b/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}, diff --git a/src/services/telegram_actions.py b/src/services/telegram_actions.py index aab1437f..e606e6b9 100644 --- a/src/services/telegram_actions.py +++ b/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") diff --git a/src/telegram/collector_mixins/stats.py b/src/telegram/collector_mixins/stats.py index 8459ee30..961da4fa 100644 --- a/src/telegram/collector_mixins/stats.py +++ b/src/telegram/collector_mixins/stats.py @@ -42,22 +42,39 @@ class StatsMixin: def is_stats_running(self: "Collector") -> bool: return self._stats_running or self._stats_all_running - def stats_worker_count(self: "Collector") -> int: - configured = max(1, int(getattr(self._config, "stats_worker_count", 3) or 1)) + def _configured_worker_count( + self: "Collector", attr: str, default: int + ) -> int: + """Configured worker count for an attr, clamped to connected clients. + + Shared by :meth:`stats_worker_count` / :meth:`stats_all_worker_count`: + ``max(1, min(configured, connected))`` when clients are connected, else + the configured value. + """ + configured = max(1, int(getattr(self._config, attr, default) or 1)) connected = len(getattr(self._pool, "clients", {}) or {}) if connected <= 0: return configured return max(1, min(configured, connected)) - def stats_all_worker_count(self: "Collector") -> int: - configured = max(1, int(getattr(self._config, "stats_all_worker_count", 1) or 1)) - connected = len(getattr(self._pool, "clients", {}) or {}) - if connected <= 0: - return configured - return max(1, min(configured, connected)) + def stats_worker_count(self: "Collector") -> int: + return self._configured_worker_count("stats_worker_count", 3) - async def available_stats_worker_count(self: "Collector") -> int: - configured = max(1, int(getattr(self._config, "stats_worker_count", 3) or 1)) + def stats_all_worker_count(self: "Collector") -> int: + return self._configured_worker_count("stats_all_worker_count", 1) + + async def _available_worker_count( + self: "Collector", *, attr: str, default: int, fallback: int, log_label: str + ) -> int: + """Live-available worker count for an attr, falling back to the configured value. + + Shared by :meth:`available_stats_worker_count` / + :meth:`available_stats_all_worker_count`: read + ``available_stats_client_count`` from the pool (sync or async), and when + positive return ``max(1, min(configured, available))`` (or ``1`` when + the pool reports none), else the configured ``fallback``. + """ + configured = max(1, int(getattr(self._config, attr, default) or 1)) counter = getattr(self._pool, "available_stats_client_count", None) if callable(counter): try: @@ -69,24 +86,24 @@ async def available_stats_worker_count(self: "Collector") -> int: return max(1, min(configured, available)) return 1 except Exception: - logger.debug("Failed to read available stats client count", exc_info=True) - return self.stats_worker_count() + logger.debug("Failed to read available %s client count", log_label, exc_info=True) + return fallback + + async def available_stats_worker_count(self: "Collector") -> int: + return await self._available_worker_count( + attr="stats_worker_count", + default=3, + fallback=self.stats_worker_count(), + log_label="stats", + ) async def available_stats_all_worker_count(self: "Collector") -> int: - configured = max(1, int(getattr(self._config, "stats_all_worker_count", 1) or 1)) - counter = getattr(self._pool, "available_stats_client_count", None) - if callable(counter): - try: - count = counter() - if asyncio.iscoroutine(count): - count = await count - available = int(count) - if available > 0: - return max(1, min(configured, available)) - return 1 - except Exception: - logger.debug("Failed to read available stats-all client count", exc_info=True) - return self.stats_all_worker_count() + return await self._available_worker_count( + attr="stats_all_worker_count", + default=1, + fallback=self.stats_all_worker_count(), + log_label="stats-all", + ) def set_stats_all_running(self: "Collector", running: bool) -> None: self._stats_all_running = running