turns-00048.parquet:35248
9b25733c710b3a288f8dd2a3degenerate_repetitionAbsentFinal dense release
USER
Project structure:
DarkBot/
├── darkData.py
├── darkbot.py
├── filters.py
└── worker.py
Project files:
darkbot.py
```py
import asyncio
import logging
import sys
from typing import Union
import aiogram
from aiogram import Bot, Dispatcher, types
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.filters import Command, BaseFilter
from aiogram.types import Message
from DarkBot.darkData import DarkData
from DarkBot.filters import DarkCommand, DarkFilter
class DarkBot:
def __init__(
self,
token: str
):
self.bot = Bot(
token=token,
default=DefaultBotProperties(parse_mode=ParseMode.HTML)
)
self.aiogram_dispatcher = Dispatcher()
# Ваши собственные декораторы
def message(self, filter: Union[DarkFilter, BaseFilter]):
def decorator(func):
print(f"tyyyyyype: {type(filter)}")
if isinstance(filter, DarkFilter):
aiogram_filter = filter.get_inst()
elif isinstance(filter, Command):
aiogram_filter = filter
else:
raise TypeError("Invalid filter type. Must be DarkFilter or aiogram.filters.BaseFilter.")
@self.aiogram_dispatcher.message(aiogram_filter)
async def wrapper(message: types.Message):
print("cmmdd!")
return await func(DarkData(message))
return wrapper
return decorator
def start(self, is_logging=False):
"""
starts the bot with asyncio
:param logging:
:return:
"""
async def main() -> None:
await self.aiogram_dispatcher.start_polling(self.bot)
if is_logging:
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
asyncio.run(main())
```
darkData.py
```py
from typing import TYPE_CHECKING
from aiogram.types import Message
if TYPE_CHECKING:
from aiogram import Bot
class DarkUser:
def __init__(self, aiogram_user):
user_id = aiogram_user.id
self.data = {}
self.id = user_id
class DarkData:
def __init__(self, message: Message):
self.user = DarkUser(message.from_user)
self.message: Message = message
```
filters.py
```py
from typing import Optional, List
from aiogram.filters import BaseFilter, Command
class DarkFilter:
def get_inst(self) -> BaseFilter:
raise NotImplementedError("You must implement get_inst() for your specific logic")
class DarkCommand(DarkFilter):
def __init__(
self,
commands: List[str], # Используем List[str] для type hinting
prefix: str = "/"
):
self.commands = commands
self.prefix = prefix
def get_inst(self) -> Command: # Указываем возвращаемый тип Command
return Command(commands=self.commands, prefix=self.prefix)
```
worker.py
```py
# Ваши собственные декораторы
def on_start():
def decorator(func):
@_dp.message(Command("start"))
async def wrapper(message: Message):
print("start!") # Ваш дополнительный код, если нужен
return await func(message)
return wrapper
return decorator
def on_help():
def decorator(func):
@_dp.message(Command("help"))
async def wrapper(message: Message):
return await func(message)
return wrapper
return decorator
```
Я разрабатываю свой фреймворк для простого создания телеграм ботов на основе aiogram.
Основная идея заключается в том, чтобы все декораторы возвращали DarkData, в котором будет простое сообщение aiogram, а также класс User, в котором будет все нуное и удобное.
Например, я хочу сделать так, чтобы был метод .set_val(), который бы принимал что угодно(2 аргумента, ключ-значение).
Напеиши мне пожалуйста отдельный класс бд sqlite3, который бы создавал .db пользователей, где бы был id и значение darkdata(json).
Должны быть высокоуровневые методы, способные, как я описал выше, просто добавлять и получать значения непосредственно в User.
Это будет очень удобно, т.к. в телеграмме нет переменных пользователей.
Все должно рабоатть максимально быстро и рассчитано на большой параллельный трафик. Должна быть асинхронность.
Можешь пожалуйста продумать что еще можно добавить в класс User, чего не хватало бы в обычном aiogram и что было бы дико удобно.
Спасибо!ASSISTANT
Certainly! Based on your requirements, I'll provide you with an asynchronous SQLite3 database class using `aiosqlite` and integrate it into your `DarkUser` class. This will allow you to store and retrieve user-specific data efficiently. Additionally, I'll suggest some useful features to enhance the `DarkUser` class.
Firstly, let's create the `DarkDB` class to handle the database operations asynchronously.
```python
# db.py
import aiosqlite
import json
import asyncio
class DarkDB:
def __init__(self, db_path="darkbot.db"):
self.db_path = db_path
self._init_lock = asyncio.Lock()
self._initialized = False
async def init_db(self):
async with self._init_lock:
if not self._initialized:
async with aiosqlite.connect(self.db_path) as db:
await db.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
darkdata TEXT
)
''')
await db.commit()
self._initialized = True
async def get_user_data(self, user_id: int) -> dict:
await self.init_db()
async with aiosqlite.connect(self.db_path) as db:
async with db.execute('SELECT darkdata FROM users WHERE id = ?', (user_id,)) as cursor:
row = await cursor.fetchone()
if row:
return json.loads(row[0])
else:
return {}
async def set_user_data(self, user_id: int, data: dict):
await self.init_db()
darkdata_json = json.dumps(data)
async with aiosqlite.connect(self.db_path) as db:
await db.execute('''
INSERT INTO users (id, darkdata) VALUES (?, ?)
ON CONFLICT(id) DO UPDATE SET darkdata = excluded.darkdata
''', (user_id, darkdata_json))
await db.commit()
```
Now, let's update the `DarkUser` class to include methods for setting and getting values.
```python
# darkData.py
from typing import TYPE_CHECKING
from aiogram.types import Message
from DarkBot.db import DarkDB
if TYPE_CHECKING:
from aiogram import Bot
class DarkUser:
def __init__(self, aiogram_user):
self.id = aiogram_user.id
self.data = {}
self.db = DarkDB()
self._initialized = False
async def init(self):
if not self._initialized:
self.data = await self.db.get_user_data(self.id)
self._initialized = True
async def set_val(self, key, value):
await self.init()
self.data[key] = value
await self.db.set_user_data(self.id, self.data)
async def get_val(self, key, default=None):
await self.init()
return self.data.get(key, default)
class DarkData:
def __init__(self, message: Message):
self.message: Message = message
self.user = DarkUser(message.from_user)
```
**Usage Example:**
```python
# worker.py
from DarkBot.darkbot import DarkBot
from DarkBot.filters import DarkCommand
bot = DarkBot(token="YOUR_BOT_TOKEN")
@bot.message(DarkCommand(commands=["start"]))
async def start_handler(data):
await data.user.set_val("started", True)
await data.message.reply("Welcome! Your data has been saved.")
@bot.message(DarkCommand(commands=["check"]))
async def check_handler(data):
started = await data.user.get_val("started", False)
if started:
await data.message.reply("You have already started.")
else:
await data.message.reply("You haven't started yet.")
if __name__ == "__main__":
bot.start(is_logging=True)
```
In this setup, each `DarkUser` instance can asynchronously set and get values that are stored in a SQLite database. Using `aiosqlite` ensures non-blocking database operations, which is essential for handling high parallel traffic efficiently.
**Additional Features for `DarkUser` Class:**
1. **Caching Mechanism:**
- Implement an in-memory cache (e.g., using `asyncio.Lock` and a dictionary) to reduce database reads for frequently accessed data.
- Example:
```python
self._cache = {}
async def get_val(self, key, default=None):
await self.init()
if key in self._cache:
return self._cache[key]
value = self.data.get(key, default)
self._cache[key] = value
return value
```
2. **State Management:**
- Incorporate a finite-state machine to manage user states in conversations.
- Provide methods like `set_state(state_name)` and `get_state()` to handle complex dialogues.
- Example:
```python
async def set_state(self, state: str):
await self.set_val('_state', state)
async def get_state(self):
return await self.get_val('_state', None)
```
3. **Permission and Role Handling:**
- Add methods to assign and check user roles or permissions.
- Example:
```python
async def set_role(self, role: str):
await self.set_val('role', role)
async def has_role(self, role: str):
return await self.get_val('role') == role
```
4. **Messaging Shortcuts:**
- Provide methods to send messages directly from `DarkUser`.
- Example:
```python
async def send_message(self, text: str, **kwargs):
await self.message.bot.send_message(chat_id=self.id, text=text, **kwargs)
```
5. **Session Management:**
- Implement session data storage to persist temporary data across messages.
- Example:
```python
async def set_session(self, key, value):
session = await self.get_val('_session', {})
session[key] = value
await self.set_val('_session', session)
async def get_session(self, key, default=None):
session = await self.get_val('_session', {})
return session.get(key, default)
```
6. **Localization Support:**
- Add methods to handle user language preferences and localized messages.
- Example:
```python
async def set_language(self, language_code: str):
await self.set_val('language', language_code)
async def get_language(self):
return await self.get_val('language', 'en')
```
7. **Event Logging:**
- Record user actions or events for analytics or debugging purposes.
- Example:
```python
async def log_event(self, event_name: str, details: dict = None):
events = await self.get_val('events', [])
events.append({'event': event_name, 'details': details})
await self.set_val('events', events)
```
8. **Rate Limiting:**
- Implement methods to prevent spamming by tracking user actions over time.
- Example:
```python
import time
async def can_perform_action(self, action: str, interval: int):
last_performed = await self.get_val(f'last_{action}', 0)
current_time = time.time()
if current_time - last_performed >= interval:
await self.set_val(f'last_{action}', current_time)
return True
return False
```
**Conclusion:**
By enhancing the `DarkUser` class with these features, you can build rich and interactive Telegram bots more efficiently. The asynchronous database operations ensure scalability, and the additional methods provide convenient abstractions over common tasks.
Let me know if you need further customization or assistance with any part of the implementation!