Skip to content

Instantly share code, notes, and snippets.

@Notookk

Notookk/config.py

Created February 24, 2025 02:17
Show Gist options
  • Save Notookk/123b6322b6ea91b43e19a2d4abc66914 to your computer and use it in GitHub Desktop.
Save Notookk/123b6322b6ea91b43e19a2d4abc66914 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
config.py
import os
# Bot credentials
API_ID = '25193832'
API_HASH = 'e154b1ccb0195edec0bc91ae7efebc2f'
# MongoDB connection string
DB_URI = 'mongodb+srv://copy:[email protected]/?retryWrites=true&w=majority&appName=Cluster0'
LOG_CHANNEL = '-1002240372506'
TOKEN = "7691103794:AAG453Q33UQ1cByVMQ-22ZneBtMHIbP3KZ0"
OWNER_ID = 7875192045 # Change to actual owner ID
ALERT_CHANNEL_ID = "-1002329693689"
MEDIA_DIR = "../media"
DB_PATH = "nsfw_bot.db"
AUTH_USERS = [7875192045]
DB_NAME = "BroadcastBot"
BROADCAST_AS_COPY = True
BOT_OWNER_ID = 7875192045
OWNER_IDS = [7875192045, 6656608288, 6545754981] # Add as many as you want!
Raw
database.py
import sqlite3
from config import DB_PATH
import aiosqlite
import datetime
import asyncio
import motor.motor_asyncio
def init_db():
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("CREATE TABLE IF NOT EXISTS approved_users (user_id INTEGER PRIMARY KEY)")
cursor.execute("CREATE TABLE IF NOT EXISTS user_violations (user_id INTEGER, category TEXT, count INTEGER, PRIMARY KEY(user_id, category))")
cursor.execute("CREATE TABLE IF NOT EXISTS users (user_id INTEGER PRIMARY KEY)")
conn.commit()
conn.close()
def is_approved(user_id):
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("SELECT user_id FROM approved_users WHERE user_id = ?", (user_id,))
result = cursor.fetchone()
conn.close()
return result is not None
def update_violations(user_id, category):
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("INSERT INTO user_violations (user_id, category, count) VALUES (?, ?, 1) ON CONFLICT(user_id, category) DO UPDATE SET count = count + 1", (user_id, category))
conn.commit()
conn.close()
def add_approved_user(user_id):
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("INSERT OR IGNORE INTO approved_users (user_id) VALUES (?)", (user_id,))
conn.commit()
conn.close()
def remove_approved_user(user_id):
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("DELETE FROM approved_users WHERE user_id = ?", (user_id,))
conn.commit()
conn.close()
def get_user_violations(user_id):
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("SELECT category, count FROM user_violations WHERE user_id = ?", (user_id,))
results = cursor.fetchall()
conn.close()
return results
def get_all_users():
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("SELECT user_id FROM approved_users")
result = cursor.fetchall()
conn.close()
return [user[0] for user in result]
class Database:
"""SQLite3 database handler for managing Telegram bot users & groups."""
def __init__(self, db_path="users.db"):
self.db_path = db_path
async def init_db(self):
"""Initialize the database and ensure all necessary columns exist."""
async with aiosqlite.connect(self.db_path) as db:
# ✅ Create users table with started_bot column
await db.execute(
"""CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
username TEXT,
started_bot INTEGER DEFAULT 0
)"""
)
# ✅ Ensure `started_bot` column exists (for old databases)
try:
await db.execute("SELECT started_bot FROM users LIMIT 1")
except aiosqlite.OperationalError:
print("🛠 Adding missing 'started_bot' column to users table...")
await db.execute("ALTER TABLE users ADD COLUMN started_bot INTEGER DEFAULT 0")
# ✅ Create groups table
await db.execute(
"""CREATE TABLE IF NOT EXISTS groups (
id INTEGER PRIMARY KEY,
title TEXT
)"""
)
await db.commit()
async def add_user(self, user_id: int, username: str = None):
"""Add a user when they interact with the bot (avoid duplicates)."""
async with aiosqlite.connect(self.db_path) as db:
await db.execute(
"INSERT OR IGNORE INTO users (id, username) VALUES (?, ?)",
(user_id, username),
)
await db.commit()
async def mark_user_started(self, user_id: int):
"""Mark a user as having started the bot."""
async with aiosqlite.connect(self.db_path) as db:
await db.execute(
"UPDATE users SET started_bot = 1 WHERE id = ?",
(user_id,),
)
await db.commit()
async def add_group(self, group_id: int, title: str):
"""Add a new group when a message is sent there (avoiding duplicates)."""
async with aiosqlite.connect(self.db_path) as db:
await db.execute(
"INSERT OR IGNORE INTO groups (id, title) VALUES (?, ?)",
(group_id, title),
)
await db.commit()
async def get_all_users(self) -> list[int]:
"""Get all user IDs stored in the database."""
async with aiosqlite.connect(self.db_path) as db:
cursor = await db.execute("SELECT id FROM users")
users = await cursor.fetchall()
return [row[0] for row in users] if users else []
async def get_users_who_started(self) -> list[int]:
"""Get all users who have started the bot."""
async with aiosqlite.connect(self.db_path) as db:
cursor = await db.execute("SELECT id FROM users WHERE started_bot = 1")
users = await cursor.fetchall()
return [row[0] for row in users] if users else []
async def get_all_groups(self) -> list[int]:
"""Get all stored group IDs."""
async with aiosqlite.connect(self.db_path) as db:
cursor = await db.execute("SELECT id FROM groups")
groups = await cursor.fetchall()
return [row[0] for row in groups] if groups else []
async def delete_user(self, user_id: int):
"""Delete a user from the database."""
async with aiosqlite.connect(self.db_path) as db:
await db.execute("DELETE FROM users WHERE id = ?", (user_id,))
await db.commit()
async def get_all_notif_user(self) -> list[int]:
"""Retrieve all user IDs who have started the bot."""
async with aiosqlite.connect(self.db_path) as db:
cursor = await db.execute("SELECT id FROM users WHERE started_bot = 1")
users = await cursor.fetchall()
return [row[0] for row in users] if users else []
async def get_total_counts(self) -> dict:
"""Get the total count of users and groups in the database."""
async with aiosqlite.connect(self.db_path) as db:
user_cursor = await db.execute("SELECT COUNT(*) FROM users")
total_users = (await user_cursor.fetchone())[0]
group_cursor = await db.execute("SELECT COUNT(*) FROM groups")
total_groups = (await group_cursor.fetchone())[0]
return {"users": total_users, "groups": total_groups}
Raw
main.py
import logging
import asyncio
import nest_asyncio
from telegram.ext import ApplicationBuilder, Application, CommandHandler, MessageHandler, CallbackQueryHandler, filters
from config import TOKEN
from database import Database
from handlers.copyright import (
start_command,
button_handler,
mute_user,
unmute_user,
ping_u,
get_user_id_from_username,
delete_edited_messages,
delete_invalid_messages,
commands
)
from handlers.nsfw import *
from handlers.utils import *
from handlers.broadcast import *
# ✅ Fix event loop conflict
nest_asyncio.apply()
# ✅ Configure Logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
# ✅ Initialize Database
db = Database()
async def main():
"""Main function to initialize the bot and start polling."""
await db.init_db() # ✅ Ensure database is ready before starting bot
application = ApplicationBuilder().token(TOKEN).build()
# ✅ Register Handlers
application.add_handler(CommandHandler("start", start_command))
application.add_handler(CallbackQueryHandler(button_handler))
application.add_handler(CommandHandler("command", commands))
application.add_handler(CommandHandler("mute", mute_user))
application.add_handler(CommandHandler("unmute", unmute_user))
application.add_handler(CommandHandler("ping", ping_u))
application.add_handler(CommandHandler("info", get_user_id_from_username))
application.add_handler(CommandHandler("userinfo", user_info))
application.add_handler(CommandHandler("myinfo", my_info))
application.add_handler(CommandHandler("sudolist", get_approved_users_list))
application.add_handler(CommandHandler("add", add_approved))
application.add_handler(CommandHandler("remove", remove_approved))
application.add_handler(CommandHandler("broadcast", broadcast))
#application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, delete_slang_words))
application.add_handler(MessageHandler(filters.ALL, handle_media))
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, track_user))
application.add_handler(MessageHandler(filters.ALL, track_group))
application.add_handler(CommandHandler("listusers", list_users_and_groups))
application.add_handler(MessageHandler(filters.UpdateType.EDITED_MESSAGE, delete_edited_messages))
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, delete_invalid_messages))
logger.info("🤖 Bot is running...✅")
await application.run_polling(timeout=30) # Increase timeout to 30 seconds
logger.info("🤖 Bot stopped...✅")
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main()) # ✅ Uses existing event loop (NO conflict)
Raw
nsfw.py
import os
import logging
import ffmpeg
import imageio
import numpy as np
from PIL import Image
from lottie import parsers
from telegram import constants
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters, CallbackContext
from config import TOKEN, OWNER_ID, ALERT_CHANNEL_ID, MEDIA_DIR
from database import is_approved, update_violations, add_approved_user, remove_approved_user, get_user_violations, get_all_users
from .predict import detect_nsfw
from database.database import init_db
# Ensure media directory exists
os.makedirs(MEDIA_DIR, exist_ok=True)
# Setup database
init_db()
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)
# ✅ Convert WebP static stickers to PNG
def convert_webp_to_png(file_path):
try:
png_path = file_path.replace(".webp", ".png")
img = Image.open(file_path).convert("RGB")
img.save(png_path, "PNG")
return png_path
except Exception as e:
logger.error(f"Error converting WebP to PNG: {e}")
return None
# ✅ Extract frame from animated WEBM stickers
def extract_frame_from_webm(input_path):
"""Extracts the first frame from a .webm video file and saves it as a .jpg image."""
output_path = input_path.replace(".webm", ".jpg") # Set output as JPG
try:
reader = imageio.get_reader(input_path, format="webm")
frame = reader.get_next_data() # Get the first frame
reader.close()
# Convert frame to numpy array (ensure it's writable)
frame = np.array(frame, dtype=np.uint8)
# Save frame as JPEG
imageio.imwrite(output_path, frame, format="JPEG")
return output_path
except Exception as e:
print(f"Error extracting frame from WEBM: {e}")
return None
# ✅ Convert TGS stickers (Lottie) to PNG
def convert_tgs_to_png(file_path):
output_path = file_path.replace(".tgs", ".png")
try:
animation = parse_tgs(file_path)
frame = animation.render_frame(0)
frame.save(output_path, "PNG")
return output_path
except Exception as e:
logger.error(f"Error converting TGS to PNG: {e}")
return None
# ✅ Handle all media (images, videos, stickers, GIFs, animated stickers)
async def handle_media(update: Update, context: CallbackContext):
"""Handles all media types and scans for NSFW content."""
user = update.message.from_user
chat_id = update.message.chat_id
media_files = update.message.effective_attachment
if is_approved(user.id):
return # Skip NSFW scan for approved users
if not isinstance(media_files, (list, tuple)):
media_files = [media_files]
for file in media_files:
if not hasattr(file, "file_id"):
continue
file_obj = await context.bot.get_file(file.file_id)
file_path = os.path.join(MEDIA_DIR, f"{user.id}_{file.file_id}")
await file_obj.download_to_drive(file_path)
if not os.path.exists(file_path):
logger.error(f"Downloaded file missing: {file_path}")
continue
# Handle Stickers Separately
if update.message.sticker:
if update.message.sticker.is_animated:
# Convert `.tgs` animated sticker to PNG
new_path = convert_tgs_to_png(file_path)
elif update.message.sticker.is_video:
# Convert `.webm` video sticker to an image frame
new_path = extract_frame_from_webm(file_path)
else:
# Convert `.webp` static sticker to PNG
new_path = convert_webp_to_png(file_path)
if new_path:
file_path = new_path
else:
continue
# Run NSFW detection
result = detect_nsfw(file_path)
if result is None:
continue
max_category = max(result, key=result.get)
if max_category in ["porn", "sexy", "hentai"]:
await update.message.delete()
update_violations(user.id, max_category)
alert_message = f"""
╭─────────────────
╰──●𝙽𝚂𝙵𝚆 𝙳𝙴𝚃𝙴𝙲𝚃𝙴𝙳 🔞
╭✠╼━━━━━━❖━━━━━━━✠╮
│☾𝚄𝚜𝚎𝚛: {user.id}
│☾𝚄𝚜𝚎𝚛𝚗𝚊𝚖𝚎: @{user.username if user.username else 'None'}
│☾𝙳𝚎𝚝𝚊𝚒𝚕𝚜:
│☾𝙳𝚛𝚊𝚠𝚒𝚗𝚐𝚜: {result['drawings']:.2f}
│☾𝙽𝚎𝚞𝚝𝚛𝚊𝚕: {result['neutral']:.2f}
│☾𝙿𝚘𝚛𝚗: {result['porn']:.2f}
│☾𝙷𝚎𝚗𝚝𝚊𝚒: {result['hentai']:.2f}
│☾𝚂𝚎𝚡𝚢: {result['sexy']:.2f}
╰✠╼━━━━━━❖━━━━━━━✠╯"""
await context.bot.send_message(chat_id, alert_message, parse_mode="Markdown")
channel_alert = f"""
NSFW DETECTED 🔞
User: {user.id}
Username: @{user.username if user.username else 'None'}
Details:
Drawings: {result['drawings']:.2f}
Neutral: {result['neutral']:.2f}
Porn: {result['porn']:.2f}
Hentai: {result['hentai']:.2f}
Sexy: {result['sexy']:.2f}
Chat: {chat_id}
"""
keyboard = [[InlineKeyboardButton("👤 View Profile", url=f"tg://user?id={user.id}")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await context.bot.send_message(ALERT_CHANNEL_ID, channel_alert, reply_markup=reply_markup)
# Delete the scanned file
if os.path.exists(file_path):
os.remove(file_path)
else:
logger.warning(f"File not found for deletion: {file_path}")
# ✅ Owner Commands
async def add_approved(update: Update, context: CallbackContext):
if update.message.from_user.id != OWNER_ID:
await update.message.reply_text("» ᴀᴡᴡ, ᴛʜɪs ɪs ɴᴏᴛ ғᴏʀ ʏᴏᴜ ʙᴀʙʏ.")
return
try:
user_id = int(context.args[0])
add_approved_user(user_id)
await update.message.reply_text(f"✅ User {user_id} added to approved list.")
except (IndexError, ValueError):
await update.message.reply_text("❌ Usage: /approve <user_id>")
async def remove_approved(update: Update, context: CallbackContext):
if update.message.from_user.id != OWNER_ID:
await update.message.reply_text("» ᴀᴡᴡ, ᴛʜɪs ɪs ɴᴏᴛ ғᴏʀ ʏᴏᴜ ʙᴀʙʏ.")
return
try:
user_id = int(context.args[0])
remove_approved_user(user_id)
await update.message.reply_text(f"❌ User {user_id} removed from approved list.")
except (IndexError, ValueError):
await update.message.reply_text("❌ Usage: /remove <user_id>")
async def my_info(update: Update, context: CallbackContext):
user_id = update.message.from_user.id
violations = get_user_violations(user_id)
if not violations:
await update.message.reply_text("✅ You have a clean record.")
else:
info = "📊 **NSFW Violation History**\n"
for category, count in violations:
info += f"🔸 {category}: {count} times\n"
await update.message.reply_text(info, parse_mode="Markdown")
async def user_info(update: Update, context: CallbackContext):
message = update.message
args = context.args
if len(args) == 0:
username_or_userid = message.from_user.id
else:
username_or_userid = args[0]
violation_history = get_user_violations(username_or_userid)
if violation_history:
reply_text = f"User {username_or_userid} has violation history:\n"
reply_text += "\n".join(f"{violation[0]}: {violation[1]}" for violation in violation_history)
else:
reply_text = f"User {username_or_userid} is clear."
# Send the reply to the user
await message.reply_text(reply_text)
from telegram.constants import ParseMode
async def get_approved_users_list(update: Update, context: CallbackContext):
approved_users_list = []
for i, user_id in enumerate(get_all_users()):
user = await context.bot.get_chat(user_id)
if is_approved(user_id):
user_name = user.first_name
if user.username:
user_url = f"https://t.me/{user.username}"
else:
user_url = f"https://t.me/{user_id}"
user_link = f"{i+1}. [{user_name}]({user_url})"
approved_users_list.append(user_link)
if not approved_users_list:
await update.message.reply_text("❌ No approved users found.")
else:
approved_users_str = "\n".join(approved_users_list)
message = f"""``` ✨Approved Users✨ ```
╭✠╼━━━━━━❖━━━━━━━✠━━━━━━
│
{approved_users_str}
│
╰✠╼━━━━━━❖━━━━━━━✠━━━━━━
💫Total Approved Users: {len(approved_users_list)}
"""
await update.message.reply_text(f""""{message}""", parse_mode=ParseMode.MARKDOWN, disable_web_page_preview=True)
Raw
predict.py
import os
import tensorflow as tf
import tensorflow_hub as hub
from tensorflow import keras
import numpy as np
from PIL import Image
MODEL_PATH = os.path.join(os.path.dirname(__file__), 'nsfw_model', 'nsfw_mobilenet2.224x224.h5')
IMAGE_DIM = 224
# Load Model
def load_model():
return tf.keras.models.load_model(MODEL_PATH, custom_objects={'KerasLayer': hub.KerasLayer}, compile=False)
model = load_model()
def classify(model, image_path):
img = keras.preprocessing.image.load_img(image_path, target_size=(IMAGE_DIM, IMAGE_DIM))
img = keras.preprocessing.image.img_to_array(img) / 255.0
img = np.expand_dims(img, axis=0)
categories = ['drawings', 'hentai', 'neutral', 'porn', 'sexy']
predictions = model.predict(img)[0]
os.remove(image_path)
return {category: float(predictions[i]) for i, category in enumerate(categories)}
def detect_nsfw(image_path):
return classify(model, image_path)
Raw
utils.py
import os
import logging
from telegram import User
from config import MEDIA_DIR
# Set up logging
logging.basicConfig(
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
level=logging.INFO,
)
logger = logging.getLogger(__name__)
def mention_user(user: User):
"""Creates a clickable mention for a Telegram user."""
return f"[{user.first_name}](tg://user?id={user.id})"
def clean_media_folder():
"""Ensures media directory is clean before storing new files."""
if not os.path.exists(MEDIA_DIR):
os.makedirs(MEDIA_DIR)
for file in os.listdir(MEDIA_DIR):
os.remove(os.path.join(MEDIA_DIR, file))
def log_message(update, nsfw_category=None):
"""Logs incoming messages and NSFW detections."""
user = update.message.from_user
chat = update.message.chat
content_type = update.message.effective_attachment
log_msg = (f"User: {user.id} | Chat: {chat.id} | "
f"Media Type: {content_type} | NSFW: {nsfw_category if nsfw_category else 'Safe'}")
logger.info(log_msg)
Raw
video_utils.py
import os
import cv2
def capture_screenshot(video_path: str, output_dir: str = "screenshots"):
# Ensure the output directory exists
os.makedirs(output_dir, exist_ok=True)
# Load the video
vid_obj = cv2.VideoCapture(video_path)
fps = vid_obj.get(cv2.CAP_PROP_FPS)
if not fps or fps <= 0:
fps = 30 # Default to 30 FPS if unable to read
frame_interval = int(fps * 10) # Capture every 10 seconds
frames = []
count = 0
try:
while True:
success, frame = vid_obj.read()
if not success:
break
if count % frame_interval == 0:
frame_path = os.path.join(output_dir, f"screenshot_{count}.png")
cv2.imwrite(frame_path, frame)
frames.append(frame_path)
count += 1
except Exception as e:
print(f"Error capturing screenshots: {e}")
finally:
vid_obj.release()
return frames
import subprocess
import os
def convert_webm_to_png(webm_path: str) -> str:
try:
png_path = os.path.splitext(webm_path)[0] + ".png"
command = [
"ffmpeg",
"-i", webm_path,
"-vf", "fps=1",
"-vframes", "1",
png_path
]
subprocess.run(command, check=True)
return png_path if os.path.exists(png_path) else None
except Exception as e:
logging.error(f"Failed to convert webm to PNG: {e}")
return None
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment