import asyncio asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy()) import json import logging import os import tempfile import time import uuid from concurrent.futures import ThreadPoolExecutor from pathlib import Path from threading import Lock from typing import Optional, Tuple, Dict, Callable, List, Any import av import uvicorn from fastapi import FastAPI from aiogram import Bot, Dispatcher, F from aiogram.client.session.aiohttp import AiohttpSession from aiogram.client.telegram import TelegramAPIServer from aiogram.enums import ChatAction from aiogram.filters import CommandStart from aiogram.types import ( Message, CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton, BufferedInputFile, ) from faster_whisper import WhisperModel # ============================================================ # 1. ЛОГИ # ============================================================ logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", ) logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("huggingface_hub").setLevel(logging.WARNING) logging.getLogger("faster_whisper").setLevel(logging.WARNING) logger = logging.getLogger("transcriber-bot") # ============================================================ # 2. НАСТРОЙКИ # ============================================================ BOT_TOKEN = os.getenv("BOT_TOKEN") TELEGRAM_API_PROXY = os.getenv("TELEGRAM_API_PROXY") if not BOT_TOKEN: raise RuntimeError("BOT_TOKEN не найден. Добавь BOT_TOKEN в Secrets на Hugging Face.") MODEL_NAME = "dropbox-dash/faster-whisper-large-v3-turbo" DEVICE = "cpu" COMPUTE_TYPE = "int8" LANGUAGE = "ru" MAX_FILE_SIZE_MB = 200 TELEGRAM_MESSAGE_LIMIT = 3900 MAX_WORKERS = 1 # Главный ускоритель. Было 5. Для Hugging Face CPU это слишком тяжело. # 3 обычно даёт хороший баланс качества/скорости. BEAM_SIZE = 3 # Если хочешь вернуть максимум качества ценой скорости — поставь 5. # BEAM_SIZE = 5 BEST_OF = 1 PATIENCE = 1.0 # Тайминги. MAX_TIMED_BLOCK_SECONDS = 9 FILES_DIR = Path(tempfile.gettempdir()) / "tg_transcriber_files" FILES_DIR.mkdir(parents=True, exist_ok=True) STATS_PATH = FILES_DIR / "runtime_stats.json" # Базовые оценки для Hugging Face CPU. DEFAULT_SECONDS_PER_AUDIO_SECOND = 1.25 DEFAULT_COLD_START_EXTRA_SECONDS = 90.0 INITIAL_PROMPT = ( "Русская разговорная речь. Дословная транскрибация без пересказа. " "Темы: криптовалюта, трейдинг, инвестиции, Telegram, вебинар, обучение, " "воронка, прогрев, рассылка, GetCourse, Bybit, Binance, DeFi." ) # ============================================================ # 3. FASTAPI ДЛЯ HUGGING FACE SPACE # ============================================================ app = FastAPI() @app.get("/") def health_check() -> dict: return { "status": "ok", "service": "telegram-transcriber-bot", "model": MODEL_NAME, "telegram_proxy_enabled": bool(TELEGRAM_API_PROXY), } # ============================================================ # 4. TELEGRAM-БОТ # ============================================================ if TELEGRAM_API_PROXY: logger.info("Using Telegram API proxy: %s", TELEGRAM_API_PROXY) api_server = TelegramAPIServer( base="{}/bot{{token}}/{{method}}".format(TELEGRAM_API_PROXY.rstrip("/")), file="{}/file/bot{{token}}/{{path}}".format(TELEGRAM_API_PROXY.rstrip("/")), ) session = AiohttpSession(api=api_server) bot = Bot(token=BOT_TOKEN, session=session) else: logger.warning("TELEGRAM_API_PROXY is not set. Using direct Telegram API.") bot = Bot(token=BOT_TOKEN) dp = Dispatcher() executor = ThreadPoolExecutor(max_workers=MAX_WORKERS) model: Optional[WhisperModel] = None model_lock = Lock() user_files: Dict[int, Dict[str, Any]] = {} class UserVisibleError(Exception): pass # ============================================================ # 5. UI # ============================================================ def format_choice_keyboard() -> InlineKeyboardMarkup: return InlineKeyboardMarkup( inline_keyboard=[ [ InlineKeyboardButton( text="Без таймингов", callback_data="choice_plain", ) ], [ InlineKeyboardButton( text="С таймингами", callback_data="choice_timed", ) ], ] ) def split_choice_keyboard() -> InlineKeyboardMarkup: return InlineKeyboardMarkup( inline_keyboard=[ [ InlineKeyboardButton( text="1 часть", callback_data="split_1", ) ], [ InlineKeyboardButton( text="2 части", callback_data="split_2", ) ], [ InlineKeyboardButton( text="3 части", callback_data="split_3", ) ], [ InlineKeyboardButton( text="4 части", callback_data="split_4", ) ], ] ) def progress_bar(percent: int) -> str: total_blocks = 10 filled_blocks = round(percent / 100 * total_blocks) empty_blocks = total_blocks - filled_blocks return "█" * filled_blocks + "░" * empty_blocks # ============================================================ # 6. БАЗОВЫЕ ФУНКЦИИ # ============================================================ def format_seconds(seconds: float) -> str: seconds = int(max(0, seconds)) minutes, sec = divmod(seconds, 60) hours, minutes = divmod(minutes, 60) if hours: return "{:02d}:{:02d}:{:02d}".format(hours, minutes, sec) return "{:02d}:{:02d}".format(minutes, sec) def format_seconds_for_filename(seconds: float) -> str: return format_seconds(seconds).replace(":", "-") def load_runtime_stats() -> Dict[str, float]: if not STATS_PATH.exists(): return { "seconds_per_audio_second": DEFAULT_SECONDS_PER_AUDIO_SECOND, "cold_start_extra_seconds": DEFAULT_COLD_START_EXTRA_SECONDS, } try: with open(STATS_PATH, "r", encoding="utf-8") as file: data = json.load(file) return { "seconds_per_audio_second": float( data.get("seconds_per_audio_second", DEFAULT_SECONDS_PER_AUDIO_SECOND) ), "cold_start_extra_seconds": float( data.get("cold_start_extra_seconds", DEFAULT_COLD_START_EXTRA_SECONDS) ), } except Exception: return { "seconds_per_audio_second": DEFAULT_SECONDS_PER_AUDIO_SECOND, "cold_start_extra_seconds": DEFAULT_COLD_START_EXTRA_SECONDS, } def save_runtime_stats(stats: Dict[str, float]) -> None: try: with open(STATS_PATH, "w", encoding="utf-8") as file: json.dump(stats, file, ensure_ascii=False) except Exception: pass def update_runtime_stats( duration_seconds: Optional[float], elapsed_seconds: float, model_was_loaded_before_job: bool, ) -> None: if not duration_seconds or duration_seconds <= 0: return stats = load_runtime_stats() measured_factor = elapsed_seconds / duration_seconds # Если модель уже была загружена, это честный замер скорости транскрибации. # Если модель была холодная, не даём этому полностью испортить будущие оценки. if model_was_loaded_before_job: old_factor = stats["seconds_per_audio_second"] new_factor = old_factor * 0.65 + measured_factor * 0.35 stats["seconds_per_audio_second"] = max(0.3, min(new_factor, 4.0)) else: estimated_transcribe = duration_seconds * stats["seconds_per_audio_second"] cold_extra = max(0.0, elapsed_seconds - estimated_transcribe) old_cold = stats["cold_start_extra_seconds"] new_cold = old_cold * 0.6 + cold_extra * 0.4 stats["cold_start_extra_seconds"] = max(10.0, min(new_cold, 240.0)) save_runtime_stats(stats) def estimate_transcription_time( duration_seconds: Optional[float], parts_count: int, with_timestamps: bool, ) -> str: if not duration_seconds: return "зависит от файла" stats = load_runtime_stats() factor = stats["seconds_per_audio_second"] # Если нужны части или тайминги, модели нужны временные сегменты. # Это обычно чуть медленнее, чем чистый текст одним файлом. if parts_count > 1 or with_timestamps: factor += 0.18 estimated_seconds = duration_seconds * factor if model is None: estimated_seconds += stats["cold_start_extra_seconds"] else: estimated_seconds += 8 # Небольшой запас на Telegram / отправку файлов. estimated_seconds += 8 + parts_count * 2 return format_seconds(estimated_seconds) def get_media_info(input_path: Path) -> Tuple[Optional[float], Optional[str]]: try: container = av.open(str(input_path)) duration = None if container.duration is not None: duration = float(container.duration / av.time_base) has_audio = any(stream.type == "audio" for stream in container.streams) has_video = any(stream.type == "video" for stream in container.streams) if has_audio and has_video: media_type = "видео с аудио" elif has_audio: media_type = "аудио" elif has_video: media_type = "видео без аудио" else: media_type = "неизвестный файл" container.close() return duration, media_type except Exception: return None, None def get_message_media(message: Message) -> Tuple[Optional[str], str, Optional[int]]: if message.voice: return message.voice.file_id, "voice.ogg", message.voice.file_size if message.audio: filename = message.audio.file_name or "audio" return message.audio.file_id, filename, message.audio.file_size if message.video: filename = message.video.file_name or "video.mp4" return message.video.file_id, filename, message.video.file_size if message.video_note: return message.video_note.file_id, "video_note.mp4", message.video_note.file_size if message.document: filename = message.document.file_name or "document" return message.document.file_id, filename, message.document.file_size return None, "file", None def ensure_file_size_allowed(file_size: Optional[int]) -> None: if file_size is None: return max_bytes = MAX_FILE_SIZE_MB * 1024 * 1024 if file_size > max_bytes: size_mb = file_size / 1024 / 1024 raise UserVisibleError( "Файл слишком большой: {:.1f} МБ. Лимит: {} МБ.".format( size_mb, MAX_FILE_SIZE_MB, ) ) async def download_telegram_file(file_id: str, destination: Path) -> None: tg_file = await bot.get_file(file_id) await bot.download_file(tg_file.file_path, destination=destination) def split_text_for_telegram(text: str, limit: int = TELEGRAM_MESSAGE_LIMIT) -> List[str]: if len(text) <= limit: return [text] chunks = [] current = "" parts = text.split("\n") for part in parts: part = part.strip() if not part: continue if len(current) + len(part) + 1 <= limit: current = part if not current else current + "\n" + part else: if current: chunks.append(current) current = part if current: chunks.append(current) return chunks async def edit_status( status_message: Message, title: str, percent: int, duration_text: str, estimate_text: str, extra: str = "", ) -> None: percent = max(0, min(100, percent)) text = ( "{}\n\n" "{} {}%\n\n" "Длительность: {}\n" "Ориентир ожидания: {}" ).format( title, progress_bar(percent), percent, duration_text, estimate_text, ) if extra: text += "\n" + extra try: await status_message.edit_text(text) except Exception: pass # ============================================================ # 7. ТАЙМИНГИ # ============================================================ def split_text_into_phrases(text: str) -> List[str]: text = text.strip() if not text: return [] separators = [". ", "! ", "? ", "; ", ", "] parts = [text] for separator in separators: new_parts = [] for part in parts: if len(part) <= 140: new_parts.append(part) continue split_items = part.split(separator) if len(split_items) == 1: new_parts.append(part) continue for item in split_items: item = item.strip() if item: new_parts.append(item) parts = new_parts final_parts = [] for part in parts: part = part.strip() if not part: continue if len(part) <= 170: final_parts.append(part) continue words = part.split() current = "" for word in words: if len(current) + len(word) + 1 <= 150: current = word if not current else current + " " + word else: if current: final_parts.append(current) current = word if current: final_parts.append(current) return final_parts def make_dense_timed_lines(segment_start: float, segment_end: float, text: str) -> List[str]: duration = max(0.1, segment_end - segment_start) text = text.strip() if not text: return [] if duration <= MAX_TIMED_BLOCK_SECONDS: return [ "[{} — {}] {}".format( format_seconds(segment_start), format_seconds(segment_end), text, ) ] phrases = split_text_into_phrases(text) if not phrases: return [] target_blocks = max(1, round(duration / MAX_TIMED_BLOCK_SECONDS)) if len(phrases) < target_blocks: words = text.split() words_per_block = max(1, round(len(words) / target_blocks)) phrases = [] for i in range(0, len(words), words_per_block): phrases.append(" ".join(words[i:i + words_per_block])) total_chars = sum(max(1, len(phrase)) for phrase in phrases) lines = [] current_time = segment_start for index, phrase in enumerate(phrases): phrase = phrase.strip() if not phrase: continue if index == len(phrases) - 1: next_time = segment_end else: share = max(1, len(phrase)) / total_chars next_time = current_time + duration * share next_time = min(next_time, segment_end) lines.append( "[{} — {}] {}".format( format_seconds(current_time), format_seconds(next_time), phrase, ) ) current_time = next_time return lines # ============================================================ # 8. МОДЕЛЬ И ТРАНСКРИБАЦИЯ # ============================================================ def load_model_once() -> WhisperModel: global model if model is not None: return model with model_lock: if model is None: logger.info("Loading model: %s", MODEL_NAME) model = WhisperModel( MODEL_NAME, device=DEVICE, compute_type=COMPUTE_TYPE, cpu_threads=max(os.cpu_count() or 2, 2), num_workers=1, ) logger.info("Model loaded") return model def split_audio_boundaries(duration_seconds: Optional[float], parts_count: int) -> List[float]: if not duration_seconds or parts_count <= 1: return [duration_seconds or 0] return [ duration_seconds * index / parts_count for index in range(1, parts_count + 1) ] def send_event_to_loop( loop: asyncio.AbstractEventLoop, queue: asyncio.Queue, event: Dict[str, Any], ) -> None: loop.call_soon_threadsafe(queue.put_nowait, event) def transcribe_stream_sync( input_path: Path, duration_seconds: Optional[float], parts_count: int, with_timestamps: bool, loop: asyncio.AbstractEventLoop, queue: asyncio.Queue, ) -> str: started_at = time.perf_counter() model_was_loaded_before_job = model is not None whisper = load_model_once() need_internal_timestamps = with_timestamps or parts_count > 1 boundaries = split_audio_boundaries(duration_seconds, parts_count) current_part_index = 1 current_part_lines: List[str] = [] all_lines: List[str] = [] last_progress = 0 last_progress_time = time.perf_counter() try: send_event_to_loop( loop, queue, { "type": "status", "percent": 12, "stage": "Подготовка", }, ) segments, info = whisper.transcribe( str(input_path), language=LANGUAGE, task="transcribe", beam_size=BEAM_SIZE, best_of=BEST_OF, patience=PATIENCE, condition_on_previous_text=False, temperature=0.0, initial_prompt=INITIAL_PROMPT, vad_filter=False, without_timestamps=not need_internal_timestamps, word_timestamps=False, ) for segment in segments: raw_text = segment.text.strip() if not raw_text: continue if with_timestamps: segment_lines = make_dense_timed_lines( segment_start=segment.start, segment_end=segment.end, text=raw_text, ) else: segment_lines = [raw_text] current_part_lines.extend(segment_lines) all_lines.extend(segment_lines) if duration_seconds and need_internal_timestamps: progress = int(15 + min(segment.end / duration_seconds, 1) * 80) now = time.perf_counter() if progress >= last_progress + 10 or now - last_progress_time >= 15: last_progress = progress last_progress_time = now send_event_to_loop( loop, queue, { "type": "status", "percent": progress, "stage": "Распознавание", }, ) if parts_count > 1 and duration_seconds: while ( current_part_index <= parts_count and segment.end >= boundaries[current_part_index - 1] ): part_text = ( "\n".join(current_part_lines).strip() if with_timestamps else " ".join(current_part_lines).strip() ) if not part_text: part_text = "В этой части речь не распознана." boundary_time = boundaries[current_part_index - 1] send_event_to_loop( loop, queue, { "type": "part", "part_index": current_part_index, "parts_count": parts_count, "end_time": boundary_time, "text": part_text, }, ) current_part_lines = [] current_part_index += 1 if parts_count <= 1: final_text = ( "\n".join(all_lines).strip() if with_timestamps else " ".join(all_lines).strip() ) else: while current_part_index <= parts_count: part_text = ( "\n".join(current_part_lines).strip() if with_timestamps else " ".join(current_part_lines).strip() ) if part_text: boundary_time = boundaries[current_part_index - 1] send_event_to_loop( loop, queue, { "type": "part", "part_index": current_part_index, "parts_count": parts_count, "end_time": boundary_time, "text": part_text, }, ) current_part_lines = [] current_part_index += 1 final_text = ( "\n".join(all_lines).strip() if with_timestamps else " ".join(all_lines).strip() ) if not final_text: final_text = "Не удалось распознать речь." except Exception: logger.exception("Transcription error") send_event_to_loop( loop, queue, { "type": "error", "message": "Не получилось распознать файл. Возможно, файл повреждён или внутри нет нормальной аудиодорожки.", }, ) return "" elapsed = time.perf_counter() - started_at update_runtime_stats( duration_seconds=duration_seconds, elapsed_seconds=elapsed, model_was_loaded_before_job=model_was_loaded_before_job, ) send_event_to_loop( loop, queue, { "type": "done", "text": final_text, }, ) return final_text async def transcribe_and_stream( input_path: Path, duration_seconds: Optional[float], duration_text: str, parts_count: int, with_timestamps: bool, status_message: Message, ) -> None: loop = asyncio.get_running_loop() queue: asyncio.Queue = asyncio.Queue() estimate_text = estimate_transcription_time( duration_seconds=duration_seconds, parts_count=parts_count, with_timestamps=with_timestamps, ) future = loop.run_in_executor( executor, transcribe_stream_sync, input_path, duration_seconds, parts_count, with_timestamps, loop, queue, ) final_text = "" await edit_status( status_message=status_message, title="Идёт транскрибация", percent=10, duration_text=duration_text, estimate_text=estimate_text, extra="Этап: запуск", ) while True: if future.done() and queue.empty(): break try: event = await asyncio.wait_for(queue.get(), timeout=0.5) except asyncio.TimeoutError: continue event_type = event.get("type") if event_type == "status": await edit_status( status_message=status_message, title="Идёт транскрибация", percent=int(event.get("percent", 10)), duration_text=duration_text, estimate_text=estimate_text, extra="Этап: {}".format(event.get("stage", "")), ) elif event_type == "part": part_index = int(event["part_index"]) parts_total = int(event["parts_count"]) end_time = float(event["end_time"]) text = str(event["text"]) filename = "Часть {} из {} — {}.txt".format( part_index, parts_total, format_seconds_for_filename(end_time), ) document = BufferedInputFile( file=text.encode("utf-8"), filename=filename, ) await status_message.answer_document(document=document) elif event_type == "error": await status_message.edit_text(str(event.get("message", "Ошибка транскрибации."))) return elif event_type == "done": final_text = str(event.get("text", "")) try: await future except Exception: logger.exception("Executor transcription error") if not final_text: final_text = "Не удалось распознать речь." await edit_status( status_message=status_message, title="Транскрибация готова", percent=100, duration_text=duration_text, estimate_text=estimate_text, ) final_filename = "Транскрипция — {}.txt".format( format_seconds_for_filename(duration_seconds or 0) ) final_document = BufferedInputFile( file=final_text.encode("utf-8"), filename=final_filename, ) await status_message.answer_document(document=final_document) async def preload_model_background() -> None: loop = asyncio.get_running_loop() try: logger.info("Background model preload started") await loop.run_in_executor(executor, load_model_once) logger.info("Background model preload finished") except Exception: logger.exception("Background model preload failed") # ============================================================ # 9. TELEGRAM-ХЭНДЛЕРЫ # ============================================================ @dp.message(CommandStart()) async def start_handler(message: Message) -> None: await message.answer( "Отправь голосовое, аудио, видео, кружок или файл — я сделаю транскрибацию." ) @dp.message(F.voice | F.audio | F.video | F.video_note | F.document) async def media_handler(message: Message) -> None: file_id, filename, file_size = get_message_media(message) if not file_id: await message.answer("Не вижу файла для транскрибации.") return try: ensure_file_size_allowed(file_size) except UserVisibleError as error: await message.answer(str(error)) return user_id = message.from_user.id file_suffix = Path(filename).suffix or ".bin" input_path = FILES_DIR / "{}_{}{}".format(user_id, uuid.uuid4().hex, file_suffix) status_message = await message.answer( "Скачиваю файл\n\n{} 5%".format(progress_bar(5)) ) try: await bot.send_chat_action(message.chat.id, ChatAction.TYPING) await download_telegram_file(file_id, input_path) duration_seconds, media_type = get_media_info(input_path) if media_type == "видео без аудио": await status_message.edit_text("В файле не найдена аудиодорожка.") try: input_path.unlink(missing_ok=True) except Exception: pass return duration_text = ( format_seconds(duration_seconds) if duration_seconds else "не удалось определить" ) user_files[user_id] = { "path": input_path, "duration_seconds": duration_seconds, "duration_text": duration_text, } await status_message.edit_text( "Файл загружен\n\n" "Длительность: {}\n\n" "Выбери формат транскрибации:".format(duration_text), reply_markup=format_choice_keyboard(), ) except Exception: logger.exception("File processing error") await status_message.edit_text("Не получилось обработать файл. Попробуй отправить другой файл.") try: input_path.unlink(missing_ok=True) except Exception: pass @dp.callback_query(F.data == "choice_plain") async def choice_plain_handler(callback: CallbackQuery) -> None: user_id = callback.from_user.id if user_id not in user_files: await callback.message.answer("Файл не найден. Отправь его ещё раз.") await callback.answer() return user_files[user_id]["with_timestamps"] = False await callback.message.edit_text( "Выбери, на сколько частей разделить результат.\n\n" "Если файл длинный, части помогут начать работать с текстом раньше. " "После обработки бот всё равно отправит полную транскрипцию одним файлом.", reply_markup=split_choice_keyboard(), ) await callback.answer() @dp.callback_query(F.data == "choice_timed") async def choice_timed_handler(callback: CallbackQuery) -> None: user_id = callback.from_user.id if user_id not in user_files: await callback.message.answer("Файл не найден. Отправь его ещё раз.") await callback.answer() return user_files[user_id]["with_timestamps"] = True await callback.message.edit_text( "Выбери, на сколько частей разделить результат.\n\n" "Если файл длинный, части помогут начать работать с текстом раньше. " "После обработки бот всё равно отправит полную транскрипцию одним файлом.", reply_markup=split_choice_keyboard(), ) await callback.answer() @dp.callback_query(F.data.startswith("split_")) async def split_handler(callback: CallbackQuery) -> None: user_id = callback.from_user.id file_data = user_files.get(user_id) if not file_data: await callback.message.answer("Файл не найден. Отправь его ещё раз.") await callback.answer() return try: parts_count = int(callback.data.split("_")[1]) except Exception: parts_count = 1 parts_count = max(1, min(4, parts_count)) input_path = file_data["path"] duration_seconds = file_data["duration_seconds"] duration_text = file_data["duration_text"] with_timestamps = bool(file_data.get("with_timestamps", False)) if not isinstance(input_path, Path) or not input_path.exists(): await callback.message.answer("Файл уже недоступен. Отправь его ещё раз.") await callback.answer() return await callback.answer() try: await transcribe_and_stream( input_path=input_path, duration_seconds=duration_seconds, duration_text=duration_text, parts_count=parts_count, with_timestamps=with_timestamps, status_message=callback.message, ) finally: try: input_path.unlink(missing_ok=True) except Exception: pass user_files.pop(user_id, None) @dp.message() async def fallback_handler(message: Message) -> None: await message.answer( "Отправь голосовое, аудио, видео, кружок или файл — я сделаю транскрибацию." ) # ============================================================ # 10. ЗАПУСК # ============================================================ async def run_web_server() -> None: port = int(os.getenv("PORT", "7860")) config = uvicorn.Config( app, host="0.0.0.0", port=port, log_level="warning", loop="asyncio", ) server = uvicorn.Server(config) await server.serve() async def main() -> None: asyncio.create_task(run_web_server()) # Модель начинает грузиться сразу после старта Space. # Это уменьшает ожидание при первом файле после перезапуска. asyncio.create_task(preload_model_background()) logger.info("Bot started. Waiting for Telegram messages...") await dp.start_polling(bot) if __name__ == "__main__": asyncio.run(main())