ars/chatbottren
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: telegram bot base

Browse Source
This commit is contained in:
Ахметзянов Арсен
2026-02-16 10:45:01 +05:00
commit 3627117ea2
9 changed files with 809 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
.env.example Normal file
View File

@@ -0,0 +1,19 @@
# Gitea
GITEA_URL=https://lab.iieasy.ru
GITEA_TOKEN=your_gitea_api_token
REPO_OWNER=ars
REPO_NAME=task_org
# LLM (Ollama)
OPENAI_BASE_URL=http://192.168.88.160:11434/v1
OPENAI_API_KEY=ollama
OPENAI_MODEL=llama3.2
# Telegram
TELEGRAM_BOT_TOKEN=your_bot_token
# Опционально: Gitea username для «на меня» (или JSON: {"telegram_username": "gitea_username"})
# TELEGRAM_TO_GITEA_USERNAME=
# Whisper (голосовые сообщения): base, small, medium, large-v3
# WHISPER_MODEL=small

6
.gitignore vendored Normal file
View File

@@ -0,0 +1,6 @@
.env
.venv
venv
__pycache__
*.pyc
*.pyo

18
README.md Normal file
View File

@@ -0,0 +1,18 @@
# Telegram-бот для Issues в Gitea
Текстовый интерфейс: сообщения обрабатываются LLM (Ollama) через Function Calling, создание и список задач — через Gitea API.
## Запуск
```bash
cd /home/its/telegrambottask
python3 -m venv .venv
.venv/bin/pip install -r requirements.txt
.venv/bin/python bot.py
```
Перед запуском проверь `.env`: Gitea, токен бота, `OPENAI_BASE_URL` и модель (по умолчанию `llama3.2`; для tool calling подойдут также `qwen2.5`, `mistral` и др.).
## «На меня» / мои задачи
Если в Telegram username совпадает с логином в Gitea — запросы «какие задачи на мне» будут работать без доп. настройки. Иначе задай в `.env` переменную `TELEGRAM_TO_GITEA_USERNAME` (один логин или JSON маппинг).

82
bot.py Normal file
View File

@@ -0,0 +1,82 @@
"""Telegram-бот: текстовые и голосовые сообщения → LLM → ответ."""
import asyncio
import logging
import tempfile
from pathlib import Path
from aiogram import Bot, Dispatcher, F, Router
from aiogram.types import Message
import config
from config import get_gitea_username
from llm_handler import process_message
from transcribe import transcribe_audio
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
config.validate()
bot = Bot(token=config.TELEGRAM_BOT_TOKEN)
dp = Dispatcher()
router = Router()
dp.include_router(router)
@router.message(F.text)
async def on_text(message: Message) -> None:
text = (message.text or "").strip()
if not text:
return
gitea_username = get_gitea_username(
message.from_user.username if message.from_user else None,
message.from_user.id if message.from_user else None,
)
try:
reply = await asyncio.wait_for(process_message(text, gitea_username), timeout=60.0)
await message.answer(reply or "Готово.")
except asyncio.TimeoutError:
await message.answer("Таймаут. Попробуй короче или позже.")
except Exception as e:
logger.exception("Handler error")
await message.answer(f"Ошибка: {e!s}")
@router.message(F.voice)
async def on_voice(message: Message) -> None:
if not message.voice:
return
gitea_username = get_gitea_username(
message.from_user.username if message.from_user else None,
message.from_user.id if message.from_user else None,
)
tmp = None
try:
file = await message.bot.get_file(message.voice.file_id)
tmp = tempfile.NamedTemporaryFile(suffix=".ogg", delete=False)
tmp.close()
await message.bot.download_file(file.file_path, tmp.name)
text = await asyncio.to_thread(transcribe_audio, tmp.name)
Path(tmp.name).unlink(missing_ok=True)
tmp = None
if not (text or "").strip():
await message.answer("Не удалось распознать речь.")
return
reply = await asyncio.wait_for(process_message(text.strip(), gitea_username), timeout=60.0)
await message.answer(reply or "Готово.")
except asyncio.TimeoutError:
await message.answer("Таймаут. Попробуй короче или позже.")
except Exception as e:
logger.exception("Voice handler error")
await message.answer(f"Ошибка: {e!s}")
finally:
if tmp is not None and getattr(tmp, "name", None):
Path(tmp.name).unlink(missing_ok=True)
async def main() -> None:
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())

59
config.py Normal file
View File

@@ -0,0 +1,59 @@
"""Конфигурация из переменных окружения."""
import os
from pathlib import Path
from dotenv import load_dotenv
load_dotenv(Path(__file__).resolve().parent / ".env")
# Gitea
GITEA_URL = os.getenv("GITEA_URL", "").rstrip("/")
GITEA_TOKEN = os.getenv("GITEA_TOKEN") or os.getenv("TOKEN")
REPO_OWNER = os.getenv("REPO_OWNER", "")
REPO_NAME = os.getenv("REPO_NAME", "")
# LLM (Ollama / OpenAI-compatible)
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "ollama")
OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", "")
OPENAI_MODEL = os.getenv("OPENAI_MODEL", "qwen3")
# Telegram
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "")
TELEGRAM_TO_GITEA_USERNAME = os.getenv("TELEGRAM_TO_GITEA_USERNAME", "")
# Whisper (голосовые)
WHISPER_MODEL = os.getenv("WHISPER_MODEL", "small")
def validate() -> None:
"""Проверка обязательных переменных при старте."""
missing = []
if not GITEA_URL:
missing.append("GITEA_URL")
if not GITEA_TOKEN:
missing.append("GITEA_TOKEN")
if not REPO_OWNER:
missing.append("REPO_OWNER")
if not REPO_NAME:
missing.append("REPO_NAME")
if not TELEGRAM_BOT_TOKEN:
missing.append("TELEGRAM_BOT_TOKEN")
if missing:
raise RuntimeError(f"Не заданы: {', '.join(missing)}. Проверьте .env")
def get_gitea_username(telegram_username: str | None, telegram_id: int | None) -> str:
"""Gitea username для текущего пользователя Telegram (для «на меня»)."""
if TELEGRAM_TO_GITEA_USERNAME:
# Один username для всех или JSON — пока только одна строка
try:
import json
m = json.loads(TELEGRAM_TO_GITEA_USERNAME)
if isinstance(m, dict) and telegram_username and telegram_username in m:
return m[telegram_username]
if isinstance(m, dict) and telegram_id is not None and str(telegram_id) in m:
return m[str(telegram_id)]
except (json.JSONDecodeError, TypeError):
pass
return TELEGRAM_TO_GITEA_USERNAME
return telegram_username or ""

195
gitea_client.py Normal file
View File

@@ -0,0 +1,195 @@
"""Клиент Gitea API для работы с issues."""
import re
from datetime import datetime, timedelta, timezone
import httpx
import config
BASE = f"{config.GITEA_URL}/api/v1"
HEADERS = {"Authorization": f"token {config.GITEA_TOKEN}"}
def _url(path: str) -> str:
return f"{BASE}{path}"
async def get_assignees() -> list[dict]:
"""Список пользователей, которых можно назначить на issues (username, full_name, id)."""
async with httpx.AsyncClient() as client:
r = await client.get(
_url(f"/repos/{config.REPO_OWNER}/{config.REPO_NAME}/assignees"),
headers=HEADERS,
timeout=15.0,
)
r.raise_for_status()
return r.json()
def resolve_assignee_by_name(name: str, assignees: list[dict]) -> str | None:
"""Найти username по имени/подстроке в assignees. Возвращает login или None."""
name = (name or "").strip().lower()
if not name:
return None
for u in assignees:
login = (u.get("login") or "").lower()
full = (u.get("full_name") or "").lower()
if name in login or name in full or (name and (login == name or full == name)):
return u.get("login")
return None
def parse_deadline(s: str) -> str | None:
"""Парсит дату из строки (завтра, ISO date). Возвращает ISO date YYYY-MM-DD или None."""
if not s or not isinstance(s, str):
return None
s = s.strip().lower()
if s == "завтра":
d = datetime.now(timezone.utc) + timedelta(days=1)
return d.strftime("%Y-%m-%d")
if s == "сегодня":
d = datetime.now(timezone.utc)
return d.strftime("%Y-%m-%d")
# ISO-like date
m = re.match(r"(\d{4})-(\d{2})-(\d{2})", s)
if m:
return f"{m.group(1)}-{m.group(2)}-{m.group(3)}"
return None
async def create_issue(
title: str,
body: str = "",
assignees: list[str] | None = None,
deadline: str | None = None,
) -> dict:
"""Создать issue. assignees — список Gitea username. deadline — ISO date YYYY-MM-DD."""
payload = {"title": title}
if body:
payload["body"] = body
if assignees:
payload["assignees"] = assignees
due = parse_deadline(deadline) if deadline else None
if due:
payload["due_date"] = due + "T12:00:00Z"
async with httpx.AsyncClient() as client:
r = await client.post(
_url(f"/repos/{config.REPO_OWNER}/{config.REPO_NAME}/issues"),
headers={**HEADERS, "Content-Type": "application/json"},
json=payload,
timeout=15.0,
)
r.raise_for_status()
return r.json()
async def get_issues(
state: str = "open",
assignee: str | None = None,
page: int = 1,
limit: int = 20,
) -> list[dict]:
"""Список issues. state: open|closed. assignee: Gitea username."""
params = {"state": state, "page": page, "limit": limit}
if assignee:
params["assignee"] = assignee
async with httpx.AsyncClient() as client:
r = await client.get(
_url(f"/repos/{config.REPO_OWNER}/{config.REPO_NAME}/issues"),
headers=HEADERS,
params=params,
timeout=15.0,
)
r.raise_for_status()
return r.json()
async def add_comment(issue_number: int, body: str) -> dict:
"""Добавить комментарий к задаче."""
async with httpx.AsyncClient() as client:
r = await client.post(
_url(f"/repos/{config.REPO_OWNER}/{config.REPO_NAME}/issues/{issue_number}/comments"),
headers={**HEADERS, "Content-Type": "application/json"},
json={"body": body},
timeout=15.0,
)
r.raise_for_status()
return r.json()
async def close_issue(issue_number: int) -> dict:
"""Закрыть задачу."""
async with httpx.AsyncClient() as client:
r = await client.patch(
_url(f"/repos/{config.REPO_OWNER}/{config.REPO_NAME}/issues/{issue_number}"),
headers={**HEADERS, "Content-Type": "application/json"},
json={"state": "closed"},
timeout=15.0,
)
r.raise_for_status()
return r.json()
async def start_stopwatch(issue_number: int) -> dict:
"""Запустить таймер по задаче."""
async with httpx.AsyncClient() as client:
r = await client.post(
_url(f"/repos/{config.REPO_OWNER}/{config.REPO_NAME}/issues/{issue_number}/stopwatch/start"),
headers=HEADERS,
timeout=15.0,
)
r.raise_for_status()
return r.json()
async def stop_stopwatch(issue_number: int) -> dict:
"""Остановить таймер по задаче."""
async with httpx.AsyncClient() as client:
r = await client.post(
_url(f"/repos/{config.REPO_OWNER}/{config.REPO_NAME}/issues/{issue_number}/stopwatch/stop"),
headers=HEADERS,
timeout=15.0,
)
r.raise_for_status()
return r.json()
async def add_tracked_time(issue_number: int, seconds: int) -> dict:
"""Добавить учтённое время к задаче (в секундах)."""
async with httpx.AsyncClient() as client:
r = await client.post(
_url(f"/repos/{config.REPO_OWNER}/{config.REPO_NAME}/issues/{issue_number}/times"),
headers={**HEADERS, "Content-Type": "application/json"},
json={"time": seconds},
timeout=15.0,
)
r.raise_for_status()
return r.json()
async def set_issue_deadline(issue_number: int, deadline: str | None) -> dict:
"""Установить или снять срок задачи. deadline — дата YYYY-MM-DD или None."""
parsed = parse_deadline(deadline) if deadline else None
due = (parsed + "T12:00:00Z") if parsed else None
async with httpx.AsyncClient() as client:
r = await client.post(
_url(f"/repos/{config.REPO_OWNER}/{config.REPO_NAME}/issues/{issue_number}/deadline"),
headers={**HEADERS, "Content-Type": "application/json"},
json={"due_date": due},
timeout=15.0,
)
r.raise_for_status()
return r.json()
async def set_issue_assignees(issue_number: int, assignees: list[str]) -> dict:
"""Заменить ответственных по задаче (список Gitea username)."""
async with httpx.AsyncClient() as client:
r = await client.patch(
_url(f"/repos/{config.REPO_OWNER}/{config.REPO_NAME}/issues/{issue_number}"),
headers={**HEADERS, "Content-Type": "application/json"},
json={"assignees": assignees},
timeout=15.0,
)
r.raise_for_status()
return r.json()

395
llm_handler.py Normal file
View File

@@ -0,0 +1,395 @@
"""Обработка сообщений через LLM и вызов инструментов (Gitea)."""
import json
from openai import AsyncOpenAI
import config
import gitea_client
client = AsyncOpenAI(
api_key=config.OPENAI_API_KEY,
base_url=config.OPENAI_BASE_URL or None,
)
TOOLS = [
{
"type": "function",
"function": {
"name": "create_issue",
"description": "Создать задачу (issue) в репозитории. Используй, когда пользователь просит создать задачу, назначить, поставить дедлайн.",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "Заголовок задачи"},
"body": {"type": "string", "description": "Описание задачи (необязательно)"},
"assignee": {"type": "string", "description": "Имя или логин исполнителя (необязательно)"},
"deadline": {"type": "string", "description": "Дедлайн: дата в формате YYYY-MM-DD или «завтра», «сегодня»"},
},
"required": ["title"],
},
},
},
{
"type": "function",
"function": {
"name": "get_issues",
"description": "Получить список задач. Используй для запросов «какие задачи», «что на меня», «открытые задачи» и т.п.",
"parameters": {
"type": "object",
"properties": {
"assignee": {"type": "string", "description": "Логин исполнителя. Для «на меня» подставь текущего пользователя Gitea (gitea_username)."},
"state": {"type": "string", "enum": ["open", "closed"], "description": "Состояние: open или closed. По умолчанию open."},
},
},
},
},
{
"type": "function",
"function": {
"name": "add_comment",
"description": "Написать комментарий к задаче по номеру (#N).",
"parameters": {
"type": "object",
"properties": {
"issue_number": {"type": "integer", "description": "Номер задачи (например 5 для #5)."},
"body": {"type": "string", "description": "Текст комментария."},
},
"required": ["issue_number", "body"],
},
},
},
{
"type": "function",
"function": {
"name": "close_issue",
"description": "Закрыть задачу по номеру (#N).",
"parameters": {
"type": "object",
"properties": {
"issue_number": {"type": "integer", "description": "Номер задачи."},
},
"required": ["issue_number"],
},
},
},
{
"type": "function",
"function": {
"name": "start_stopwatch",
"description": "Запустить таймер (отслеживание времени) по задаче #N.",
"parameters": {
"type": "object",
"properties": {
"issue_number": {"type": "integer", "description": "Номер задачи."},
},
"required": ["issue_number"],
},
},
},
{
"type": "function",
"function": {
"name": "stop_stopwatch",
"description": "Остановить таймер по задаче #N.",
"parameters": {
"type": "object",
"properties": {
"issue_number": {"type": "integer", "description": "Номер задачи."},
},
"required": ["issue_number"],
},
},
},
{
"type": "function",
"function": {
"name": "add_tracked_time",
"description": "Добавить учтённое время к задаче (часы или минуты).",
"parameters": {
"type": "object",
"properties": {
"issue_number": {"type": "integer", "description": "Номер задачи."},
"hours": {"type": "number", "description": "Часы (необязательно, если указаны минуты)."},
"minutes": {"type": "number", "description": "Минуты (необязательно, если указаны часы)."},
},
"required": ["issue_number"],
},
},
},
{
"type": "function",
"function": {
"name": "set_issue_deadline",
"description": "Установить или снять срок выполнения задачи. Передай пустой deadline или remove=true чтобы снять срок.",
"parameters": {
"type": "object",
"properties": {
"issue_number": {"type": "integer", "description": "Номер задачи."},
"deadline": {"type": "string", "description": "Дата: YYYY-MM-DD или «завтра», «сегодня». Пусто — снять срок."},
},
"required": ["issue_number"],
},
},
},
{
"type": "function",
"function": {
"name": "set_issue_assignees",
"description": "Заменить ответственных по задаче. Передай список имён/логинов. Пустой массив — снять всех.",
"parameters": {
"type": "object",
"properties": {
"issue_number": {"type": "integer", "description": "Номер задачи."},
"assignees": {"type": "array", "items": {"type": "string"}, "description": "Список имён или логинов исполнителей."},
},
"required": ["issue_number", "assignees"],
},
},
},
]
def _system_prompt(gitea_username: str) -> str:
return (
"Ты помощник по задачам в Gitea. Отвечай кратко, по делу, без лишней вежливости. "
"Можешь создавать и закрывать задачи, писать комментарии, запускать/останавливать таймер, добавлять время, ставить срок и менять ответственных по номеру задачи (#N). "
f"Текущий пользователь в Gitea (для запросов «на меня», «мои задачи»): {gitea_username or 'не задан'}."
)
async def _run_create_issue(args: dict) -> str:
title = (args.get("title") or "").strip()
if not title:
return "Ошибка: не указан заголовок задачи."
body = (args.get("body") or "").strip()
assignee_raw = (args.get("assignee") or "").strip()
deadline = (args.get("deadline") or "").strip() or None
assignees = []
if assignee_raw:
assignees_list = await gitea_client.get_assignees()
resolved = gitea_client.resolve_assignee_by_name(assignee_raw, assignees_list)
if resolved:
assignees = [resolved]
else:
return f"Исполнитель не найден: «{assignee_raw}». Проверь имя или логин."
try:
issue = await gitea_client.create_issue(
title=title,
body=body,
assignees=assignees if assignees else None,
deadline=deadline,
)
num = issue.get("number")
return f"Создана задача #{num}. " + (f"Назначена на {assignees[0]}, " if assignees else "") + (f"дедлайн {deadline}." if deadline else "")
except Exception as e:
return f"Ошибка Gitea: {e!s}"
async def _run_get_issues(args: dict) -> str:
assignee = (args.get("assignee") or "").strip() or None
state = (args.get("state") or "open").strip().lower() or "open"
if state not in ("open", "closed"):
state = "open"
try:
issues = await gitea_client.get_issues(state=state, assignee=assignee)
except Exception as e:
return f"Ошибка Gitea: {e!s}"
if not issues:
return "Нет задач."
lines = [f"#{i.get('number')} {i.get('title', '')}" for i in issues]
return "\n".join(lines)
def _parse_issue_number(args: dict, key: str = "issue_number") -> int | None:
"""Извлечь номер задачи из аргументов. Возвращает int или None при ошибке."""
v = args.get(key)
if v is None:
return None
try:
return int(v)
except (TypeError, ValueError):
return None
async def _run_add_comment(args: dict) -> str:
num = _parse_issue_number(args)
if num is None:
return "Ошибка: укажи номер задачи (issue_number)."
body = (args.get("body") or "").strip()
if not body:
return "Ошибка: укажи текст комментария (body)."
try:
await gitea_client.add_comment(num, body)
return f"Комментарий добавлен к задаче #{num}."
except Exception as e:
return f"Ошибка Gitea: {e!s}"
async def _run_close_issue(args: dict) -> str:
num = _parse_issue_number(args)
if num is None:
return "Ошибка: укажи номер задачи (issue_number)."
try:
await gitea_client.close_issue(num)
return f"Задача #{num} закрыта."
except Exception as e:
return f"Ошибка Gitea: {e!s}"
async def _run_start_stopwatch(args: dict) -> str:
num = _parse_issue_number(args)
if num is None:
return "Ошибка: укажи номер задачи (issue_number)."
try:
await gitea_client.start_stopwatch(num)
return f"Таймер по задаче #{num} запущен."
except Exception as e:
return f"Ошибка Gitea: {e!s}"
async def _run_stop_stopwatch(args: dict) -> str:
num = _parse_issue_number(args)
if num is None:
return "Ошибка: укажи номер задачи (issue_number)."
try:
await gitea_client.stop_stopwatch(num)
return f"Таймер по задаче #{num} остановлен."
except Exception as e:
return f"Ошибка Gitea: {e!s}"
async def _run_add_tracked_time(args: dict) -> str:
num = _parse_issue_number(args)
if num is None:
return "Ошибка: укажи номер задачи (issue_number)."
hours = args.get("hours")
minutes = args.get("minutes")
if hours is not None:
seconds = int(float(hours) * 3600)
elif minutes is not None:
seconds = int(float(minutes) * 60)
else:
return "Ошибка: укажи hours или minutes."
if seconds <= 0:
return "Ошибка: время должно быть больше нуля."
try:
await gitea_client.add_tracked_time(num, seconds)
return f"Добавлено {seconds // 3600}ч {(seconds % 3600) // 60}м к задаче #{num}."
except Exception as e:
return f"Ошибка Gitea: {e!s}"
async def _run_set_issue_deadline(args: dict) -> str:
num = _parse_issue_number(args)
if num is None:
return "Ошибка: укажи номер задачи (issue_number)."
deadline = (args.get("deadline") or "").strip() or None
try:
await gitea_client.set_issue_deadline(num, deadline)
if deadline:
return f"Срок задачи #{num} установлен: {deadline}."
return f"Срок задачи #{num} снят."
except Exception as e:
return f"Ошибка Gitea: {e!s}"
async def _run_set_issue_assignees(args: dict) -> str:
num = _parse_issue_number(args)
if num is None:
return "Ошибка: укажи номер задачи (issue_number)."
raw_list = args.get("assignees") or []
if not isinstance(raw_list, list):
raw_list = []
assignees_list = await gitea_client.get_assignees()
resolved = []
for name in raw_list:
s = (name or "").strip()
if not s:
continue
r = gitea_client.resolve_assignee_by_name(s, assignees_list)
if r:
resolved.append(r)
else:
return f"Исполнитель не найден: «{s}»."
try:
await gitea_client.set_issue_assignees(num, resolved)
if resolved:
return f"Ответственные по задаче #{num}: {', '.join(resolved)}."
return f"Ответственные по задаче #{num} сняты."
except Exception as e:
return f"Ошибка Gitea: {e!s}"
async def _execute_tool(name: str, arguments: str) -> str:
try:
args = json.loads(arguments) if arguments else {}
except json.JSONDecodeError:
return "Ошибка: неверные аргументы."
if name == "create_issue":
return await _run_create_issue(args)
if name == "get_issues":
return await _run_get_issues(args)
if name == "add_comment":
return await _run_add_comment(args)
if name == "close_issue":
return await _run_close_issue(args)
if name == "start_stopwatch":
return await _run_start_stopwatch(args)
if name == "stop_stopwatch":
return await _run_stop_stopwatch(args)
if name == "add_tracked_time":
return await _run_add_tracked_time(args)
if name == "set_issue_deadline":
return await _run_set_issue_deadline(args)
if name == "set_issue_assignees":
return await _run_set_issue_assignees(args)
return f"Неизвестный инструмент: {name}"
async def process_message(user_message: str, gitea_username: str) -> str:
"""Обработать текст пользователя: LLM + выполнение tool_calls. Возвращает ответ для чата."""
messages = [
{"role": "system", "content": _system_prompt(gitea_username)},
{"role": "user", "content": user_message},
]
max_tool_rounds = 5
for _ in range(max_tool_rounds):
kwargs = {
"model": config.OPENAI_MODEL,
"messages": messages,
}
if _ == 0:
kwargs["tools"] = TOOLS
else:
kwargs["tools"] = TOOLS
response = await client.chat.completions.create(**kwargs)
choice = response.choices[0] if response.choices else None
if not choice:
return "Нет ответа от модели."
msg = choice.message
if getattr(msg, "content", None) and msg.content:
return (msg.content or "").strip() or "Готово."
tool_calls = getattr(msg, "tool_calls", None) or []
if not tool_calls:
return (getattr(msg, "content", None) or "").strip() or "Готово."
assistant_tool_calls = [
{"id": tc.id, "type": "function", "function": {"name": tc.function.name, "arguments": tc.function.arguments or "{}"}}
for tc in tool_calls
]
messages.append(
{
"role": "assistant",
"content": None,
"tool_calls": assistant_tool_calls,
}
)
for tc in tool_calls:
result = await _execute_tool(tc.function.name, tc.function.arguments or "{}")
messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})
return "Слишком много шагов. Повтори запрос."

5
requirements.txt Normal file
View File

@@ -0,0 +1,5 @@
aiogram>=3.4.0
openai>=1.12.0
httpx>=0.27.0
python-dotenv>=1.0.0
faster-whisper>=1.0.0

30
transcribe.py Normal file
View File

@@ -0,0 +1,30 @@
"""Транскрибация аудио через faster-whisper (модель small по умолчанию)."""
import logging
from pathlib import Path
import config
logger = logging.getLogger(__name__)
_model = None
def _get_model():
global _model
if _model is None:
from faster_whisper import WhisperModel
logger.info("Загрузка Whisper модели %s...", config.WHISPER_MODEL)
_model = WhisperModel(config.WHISPER_MODEL)
logger.info("Whisper модель загружена.")
return _model
def transcribe_audio(audio_path: str | Path) -> str:
"""Перевести аудиофайл в текст. Путь до .ogg, .mp3, .wav и т.д."""
path = Path(audio_path)
if not path.exists():
return ""
model = _get_model()
segments, _ = model.transcribe(str(path), language="ru", vad_filter=True)
text = " ".join(s.text.strip() for s in segments if s.text).strip()
return text