# buttons_db.py version: 1.2 # script was programmed by: Kodo # For any inquiries: @DevKodo # Data source and operation: @SourceKodo import os import aiosqlite from datetime import datetime DB_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "database") _INIT_DONE: set = set() # يمنع إعادة فحص المخطط لكل تحديث (kodo.py يستدعي init_db لكل تحديث) RESPONSE_TYPES = ("text", "photo", "video", "document", "audio", "url", "submenu") TYPE_ICONS = { "text": "📝", "photo": "🖼", "video": "🎬", "document": "📎", "audio": "🎵", "url": "🔗", "submenu": "📂", } def get_db_path(bot_username: str) -> str: os.makedirs(DB_DIR, exist_ok=True) return os.path.join(DB_DIR, f"{bot_username.lower()}.db") async def init_db(bot_username: str): key = bot_username.lower() if key in _INIT_DONE: return async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute(""" CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT ) """) await db.execute(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY, username TEXT, first_name TEXT, is_banned INTEGER DEFAULT 0, is_blocked INTEGER DEFAULT 0, last_active TEXT, joined_at TEXT ) """) await db.execute(""" CREATE TABLE IF NOT EXISTS admins ( id INTEGER PRIMARY KEY, username TEXT, first_name TEXT, added_at TEXT ) """) await db.execute(""" CREATE TABLE IF NOT EXISTS mandatory_channels ( id INTEGER PRIMARY KEY AUTOINCREMENT, channel_id TEXT UNIQUE NOT NULL, channel_username TEXT, channel_title TEXT NOT NULL, invite_link TEXT, display_type TEXT DEFAULT 'buttons', is_active INTEGER DEFAULT 1, added_at TEXT ) """) await db.execute(""" CREATE TABLE IF NOT EXISTS funded_channels ( id INTEGER PRIMARY KEY AUTOINCREMENT, channel_id TEXT UNIQUE NOT NULL, channel_username TEXT, channel_title TEXT NOT NULL, invite_link TEXT, display_type TEXT DEFAULT 'buttons', target_count INTEGER NOT NULL DEFAULT 100, current_count INTEGER DEFAULT 0, is_active INTEGER DEFAULT 1, added_at TEXT, completed_at TEXT ) """) await db.execute(""" CREATE TABLE IF NOT EXISTS funded_subscriptions ( id INTEGER PRIMARY KEY AUTOINCREMENT, funded_channel_id INTEGER NOT NULL, user_id INTEGER NOT NULL, subscribed_at TEXT, UNIQUE(funded_channel_id, user_id) ) """) await db.execute(""" CREATE TABLE IF NOT EXISTS buttons ( id INTEGER PRIMARY KEY AUTOINCREMENT, parent_id INTEGER DEFAULT NULL, label TEXT NOT NULL, response_type TEXT NOT NULL, response_data TEXT DEFAULT '', response_caption TEXT DEFAULT '', columns INTEGER DEFAULT 5, row_number INTEGER DEFAULT 0, sort_order INTEGER DEFAULT 0, press_count INTEGER DEFAULT 0, is_active INTEGER DEFAULT 1, created_at TEXT ) """) try: await db.execute("ALTER TABLE buttons ADD COLUMN row_number INTEGER DEFAULT 0") await db.execute("UPDATE buttons SET row_number = sort_order") await db.commit() except Exception: pass defaults = [ ("is_open", "1"), ("closed_message", "🔴 البوت مقفول حالياً."), ("welcome_message", ""), ("welcome_image", ""), ("notify_new_users", "1"), ("notify_blocked", "1"), ("fund_notify", "1"), ("root_columns", "5"), ] for dkey, value in defaults: await db.execute( "INSERT OR IGNORE INTO settings (key, value) VALUES (?, ?)", (dkey, value) ) await db.commit() _INIT_DONE.add(bot_username.lower()) async def get_setting(bot_username: str, key: str, default: str = "") -> str: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute("SELECT value FROM settings WHERE key=?", (key,)) as cur: row = await cur.fetchone() return row[0] if row else default async def set_setting(bot_username: str, key: str, value: str): async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute( "INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)", (key, value) ) await db.commit() async def get_all_settings(bot_username: str) -> dict: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute("SELECT key, value FROM settings") as cur: return {r[0]: r[1] for r in await cur.fetchall()} def _row_to_btn(row) -> dict: return { "id": row[0], "parent_id": row[1], "label": row[2], "response_type": row[3], "response_data": row[4], "response_caption": row[5], "columns": row[6], "row_number": row[7], "sort_order": row[8], "press_count": row[9], "is_active": bool(row[10]), "created_at": row[11], } _BTN_FIELDS = ( "id,parent_id,label,response_type,response_data,response_caption," "columns,row_number,sort_order,press_count,is_active,created_at" ) async def get_active_buttons(bot_username: str, parent_id: int | None = None) -> list[dict]: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute( f"SELECT {_BTN_FIELDS} FROM buttons " "WHERE is_active=1 AND parent_id IS ? ORDER BY row_number, sort_order", (parent_id,) ) as cur: return [_row_to_btn(r) for r in await cur.fetchall()] async def get_all_buttons_at_level(bot_username: str, parent_id: int | None = None) -> list[dict]: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute( f"SELECT {_BTN_FIELDS} FROM buttons " "WHERE parent_id IS ? ORDER BY row_number, sort_order", (parent_id,) ) as cur: return [_row_to_btn(r) for r in await cur.fetchall()] async def get_button(bot_username: str, button_id: int) -> dict | None: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute( f"SELECT {_BTN_FIELDS} FROM buttons WHERE id=?", (button_id,) ) as cur: row = await cur.fetchone() return _row_to_btn(row) if row else None async def get_rows_info(bot_username: str, parent_id: int | None) -> list[dict]: buttons = await get_all_buttons_at_level(bot_username, parent_id) rows: dict[int, int] = {} for btn in buttons: rn = btn["row_number"] rows[rn] = rows.get(rn, 0) + 1 return [{"row_number": rn, "count": cnt} for rn, cnt in sorted(rows.items())] async def get_next_row_number(bot_username: str, parent_id: int | None) -> int: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute( "SELECT COALESCE(MAX(row_number), -1) + 1 FROM buttons WHERE parent_id IS ?", (parent_id,) ) as cur: row = await cur.fetchone() return row[0] if row else 0 async def get_next_sort_in_row(bot_username: str, parent_id: int | None, row_number: int) -> int: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute( "SELECT COALESCE(MAX(sort_order), -1) + 1 FROM buttons " "WHERE parent_id IS ? AND row_number=?", (parent_id, row_number) ) as cur: row = await cur.fetchone() return row[0] if row else 0 async def add_button(bot_username: str, parent_id: int | None, label: str, response_type: str, response_data: str = "", response_caption: str = "", columns: int = 5, row_number: int = 0) -> int: now = datetime.utcnow().isoformat() sort_order = await get_next_sort_in_row(bot_username, parent_id, row_number) async with aiosqlite.connect(get_db_path(bot_username)) as db: cur = await db.execute( "INSERT INTO buttons (parent_id,label,response_type,response_data," "response_caption,columns,row_number,sort_order,press_count,is_active,created_at) " "VALUES (?,?,?,?,?,?,?,?,0,1,?)", (parent_id, label, response_type, response_data, response_caption, columns, row_number, sort_order, now) ) await db.commit() return cur.lastrowid async def update_button_label(bot_username: str, button_id: int, label: str): async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute("UPDATE buttons SET label=? WHERE id=?", (label, button_id)) await db.commit() async def update_button_content(bot_username: str, button_id: int, response_data: str, response_caption: str = ""): async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute( "UPDATE buttons SET response_data=?, response_caption=? WHERE id=?", (response_data, response_caption, button_id) ) await db.commit() async def update_button_columns(bot_username: str, button_id: int, columns: int): async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute("UPDATE buttons SET columns=? WHERE id=?", (columns, button_id)) await db.commit() async def update_button_row(bot_username: str, button_id: int, row_number: int): btn = await get_button(bot_username, button_id) if not btn: return sort_order = await get_next_sort_in_row(bot_username, btn["parent_id"], row_number) async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute( "UPDATE buttons SET row_number=?, sort_order=? WHERE id=?", (row_number, sort_order, button_id) ) await db.commit() async def move_button_to(bot_username: str, button_id: int, new_parent_id: int | None, new_row: int): sort_order = await get_next_sort_in_row(bot_username, new_parent_id, new_row) async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute( "UPDATE buttons SET parent_id=?, row_number=?, sort_order=? WHERE id=?", (new_parent_id, new_row, sort_order, button_id) ) await db.commit() async def toggle_button(bot_username: str, button_id: int): async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute( "UPDATE buttons SET is_active = 1 - is_active WHERE id=?", (button_id,) ) await db.commit() async def delete_button_cascade(bot_username: str, button_id: int): async with aiosqlite.connect(get_db_path(bot_username)) as db: to_delete = [button_id] queue = [button_id] while queue: cur_id = queue.pop(0) async with db.execute( "SELECT id FROM buttons WHERE parent_id=?", (cur_id,) ) as cur: children = [r[0] for r in await cur.fetchall()] to_delete.extend(children) queue.extend(children) for bid in to_delete: await db.execute("DELETE FROM buttons WHERE id=?", (bid,)) await db.commit() async def move_button(bot_username: str, button_id: int, direction: str): btn = await get_button(bot_username, button_id) if not btn: return all_btns = await get_all_buttons_at_level(bot_username, btn["parent_id"]) idx = next((i for i, b in enumerate(all_btns) if b["id"] == button_id), None) if idx is None: return if direction == "up" and idx > 0: neighbor = all_btns[idx - 1] elif direction == "down" and idx < len(all_btns) - 1: neighbor = all_btns[idx + 1] else: return async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute( "UPDATE buttons SET row_number=?, sort_order=? WHERE id=?", (neighbor["row_number"], neighbor["sort_order"], button_id) ) await db.execute( "UPDATE buttons SET row_number=?, sort_order=? WHERE id=?", (btn["row_number"], btn["sort_order"], neighbor["id"]) ) await db.commit() async def increment_press(bot_username: str, button_id: int): async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute( "UPDATE buttons SET press_count = press_count + 1 WHERE id=?", (button_id,) ) await db.commit() async def get_top_buttons(bot_username: str, limit: int = 5) -> list[dict]: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute( "SELECT id,label,response_type,press_count FROM buttons " "ORDER BY press_count DESC LIMIT ?", (limit,) ) as cur: return [ {"id": r[0], "label": r[1], "response_type": r[2], "press_count": r[3]} for r in await cur.fetchall() ] async def count_buttons(bot_username: str) -> int: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute("SELECT COUNT(*) FROM buttons") as cur: return (await cur.fetchone())[0] async def get_all_submenus(bot_username: str, exclude_id: int | None = None) -> list[dict]: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute( "SELECT id,parent_id,label FROM buttons WHERE response_type='submenu'" ) as cur: rows = [{"id": r[0], "parent_id": r[1], "label": r[2]} for r in await cur.fetchall()] if exclude_id: excluded = {exclude_id} queue = [exclude_id] while queue: cur_id = queue.pop(0) children = [r["id"] for r in rows if r["parent_id"] == cur_id] excluded.update(children) queue.extend(children) rows = [r for r in rows if r["id"] not in excluded] return rows async def build_tree_text(bot_username: str, parent_id: int | None = None, prefix: str = "") -> str: buttons = await get_all_buttons_at_level(bot_username, parent_id) lines = [] for i, btn in enumerate(buttons): is_last = (i == len(buttons) - 1) connector = "└─" if is_last else "├─" icon = TYPE_ICONS.get(btn["response_type"], "•") status = "🟢" if btn["is_active"] else "🔴" presses = f" ({btn['press_count']} ✦)" if btn["press_count"] > 0 else "" row_tag = f" ┤{btn['row_number']+1}├" lines.append(f"{prefix}{connector} {status} {icon} {btn['label']}{row_tag}{presses}") if btn["response_type"] == "submenu": child_prefix = prefix + (" " if is_last else "│ ") child_tree = await build_tree_text(bot_username, btn["id"], child_prefix) if child_tree: lines.append(child_tree) return "\n".join(lines) async def get_or_create_user(bot_username: str, user_id: int, username: str | None, first_name: str) -> dict: now = datetime.utcnow().isoformat() async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute( "SELECT id,username,first_name,is_banned,is_blocked FROM users WHERE id=?", (user_id,) ) as cur: row = await cur.fetchone() if not row: await db.execute( "INSERT INTO users (id,username,first_name,last_active,joined_at) VALUES(?,?,?,?,?)", (user_id, username, first_name, now, now) ) await db.commit() return {"id": user_id, "username": username, "first_name": first_name, "is_banned": 0, "is_blocked": 0, "is_new": True} await db.execute( "UPDATE users SET username=?,first_name=?,last_active=? WHERE id=?", (username, first_name, now, user_id) ) await db.commit() return {"id": row[0], "username": row[1], "first_name": row[2], "is_banned": row[3], "is_blocked": row[4], "is_new": False} async def get_all_active_users(bot_username: str) -> list[dict]: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute( "SELECT id,username,first_name FROM users WHERE is_banned=0 AND is_blocked=0" ) as cur: return [{"id": r[0], "username": r[1], "first_name": r[2]} for r in await cur.fetchall()] async def ban_user(bot_username: str, user_id: int) -> bool: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute("SELECT id FROM users WHERE id=?", (user_id,)) as cur: if not await cur.fetchone(): return False await db.execute("UPDATE users SET is_banned=1 WHERE id=?", (user_id,)) await db.commit() return True async def unban_user(bot_username: str, user_id: int): async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute("UPDATE users SET is_banned=0 WHERE id=?", (user_id,)) await db.commit() async def get_statistics(bot_username: str) -> dict: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute("SELECT COUNT(*) FROM users") as cur: total = (await cur.fetchone())[0] async with db.execute("SELECT COUNT(*) FROM users WHERE is_banned=1") as cur: banned = (await cur.fetchone())[0] async with db.execute("SELECT COUNT(*) FROM users WHERE is_blocked=1") as cur: blocked = (await cur.fetchone())[0] return {"total": total, "banned": banned, "blocked": blocked, "active": total - banned - blocked} async def get_admins(bot_username: str) -> list[dict]: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute("SELECT id,username,first_name FROM admins") as cur: return [{"id": r[0], "username": r[1], "first_name": r[2]} for r in await cur.fetchall()] async def is_admin(bot_username: str, user_id: int) -> bool: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute("SELECT id FROM admins WHERE id=?", (user_id,)) as cur: return bool(await cur.fetchone()) async def add_admin(bot_username: str, user_id: int, username: str | None, first_name: str): now = datetime.utcnow().isoformat() async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute( "INSERT OR REPLACE INTO admins (id,username,first_name,added_at) VALUES(?,?,?,?)", (user_id, username, first_name, now) ) await db.commit() async def remove_admin(bot_username: str, user_id: int): async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute("DELETE FROM admins WHERE id=?", (user_id,)) await db.commit() async def get_mandatory_channels(bot_username: str) -> list[dict]: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute( "SELECT id,channel_id,channel_username,channel_title,invite_link,display_type " "FROM mandatory_channels WHERE is_active=1" ) as cur: return [{"id": r[0], "channel_id": r[1], "channel_username": r[2], "channel_title": r[3], "invite_link": r[4], "display_type": r[5]} for r in await cur.fetchall()] async def add_mandatory_channel(bot_username: str, channel_id: str, channel_title: str, channel_username: str | None, invite_link: str | None, display_type: str = "buttons"): now = datetime.utcnow().isoformat() async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute( "INSERT OR REPLACE INTO mandatory_channels " "(channel_id,channel_username,channel_title,invite_link,display_type,is_active,added_at) " "VALUES(?,?,?,?,?,1,?)", (channel_id, channel_username, channel_title, invite_link, display_type, now) ) await db.commit() async def remove_mandatory_channel(bot_username: str, channel_db_id: int): async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute( "UPDATE mandatory_channels SET is_active=0 WHERE id=?", (channel_db_id,) ) await db.commit() async def get_funded_channels(bot_username: str) -> list[dict]: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute( "SELECT id,channel_id,channel_username,channel_title,invite_link," "display_type,target_count,current_count FROM funded_channels WHERE is_active=1" ) as cur: return [{"id": r[0], "channel_id": r[1], "channel_username": r[2], "channel_title": r[3], "invite_link": r[4], "display_type": r[5], "target_count": r[6], "current_count": r[7]} for r in await cur.fetchall()] async def add_funded_channel(bot_username: str, channel_id: str, channel_title: str, channel_username: str | None, invite_link: str | None, display_type: str, target_count: int): now = datetime.utcnow().isoformat() async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute( "INSERT OR REPLACE INTO funded_channels " "(channel_id,channel_username,channel_title,invite_link,display_type," "target_count,current_count,is_active,added_at) VALUES(?,?,?,?,?,?,0,1,?)", (channel_id, channel_username, channel_title, invite_link, display_type, target_count, now) ) await db.commit() async def remove_funded_channel(bot_username: str, channel_db_id: int): async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute( "UPDATE funded_channels SET is_active=0 WHERE id=?", (channel_db_id,) ) await db.commit() async def record_funded_sub(bot_username: str, channel_db_id: int, user_id: int) -> bool: now = datetime.utcnow().isoformat() async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute( "SELECT id FROM funded_subscriptions WHERE funded_channel_id=? AND user_id=?", (channel_db_id, user_id) ) as cur: if await cur.fetchone(): return False await db.execute( "INSERT INTO funded_subscriptions (funded_channel_id,user_id,subscribed_at) VALUES(?,?,?)", (channel_db_id, user_id, now) ) await db.execute( "UPDATE funded_channels SET current_count=current_count+1 WHERE id=?", (channel_db_id,) ) await db.commit() return True async def get_funded_channel_count(bot_username: str, channel_db_id: int) -> tuple[int, int]: async with aiosqlite.connect(get_db_path(bot_username)) as db: async with db.execute( "SELECT current_count,target_count FROM funded_channels WHERE id=?", (channel_db_id,) ) as cur: row = await cur.fetchone() return (row[0], row[1]) if row else (0, 0) async def mark_funded_complete(bot_username: str, channel_db_id: int): now = datetime.utcnow().isoformat() async with aiosqlite.connect(get_db_path(bot_username)) as db: await db.execute( "UPDATE funded_channels SET is_active=0, completed_at=? WHERE id=?", (now, channel_db_id) ) await db.commit()