import motor.motor_asyncio from datetime import datetime, timedelta class MongoDB: _instances = {} def __new__(cls, uri: str, db_name: str): if (uri, db_name) not in cls._instances: instance = super().__new__(cls) instance.client = motor.motor_asyncio.AsyncIOMotorClient(uri) instance.db = instance.client[db_name] instance.user_data = instance.db["users"] instance.channel_data = instance.db["channels"] instance.premium_users = instance.db['pros'] instance.fsub_status = instance.db['fsub_status'] # New collection for fsub status tracking instance.request_sub = instance.db['request_sub'] # New collection for join request tracking cls._instances[(uri, db_name)] = instance return cls._instances[(uri, db_name)] async def set_channels(self, channels: list[int]): await self.user_data.update_one( {"_id": 1}, {"$set": {"channels": channels}}, upsert=True ) async def get_channels(self) -> list[int]: data = await self.user_data.find_one({"_id": 1}) return data.get("channels", []) if data else [] async def add_channel_user(self, channel_id: int, user_id: int): await self.channel_data.update_one( {"_id": channel_id}, {"$addToSet": {"users": user_id}}, upsert=True ) async def remove_channel_user(self, channel_id: int, user_id: int): await self.channel_data.update_one( {"_id": channel_id}, {"$pull": {"users": user_id}} ) async def get_channel_users(self, channel_id: int) -> list[int]: doc = await self.channel_data.find_one({"_id": channel_id}) return doc.get("users", []) if doc else [] async def is_user_in_channel(self, channel_id: int, user_id: int) -> bool: doc = await self.channel_data.find_one( {"_id": channel_id, "users": {"$in": [user_id]}}, {"_id": 1} ) return doc is not None # ✅ PRO FEATURES async def add_pro(self, user_id: int, expiry_date: datetime = None): try: await self.premium_users.update_one( {'_id': user_id}, {'$set': {'expiry_date': expiry_date}}, upsert=True ) return True except Exception as e: print(f"Failed to add premium user: {e}") return False async def remove_pro(self, user_id: int): try: await self.premium_users.delete_one({'_id': user_id}) return True except Exception as e: print(f"Failed to remove premium user: {e}") return False async def is_pro(self, user_id: int): doc = await self.premium_users.find_one({'_id': user_id}) if not doc: return False if 'expiry_date' not in doc: return True # Legacy premium users without expiry date if doc['expiry_date'] is None: return True # Permanent premium users return doc['expiry_date'] > datetime.now() async def get_pros_list(self): current_time = datetime.now() cursor = self.premium_users.find({ '$or': [ {'expiry_date': None}, # Permanent premium users {'expiry_date': {'$exists': False}}, # Legacy premium users {'expiry_date': {'$gt': current_time}} # Active premium users ] }) return [doc['_id'] async for doc in cursor] async def get_expiry_date(self, user_id: int) -> datetime: doc = await self.premium_users.find_one({'_id': user_id}) return doc.get('expiry_date') if doc else None # ✅ USER FUNCTIONS async def present_user(self, user_id: int) -> bool: found = await self.user_data.find_one({'_id': user_id}) return bool(found) async def add_user(self, user_id: int, ban: bool = False): await self.user_data.insert_one({'_id': user_id, 'ban': ban}) async def full_userbase(self) -> list[int]: # Filter for actual Telegram user IDs (real users have numeric positive IDs) # The 'users' collection also stores system settings documents where _id is a string: # e.g., "fsub_channels", "shortner_settings", "db_channels", "bot_settings", "messages_settings", "admins_list" # and _id: 1 is the channel list. We filter them out using: _id > 1 and type of _id is numeric. cursor = self.user_data.find({"_id": {"$gt": 1, "$type": ["int", "long"]}}) return [doc['_id'] async for doc in cursor] async def del_user(self, user_id: int): await self.user_data.delete_one({'_id': user_id}) async def ban_user(self, user_id: int): await self.user_data.update_one({'_id': user_id}, {'$set': {'ban': True}}) async def unban_user(self, user_id: int): await self.user_data.update_one({'_id': user_id}, {'$set': {'ban': False}}) async def is_banned(self, user_id: int) -> bool: user = await self.user_data.find_one({'_id': user_id}) return user.get('ban', False) if user else False # ✅ FSUB CHANNELS FUNCTIONS async def set_fsub_channels(self, fsub_data: dict): """Store fsub channels data to database for persistence across bot restarts""" await self.user_data.update_one( {"_id": "fsub_channels"}, {"$set": {"channels": fsub_data}}, upsert=True ) async def get_fsub_channels(self) -> dict: """Get fsub channels data from database""" data = await self.user_data.find_one({"_id": "fsub_channels"}) return data.get("channels", {}) if data else {} async def add_fsub_channel(self, channel_id: int, channel_data: list): """Add a single fsub channel to database""" current_data = await self.get_fsub_channels() current_data[str(channel_id)] = channel_data await self.set_fsub_channels(current_data) async def remove_fsub_channel(self, channel_id: int): """Remove a single fsub channel from database""" current_data = await self.get_fsub_channels() current_data.pop(str(channel_id), None) await self.set_fsub_channels(current_data) # ✅ SHORTNER SETTINGS FUNCTIONS async def set_shortner_settings(self, shortner_data: dict): """Store shortner settings to database for persistence across bot restarts""" await self.user_data.update_one( {"_id": "shortner_settings"}, {"$set": {"settings": shortner_data}}, upsert=True ) async def get_shortner_settings(self) -> dict: """Get shortner settings from database""" data = await self.user_data.find_one({"_id": "shortner_settings"}) return data.get("settings", {}) if data else {} async def update_shortner_setting(self, key: str, value: str): """Update a single shortner setting""" current_data = await self.get_shortner_settings() current_data[key] = value await self.set_shortner_settings(current_data) async def get_shortner_status(self) -> bool: """Get shortner on/off status""" settings = await self.get_shortner_settings() return settings.get('enabled', True) # Default is enabled async def set_shortner_status(self, enabled: bool): """Set shortner on/off status""" await self.update_shortner_setting('enabled', enabled) # ✅ FSUB STATUS COLLECTION FUNCTIONS async def update_fsub_status(self, user_id: int, channel_id: int, status: str): """Update user's subscription status for a specific channel""" await self.fsub_status.update_one( {"user_id": user_id, "channel_id": channel_id}, {"$set": {"status": status, "last_updated": datetime.now()}}, upsert=True ) async def get_fsub_status(self, user_id: int, channel_id: int) -> str: """Get user's subscription status for a specific channel""" doc = await self.fsub_status.find_one({"user_id": user_id, "channel_id": channel_id}) return doc.get("status") if doc else None async def remove_fsub_status(self, user_id: int, channel_id: int): """Remove user's fsub status record""" await self.fsub_status.delete_one({"user_id": user_id, "channel_id": channel_id}) async def get_user_fsub_statuses(self, user_id: int) -> dict: """Get all fsub statuses for a user""" cursor = self.fsub_status.find({"user_id": user_id}) statuses = {} async for doc in cursor: statuses[doc["channel_id"]] = doc["status"] return statuses async def clear_expired_fsub_statuses(self, days: int = 7): """Clear fsub status records older than specified days""" cutoff_date = datetime.now() - timedelta(days=days) await self.fsub_status.delete_many({"last_updated": {"$lt": cutoff_date}}) # ✅ REQUEST SUB COLLECTION FUNCTIONS async def add_join_request(self, user_id: int, channel_id: int, request_id: int = None): """Record a join request submission""" await self.request_sub.update_one( {"user_id": user_id, "channel_id": channel_id}, {"$set": { "request_id": request_id, "status": "pending", "submitted_at": datetime.now(), "last_updated": datetime.now() }}, upsert=True ) async def update_join_request_status(self, user_id: int, channel_id: int, status: str): """Update join request status (pending, approved, rejected)""" await self.request_sub.update_one( {"user_id": user_id, "channel_id": channel_id}, {"$set": {"status": status, "last_updated": datetime.now()}} ) async def get_join_request_status(self, user_id: int, channel_id: int) -> str: """Get join request status""" doc = await self.request_sub.find_one({"user_id": user_id, "channel_id": channel_id}) return doc.get("status") if doc else None async def has_submitted_join_request(self, user_id: int, channel_id: int) -> bool: """Check if user has submitted a join request for channel""" doc = await self.request_sub.find_one({"user_id": user_id, "channel_id": channel_id}) return doc is not None async def remove_join_request(self, user_id: int, channel_id: int): """Remove join request record""" await self.request_sub.delete_one({"user_id": user_id, "channel_id": channel_id}) async def get_pending_requests_for_channel(self, channel_id: int) -> list: """Get all pending join requests for a channel""" cursor = self.request_sub.find({"channel_id": channel_id, "status": "pending"}) requests = [] async for doc in cursor: requests.append({ "user_id": doc["user_id"], "request_id": doc.get("request_id"), "submitted_at": doc["submitted_at"] }) return requests async def clear_old_join_requests(self, days: int = 30): """Clear join request records older than specified days""" cutoff_date = datetime.now() - timedelta(days=days) await self.request_sub.delete_many({"submitted_at": {"$lt": cutoff_date}}) async def cleanup_database(self): """Perform comprehensive database maintenance - clean old records and validate data""" try: # Clean old fsub status records (older than 7 days) await self.clear_expired_fsub_statuses(7) # Clean old join request records (older than 30 days) await self.clear_old_join_requests(30) # Additional cleanup for orphaned records await self.cleanup_orphaned_records() return True except Exception as e: print(f"Database cleanup error: {e}") return False async def cleanup_orphaned_records(self): """Clean up records that are no longer valid""" try: # Remove fsub status records for users who no longer exist users = await self.full_userbase() user_ids_set = set(users) # Clean fsub_status collection async for doc in self.fsub_status.find({"user_id": {"$nin": users}}): await self.fsub_status.delete_one({"_id": doc["_id"]}) # Clean request_sub collection async for doc in self.request_sub.find({"user_id": {"$nin": users}}): await self.request_sub.delete_one({"_id": doc["_id"]}) return True except Exception as e: print(f"Error cleaning orphaned records: {e}") return False async def get_comprehensive_fsub_statistics(self): """Get detailed statistics about fsub system""" try: # Basic counts fsub_count = await self.fsub_status.count_documents({}) request_count = await self.request_sub.count_documents({}) pending_requests = await self.request_sub.count_documents({"status": "pending"}) approved_requests = await self.request_sub.count_documents({"status": "approved"}) rejected_requests = await self.request_sub.count_documents({"status": "rejected"}) # Status breakdown status_breakdown = {} async for doc in self.fsub_status.aggregate([ {"$group": {"_id": "$status", "count": {"$sum": 1}}} ]): status_breakdown[doc["_id"]] = doc["count"] # Channel-wise statistics channel_stats = {} async for doc in self.fsub_status.aggregate([ {"$group": {"_id": "$channel_id", "count": {"$sum": 1}}} ]): channel_stats[doc["_id"]] = doc["count"] # Recent activity (last 24 hours) from datetime import datetime, timedelta yesterday = datetime.now() - timedelta(days=1) recent_fsub_updates = await self.fsub_status.count_documents( {"last_updated": {"$gte": yesterday}} ) recent_requests = await self.request_sub.count_documents( {"submitted_at": {"$gte": yesterday}} ) return { "total_fsub_records": fsub_count, "total_join_requests": request_count, "pending_requests": pending_requests, "approved_requests": approved_requests, "rejected_requests": rejected_requests, "status_breakdown": status_breakdown, "channel_statistics": channel_stats, "recent_activity": { "fsub_updates_24h": recent_fsub_updates, "join_requests_24h": recent_requests } } except Exception as e: print(f"Error getting comprehensive fsub statistics: {e}") return {} async def get_user_activity_summary(self, user_id: int): """Get comprehensive activity summary for a specific user""" try: # Get all fsub statuses fsub_statuses = await self.get_user_fsub_statuses(user_id) # Get join request history join_requests = [] async for doc in self.request_sub.find({"user_id": user_id}): join_requests.append({ "channel_id": doc["channel_id"], "status": doc["status"], "submitted_at": doc["submitted_at"], "last_updated": doc["last_updated"] }) # Check if user is banned is_banned = await self.is_banned(user_id) # Check if user is premium is_premium = await self.is_pro(user_id) return { "user_id": user_id, "is_banned": is_banned, "is_premium": is_premium, "fsub_statuses": fsub_statuses, "join_request_history": join_requests, "total_channels_joined": len([s for s in fsub_statuses.values() if s == "joined"]), "total_requests_submitted": len(join_requests) } except Exception as e: print(f"Error getting user activity summary for {user_id}: {e}") return {} async def get_channel_activity_summary(self, channel_id: int): """Get comprehensive activity summary for a specific channel""" try: # Get all users in channel channel_users = await self.get_channel_users(channel_id) # Get fsub status breakdown for this channel status_counts = {} async for doc in self.fsub_status.aggregate([ {"$match": {"channel_id": channel_id}}, {"$group": {"_id": "$status", "count": {"$sum": 1}}} ]): status_counts[doc["_id"]] = doc["count"] # Get join request stats for this channel request_stats = {} async for doc in self.request_sub.aggregate([ {"$match": {"channel_id": channel_id}}, {"$group": {"_id": "$status", "count": {"$sum": 1}}} ]): request_stats[doc["_id"]] = doc["count"] # Get recent activity (last 7 days) from datetime import datetime, timedelta week_ago = datetime.now() - timedelta(days=7) recent_joins = await self.fsub_status.count_documents({ "channel_id": channel_id, "status": "joined", "last_updated": {"$gte": week_ago} }) recent_requests = await self.request_sub.count_documents({ "channel_id": channel_id, "submitted_at": {"$gte": week_ago} }) return { "channel_id": channel_id, "total_users": len(channel_users), "status_breakdown": status_counts, "request_statistics": request_stats, "recent_activity_7d": { "new_joins": recent_joins, "new_requests": recent_requests } } except Exception as e: print(f"Error getting channel activity summary for {channel_id}: {e}") return {} async def bulk_update_user_statuses(self, updates: list): """Perform bulk updates for user statuses - useful for synchronization""" try: operations = [] for update in updates: user_id = update["user_id"] channel_id = update["channel_id"] status = update["status"] operations.append({ "updateOne": { "filter": {"user_id": user_id, "channel_id": channel_id}, "update": { "$set": { "status": status, "last_updated": datetime.now() } }, "upsert": True } }) if operations: result = await self.fsub_status.bulk_write(operations) return result return None except Exception as e: print(f"Error in bulk update user statuses: {e}") return None async def sync_channel_members(self, channel_id: int, current_members: list): """Synchronize database with actual channel members""" try: # Get stored users for this channel stored_users = await self.get_channel_users(channel_id) # Find users to add (in channel but not in database) users_to_add = set(current_members) - set(stored_users) # Find users to remove (in database but not in channel) users_to_remove = set(stored_users) - set(current_members) # Add new users for user_id in users_to_add: await self.add_channel_user(channel_id, user_id) await self.update_fsub_status(user_id, channel_id, "joined") # Remove old users for user_id in users_to_remove: await self.remove_channel_user(channel_id, user_id) await self.update_fsub_status(user_id, channel_id, "left") return { "added": len(users_to_add), "removed": len(users_to_remove), "synced": True } except Exception as e: print(f"Error syncing channel members for {channel_id}: {e}") return {"synced": False, "error": str(e)} async def export_fsub_data(self, channel_id: int = None): """Export force subscription data for backup or analysis""" try: export_data = { "export_timestamp": datetime.now().isoformat(), "fsub_statuses": [], "join_requests": [] } # Filter by channel if specified filter_query = {"channel_id": channel_id} if channel_id else {} # Export fsub statuses async for doc in self.fsub_status.find(filter_query): doc["_id"] = str(doc["_id"]) # Convert ObjectId to string if "last_updated" in doc: doc["last_updated"] = doc["last_updated"].isoformat() export_data["fsub_statuses"].append(doc) # Export join requests async for doc in self.request_sub.find(filter_query): doc["_id"] = str(doc["_id"]) # Convert ObjectId to string if "submitted_at" in doc: doc["submitted_at"] = doc["submitted_at"].isoformat() if "last_updated" in doc: doc["last_updated"] = doc["last_updated"].isoformat() export_data["join_requests"].append(doc) return export_data except Exception as e: print(f"Error exporting fsub data: {e}") return None async def get_fsub_statistics(self): """Get statistics about fsub collections""" try: fsub_count = await self.fsub_status.count_documents({}) request_count = await self.request_sub.count_documents({}) pending_requests = await self.request_sub.count_documents({"status": "pending"}) return { "fsub_status_records": fsub_count, "join_request_records": request_count, "pending_requests": pending_requests } except Exception as e: print(f"Error getting fsub statistics: {e}") return {} # ✅ DB CHANNELS FUNCTIONS async def set_db_channels(self, db_channels_data: dict): """Store DB channels data to database for persistence across bot restarts""" await self.user_data.update_one( {"_id": "db_channels"}, {"$set": {"channels": db_channels_data}}, upsert=True ) async def get_db_channels(self) -> dict: """Get DB channels data from database""" data = await self.user_data.find_one({"_id": "db_channels"}) return data.get("channels", {}) if data else {} async def add_db_channel(self, channel_id: int, channel_data: dict): """Add a single DB channel to database""" current_data = await self.get_db_channels() current_data[str(channel_id)] = channel_data await self.set_db_channels(current_data) async def remove_db_channel(self, channel_id: int): """Remove a single DB channel from database""" current_data = await self.get_db_channels() current_data.pop(str(channel_id), None) await self.set_db_channels(current_data) async def update_db_channel(self, channel_id: int, channel_data: dict): """Update a single DB channel in database""" current_data = await self.get_db_channels() if str(channel_id) in current_data: current_data[str(channel_id)].update(channel_data) await self.set_db_channels(current_data) async def get_primary_db_channel(self) -> int: """Get the primary DB channel ID""" db_channels = await self.get_db_channels() for channel_id_str, channel_data in db_channels.items(): if channel_data.get('is_primary', False): return int(channel_id_str) return None async def set_primary_db_channel(self, channel_id: int): """Set a DB channel as primary (remove primary from others)""" db_channels = await self.get_db_channels() # Remove primary status from all channels for ch_id, ch_data in db_channels.items(): ch_data['is_primary'] = False # Set new primary channel if str(channel_id) in db_channels: db_channels[str(channel_id)]['is_primary'] = True await self.set_db_channels(db_channels) async def get_active_db_channels(self) -> dict: """Get all active DB channels""" db_channels = await self.get_db_channels() active_channels = {} for channel_id_str, channel_data in db_channels.items(): if channel_data.get('is_active', True): active_channels[channel_id_str] = channel_data return active_channels async def toggle_db_channel_status(self, channel_id: int): """Toggle DB channel active/inactive status""" db_channels = await self.get_db_channels() if str(channel_id) in db_channels: current_status = db_channels[str(channel_id)].get('is_active', True) db_channels[str(channel_id)]['is_active'] = not current_status await self.set_db_channels(db_channels) return not current_status return None # ✅ BOT SETTINGS FUNCTIONS async def set_bot_settings(self, settings_data: dict): """Store bot settings to database for persistence across bot restarts""" await self.user_data.update_one( {"_id": "bot_settings"}, {"$set": {"settings": settings_data}}, upsert=True ) async def get_bot_settings(self) -> dict: """Get bot settings from database""" data = await self.user_data.find_one({"_id": "bot_settings"}) return data.get("settings", {}) if data else {} async def update_bot_setting(self, key: str, value): """Update a single bot setting""" current_data = await self.get_bot_settings() current_data[key] = value await self.set_bot_settings(current_data) async def get_bot_setting(self, key: str, default=None): """Get a single bot setting with default fallback""" settings = await self.get_bot_settings() return settings.get(key, default) # ✅ PROTECT CONTENT FUNCTIONS async def set_protect_content(self, enabled: bool): """Set protect content on/off status (permanently forced False by the bot)""" await self.update_bot_setting('protect_content', enabled) async def get_protect_content(self) -> bool: """Get protect content on/off status""" return await self.get_bot_setting('protect_content', False) # ✅ MESSAGES SETTINGS FUNCTIONS async def set_messages_settings(self, messages_data: dict): """Store messages settings to database for persistence across bot restarts""" await self.user_data.update_one( {"_id": "messages_settings"}, {"$set": {"messages": messages_data}}, upsert=True ) async def get_messages_settings(self) -> dict: """Get messages settings from database""" data = await self.user_data.find_one({"_id": "messages_settings"}) return data.get("messages", {}) if data else {} async def update_message_setting(self, key: str, value: str): """Update a single message setting""" current_data = await self.get_messages_settings() current_data[key] = value await self.set_messages_settings(current_data) async def get_message_setting(self, key: str, default: str = ""): """Get a single message setting with default fallback""" messages = await self.get_messages_settings() return messages.get(key, default) # ✅ ADMIN SETTINGS FUNCTIONS async def set_admins_list(self, admins_list: list): """Store admins list to database for persistence across bot restarts""" await self.user_data.update_one( {"_id": "admins_list"}, {"$set": {"admins": admins_list}}, upsert=True ) async def get_admins_list(self) -> list: """Get admins list from database""" data = await self.user_data.find_one({"_id": "admins_list"}) return data.get("admins", []) if data else [] async def add_admin(self, admin_id: int): """Add an admin to the database""" current_admins = await self.get_admins_list() if admin_id not in current_admins: current_admins.append(admin_id) await self.set_admins_list(current_admins) return True return False async def remove_admin(self, admin_id: int): """Remove an admin from the database""" current_admins = await self.get_admins_list() if admin_id in current_admins: current_admins.remove(admin_id) await self.set_admins_list(current_admins) return True return False # ✅ BATCH SETTINGS FUNCTIONS async def save_all_settings(self, bot_settings: dict, messages: dict, admins: list): """Save all settings in a single transaction for efficiency""" try: await self.set_bot_settings(bot_settings) await self.set_messages_settings(messages) await self.set_admins_list(admins) return True except Exception as e: print(f"Error saving all settings: {e}") return False async def load_all_settings(self) -> dict: """Load all settings in a single call for efficiency""" try: bot_settings = await self.get_bot_settings() messages = await self.get_messages_settings() admins = await self.get_admins_list() shortner_settings = await self.get_shortner_settings() return { "bot_settings": bot_settings, "messages": messages, "admins": admins, "shortner_settings": shortner_settings } except Exception as e: print(f"Error loading all settings: {e}") return { "bot_settings": {}, "messages": {}, "admins": [], "shortner_settings": {} }