import time import asyncio import random import json import logging import os import datetime import shlex import shutil import threading from rubpy import BotClient from rubpy.bot import filters from rubpy.bot.models import Update from db import ( get_user, update_grow, leaderboard, leaderboard_for_chat, transfer_size, create_challenge, get_challenge_by_message, delete_challenge, update_pvp_stats, create_promo, get_promo, has_redeemed, redeem_promo, ban_user, unban_user, is_banned, bot_stats, set_size, get_all_user_ids, check_and_unlock_achievements, record_grow_event, record_user_chat, get_user_achievements_map, resolve_user_id, get_ban_history, get_current_ban, get_user_rank, get_active_event_multipliers, create_event, list_events, get_active_events, delete_event, get_user_group_rank, get_user_group_size, get_user_streak, log_admin_action, get_admin_log, search_users, get_banned_user_ids, get_active_user_ids, get_all_private_chat_ids, get_pending_broadcasts, mark_broadcast_sending, finish_broadcast, ) from config import ( ADMIN_IDS, is_admin, GROW_COOLDOWN, GOLDEN_EGGPLANT_CHANCE, GOLDEN_EGGPLANT_BONUS, ACHIEVEMENTS, EVENT_TEMPLATES, EVENT_TEMPLATES_BY_ID, ) class StatusFormatter(logging.Formatter): ICONS = { logging.DEBUG: "🔵", logging.INFO: "🟢", logging.WARNING: "🟡", logging.ERROR: "🔴", logging.CRITICAL: "🔴", } def format(self, record): icon = self.ICONS.get(record.levelno, "🟢") ts = self.formatTime(record, "%H:%M:%S") return f"{icon} | [{ts}] {record.getMessage()}" _handler = logging.StreamHandler() _handler.setFormatter(StatusFormatter()) logging.basicConfig(level=logging.INFO, handlers=[_handler]) for _noisy_logger in ("httpx", "httpcore", "aiohttp"): logging.getLogger(_noisy_logger).setLevel(logging.WARNING) log = logging.getLogger("bot") BOT_TOKEN = os.environ.get("RUBIKA_BOT_TOKEN", "HBFJD0HOQQDBBQWORNGYJRCCYJCJIBEWTJQNQJFXYKLKTLPSJZRMBERKAINWEJAX") WEBHOOK_URL = os.environ.get("WEBHOOK_URL", "") WEBHOOK_PORT = int(os.environ.get("WEBHOOK_PORT", "8080")) if not BOT_TOKEN: raise SystemExit("RUBIKA_BOT_TOKEN env var is not set.") SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) LANGS = {} _lang_path = os.path.join(SCRIPT_DIR, "lang_fa.json") try: with open(_lang_path, "r", encoding="utf-8") as f: LANGS["fa"] = json.load(f) except FileNotFoundError: log.error("lang_fa.json not found!") LANGS["fa"] = {} except json.JSONDecodeError: log.error("lang_fa.json is corrupted!") LANGS["fa"] = {} def t(key: str, **kwargs) -> str: template = LANGS.get("fa", {}).get(key) or f"[missing: {key}]" try: return template.format(**kwargs) except KeyError: return template def log_id(user_id: str) -> str: return str(user_id) _last_command: dict = {} SPAM_COOLDOWN = 3 def check_spam(user_id: str) -> bool: now = time.time() last = _last_command.get(user_id, 0) if now - last < SPAM_COOLDOWN: return True _last_command[user_id] = now return False async def blocked_if_banned(update: Update) -> bool: sender_id = str(update.new_message.sender_id) if is_banned(sender_id): await update.reply(t("banned_message")) return True return False def parse_args(update: Update): text = update.new_message.text or "" parts = text.split(maxsplit=1) if len(parts) < 2: return [] try: return shlex.split(parts[1]) except (IndexError, ValueError): return parts[1].split() async def sync_user(client, update: Update) -> dict: sender_id = str(update.new_message.sender_id) user = get_user(sender_id) chat_id = str(update.chat_id) try: chat = await client.get_chat(chat_id) log.info(f"get_chat({chat_id}): first_name={chat.first_name}, last_name={chat.last_name}, username={getattr(chat, 'username', None)}") name = (chat.first_name or "") + " " + (chat.last_name or "") name = name.strip() if not name and getattr(chat, "username", None): name = chat.username if not name: name = None except Exception as e: log.error(f"get_chat({chat_id}) failed: {e}") name = None if name: get_user(sender_id, name) user = get_user(sender_id) return user async def sync_user_check(client, update: Update) -> dict: sender_id = str(update.new_message.sender_id) user = await sync_user(client, update) if not is_registered(user): await update.reply(t("not_registered")) return None return user async def announce_new_achievements(update: Update, sender_id: str): for a in check_and_unlock_achievements(sender_id): await update.reply(t( "achievement_unlocked_announcement", emoji=a["emoji"], title=t(f"ach_{a['id']}_title"), desc=t(f"ach_{a['id']}_desc"), )) log.info(f"User {log_id(sender_id)} unlocked achievement: {a['id']}") def leaderboard_lines(tops) -> list: lines = [] for i, (name, size) in enumerate(tops, 1): emoji = "🥇" if i == 1 else "🥈" if i == 2 else "🥉" if i == 3 else "🥕" lines.append(f"{i}. {name}: {size} {emoji}") return lines def is_registered(user: dict) -> bool: return not user["username"].startswith("هویج_") app = BotClient(token=BOT_TOKEN, use_webhook=bool(WEBHOOK_URL)) # ──── Player commands ──── @app.on_update(filters.commands("start")) async def cmd_start(client, update: Update): sender_id = str(update.new_message.sender_id) if check_spam(sender_id): return user = await sync_user(client, update) record_user_chat(sender_id, str(update.chat_id)) await update.reply(t("welcome", username=user["username"])) @app.on_update(filters.commands("grow")) async def cmd_grow(client, update: Update): sender_id = str(update.new_message.sender_id) if check_spam(sender_id): return if await blocked_if_banned(update): return user = await sync_user_check(client, update) if not user: return record_user_chat(sender_id, str(update.chat_id)) mult = get_active_event_multipliers() effective_cooldown = int(GROW_COOLDOWN * mult["cooldown_multiplier"]) now = int(time.time()) if now - user["last_grow"] < effective_cooldown: remaining = effective_cooldown - (now - user["last_grow"]) hours = remaining // 3600 minutes = (remaining % 3600) // 60 await update.reply(t("grow_cooldown", hours=hours, minutes=minutes)) return effective_golden_chance = min(1.0, GOLDEN_EGGPLANT_CHANCE * mult["golden_chance_multiplier"]) effective_golden_bonus = int(GOLDEN_EGGPLANT_BONUS * mult["golden_bonus_multiplier"]) is_golden = random.random() < effective_golden_chance if is_golden: grow_amount = effective_golden_bonus else: grow_amount = round(random.randint(-2, 10) * mult["grow_multiplier"]) old_size = user["size"] if not update_grow(sender_id, grow_amount): await update.reply(t("grow_error")) return record_grow_event(sender_id, is_golden) user = get_user(sender_id) if user["size"] < 0: update_grow(sender_id, -user["size"]) user = get_user(sender_id) tag = "GOLD " if is_golden else "" log.info(f"User {log_id(sender_id)} {tag}grew {old_size} -> {user['size']}") if is_golden: await update.reply(t("grow_gold", amount=effective_golden_bonus, size=user["size"])) elif grow_amount > 0: await update.reply(t("grow_success", amount=grow_amount, size=user["size"])) elif grow_amount < 0: await update.reply(t("grow_shrink", amount=abs(grow_amount), size=user["size"])) else: await update.reply(t("grow_zero")) active_events = get_active_events() if active_events: names = "، ".join(t(f"evt_{e['template_id']}_title") for e in active_events) await update.reply(t("grow_event_active", names=names)) await announce_new_achievements(update, sender_id) @app.on_update(filters.commands("stats")) async def cmd_stats(client, update: Update): sender_id = str(update.new_message.sender_id) if check_spam(sender_id): return user = await sync_user_check(client, update) if not user: return record_user_chat(sender_id, str(update.chat_id)) chat_id = str(update.chat_id) last_grow_str = time.ctime(user["last_grow"]) if user["last_grow"] > 0 else t("never_grown") world_rank = get_user_rank(sender_id) group_rank = get_user_group_rank(sender_id, chat_id) group_size = get_user_group_size(sender_id, chat_id) streak = get_user_streak(sender_id) win_rate = f"{(user['wins'] / user['total_pvp'] * 100):.0f}" if user["total_pvp"] > 0 else "0" achievements = get_user_achievements_map(sender_id) if user["last_grow"] > 0: now = int(time.time()) elapsed = now - user["last_grow"] remaining = GROW_COOLDOWN - elapsed if remaining > 0: mins, secs = divmod(remaining, 60) hours, mins = divmod(mins, 60) cooldown_str = f"{hours} ساعت و {mins} دقیقه" else: cooldown_str = "✅ آماده رشد!" else: cooldown_str = "✅ آماده رشد!" await update.reply(t( "stats_full", username=user["username"], size=user["size"], last_grow=last_grow_str, world_rank=world_rank if world_rank else "—", group_rank=group_rank if group_rank else "—", group_size=group_size, streak=streak, total_pvp=user["total_pvp"], wins=user["wins"], win_rate=win_rate, golden_count=user["golden_count"], achievements_count=len(achievements), cooldown=cooldown_str, )) @app.on_update(filters.commands("top")) async def cmd_top(client, update: Update): sender_id = str(update.new_message.sender_id) if check_spam(sender_id): return chat_id = str(update.chat_id) if sender_id == chat_id: await update.reply(t("top_private_notice")) return tops = leaderboard_for_chat(chat_id, limit=10) if not tops: await update.reply(t("top_group_empty")) return await update.reply(t("top_group_header") + "\n\n" + "\n".join(leaderboard_lines(tops))) @app.on_update(filters.commands("globaltop")) async def cmd_globaltop(client, update: Update): sender_id = str(update.new_message.sender_id) if check_spam(sender_id): return user = await sync_user_check(client, update) if not user: return tops = leaderboard(limit=10) if not tops: await update.reply(t("top_empty")) return await update.reply(t("top_global_header") + "\n\n" + "\n".join(leaderboard_lines(tops))) @app.on_update(filters.commands("help")) async def cmd_help(client, update: Update): sender_id = str(update.new_message.sender_id) if check_spam(sender_id): return await update.reply(t("help")) @app.on_update(filters.commands("idwd")) async def cmd_idwd(client, update: Update): sender_id = str(update.new_message.sender_id) await update.reply(f"`{sender_id}`") @app.on_update(filters.commands("pvp")) async def cmd_pvp(client, update: Update): sender_id = str(update.new_message.sender_id) if check_spam(sender_id): return if await blocked_if_banned(update): return user = await sync_user_check(client, update) if not user: return record_user_chat(sender_id, str(update.chat_id)) args = parse_args(update) if not args: await update.reply(t("pvp_invalid_amount")) return try: amount = int(args[0]) if amount <= 0: raise ValueError except ValueError: await update.reply(t("pvp_invalid_amount")) return if user["size"] < amount: await update.reply(t("pvp_low_size", amount=amount)) return challenge_msg = await update.reply(t("pvp_start", amount=amount, username=user["username"])) create_challenge(str(update.chat_id), int(challenge_msg.message_id), int(time.time()), sender_id, amount) @app.on_update(filters.commands("accept")) async def cmd_accept(client, update: Update): sender_id = str(update.new_message.sender_id) if check_spam(sender_id): return if await blocked_if_banned(update): return if not update.new_message.reply_to_message_id: await update.reply(t("pvp_accept_need_reply")) return chat_id = str(update.chat_id) accepter_id = sender_id accepter = await sync_user_check(client, update) if not accepter: return record_user_chat(accepter_id, chat_id) replied_message_id = int(update.new_message.reply_to_message_id) challenge = get_challenge_by_message(chat_id, replied_message_id) if not challenge: await update.reply(t("pvp_accept_no_challenge")) return amount = challenge["amount"] challenger_id = challenge["challenger_id"] if challenger_id == accepter_id: await update.reply(t("pvp_self_challenge")) return if accepter["size"] < amount: await update.reply(t("pvp_accept_low_size", amount=amount)) return delete_challenge(chat_id, replied_message_id) players = [challenger_id, accepter_id] winner_id = random.choice(players) loser_id = [p for p in players if p != winner_id][0] winner = get_user(winner_id) loser = get_user(loser_id) if transfer_size(loser_id, winner_id, amount): update_pvp_stats(winner_id, True) update_pvp_stats(loser_id, False) await update.reply(t("pvp_winner", winner=winner["username"], loser=loser["username"], amount=amount)) log.info(f"PVP: {log_id(winner_id)} beat {log_id(loser_id)} for {amount}") mult = get_active_event_multipliers() if mult["pvp_bonus_percent"] > 0: bonus = round(amount * mult["pvp_bonus_percent"] / 100) if bonus > 0: update_grow(winner_id, bonus) await update.reply(t("pvp_event_bonus", winner=winner["username"], bonus=bonus)) await announce_new_achievements(update, winner_id) else: await update.reply(t("pvp_transfer_error")) @app.on_update(filters.commands("achievements")) async def cmd_achievements(client, update: Update): sender_id = str(update.new_message.sender_id) if check_spam(sender_id): return user = await sync_user_check(client, update) if not user: return unlocked_map = get_user_achievements_map(sender_id) lines = [t("achievements_header")] for a in ACHIEVEMENTS: title = t(f"ach_{a['id']}_title") desc = t(f"ach_{a['id']}_desc") if a["id"] in unlocked_map: date_str = datetime.datetime.fromtimestamp(unlocked_map[a["id"]]).strftime("%Y-%m-%d") lines.append(t("achievement_line_unlocked", emoji=a["emoji"], title=title, desc=desc, date=date_str)) else: lines.append(t("achievement_line_locked", emoji=a["emoji"], title=title, desc=desc)) lines.append(t("achievements_footer", unlocked=len(unlocked_map), total=len(ACHIEVEMENTS))) await update.reply("\n".join(lines)) # ──── Admin commands ──── @app.on_update(filters.commands("redeemcode")) async def cmd_redeemcode_create(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return args = parse_args(update) if len(args) < 3: await update.reply(t("promo_format")) return try: amount = int(args[0]) days = int(args[1]) code = args[2].strip() except ValueError: await update.reply(t("promo_not_number")) return if amount <= 0 or days <= 0 or not code: await update.reply(t("promo_not_positive")) return if create_promo(code, amount, days): await update.reply(t("promo_created", code=code, amount=amount, days=days)) log.info(f"Redeem code created: {code} (+{amount}, {days}d)") else: await update.reply(t("promo_exists")) @app.on_update(filters.commands("admin")) async def cmd_admin(client, update: Update): sender_id = str(update.new_message.sender_id) log.info(f"/admin called by {sender_id}, is_admin={is_admin(sender_id)}") if not is_admin(sender_id): await update.reply(t("unknown_command")) return await update.reply(t("admin_panel")) @app.on_update(filters.commands("eventtemplates")) async def cmd_event_templates(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return lines = [t("eventtemplates_header")] for template in EVENT_TEMPLATES: tid = template["id"] lines.append(t("eventtemplates_line", id=tid, title=t(f"evt_{tid}_title"), desc=t(f"evt_{tid}_desc"))) lines.append(t("eventtemplates_footer")) await update.reply("\n".join(lines)) @app.on_update(filters.commands("eventadd")) async def cmd_event_add(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return text = update.new_message.text or "" try: args = shlex.split(text.split(maxsplit=1)[1]) if len(text.split()) > 1 else [] except (IndexError, ValueError): args = [] if len(args) < 3: await update.reply(t("eventadd_format")) return template_id, start_str, end_str = args[0], args[1], args[2] name = " ".join(args[3:]) if len(args) > 3 else (t(f"evt_{template_id}_title") if template_id in EVENT_TEMPLATES_BY_ID else template_id) if template_id not in EVENT_TEMPLATES_BY_ID: await update.reply(t("eventadd_template_not_found", template_id=template_id)) return try: start_dt = datetime.datetime.strptime(start_str, "%Y-%m-%d %H:%M") end_dt = datetime.datetime.strptime(end_str, "%Y-%m-%d %H:%M") except ValueError: await update.reply(t("eventadd_date_format_error")) return start_at, end_at = int(start_dt.timestamp()), int(end_dt.timestamp()) if end_at <= start_at: await update.reply(t("eventadd_date_order_error")) return event_id = create_event(template_id, name, start_at, end_at, sender_id) if event_id is None: await update.reply(t("eventadd_failed")) return await update.reply(t( "eventadd_created", id=event_id, name=name, desc=t(f"evt_{template_id}_desc"), start=start_str, end=end_str, )) log.info(f"Event #{event_id} created ({template_id}) by admin {log_id(sender_id)}") @app.on_update(filters.commands("events")) async def cmd_events(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return events = list_events() if not events: await update.reply(t("events_empty")) return now = int(time.time()) lines = [t("events_header")] for e in events: if e["start_at"] <= now <= e["end_at"]: status = t("events_status_active") elif now < e["start_at"]: status = t("events_status_upcoming") else: status = t("events_status_finished") start_str = datetime.datetime.fromtimestamp(e["start_at"]).strftime("%Y-%m-%d %H:%M") end_str = datetime.datetime.fromtimestamp(e["end_at"]).strftime("%Y-%m-%d %H:%M") lines.append(t( "events_line", id=e["id"], status=status, name=e["name"], desc=t(f"evt_{e['template_id']}_desc"), start=start_str, end=end_str, )) await update.reply("\n\n".join(lines)) @app.on_update(filters.commands("eventdel")) async def cmd_event_del(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return args = parse_args(update) if not args: await update.reply(t("eventdel_format")) return try: event_id = int(args[0]) except ValueError: await update.reply(t("eventdel_invalid_id")) return if delete_event(event_id): await update.reply(t("eventdel_success", id=event_id)) else: await update.reply(t("eventdel_not_found")) async def _resolve_or_reply(update: Update, raw_identifier: str): target_id = resolve_user_id(raw_identifier) if not target_id: await update.reply(t("resolve_not_found", identifier=raw_identifier)) return None return target_id @app.on_update(filters.commands("addsize")) async def cmd_addsize(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return args = parse_args(update) if len(args) < 2: await update.reply(t("addsize_format")) return target_id = await _resolve_or_reply(update, args[0]) if not target_id: return try: amount = int(args[1]) except ValueError: await update.reply(t("addsize_not_number")) return if update_grow(target_id, amount): user = get_user(target_id) log_admin_action(sender_id, "addsize", target_id, f"+{amount} -> size={user['size']}") await update.reply(t("addsize_success", username=user["username"], size=user["size"])) else: await update.reply(t("addsize_not_found")) @app.on_update(filters.commands("setsize")) async def cmd_setsize(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return args = parse_args(update) if len(args) < 2: await update.reply(t("setsize_format")) return target_id = await _resolve_or_reply(update, args[0]) if not target_id: return try: amount = int(args[1]) except ValueError: await update.reply(t("setsize_not_number")) return if set_size(target_id, amount): user = get_user(target_id) log_admin_action(sender_id, "setsize", target_id, f"={amount}") await update.reply(t("setsize_success", username=user["username"], size=amount)) else: await update.reply(t("setsize_not_found")) @app.on_update(filters.commands("ban")) async def cmd_ban(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return args = parse_args(update) if not args: await update.reply(t("ban_format")) return target_id = await _resolve_or_reply(update, args[0]) if not target_id: return reason = " ".join(args[1:]) if len(args) > 1 else "" ban_user(target_id, reason) user = get_user(target_id) log_admin_action(sender_id, "ban", target_id, reason or None) await update.reply(t("ban_success", username=user["username"])) log.warning(f"User {log_id(target_id)} banned by admin {log_id(sender_id)}" + (f": {reason}" if reason else "")) @app.on_update(filters.commands("unban")) async def cmd_unban(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return args = parse_args(update) if not args: await update.reply(t("unban_format")) return target_id = await _resolve_or_reply(update, args[0]) if not target_id: return if unban_user(target_id): user = get_user(target_id) log_admin_action(sender_id, "unban", target_id) await update.reply(t("unban_success", username=user["username"])) log.info(f"User {log_id(target_id)} unbanned by admin {log_id(sender_id)}") else: await update.reply(t("unban_not_banned")) @app.on_update(filters.commands("userinfo")) async def cmd_userinfo(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return args = parse_args(update) if not args: await update.reply(t("userinfo_format")) return target_id = await _resolve_or_reply(update, args[0]) if not target_id: return user = get_user(target_id) rank = get_user_rank(target_id) unlocked_map = get_user_achievements_map(target_id) current_ban = get_current_ban(target_id) history = get_ban_history(target_id) def fmt_date(ts): return datetime.datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M") lines = [ t("userinfo_header", username=user["username"]), t("userinfo_id", id=target_id), "", t("userinfo_size_rank", size=user["size"], rank=rank if rank else t("userinfo_rank_unknown")), t("userinfo_grows", count=user["grows_count"]), t("userinfo_golden", count=user["golden_count"]), t("userinfo_pvp", total=user["total_pvp"], wins=user["wins"]), t("userinfo_achievements", unlocked=len(unlocked_map), total=len(ACHIEVEMENTS)), ] if current_ban: lines.append( t("userinfo_currently_banned", date=fmt_date(current_ban["banned_at"])) + (t("userinfo_ban_reason", reason=current_ban["reason"]) if current_ban["reason"] else "") ) else: lines.append(t("userinfo_not_banned")) if history: lines.append(t("userinfo_ban_history_header")) for h in history[:10]: action = t("userinfo_action_ban") if h["action"] == "ban" else t("userinfo_action_unban") reason_part = f" ({h['reason']})" if h["reason"] else "" lines.append(t("userinfo_ban_history_line", date=fmt_date(h["at"]), action=action, reason=reason_part)) await update.reply("\n".join(lines)) @app.on_update(filters.commands("botstats")) async def cmd_botstats(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return stats = bot_stats() await update.reply(t( "botstats_header", user_count=stats["user_count"], total_size=stats["total_size"], avg_size=stats["avg_size"], banned_count=stats["banned_count"], )) @app.on_update(filters.commands("broadcast")) async def cmd_broadcast(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return args = parse_args(update) if not args: await update.reply(t("broadcast_format")) return text = " ".join(args) chat_ids = get_all_private_chat_ids() sent, failed = 0, 0 status_msg = await update.reply(t("broadcast_sending", count=len(chat_ids))) for cid in chat_ids: try: msg = t("broadcast_prefix", text=text) await client.send_message(cid, msg) sent += 1 except Exception as e: failed += 1 log.error(f"Broadcast to {cid} failed: {type(e).__name__}: {e}") await asyncio.sleep(0.1) await status_msg.edit_text(t("broadcast_result", sent=sent, failed=failed)) if failed > 0: log.warning(f"Broadcast: sent {sent}, failed {failed}") else: log.info(f"Broadcast: sent to {sent} users") @app.on_update(filters.commands("adminlog")) async def cmd_adminlog(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return logs = get_admin_log(limit=10) if not logs: await update.reply(t("adminlog_empty")) return lines = [t("adminlog_header")] for admin_id, action, target_id, details, at in logs: dt = datetime.datetime.fromtimestamp(at).strftime("%m/%d %H:%M") target_part = f" → {target_id[:8]}..." if target_id else "" detail_part = f" ({details})" if details else "" lines.append(f"• {dt} | {admin_id[:8]}... | {action}{target_part}{detail_part}") await update.reply("\n".join(lines)) @app.on_update(filters.commands("search")) async def cmd_search(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return args = parse_args(update) if not args: await update.reply(t("search_format")) return query = " ".join(args) results = search_users(query, limit=10) if not results: await update.reply(t("search_not_found", query=query)) return lines = [t("search_header", query=query)] for r in results: lines.append(f"• {r['username']} | سایز: {r['size']} | آیدی: `{r['user_id']}`") await update.reply("\n".join(lines)) @app.on_update(filters.commands("broadcastto")) async def cmd_broadcastto(client, update: Update): sender_id = str(update.new_message.sender_id) if not is_admin(sender_id): await update.reply(t("unknown_command")) return args = parse_args(update) if len(args) < 2: await update.reply(t("broadcastto_format")) return target = args[0].lower() text = " ".join(args[1:]) if target == "banned": user_ids = get_banned_user_ids() elif target == "active": user_ids = get_active_user_ids() else: await update.reply(t("broadcastto_invalid_target")) return if not user_ids: await update.reply(t("broadcastto_no_users")) return chat_ids = [] for uid in user_ids: cursor.execute("SELECT chat_id FROM user_chats WHERE user_id = ? AND chat_id LIKE 'b%' LIMIT 1", (uid,)) row = cursor.fetchone() if row: chat_ids.append(row[0]) if not chat_ids: await update.reply(t("broadcastto_no_users")) return status_msg = await update.reply(t("broadcast_sending", count=len(chat_ids))) sent = 0 failed = 0 for cid in chat_ids: try: await client.send_message(cid, t("broadcast_prefix", text=text)) sent += 1 except Exception as e: failed += 1 log.error(f"Broadcastto to {cid} failed: {e}") await asyncio.sleep(0.1) await status_msg.edit_text(t("broadcast_result", sent=sent, failed=failed)) log_admin_action(sender_id, "broadcastto", None, f"target={target}, sent={sent}, failed={failed}") def auto_backup(): try: backup_dir = os.path.join(SCRIPT_DIR, "backups") os.makedirs(backup_dir, exist_ok=True) ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") backup_path = os.path.join(backup_dir, f"game_{ts}.db") shutil.copy2(os.path.join(SCRIPT_DIR, "game.db"), backup_path) log.info(f"Auto backup created: {backup_path}") backups = sorted([f for f in os.listdir(backup_dir) if f.endswith(".db")]) while len(backups) > 7: os.remove(os.path.join(backup_dir, backups.pop(0))) except Exception as e: log.error(f"Auto backup failed: {e}") def _backup_loop(): timer = threading.Timer(86400, _backup_loop) timer.daemon = True timer.start() async def process_broadcast_queue(): """Polls broadcast_queue for jobs created by the web control panel and actually delivers them, since only this running bot process holds the authenticated session. Runs forever alongside the bot's own polling.""" while True: try: for job in get_pending_broadcasts(): mark_broadcast_sending(job["id"]) target = job["target"] if target == "banned": user_ids = get_banned_user_ids() elif target == "active": user_ids = get_active_user_ids() else: user_ids = get_all_user_ids() chat_ids = [] for uid in user_ids: cursor.execute( "SELECT chat_id FROM user_chats WHERE user_id = ? AND chat_id LIKE 'b%' LIMIT 1", (uid,) ) row = cursor.fetchone() if row: chat_ids.append(row[0]) sent, failed = 0, 0 msg = t("broadcast_prefix", text=job["text"]) for cid in chat_ids: try: await app.send_message(cid, msg) sent += 1 except Exception as e: failed += 1 log.error(f"Panel broadcast #{job['id']} to {cid} failed: {e}") await asyncio.sleep(0.1) finish_broadcast(job["id"], sent, failed) log.info(f"Panel broadcast #{job['id']} ({target}): sent {sent}, failed {failed}") except Exception as e: log.error(f"process_broadcast_queue error: {e}") await asyncio.sleep(5) @app.on_update(filters.text) async def handle_text(client, update: Update): text = (update.new_message.text or "").strip() sender_id = str(update.new_message.sender_id) if text.startswith("/"): return if 3 < len(text) <= 30 and " " not in text: promo = get_promo(text) if promo: if int(time.time()) > promo["expires_at"]: await update.reply(t("promo_redeem_expired")) return if has_redeemed(text, sender_id) or not redeem_promo(text, sender_id): await update.reply(t("promo_redeem_already_used")) return update_grow(sender_id, promo["amount"]) await update.reply(t("promo_redeem_success", amount=promo["amount"], code=text)) log.info(f"User {log_id(sender_id)} redeemed code {text} (+{promo['amount']})") def main(): auto_backup() _backup_loop() if WEBHOOK_URL: log.info(f"Bot starting in WEBHOOK mode: {WEBHOOK_URL}") app.run(webhook_url=WEBHOOK_URL, port=WEBHOOK_PORT) else: log.info("Bot starting in POLLING mode...") app.run() if __name__ == "__main__": main()