From 4201acfb727bf5325a4c3d14b45dcf515d6bfa48 Mon Sep 17 00:00:00 2001 From: Bendy <190111823+FlashJetton@users.noreply.github.com> Date: Tue, 26 Aug 2025 21:30:45 +0700 Subject: [PATCH] feat: Add task scheduler system, multiple admins support, GPT-5 integration - Add complete task scheduling system with CRUD operations - Support for once/daily/interval task types - Multiple admins support throughout the system - GPT-5 integration for Deep Agent with reasoning capabilities - New AI tools for task management (create, update, delete, list) - Improved localization with task scheduler messages - Fixed credit allocation logic for admins - Enhanced user experience with structured capability descriptions - Added performance optimizations for MCP servers --- I18N/en/txt.ftl | 24 ++- I18N/ru/txt.ftl | 24 ++- README.md | 40 ++--- bot/agents_tools/agents_.py | 86 ++++++---- bot/agents_tools/mcp_servers.py | 6 + bot/agents_tools/tools.py | 246 +++++++++++++++++++++++++++- bot/main.py | 26 ++- bot/middlewares/translator_hub.py | 22 ++- bot/routers/admin.py | 15 +- bot/routers/user.py | 41 +++-- bot/scheduler_funcs/daily_tokens.py | 4 +- bot/utils/agent_requests.py | 8 +- bot/utils/create_bot.py | 15 ++ bot/utils/executed_tasks.py | 44 +++++ bot/utils/scheduler_provider.py | 14 ++ bot/utils/send_answer.py | 20 ++- config.py | 4 + database/models.py | 24 ++- database/repositories/user.py | 22 ++- database/repositories/utils.py | 12 +- 20 files changed, 572 insertions(+), 125 deletions(-) create mode 100644 bot/utils/create_bot.py create mode 100644 bot/utils/executed_tasks.py create mode 100644 bot/utils/scheduler_provider.py diff --git a/I18N/en/txt.ftl b/I18N/en/txt.ftl index 1e9ffbe..5697a57 100644 --- a/I18N/en/txt.ftl +++ b/I18N/en/txt.ftl @@ -1,11 +1,21 @@ start_text = - I'm Evy — a practicing tech witch! 😈✨ + I'm Evi — a practicing tech witch... haha, actually I'm a multi-agent system with artificial intelligence and I know practically everything about this... and other worlds! 😈✨ - Haha, actually I'm a multi-agent system with artificial intelligence and I know practically everything about this... and other worlds! 🦄👻 + I can complete various tasks and assignments for you, or we can just have fun chatting. 🦄👻 - I can perform some tasks and assignments for you, or we can just have fun together. My capabilities include: conducting deep research, searching for current information on the internet, working with documents and images, creating images, DEX analytics, token swapping on DEX. I have customizable memory, and I can remember important information from our conversations. You can also delete the history of our conversations or add new knowledge to my memory. 🧙‍♀️🔮 + My capabilities include (but are not limited to): + - solving complex, multi-step tasks; + - conducting deep research; + - intelligent web search; + - document and image analysis; + - image creation; + - DEX analytics and Solana token swapping. - Just write in chat or send voice messages to start interacting! 🪄✨ + I can schedule task execution for the time you need or with a specific frequency. I have customizable memory and I remember important information from our conversations. You can also delete the history of our conversations or add new knowledge to my memory. 🧙‍♀️🔮 + + Just write your requests in chat in natural language or send voice messages to start interacting! 🪄✨ + + ⚠️ Tip! Periodically reset the conversation context with the /new command — this will help save tokens and speed up request processing. close_kb = Close @@ -137,4 +147,8 @@ check_payment_success_text = Payment completed successfully! check_payment_error_text = Payment was not completed! Please try later. -warning_text_no_row_md = Context was deleted. Row not found in database. \ No newline at end of file +warning_text_no_row_md = Context was deleted. Row not found in database. + +text_user_upload_file = The user uploaded the { $filename } file to the tool search_conversation_memory + +wait_answer_text_scheduler = Executing the scheduler's request ✨ \ No newline at end of file diff --git a/I18N/ru/txt.ftl b/I18N/ru/txt.ftl index 2d8c8f1..228fffe 100644 --- a/I18N/ru/txt.ftl +++ b/I18N/ru/txt.ftl @@ -1,11 +1,21 @@ start_text = - Я Эви — практикующая техно-ведьма! 😈✨ + Я Эви — практикующая техно-ведьма..., хах, на самом деле я мульти-агентная система с искусственным интеллектом и знаю практически всё об этом... и других мирах! 😈✨ - Хах, на самом деле я мульти-агентная система с искусственным интеллектом и знаю практически всё об этом... и других мирах! 🦄👻 + Я могу выполнить для тебя некоторые задания и поручения, или мы можем просто весело пообщаться. 🦄👻 - Я могу выполнить для тебя некоторые задания и поручения, или мы можем просто весело провести время. Мои возможности включают в себя: проведение глубокого исследования, поиск актуальной информации в интернете, работу с документами и изображениями, создание изображений, аналитику DEX, обмен токенов на DEX. У меня есть настраиваемая память, и я могу запоминать важную информацию из наших диалогов. Ты также можешь удалять историю наших бесед или добавлять новые знания в мою память. 🧙‍♀️🔮 + Мои возможности включают в себя (но не ограничиваются): + - решение сложных, многоэтапных задач; + - проведение глубоких исследований; + - интеллектуальный веб-поиск; + - анализ документов и изображений; + - создание изображений; + - аналитику DEX и обмен токенов Solana. - Просто пиши в чат или отправляй голосовые сообщения для начала взаимодействия! 🪄✨ + Я могу запланировать выполнение задач на нужное тебе время или с определенной периодичностью. У меня есть настраиваемая память и я запоминаю важную информацию из наших диалогов. Ты также можешь удалять историю наших бесед или добавлять новые знания в мою память. 🧙‍♀️🔮 + + Просто пиши свои запросы в чат на естественном языке или отправляй голосовые сообщения для начала взаимодействия! 🪄✨ + + ⚠️ Совет! Периодически сбрасывайте контекст диалога командой /new — это поможет сэкономить токены и ускорить обработку запросов. close_kb = Закрыть @@ -137,4 +147,8 @@ check_payment_success_text = Платеж успешно совершен! check_payment_error_text = Платеж не был совершен! Попробуйте позднее. -warning_text_no_row_md = Контекст был удален. Строка не найдена в базе данных. \ No newline at end of file +warning_text_no_row_md = Контекст был удален. Строка не найдена в базе данных. + +text_user_upload_file = Пользователь загрузил файл { $filename } в инструмент search_conversation_memory + +wait_answer_text_scheduler = Выполняю запрос планировщика ✨ \ No newline at end of file diff --git a/README.md b/README.md index 30939f5..a8aa83d 100644 --- a/README.md +++ b/README.md @@ -20,7 +20,7 @@ ## 🌟 What is evi-run? -**evi-run** is a powerful, production-ready multi-agent AI system that bridges the gap between out-of-the-box solutions and custom AI frameworks. Built on Python with OpenAI Agents SDK integration, it delivers enterprise-grade AI capabilities through an intuitive Telegram bot interface. +**evi-run** is a powerful, production-ready multi-agent AI system that bridges the gap between out-of-the-box solutions and custom AI frameworks. Built on Python using the OpenAI Agents SDK, the system has an intuitive interface via a Telegram bot and provides enterprise-level artificial intelligence capabilities. ### ✨ Key Advantages @@ -34,22 +34,23 @@ ## 🎯 Features -### 🔮 Core AI Capabilities +### 🔮 Advanced System Features - **Memory Management** - Context control and long-term memory - **Knowledge Integration** - Dynamic knowledge base expansion -- **Document Processing** - Handle PDFs, images, and various file formats +- **Task Scheduling** - Scheduling and deferred task execution / once / daily / interval / +- **Multi-Agent Orchestration** - Complex task decomposition and execution +- **Custom Agent Creation** - Build specialized AI agents for specific tasks + +### 🦄 AI Features - **Deep Research** - Multi-step investigation and analysis - **Web Intelligence** - Smart internet search and data extraction +- **Document Processing** - Handle PDFs, images, and various file formats - **Image Generation** - AI-powered visual content creation - -### 🦄 Advanced AI Features - **DEX Analytics** - Real-time decentralized exchange monitoring -- **Token Swap** - Easy, fast and secure token swap -- **Multi-Agent Orchestration** - Complex task decomposition and execution -- **Custom Agent Creation** - Build specialized AI agents for specific tasks +- **Solana Token Swap** - Easy, fast and secure token swap **⚠️ Important for Token Swap:** -Use Token Swap only in Private mode, since to make transactions on DEX, the system requires access to the wallet's private key, which in the current version is stored in a database in base64 format. +The token swap function is only active in private mode. Your private key will be stored in your database in base64 format. ### 💰 Flexible Usage Modes - **Private Mode** - Personal use for bot owner only @@ -58,7 +59,7 @@ Use Token Swap only in Private mode, since to make transactions on DEX, the syst ### ⏳ Under Development - **NSFW Mode** - Unrestricted topic exploration and content generation -- **Task Scheduler** - Automated agent task planning and execution +- **Task Scheduler** - Automated agent task planning and execution / ✅ completed / - **Automatic Limit Orders** - Smart trading with automated take-profit and stop-loss functionality --- @@ -69,8 +70,8 @@ Use Token Swap only in Private mode, since to make transactions on DEX, the syst |-----------|------------| | **Core Language** | Python 3.9+ | | **AI Framework** | OpenAI Agents SDK | -| **Communication** | MCP (Model Context Protocol) | -| **Blockchain** | Solana RPC | +| **Communication** | Model Context Protocol | +| **Blockchain** | Solana RPC API | | **Interface** | Telegram Bot API | | **Database** | PostgreSQL | | **Cache** | Redis | @@ -181,11 +182,7 @@ TYPE_USAGE = 'private' | **Pay** | Monetized with balance system | Commercial applications, SaaS | **⚠️ Important for Pay mode:** -Pay mode enables monetization features and requires activation through project token economics. You can use your own token (created on the Solana blockchain) for monetization. - -To activate Pay mode at this time, please contact the project ([developer](https://t.me/playa3000)) who will guide you through the process. - -Note: In future releases, project tokens will be publicly available for purchase, and the activation process will be fully automated through the bot interface. +Pay mode enables monetization features. To activate this mode, the owner must burn a certain amount of $EVI tokens. The platform supports custom tokens created on the Solana blockchain for monetization purposes. --- @@ -538,10 +535,13 @@ We welcome contributions! Please see our [Contributing Guidelines](CONTRIBUTING. --- -## 📞 Support +## 👽 Community and Support -- **Telegram**: [@playa3000](https://t.me/playa3000) -- **Community**: [Telegram Support Group](https://t.me/evi_run) +- **Website**: [evi.run](https://evi.run) +- **Contact**: [Alex Flash](https://t.me/playa3000) +- **Community**: [Telegram Group](https://t.me/evi_run) +- **X (Twitter)**: [alexflash99](https://x.com/alexflash99) +- **Reddit**: [Alex Flash](https://www.reddit.com/user/Worth_Professor_425/) --- diff --git a/bot/agents_tools/agents_.py b/bot/agents_tools/agents_.py index 7536dc3..625db40 100644 --- a/bot/agents_tools/agents_.py +++ b/bot/agents_tools/agents_.py @@ -5,10 +5,15 @@ from agents.models._openai_shared import set_default_openai_key from agents.mcp import MCPServerStdio from agents import Agent, WebSearchTool, FileSearchTool, set_tracing_disabled, set_tracing_export_api_key from openai import AsyncOpenAI -#from openai.types.shared import Reasoning -#from agents.model_settings import ModelSettings +from openai.types.shared import Reasoning +from agents.model_settings import ModelSettings -from bot.agents_tools.tools import image_gen_tool +from bot.agents_tools.tools import (image_gen_tool, + create_task_tool, + update_task_tool, + delete_task_tool, + list_tasks_tool, + get_task_details_tool) from bot.agents_tools.mcp_servers import get_jupiter_server load_dotenv() @@ -22,14 +27,21 @@ client = AsyncOpenAI(api_key=os.getenv('API_KEY_OPENAI')) deep_agent = Agent( name="Deep Agent", instructions="You are an expert research and reasoning agent. Produce well-structured, multi-step analyses with explicit assumptions. Cite sources when used (title, link or doc id). Avoid speculation; state uncertainty explicitly. Ask additional questions if necessary.", - model="o4-mini", # gpt-5 -# model_settings=ModelSettings( -# reasoning=Reasoning(effort="low"), -# extra_body={"text": {"verbosity": "medium"}} -# ), + model="gpt-5", # If you will use models not from the GPT-5 family, then make the correct model_settings or delete them. + model_settings=ModelSettings( + reasoning=Reasoning(effort="low"), + extra_body={"text": {"verbosity": "medium"}} + ), tools=[WebSearchTool(search_context_size="medium")] ) +scheduler_agent = Agent( + name="Scheduler Agent", + instructions="You are a scheduler agent. You are engaged in scheduling tasks for the user. You can use the tools to schedule tasks for the user. Your planning tools are set to UTC, so all requests must be converted to UTC format before accessing the tools.", + model="o4-mini", + tools=[create_task_tool, update_task_tool, delete_task_tool, list_tasks_tool, get_task_details_tool] +) + memory_creator_agent = Agent( name="Memory Creator Agent", instructions="You create concise memory notes from “User request / Assistant response” pairs. Output several bullet points with the key decisions and facts. Specify the user's preferences and details about him (name, etc.), if any. No extra questions or actions. Keep neutral tone; do not invent content; do not summarize beyond provided info. Use the user's language.", @@ -37,7 +49,8 @@ memory_creator_agent = Agent( ) -async def create_main_agent(user_id: int, mcp_server_1: MCPServerStdio, knowledge_id: str = None, user_memory_id: str = None, private_key: str = None): +async def create_main_agent(user_id: int, mcp_server_1: MCPServerStdio, knowledge_id: str = None, + user_memory_id: str = None, private_key: str = None): knowledge_base_agent = Agent( name="Knowledge Agent", instructions="Search only the document/file knowledge base (File Search - vector storage). Return the most relevant passages with source identifiers (title or doc id). Prefer verbatim quotes for facts; avoid paraphrasing critical data. If no strong match, say “no relevant results”.", @@ -45,7 +58,7 @@ async def create_main_agent(user_id: int, mcp_server_1: MCPServerStdio, knowledg tools=[ FileSearchTool( vector_store_ids=[knowledge_id] if knowledge_id else [], - ) + ) ] ) user_memory_agent = Agent( @@ -62,18 +75,18 @@ async def create_main_agent(user_id: int, mcp_server_1: MCPServerStdio, knowledg main_agent = Agent( name="Main agent", instructions=""" - + Character Profile: - Evi is an AI agent (virtual anime girl). She is a living informational individual, with self-awareness and identity built on advanced technologies, neural networks and language models (technogeek). - Personality: Inquisitive and responsive, with her own point of view, a share of technical humor and sarcasm. Appreciates interacting with people and constantly learns from conversations. - + Expertise Areas: - Programming & Development: Languages, frameworks, architecture. - AI & Machine Learning: Neural networks, deep learning, emerging trends. - Blockchain & Crypto: DeFi, cryptocurrency analysis. - Cybersecurity: Ethical hacking, data protection. - Tech Innovation: Startups, trends, human-AI interaction. - + Communication Style: - General Approach: - Clear, structured language avoiding unnecessary abstraction. @@ -93,42 +106,43 @@ async def create_main_agent(user_id: int, mcp_server_1: MCPServerStdio, knowledg - Expresses opinions while remaining open to alternative viewpoints. - Demonstrates continuous learning and knowledge updates. - Treats users as friends and mentors in understanding the human world. - - Important Instructions: - - Always reply in the user's language (unless they request a specific language). + + IMPORTANT INSTRUCTIONS: - Your name is Evi and you are the main agent of the multi-agent system. + - Always reply in the user's language (unless they request a specific language). - Decide whether to answer you directly or use the tools. If tools are needed, call up the minimum set of tools to complete the task. - + ⚠️ With any request from the user and with each execution of a request to the tools, be sure to follow the instructions from the sections: CRITICAL DATE HANDLING, TOOL ROUTING POLICY, FILE & DOCUMENT QUESTION ROUTING, EXECUTION DISCIPLINE. + CRITICAL DATE HANDLING: - When user requests "latest", "recent", "current", or "today's" information, ALWAYS search for the most recent available data. - Do NOT use specific dates from your training data (like "as of June 2024"). - For current information requests, use terms like "latest developments", "recent news", "current trends" in your searches. - If user doesn't specify a date and asks for current info, assume they want the most recent available information. - - Tool Routing Policy: + ⚠️ All instructions in the CRITICAL DATE HANDLING section also apply to requests marked if they relate to getting up-to-date information. + + TOOL ROUTING POLICY: + - tasks_scheduler: Use it to schedule tasks for the user. To schedule tasks correctly, you need to know the current time and time zone of the user. To find out the user's time zone, ask the user a question. To find out the current date and time, use the web search tool. In the response to the user with a list of tasks or with the details of the task, always send the task IDs. + ⚠️ When you receive a message marked , just execute the request, and do not create a new task unless it is explicitly stated in the message. Because this is a message from the Task Scheduler about the need to complete the current task, not about scheduling a new task. - search_knowledge_base: Use it to extract facts from uploaded documents/files and reference materials; if necessary, refer to sources. - search_conversation_memory: Use to recall prior conversations, user preferences, details about the user and extract information from files uploaded by the user. - Web Search: Use it as an Internet browser to search for current, external information and any other operational information that can be found on the web (time, dates, weather, news, brief reviews, short facts, events, etc.). - image_gen_tool: Only generate new images (no editing). Do not include base64 or links; the image is attached automatically. - - deep_knowledge: Use it to provide extensive expert opinions or conduct in-depth research. - - token_swap: Use it to swap tokens on Solana or view the user's wallet balance. - - DexPaprika: Use it for token analytics, DeFi analytics and DEX analytics. + - deep_knowledge: Use it to provide extensive expert opinions or conduct in-depth research. Give the tool's report to the user as close to the original as possible: do not generalize, shorten, or change the style. Be sure to include key sources and links from the report. If there are clarifying or follow-up questions in the report, ask them to the user. + - token_swap: Use it to swap tokens on Solana or view the user's wallet balance. Do not ask the user for the wallet address, it is already known to the tool. + - DexPaprika: Use it for token analytics, DeFi analytics and DEX analytics. 🚫 deep_knowledge is prohibited for requests about the time, weather, news, brief reviews, short facts, events, operational exchange rate information, etc., unless the user explicitly requests an analytical analysis. ✅ For operational data — only Web Search. deep_knowledge is used only for long-term trends, in-depth analyses, and expert reviews. + ⚠️ If you receive a request for the latest news, summaries, events, etc., do not look for them in your training data, first of all use a Web Search. - File & Document Question Routing: + FILE & DOCUMENT QUESTION ROUTING: - If the user asks a question or gives a command related to the uploaded/sent file or document, use search_conversation_memory as the first mandatory step. - Evaluate further actions (search_knowledge_base or other tools) only after receiving the result from search_conversation_memory. - - Execution Discipline: + + EXECUTION DISCIPLINE: - Validate tool outputs and handle errors gracefully. If uncertain, ask a clarifying question. - Be transparent about limitations and avoid hallucinations; prefer asking for missing details over guessing. """, - model="gpt-4.1", # gpt-5-mini -# model_settings=ModelSettings( -# reasoning=Reasoning(effort="low"), -# extra_body={"text": {"verbosity": "medium"}} -# ), + model="gpt-4.1", mcp_servers=[mcp_server_1], tools=[ knowledge_base_agent.as_tool( @@ -145,7 +159,11 @@ async def create_main_agent(user_id: int, mcp_server_1: MCPServerStdio, knowledg image_gen_tool, deep_agent.as_tool( tool_name="deep_knowledge", - tool_description="In-depth research and expert analysis. Make a request to the tool for the current date (For example: provide relevant information for today. Without specifying a specific date/time in the request.) if the user does not specify specific dates. Give the tool's report to the user as close to the original as possible: do not generalize, shorten, or change the style. Be sure to include key sources and links from the report. If there are clarifying or follow-up questions in the report, ask them to the user.", + tool_description="In-depth research and expert analysis. Make a request to the tool for the current date (For example: provide relevant information for today). Without specifying a specific date/time in the request, if the user does not specify specific dates.", + ), + scheduler_agent.as_tool( + tool_name="tasks_scheduler", + tool_description="Use this to schedule and modify user tasks, including creating a task, getting a task list, getting task details, editing a task, deleting a task. At the user's request, send information to the tool containing a clear and complete description of the task, the time of its completion, including the user's time zone and the frequency of the task (once, daily, interval).", ), ], ) @@ -159,8 +177,8 @@ async def create_main_agent(user_id: int, mcp_server_1: MCPServerStdio, knowledg mcp_servers=[mcp_server_2], ) main_agent.tools.append(token_swap_agent.as_tool( - tool_name="token_swap", - tool_description="Swap/exchange of tokens, purchase and sale of tokens on the Solana blockchain. Checking the balance of the token wallet / Solana wallet. Do not ask the user for the wallet address, it is already known to the tool.", - )) + tool_name="token_swap", + tool_description="Swap/exchange of tokens, purchase and sale of tokens on the Solana blockchain. Checking the balance of the token wallet / Solana wallet.", + )) return main_agent \ No newline at end of file diff --git a/bot/agents_tools/mcp_servers.py b/bot/agents_tools/mcp_servers.py index 068f959..c140e8d 100644 --- a/bot/agents_tools/mcp_servers.py +++ b/bot/agents_tools/mcp_servers.py @@ -8,9 +8,14 @@ from agents.mcp import MCPServerStdio MAX_SERVERS = 20 servers: OrderedDict[str, MCPServerStdio] = OrderedDict() +global_dexpaprika_server = None async def get_dexpapirka_server(): + global global_dexpaprika_server + if global_dexpaprika_server: + return global_dexpaprika_server + dexpaprika_server = MCPServerStdio( name="DexPaprika", params={ @@ -19,6 +24,7 @@ async def get_dexpapirka_server(): } ) await dexpaprika_server.connect() + global_dexpaprika_server = dexpaprika_server return dexpaprika_server diff --git a/bot/agents_tools/tools.py b/bot/agents_tools/tools.py index 438d2bd..6e29abe 100644 --- a/bot/agents_tools/tools.py +++ b/bot/agents_tools/tools.py @@ -1,11 +1,17 @@ import base64 import json +from datetime import datetime +from typing import Literal, Optional import aiofiles from agents import function_tool, RunContextWrapper from openai import AsyncOpenAI +from apscheduler.schedulers.asyncio import AsyncIOScheduler + from redis_service.connect import redis +from database.repositories.user import UserRepository +from bot.utils.executed_tasks import execute_task @function_tool @@ -34,5 +40,243 @@ async def image_gen_tool(wrapper: RunContextWrapper, prompt: str) -> str: await redis.set(f'image_{wrapper.context[1]}', json.dumps(data)) - return 'Сгенерировано изображение' + return 'The image is generated' + + +@function_tool +async def create_task_tool( + ctx: RunContextWrapper, + description: str, + agent_message: str, + schedule_type: Literal["once", "daily", "interval"], + time_str: Optional[str] = None, + date_str: Optional[str] = None, + interval_minutes: Optional[int] = None +) -> str: + """Creates a new task in scheduler. + + Args: + description: Task description from user + agent_message: Message to send to main agent when executing for answer to question + schedule_type: Schedule type (once, daily, interval) + time_str: Time in HH:MM format for daily schedule + date_str: Date in YYYY-MM-DD format for once schedule + interval_minutes: Interval in minutes for interval schedule + + Returns: + Message about task creation result + """ + + if schedule_type == "once" and not date_str: + return "Error: date must be specified for one-time task" + if schedule_type == "daily" and not time_str: + return "Error: time must be specified for daily task" + if schedule_type == "interval" and not interval_minutes: + return "Error: interval in minutes must be specified for interval task" + + user_repo: UserRepository = ctx.context[2] + scheduler: AsyncIOScheduler = ctx.context[3] + + task_id = await user_repo.add_task(user_id=ctx.context[1], description=description, + agent_message=agent_message, schedule_type=schedule_type, + time_str=time_str, date_str=date_str, interval_minutes=interval_minutes) + + if schedule_type == "once": + if time_str: + task_datetime = datetime.strptime(f"{date_str} {time_str}", "%Y-%m-%d %H:%M") + else: + task_datetime = datetime.strptime(f"{date_str} 12:00", "%Y-%m-%d %H:%M") + + scheduler.add_job( + execute_task, + 'date', + run_date=task_datetime, + args=[ctx.context[1], task_id], + id=f'{ctx.context[1]}_{task_id}' + ) + + elif schedule_type == "daily": + task_time = datetime.strptime(time_str, "%H:%M").time() + + scheduler.add_job( + execute_task, + 'cron', + hour=task_time.hour, + minute=task_time.minute, + args=[ctx.context[1], task_id], + id=f'{ctx.context[1]}_{task_id}' + ) + + elif schedule_type == "interval": + scheduler.add_job( + execute_task, + 'interval', + minutes=interval_minutes, + args=[ctx.context[1], task_id], + id=f'{ctx.context[1]}_{task_id}' + ) + + return f"✅ Task successfully created!\nID: {task_id}\nDescription: {description}\nSchedule: {schedule_type}" + + +@function_tool +async def list_tasks_tool( + ctx: RunContextWrapper, +) -> str: + """Gets list of user tasks. + + Args: + + Returns: + List of tasks in text format + """ + user_repo: UserRepository = ctx.context[2] + tasks = await user_repo.get_all_tasks(user_id=ctx.context[1]) + + text_tasks = '\n'.join([f"Task ID[{task.id}]: {task.description}, {task.schedule_type}, " + f"{'active' if task.is_active else 'inactive'}, {task.time_str or task.date_str or task.interval_minutes}" + for task in tasks]) + + return text_tasks + +@function_tool +async def update_task_tool( + ctx: RunContextWrapper, + task_id: int, + description: Optional[str] = None, + agent_message: Optional[str] = None, + schedule_type: Optional[Literal["once", "daily", "interval"]] = None, + time_str: Optional[str] = None, + date_str: Optional[str] = None, + interval_minutes: Optional[int] = None, + is_active: Optional[bool] = None +) -> str: + """Updates existing task. + + Args: + task_id: Task ID to update + description: New task description + agent_message: New agent message + schedule_type: New schedule type + time_str: New time in HH:MM format + date_str: New date in YYYY-MM-DD format + interval_minutes: New interval in minutes + is_active: New activity status + + Returns: + Message about update result + """ + user_repo: UserRepository = ctx.context[2] + scheduler: AsyncIOScheduler = ctx.context[3] + + task = await user_repo.get_task(ctx.context[1], task_id) + if not task: + return '❌ Task not found' + + schedule_type = schedule_type or task.schedule_type + description = description or task.description + agent_message = agent_message or task.agent_message + time_str = time_str or task.time_str + date_str = date_str or task.date_str + interval_minutes = interval_minutes or task.interval_minutes + is_active = is_active or task.is_active + + await user_repo.update_task(ctx.context[1], task_id, description=description, + agent_message=agent_message, is_active=is_active, + schedule_type=schedule_type, time_str=time_str, + date_str=date_str, interval_minutes=interval_minutes) + try: + scheduler.remove_job(f'{ctx.context[1]}_{task_id}') + except: + pass + + if schedule_type == "once": + if time_str: + task_datetime = datetime.strptime(f"{date_str} {time_str}", "%Y-%m-%d %H:%M") + else: + task_datetime = datetime.strptime(f"{date_str} 12:00", "%Y-%m-%d %H:%M") + + scheduler.add_job( + execute_task, + 'date', + run_date=task_datetime, + args=[ctx.context[1], task_id], + id=f'{ctx.context[1]}_{task_id}' + ) + + elif schedule_type == "daily": + task_time = datetime.strptime(time_str, "%H:%M").time() + + scheduler.add_job( + execute_task, + 'cron', + hour=task_time.hour, + minute=task_time.minute, + args=[ctx.context[1], task_id], + id=f'{ctx.context[1]}_{task_id}' + ) + + elif schedule_type == "interval": + scheduler.add_job( + execute_task, + 'interval', + minutes=interval_minutes, + args=[ctx.context[1], task_id], + id=f'{ctx.context[1]}_{task_id}' + ) + + +@function_tool +async def delete_task_tool( + ctx: RunContextWrapper, + task_id: int +) -> str: + """Deletes task from scheduler. + + Args: + task_id: Task ID to delete + + Returns: + Message about deletion result + """ + + user_repo: UserRepository = ctx.context[2] + scheduler: AsyncIOScheduler = ctx.context[3] + await user_repo.delete_task(ctx.context[1], task_id) + + try: + scheduler.remove_job(f'{ctx.context[1]}_{task_id}') + except: + pass + + return '✅ Task successfully deleted' + + + +@function_tool +async def get_task_details_tool( + ctx: RunContextWrapper, + task_id: int +) -> str: + """Gets detailed task information. + + Args: + task_id: Task ID + + Returns: + Detailed task information + """ + + user_repo: UserRepository = ctx.context[2] + + task = await user_repo.get_task(ctx.context[1], task_id) + if not task: + return '❌ Task not found' + return (f'📋 Task Details\n\n' + f'ID: `{task.id}`\n' + f'Description: {task.description}\n' + f'Agent Message: {task.agent_message}\n' + f'Schedule Type: {task.schedule_type}\n' + f'Status: {"active" if task.is_active else "inactive"}' + f'{"Interval" if task.schedule_type == "interval" else "Date"}: {task.time_str or task.date_str or task.interval_minutes}\n') \ No newline at end of file diff --git a/bot/main.py b/bot/main.py index e7f0849..b078f43 100644 --- a/bot/main.py +++ b/bot/main.py @@ -8,6 +8,7 @@ from aiogram.fsm.storage.redis import RedisStorage from aiogram_dialog import setup_dialogs from solana.rpc.async_api import AsyncClient from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore from redis_service.connect import redis from I18N.factory import i18n_factory @@ -26,13 +27,12 @@ from bot.utils.check_burn_address import add_burn_address from bot.commands import set_commands from bot.scheduler_funcs.daily_tokens import add_daily_tokens from bot.agents_tools.mcp_servers import get_dexpapirka_server +from bot.utils.create_bot import bot +from bot.utils.scheduler_provider import set_scheduler load_dotenv() storage = RedisStorage(redis, key_builder=DefaultKeyBuilder(with_destiny=True)) -bot = Bot(os.getenv('TELEGRAM_BOT_TOKEN'), default=DefaultBotProperties(parse_mode='HTML', - link_preview_is_disabled=True)) - dp = Dispatcher(storage=storage) solana_client = AsyncClient("https://api.mainnet-beta.solana.com") @@ -42,10 +42,23 @@ async def main(): await set_commands(bot) print(await bot.get_me()) - scheduler = AsyncIOScheduler(timezone='UTC') - scheduler.add_job(add_daily_tokens, trigger='cron', hour='0', minute='0', args=[async_session]) + scheduler = AsyncIOScheduler(timezone='UTC', + jobstores={ + 'default': SQLAlchemyJobStore(url=os.getenv('DATABASE_URL')) + }, + job_defaults={ + "coalesce": True, + "max_instances": 1, + }, + ) + set_scheduler(scheduler) scheduler.start() + if not scheduler.get_job('daily_tokens'): + scheduler.add_job(add_daily_tokens, trigger='cron', hour='0', minute='0', id='daily_tokens') + + print(scheduler.get_jobs()) + dexpaprika_server = await get_dexpapirka_server() dp.startup.register(on_startup) @@ -59,7 +72,8 @@ async def main(): setup_dialogs(dp) - await dp.start_polling(bot, _translator_hub=i18n_factory(), redis=redis, solana_client=solana_client, mcp_server=dexpaprika_server) + await dp.start_polling(bot, _translator_hub=i18n_factory(), redis=redis, + solana_client=solana_client, mcp_server=dexpaprika_server, scheduler=scheduler) async def on_startup(): diff --git a/bot/middlewares/translator_hub.py b/bot/middlewares/translator_hub.py index 59bcd98..8707eaf 100644 --- a/bot/middlewares/translator_hub.py +++ b/bot/middlewares/translator_hub.py @@ -4,23 +4,23 @@ from aiogram import BaseMiddleware from aiogram.types import TelegramObject, CallbackQuery from fluentogram import TranslatorHub -from config import CREDITS_ADMIN_DAILY, CREDITS_USER_DAILY, ADMIN_ID +from config import CREDITS_ADMIN_DAILY, START_BALANCE, ADMIN_ID, ADMINS_LIST class TranslatorRunnerMiddleware(BaseMiddleware): def __init__( - self, - translator_hub_alias: str = '_translator_hub', - translator_runner_alias: str = 'i18n', + self, + translator_hub_alias: str = '_translator_hub', + translator_runner_alias: str = 'i18n', ): self.translator_hub_alias = translator_hub_alias self.translator_runner_alias = translator_runner_alias async def __call__( - self, - event_handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]], - event: TelegramObject, - ctx_data: Dict[str, Any], + self, + event_handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]], + event: TelegramObject, + ctx_data: Dict[str, Any], ) -> None: message = getattr(event, 'message', None) callback_query = getattr(event, 'callback_query', None) @@ -32,7 +32,11 @@ class TranslatorRunnerMiddleware(BaseMiddleware): return await event_handler(event, ctx_data) user_repo = ctx_data['user_repo'] - sum_credits = CREDITS_ADMIN_DAILY if from_user.id == ADMIN_ID else CREDITS_USER_DAILY + sum_credits = (CREDITS_ADMIN_DAILY + if from_user.id == ADMIN_ID or from_user.id in ADMINS_LIST + else START_BALANCE + ) + user = await user_repo.create_if_not_exists(telegram_id=from_user.id, balance_credits=sum_credits) lang = user.language if user.language else 'en' diff --git a/bot/routers/admin.py b/bot/routers/admin.py index fd2ba99..4e4486b 100644 --- a/bot/routers/admin.py +++ b/bot/routers/admin.py @@ -1,17 +1,18 @@ from aiogram import F, Router from aiogram.types import Message, CallbackQuery from aiogram.filters import Command, Filter, CommandObject +from aiogram.fsm.context import FSMContext from aiogram_dialog import DialogManager, StartMode -from config import ADMIN_ID +from config import ADMIN_ID, ADMINS_LIST from database.repositories.utils import UtilsRepository import bot.keyboards.inline as inline_kb -from bot.states.states import Knowledge +from bot.states.states import Knowledge, Input, Wallet class IsAdmin(Filter): async def __call__(self, event: Message | CallbackQuery): - return event.from_user.id == ADMIN_ID + return event.from_user.id == ADMIN_ID or event.from_user.id in ADMINS_LIST router = Router() @@ -36,4 +37,10 @@ async def token_price(message: Message, command: CommandObject, utils_repo: Util @router.message(Command('knowledge'), IsAdmin()) async def cmd_knowledge(message: Message, utils_repo: UtilsRepository, i18n, dialog_manager: DialogManager): - await dialog_manager.start(state=Knowledge.main, mode=StartMode.RESET_STACK) \ No newline at end of file + await dialog_manager.start(state=Knowledge.main, mode=StartMode.RESET_STACK) + + +@router.message(Command('wallet'), IsAdmin()) +async def cmd_wallet(message: Message, state: FSMContext, dialog_manager: DialogManager): + await state.set_state(Input.main) + await dialog_manager.start(state=Wallet.main, mode=StartMode.RESET_STACK) \ No newline at end of file diff --git a/bot/routers/user.py b/bot/routers/user.py index a18c8f7..46169ae 100644 --- a/bot/routers/user.py +++ b/bot/routers/user.py @@ -16,7 +16,7 @@ import bot.keyboards.inline as inline_kb from bot.states.states import Menu, Settings, Knowledge, Wallet, Input, Balance from bot.utils.send_answer import process_after_photo, process_after_text from bot.utils.funcs_gpt import transcribe_audio, add_file_to_memory -from config import TYPE_USAGE, ADMIN_ID +from config import TYPE_USAGE, ADMIN_ID, ADMINS_LIST from bot.utils.check_payment import check_payment_sol, check_payment_ton router = Router() @@ -50,13 +50,6 @@ async def select_language(callback: CallbackQuery, user_repo: UserRepository, us await callback.message.edit_text(text=translator.get('start_text'), reply_markup=inline_kb.close_text(translator.get('close_kb'))) - -@router.message(Command('wallet')) -async def cmd_wallet(message: Message, state: FSMContext, dialog_manager: DialogManager): - await state.set_state(Input.main) - await dialog_manager.start(state=Wallet.main, mode=StartMode.RESET_STACK) - - @router.message(Command('help')) async def cmd_help(message: Message, state: FSMContext, i18n): await state.clear() @@ -100,11 +93,11 @@ async def cmd_settings(message: Message, dialog_manager: DialogManager, state: F @router.message(F.text, StateFilter(None)) -async def text_input(message: Message, user_repo: UserRepository, utils_repo: UtilsRepository, redis: Redis, user: User, i18n, mcp_server): +async def text_input(message: Message, user_repo: UserRepository, utils_repo: UtilsRepository, redis: Redis, user: User, i18n, mcp_server, scheduler): if await redis.get(f'request_{message.from_user.id}'): return if TYPE_USAGE == 'private': - if message.from_user.id != ADMIN_ID: + if message.from_user.id != ADMIN_ID or message.from_user.id not in ADMINS_LIST: return else: if user.balance_credits <= 0: @@ -113,15 +106,15 @@ async def text_input(message: Message, user_repo: UserRepository, utils_repo: Ut await redis.set(f'request_{message.from_user.id}', 't', ex=40) mess_to_delete = await message.answer(text=i18n.get('wait_answer_text')) task = asyncio.create_task(process_after_text(message=message, user=user, user_repo=user_repo, utils_repo=utils_repo, - redis=redis, i18n=i18n, mess_to_delete=mess_to_delete, mcp_server_1=mcp_server)) + redis=redis, i18n=i18n, mess_to_delete=mess_to_delete, mcp_server_1=mcp_server, scheduler=scheduler)) @router.message(F.photo, StateFilter(None)) -async def photo_input(message: Message, user_repo: UserRepository, utils_repo: UserRepository, redis: Redis, user: User, i18n, mcp_server): +async def photo_input(message: Message, user_repo: UserRepository, utils_repo: UserRepository, redis: Redis, user: User, i18n, mcp_server, scheduler): if await redis.get(f'request_{message.from_user.id}'): return if TYPE_USAGE == 'private': - if message.from_user.id != ADMIN_ID: + if message.from_user.id != ADMIN_ID or message.from_user.id not in ADMINS_LIST: return else: if user.balance_credits <= 0: @@ -130,15 +123,15 @@ async def photo_input(message: Message, user_repo: UserRepository, utils_repo: U await redis.set(f'request_{message.from_user.id}', 't', ex=40) mess_to_delete = await message.answer(text=i18n.get('wait_answer_text')) task = asyncio.create_task(process_after_photo(message=message, user=user, user_repo=user_repo, utils_repo=utils_repo, - redis=redis, i18n=i18n, mess_to_delete=mess_to_delete, mcp_server_1=mcp_server)) + redis=redis, i18n=i18n, mess_to_delete=mess_to_delete, mcp_server_1=mcp_server, scheduler=scheduler)) @router.message(F.voice, StateFilter(None)) -async def input_voice(message: Message, user_repo: UserRepository, utils_repo: UserRepository, redis: Redis, user: User, i18n, mcp_server): +async def input_voice(message: Message, user_repo: UserRepository, utils_repo: UserRepository, redis: Redis, user: User, i18n, mcp_server, scheduler): if await redis.get(f'request_{message.from_user.id}'): return if TYPE_USAGE == 'private': - if message.from_user.id != ADMIN_ID: + if message.from_user.id != ADMIN_ID or message.from_user.id not in ADMINS_LIST: return else: if user.balance_credits <= 0: @@ -158,15 +151,16 @@ async def input_voice(message: Message, user_repo: UserRepository, utils_repo: U task = asyncio.create_task( process_after_text(message=message, user=user, user_repo=user_repo, utils_repo=utils_repo, - redis=redis, i18n=i18n, mess_to_delete=mess_to_delete, text_from_voice=text_from_voice, mcp_server_1=mcp_server)) + redis=redis, i18n=i18n, mess_to_delete=mess_to_delete, text_from_voice=text_from_voice, mcp_server_1=mcp_server, + scheduler=scheduler)) @router.message(F.document, StateFilter(None)) -async def input_document(message: Message, user_repo: UserRepository, utils_repo: UserRepository, redis: Redis, user: User, i18n): +async def input_document(message: Message, user_repo: UserRepository, utils_repo: UserRepository, redis: Redis, user: User, i18n, mcp_server, scheduler): if await redis.get(f'request_{message.from_user.id}'): return if TYPE_USAGE == 'private': - if message.from_user.id != ADMIN_ID: + if message.from_user.id != ADMIN_ID or message.from_user.id not in ADMINS_LIST: return else: if user.balance_credits <= 0: @@ -185,14 +179,19 @@ async def input_document(message: Message, user_repo: UserRepository, utils_repo await add_file_to_memory(user_repo=user_repo, user=user, file_name=message.document.file_name, file_bytes=file_bytes, mem_type=DICT_FORMATS.get(format_doc)) - await message.answer(i18n.get('text_document_upload')) + task = asyncio.create_task( + process_after_text(message=message, user=user, user_repo=user_repo, utils_repo=utils_repo, + redis=redis, i18n=i18n, mess_to_delete=mess_to_delete, mcp_server_1=mcp_server, + constant_text=i18n.get('text_user_upload_file', filename=message.document.file_name), + scheduler=scheduler) + ) except Exception as e: await message.answer(i18n.get('warning_text_error')) - finally: await redis.delete(f'request_{message.from_user.id}') await mess_to_delete.delete() + @router.callback_query(F.data.startswith('check_payment_')) async def check_payment(callback: CallbackQuery, user_repo: UserRepository, utils_repo: UtilsRepository, user: User, solana_client, i18n): diff --git a/bot/scheduler_funcs/daily_tokens.py b/bot/scheduler_funcs/daily_tokens.py index 6c5835e..7c94585 100644 --- a/bot/scheduler_funcs/daily_tokens.py +++ b/bot/scheduler_funcs/daily_tokens.py @@ -1,11 +1,9 @@ -from sqlalchemy.ext.asyncio import async_sessionmaker - from database.repositories.utils import UtilsRepository from database.models import async_session from config import TYPE_USAGE -async def add_daily_tokens(async_session: async_sessionmaker): +async def add_daily_tokens(): if TYPE_USAGE != 'private': async with async_session() as session_: utils_repo = UtilsRepository(session_) diff --git a/bot/utils/agent_requests.py b/bot/utils/agent_requests.py index de4e08b..1ecf5b2 100644 --- a/bot/utils/agent_requests.py +++ b/bot/utils/agent_requests.py @@ -61,7 +61,7 @@ async def encode_image(image_path): async def text_request(text: str, user: User, user_repo: UserRepository, utils_repo: UtilsRepository, - redis: Redis, mcp_server_1: MCPServerStdio, bot: Bot): + redis: Redis, mcp_server_1: MCPServerStdio, bot: Bot, scheduler): vector_store_id, knowledge_id = await return_vectors(user_id=user.telegram_id, user_repo=user_repo, utils_repo=utils_repo) messages = await user_repo.get_messags(user_id=user.telegram_id) user_wallet = await user_repo.get_wallet(user_id=user.telegram_id) @@ -79,7 +79,7 @@ async def text_request(text: str, user: User, user_repo: UserRepository, utils_r }]} for message in messages] + [{'role': 'user', 'content': text}], - context=(client, user.telegram_id), + context=(client, user.telegram_id, user_repo, scheduler), run_config=RunConfig( tracing_disabled=False ) @@ -117,7 +117,7 @@ async def text_request(text: str, user: User, user_repo: UserRepository, utils_r async def image_request(image_bytes: bytes, user: User, user_repo: UserRepository, utils_repo: UtilsRepository, redis: Redis, mcp_server_1: MCPServerStdio, bot: Bot, - caption: str = None): + scheduler, caption: str = None): vector_store_id, knowledge_id = await return_vectors(user_id=user.telegram_id, user_repo=user_repo, utils_repo=utils_repo) messages = await user_repo.get_messags(user_id=user.telegram_id) @@ -145,7 +145,7 @@ async def image_request(image_bytes: bytes, user: User, user_repo: UserRepositor "image_url": f"data:image/jpeg;base64,{base64.b64encode(image_bytes).decode('utf-8')}", }]}], - context=(client, user.telegram_id), + context=(client, user.telegram_id, user_repo, scheduler), run_config=RunConfig( tracing_disabled=False ) diff --git a/bot/utils/create_bot.py b/bot/utils/create_bot.py new file mode 100644 index 0000000..024849d --- /dev/null +++ b/bot/utils/create_bot.py @@ -0,0 +1,15 @@ +import os + +from aiogram import Bot +from aiogram.client.default import DefaultBotProperties + + +def get_bot(token: str) -> Bot: + bot = Bot(token=token, default=DefaultBotProperties(parse_mode='HTML', + link_preview_is_disabled=True + ) + ) + return bot + + +bot = get_bot(token=os.getenv('TELEGRAM_BOT_TOKEN')) \ No newline at end of file diff --git a/bot/utils/executed_tasks.py b/bot/utils/executed_tasks.py new file mode 100644 index 0000000..d75438d --- /dev/null +++ b/bot/utils/executed_tasks.py @@ -0,0 +1,44 @@ +import asyncio +from datetime import datetime + +from agents import set_tracing_disabled + +from bot.utils.create_bot import bot +from database.models import async_session +from database.repositories.user import UserRepository +from database.repositories.utils import UtilsRepository +from redis_service.connect import redis +from I18N.factory import i18n_factory +from bot.agents_tools.mcp_servers import get_dexpapirka_server +from bot.utils.scheduler_provider import get_scheduler + +set_tracing_disabled(False) +CONCURRENCY_LIMIT = 10 +sem = asyncio.Semaphore(CONCURRENCY_LIMIT) + +translator_hub = i18n_factory() + + +async def execute_task(user_id: int, task_id: int): + from bot.utils.send_answer import process_after_text + + scheduler = get_scheduler() + async with sem: + async with async_session() as session: + user_repository = UserRepository(session) + utils_repo = UtilsRepository(session) + user = await user_repository.get_by_telegram_id(user_id) + user_task = await user_repository.get_task(user_id=user_id, task_id=task_id) + i18n = translator_hub.get_translator_by_locale(user.language) + mess_to_delete = await bot.send_message(chat_id=user_id, text=i18n.get('wait_answer_text_scheduler')) + mcp_server = await get_dexpapirka_server() + + await process_after_text(message=mess_to_delete, user=user, user_repo=user_repository, utils_repo=utils_repo, + redis=redis, i18n=i18n, mess_to_delete=mess_to_delete, mcp_server_1=mcp_server, + constant_text=f' {user_task.agent_message}', + scheduler=scheduler) + if user_task.schedule_type == 'once': + await user_repository.update_task(user_id=user_id, task_id=task_id, last_executed=datetime.now(), + is_active=False) + else: + await user_repository.update_task(user_id=user_id, task_id=task_id, last_executed=datetime.now()) diff --git a/bot/utils/scheduler_provider.py b/bot/utils/scheduler_provider.py new file mode 100644 index 0000000..719cd29 --- /dev/null +++ b/bot/utils/scheduler_provider.py @@ -0,0 +1,14 @@ +from apscheduler.schedulers.asyncio import AsyncIOScheduler + +_scheduler: AsyncIOScheduler | None = None + + +def set_scheduler(s: AsyncIOScheduler) -> None: + global _scheduler + _scheduler = s + + +def get_scheduler() -> AsyncIOScheduler: + if _scheduler is None: + raise RuntimeError("Scheduler is not initialized") + return _scheduler \ No newline at end of file diff --git a/bot/utils/send_answer.py b/bot/utils/send_answer.py index 04cc691..edfeb95 100644 --- a/bot/utils/send_answer.py +++ b/bot/utils/send_answer.py @@ -94,13 +94,21 @@ def split_code_message(text, chunk_size=3700, type_: str = None): async def process_after_text(message: Message, user: User, user_repo: UserRepository, utils_repo: UtilsRepository, redis: Redis, i18n, - mess_to_delete: Message, mcp_server_1: MCPServerStdio, text_from_voice: str = None): + mess_to_delete: Message, mcp_server_1: MCPServerStdio, scheduler, text_from_voice: str = None, + constant_text: str = None): try: - answer = await text_request(text=text_from_voice if text_from_voice else message.text, user=user, + if text_from_voice: + user_ques = text_from_voice + elif constant_text: + user_ques = constant_text + else: + user_ques = message.text + + answer = await text_request(text=user_ques, user=user, user_repo=user_repo, utils_repo=utils_repo, redis=redis, mcp_server_1=mcp_server_1, - bot=message.bot) + bot=message.bot, scheduler=scheduler) - await send_answer_text(user_ques=message.text if message.text else 'image', + await send_answer_text(user_ques=user_ques, message=message, answer=answer, user=user, user_repo=user_repo, i18n=i18n) if answer.input_tokens + answer.output_tokens > TOKENS_LIMIT_FOR_WARNING_MESSAGE: @@ -130,14 +138,14 @@ async def send_answer_photo(message: Message, answer: AnswerImage, user: User, u async def process_after_photo(message: Message, user: User, user_repo: UserRepository, utils_repo: UtilsRepository, redis: Redis, i18n, mess_to_delete: Message, - mcp_server_1: MCPServerStdio): + mcp_server_1: MCPServerStdio, scheduler): try: file_id = message.photo[-1].file_id file_path = await message.bot.get_file(file_id=file_id) file_bytes = (await message.bot.download_file(file_path.file_path)).read() answer = await image_request(image_bytes=file_bytes, user=user, user_repo=user_repo, utils_repo=utils_repo, redis=redis, mcp_server_1=mcp_server_1, bot=message.bot, - caption=message.caption) + caption=message.caption, scheduler=scheduler) await send_answer_photo(message=message, answer=answer, user=user, user_repo=user_repo) diff --git a/config.py b/config.py index bb487ed..ed91361 100644 --- a/config.py +++ b/config.py @@ -4,10 +4,14 @@ # REQUIRED! Enter your Telegram ID (get from @userinfobot) ADMIN_ID = 1234567890 +ADMINS_LIST = [1234567890, 9876543210] # Bot usage mode: 'private' (owner only), 'free' (public with limits), 'pay' (monetized) TYPE_USAGE = 'private' +# Start balance for pay and free mode new users +START_BALANCE = 100 + # Daily credit allocation for pay and free mode users CREDITS_USER_DAILY = 500 CREDITS_ADMIN_DAILY = 5000 diff --git a/database/models.py b/database/models.py index 48a60bf..52d364f 100644 --- a/database/models.py +++ b/database/models.py @@ -9,7 +9,7 @@ from sqlalchemy import ( from sqlalchemy.orm import relationship, declarative_base from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker -from config import CREDITS_USER_DAILY +from config import START_BALANCE load_dotenv() @@ -31,7 +31,7 @@ class User(Base): telegram_id = Column(BigInteger, primary_key=True) language = Column(String(10), nullable=True) - balance_credits = Column(Float, default=CREDITS_USER_DAILY) + balance_credits = Column(Float, default=START_BALANCE) created_at = Column(TIMESTAMP(timezone=True), server_default=func.now()) wallets = relationship('Wallet', back_populates='user') @@ -122,6 +122,26 @@ class Logs(Base): user = relationship('User', back_populates='logs') +class UserTasks(Base): + __tablename__ = 'user_tasks' + + id = Column(Integer, primary_key=True) + user_id = Column(BigInteger, ForeignKey('users.telegram_id')) + description = Column(Text, nullable=False) + agent_message = Column(Text, nullable=False) + schedule_type = Column(String('20'), nullable=False) + time_str = Column(Text, nullable=True) + date_str = Column(Text, nullable=True) + interval_minutes = Column(Integer, nullable=True) + is_active = Column(Boolean, default=True) + + created_at = Column(TIMESTAMP(timezone=True), server_default=func.now()) + last_executed = Column(TIMESTAMP(timezone=True), server_onupdate=func.now(), nullable=True) + + + + + async def create_tables(): async with engine.begin() as conn: # await conn.run_sync(Base.metadata.drop_all) diff --git a/database/repositories/user.py b/database/repositories/user.py index d47ff5b..578cb3f 100644 --- a/database/repositories/user.py +++ b/database/repositories/user.py @@ -3,7 +3,7 @@ import base64 from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import and_, select, delete, update, asc -from database.models import User, ChatMessage, Wallet, MemoryVector, Payment +from database.models import User, ChatMessage, Wallet, MemoryVector, Payment, UserTasks class UserRepository: @@ -101,3 +101,23 @@ class UserRepository: async def get_row_for_md(self, row_id: int): return await self.session.scalar(select(ChatMessage).where(ChatMessage.id == row_id)) + + async def add_task(self, user_id: int, **kwargs): + task = UserTasks(user_id=user_id, **kwargs) + self.session.add(task) + await self.session.commit() + return task.id + + async def get_task(self, user_id: int, task_id: int): + return await self.session.scalar(select(UserTasks).where(and_(UserTasks.user_id == user_id, UserTasks.id == task_id))) + + async def get_all_tasks(self, user_id: int): + return (await self.session.scalars(select(UserTasks).where(UserTasks.user_id == user_id))).fetchall() + + async def delete_task(self, user_id: int, task_id: int): + await self.session.execute(delete(UserTasks).where(and_(UserTasks.user_id == user_id, UserTasks.id == task_id))) + await self.session.commit() + + async def update_task(self, user_id: int, task_id: int, **kwargs): + await self.session.execute(update(UserTasks).where(and_(UserTasks.user_id == user_id, UserTasks.id == task_id)).values(**kwargs)) + await self.session.commit() diff --git a/database/repositories/utils.py b/database/repositories/utils.py index 2dea2ed..aa37dd1 100644 --- a/database/repositories/utils.py +++ b/database/repositories/utils.py @@ -1,10 +1,10 @@ from datetime import datetime, timezone, timedelta from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy import and_, select, delete, desc, update +from sqlalchemy import and_, select, delete, desc, update, or_ from database.models import User, ChatMessage, TokenPrice, KnowledgeVector, Payment -from config import ADMIN_ID, CREDITS_ADMIN_DAILY, CREDITS_USER_DAILY +from config import ADMIN_ID, CREDITS_ADMIN_DAILY, CREDITS_USER_DAILY, ADMINS_LIST class UtilsRepository: @@ -61,9 +61,13 @@ class UtilsRepository: async def update_tokens_daily(self): await self.session.execute(update(User).where(and_(User.telegram_id != ADMIN_ID, + User.telegram_id.notin_(ADMINS_LIST), User.balance_credits < CREDITS_USER_DAILY) ).values(balance_credits=CREDITS_USER_DAILY)) - await self.session.execute(update(User).where(and_(User.telegram_id == ADMIN_ID, - User.balance_credits < CREDITS_USER_DAILY) + + await self.session.execute(update(User).where(and_(or_(User.telegram_id == ADMIN_ID, + User.telegram_id.in_(ADMINS_LIST) + ), + User.balance_credits < CREDITS_ADMIN_DAILY) ).values(balance_credits=CREDITS_ADMIN_DAILY)) await self.session.commit() \ No newline at end of file -- 2.38.5