init

app/calling.py

@@ -0,0 +1,110 @@
+import asyncio
+import logging
+import os
+
+from pyrogram import Client
+from pytgcalls import PyTgCalls, filters as pfl
+from pytgcalls.exceptions import CallBusy, CallDeclined, CallDiscarded, TimedOutAnswer
+from pytgcalls.types import CallConfig, MediaStream, StreamEnded
+
+from app.config import settings
+from app.tts import synthesize
+
+logger = logging.getLogger("callsvc.calling")
+
+pyro_client = Client(
+    settings.session_name,
+    api_id=settings.api_id,
+    api_hash=settings.api_hash,
+    phone_number=settings.phone_number,
+    workdir=settings.sessions_dir,
+)
+
+call_py = PyTgCalls(pyro_client)
+
+# per-цель "stream finished" события, заполняются on_stream_end хендлером
+_stream_finished: dict[int, asyncio.Event] = {}
+# один вызов на цель одновременно, чтобы не пересекались звонки
+_locks: dict[str, asyncio.Lock] = {}
+
+
+def _lock_for(key: str) -> asyncio.Lock:
+    if key not in _locks:
+        _locks[key] = asyncio.Lock()
+    return _locks[key]
+
+
+@call_py.on_update(pfl.stream_end)
+async def _on_stream_end(_: PyTgCalls, update: StreamEnded):
+    event = _stream_finished.get(update.chat_id)
+    if event is not None:
+        event.set()
+
+
+async def start():
+    await pyro_client.start()
+    await call_py.start()
+    logger.info("Telegram user client + PyTgCalls started")
+
+
+async def stop():
+    # PyTgCalls не имеет отдельного stop() — он поднимает/использует тот же
+    # pyrogram-клиент, поэтому останавливаем только клиент.
+    await pyro_client.stop()
+    logger.info("Telegram user client + PyTgCalls stopped")
+
+
+async def make_call(text: str, target: int | str | None = None) -> dict:
+    """Звонит пользователю настоящим p2p-звонком (с гудком и ожиданием ответа)
+    и озвучивает текст (TTS) сразу после того, как собеседник ответил.
+    """
+    target = target if target is not None else settings.call_target
+    lock = _lock_for(str(target))
+
+    if lock.locked():
+        raise RuntimeError(f"Call to {target} is already in progress")
+
+    async with lock:
+        audio_path, duration = synthesize(text)
+        resolved_id = await call_py.resolve_chat_id(target)
+        if resolved_id <= 0:
+            raise RuntimeError(
+                f"{target} resolves to a group/channel id ({resolved_id}). "
+                "This service only makes direct p2p calls to users."
+            )
+        event = asyncio.Event()
+        _stream_finished[resolved_id] = event
+
+        try:
+            config = CallConfig(timeout=settings.call_ring_timeout)
+            try:
+                logger.info("Calling %s (resolved id %s)...", target, resolved_id)
+                await call_py.play(resolved_id, MediaStream(audio_path), config=config)
+            except TimedOutAnswer as exc:
+                raise RuntimeError(f"{target} did not answer the call in time") from exc
+            except CallDeclined as exc:
+                raise RuntimeError(f"{target} declined the call") from exc
+            except CallBusy as exc:
+                raise RuntimeError(f"{target} is busy") from exc
+            except CallDiscarded as exc:
+                raise RuntimeError(f"Call to {target} was discarded") from exc
+
+            logger.info("Call answered, playing TTS (%.1fs) to %s", duration, target)
+
+            # ждём сигнал о завершении стрима, но не дольше duration + запас
+            timeout = duration + 10
+            try:
+                await asyncio.wait_for(event.wait(), timeout=timeout)
+            except asyncio.TimeoutError:
+                logger.warning(
+                    "Stream end event not received for %s within %.1fs, hanging up anyway",
+                    target,
+                    timeout,
+                )
+
+            await call_py.leave_call(resolved_id)
+            return {"target": target, "duration": duration}
+        finally:
+            _stream_finished.pop(resolved_id, None)
+            if os.path.exists(audio_path):
+                os.remove(audio_path)