import os
import shutil
from telegram import Update
from telegram.ext import ContextTypes
from config.settings import OWNER_ID
from utils.markdown import escape_markdown
# Diretório base para arquivos temporários
TEMP_DIRS = [os.path.join(os.sep, "tmp")]
# Prefixos das pastas que o bot cria e deve gerenciar
BOT_FOLDER_PREFIXES = ("yt_audio_", "yt_video_", "fb_", "insta_", "tt_")
# --- Comando /help ---
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Envia mensagem de ajuda para qualquer usuário."""
help_text = (
"🤖 *Comandos disponíveis:*\n\n"
"/start - Inicia o bot\n"
"/help - Mostra esta mensagem\n"
"/clear - Limpa arquivos temporários do bot (dono)\n"
"/ls - Lista arquivos temporários (dono)\n"
"/stats - Mostra estatísticas do bot (dono)\n\n"
"Para baixar, escolha uma opção no menu ou envie um link diretamente."
)
await update.message.reply_text(help_text, parse_mode="MarkdownV2")
# --- Comando /clear ---
async def clear(update: Update, context: ContextTypes.DEFAULT_TYPE):
user = update.effective_user
if user.id != OWNER_ID:
await update.message.reply_text("❌ Você não tem permissão para usar este comando.")
return
removed_count = 0
for dir_path in TEMP_DIRS:
if not os.path.exists(dir_path):
continue
try:
for item in os.listdir(dir_path):
if item.startswith(BOT_FOLDER_PREFIXES):
item_path = os.path.join(dir_path, item)
if os.path.isdir(item_path):
shutil.rmtree(item_path)
removed_count += 1
except Exception as e:
await update.message.reply_text(f"⚠️ Erro ao limpar `{escape_markdown(dir_path)}`: {escape_markdown(str(e))}", parse_mode="MarkdownV2")
await update.message.reply_text(f"✅ {removed_count} pastas temporárias removidas.")
# --- Comando /ls ---
async def ls(update: Update, context: ContextTypes.DEFAULT_TYPE):
user = update.effective_user
if user.id != OWNER_ID:
await update.message.reply_text("❌ Você não tem permissão para usar este comando.")
return
all_items = []
for dir_path in TEMP_DIRS:
if not os.path.exists(dir_path):
continue
items = [f for f in os.listdir(dir_path) if f.startswith(BOT_FOLDER_PREFIXES) and os.path.isdir(os.path.join(dir_path, f))]
if items:
all_items.extend(items)
if not all_items:
await update.message.reply_text("ℹ️ Nenhuma pasta temporária encontrada.")
return
# Escapa os nomes para evitar erros de markdown
escaped_items = [escape_markdown(item) for item in all_items]
message = "📁 *Pastas temporárias encontradas:*\n\n" + "\n".join(escaped_items)
await update.message.reply_text(message, parse_mode="MarkdownV2")
# --- Comando /stats ---
DOWNLOAD_COUNT = 0 # contador simples em memória
async def stats(update: Update, context: ContextTypes.DEFAULT_TYPE):
user = update.effective_user
if user.id != OWNER_ID:
await update.message.reply_text("❌ Você não tem permissão para usar este comando.")
return
total_files = 0
total_size_mb = 0
for dir_path in TEMP_DIRS:
if not os.path.exists(dir_path):
continue
# Itera apenas nas pastas do bot, não em todo o /tmp
for item in os.listdir(dir_path):
if item.startswith(BOT_FOLDER_PREFIXES):
item_path = os.path.join(dir_path, item)
if not os.path.isdir(item_path):
continue
for root, _, files in os.walk(item_path):
for f in files:
total_files += 1
try:
total_size_mb += os.path.getsize(os.path.join(root, f))
except OSError:
continue # Ignora arquivos que podem ser removidos durante a contagem
total_size_mb /= (1024 * 1024)
msg = (
f"📊 *Estatísticas do bot:*\n\n"
f"Total de downloads na sessão: `{DOWNLOAD_COUNT}`\n"
f"Pastas temporárias ativas: `{total_files}`\n"
f"Tamanho ocupado: `{total_size_mb:.2f} MB`"
)
await update.message.reply_text(msg, parse_mode="MarkdownV2")