"""Обработка сообщений через 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 "Слишком много шагов. Повтори запрос."