import streamlit as st import pandas as pd import requests import json import os import time import re from datetime import datetime, timedelta, timezone from dotenv import load_dotenv # Chargement du fichier .env pour le token Telegram load_dotenv() API_TOKEN = os.getenv("API_TOKEN") # --- CONFIGURATION --- st.set_page_config(page_title="ModoBot Admin", layout="wide") NOCODB_BASE_URL = "https://dbtest.kaelstudio.tech/api/v2/tables/" TABLE_BANNIES_ID = "m7mwhc3qnlmgahy" TABLE_HISTORY_ID = "m65949r27wwrawn" NOCODB_TOKEN = os.getenv("NOCODB_TOKEN", "dbgK7YJNZuW2mTK4O3umr9j08sr9GwjaKZj5cN5p") HEADERS = {"xc-token": NOCODB_TOKEN} def fetch_data(table_id): url = f"{NOCODB_BASE_URL}{table_id}/records" response = requests.get(url, headers=HEADERS, params={"limit": 100}) if response.status_code == 200: return response.json().get("list", []) return [] def add_keyword(keyword): url = f"{NOCODB_BASE_URL}{TABLE_BANNIES_ID}/records" data = {"MotBannies": keyword} requests.post(url, headers=HEADERS, json=data) def delete_keyword(id): url = f"{NOCODB_BASE_URL}{TABLE_BANNIES_ID}/records" requests.delete(url, headers=HEADERS, json={"Id": id}) def telegram_api_call(method, params): """Envoie une requête directement à l'API Telegram""" url = f"https://api.telegram.org/bot{API_TOKEN}/{method}" try: resp = requests.post(url, json=params) return resp.json() except Exception as e: return {"ok": False, "description": str(e)} def extract_chat_id(channel_str): """Extrait l'ID entre parenthèses à la fin d'une chaîne comme 'Mon Salon (-100123)'""" match = re.search(r'\( (-?\d+) \)$'.replace(' ',''), str(channel_str)) if match: return match.group(1) return None # --- UI --- st.title("🛡️ ModoBot Dashboard") st.markdown("---") tab1, tab2, tab3, tab4 = st.tabs(["📊 Statistiques", "📜 Historique", "🚫 Gestion des Mots", "🛡️ Modération"]) # Chargement initial des données history_data = fetch_data(TABLE_HISTORY_ID) bannies_data = fetch_data(TABLE_BANNIES_ID) # --- TAB 1: STATISTICS --- with tab1: st.header("État du système") col1, col2, col3 = st.columns(3) col1.metric("Total Suppressions", len(history_data)) col2.metric("Mots Filtrés", len(bannies_data)) col3.metric("Dernière Activité", history_data[0]['Timestamp'] if history_data else "N/A") if history_data: df_hist = pd.DataFrame(history_data) st.subheader("Volume de suppressions par salon") st.bar_chart(df_hist['Channel'].value_counts()) # --- TAB 2: HISTORY --- with tab2: st.header("📜 Historique des messages supprimés") if history_data: df_hist = pd.DataFrame(history_data) cols = ['Timestamp', 'User', 'UserID', 'Message', 'Channel', 'Action'] display_cols = [c for c in cols if c in df_hist.columns] # Utilisation de width='stretch' au lieu de use_container_width st.dataframe(df_hist[display_cols].sort_values(by='Timestamp', ascending=False), width='stretch') else: st.info("Aucun message dans l'historique.") # --- TAB 3: MANAGEMENT --- with tab3: st.header("🚫 Liste des mots bannis") new_word = st.text_input("Ajouter un nouveau mot interdit") if st.button("Ajouter"): if new_word: add_keyword(new_word) st.success(f"Mot '{new_word}' ajouté !") st.rerun() st.subheader("Mots actifs") if bannies_data: for item in bannies_data: col_word, col_btn = st.columns([4, 1]) col_word.write(item['MotBannies']) if col_btn.button("Supprimer", key=f"del_{item['Id']}"): delete_keyword(item['Id']) st.rerun() else: st.write("La liste est vide.") # --- TAB 4: MANUAL MODERATION --- with tab4: st.header("🛡️ Gestion des Utilisateurs") if not history_data: st.warning("Aucun utilisateur dans l'historique pour le moment.") else: # Création d'une copie propre pour éviter le SettingWithCopyWarning df_full = pd.DataFrame(history_data).copy() # On extrait l'ID du salon pour chaque ligne df_full['ExtractedChatID'] = df_full['Channel'].apply(extract_chat_id) # Filtrage pour n'avoir que des données exploitables df_mod = df_full[df_full['ExtractedChatID'].notnull() & df_full['UserID'].notnull()].copy() if df_mod.empty: st.info("Aucun ID utilisateur/salon exploitable trouvé dans l'historique.") else: # Création d'un label clair df_mod.loc[:, 'label'] = df_mod['User'] + " | Salon: " + df_mod['Channel'] # On ne garde que les doublons UserID + ChatID unique_targets = df_mod.drop_duplicates(['UserID', 'ExtractedChatID']) selected_label = st.selectbox("Choisir une cible de modération", unique_targets['label']) target_row = unique_targets[unique_targets['label'] == selected_label].iloc[0] target_user_id = target_row['UserID'] target_chat_id = target_row['ExtractedChatID'] st.info(f"👉 **Cible sélectionnée** : User `{target_user_id}` sur Chat `{target_chat_id}`") col_act, col_timer = st.columns([2, 2]) action = col_act.selectbox("Action à effectuer", ["Kick (Expulser)", "Ban (Bannir)", "Mute (Silencier)", "Unban (Débannir)"]) duration = 0 if action in ["Ban (Bannir)", "Mute (Silencier)"]: duration = col_timer.select_slider( "Durée / Timer", options=[0, 15, 60, 360, 1440, 10080], format_func=lambda x: "Permanent" if x == 0 else (f"{x} min" if x < 60 else (f"{x//60}h" if x < 1440 else f"{x//1440}j")) ) confirm = st.checkbox(f"Confirmer {action} sur cet utilisateur") if st.button("🚀 Appliquer la sanction", disabled=not confirm): result = {"ok": False} until_date = int(time.time() + (duration * 60)) if duration > 0 else None with st.spinner("Action en cours..."): if "Kick" in action: result = telegram_api_call("banChatMember", {"chat_id": target_chat_id, "user_id": target_user_id}) telegram_api_call("unbanChatMember", {"chat_id": target_chat_id, "user_id": target_user_id}) elif "Ban" in action: p = {"chat_id": target_chat_id, "user_id": target_user_id} if until_date: p["until_date"] = until_date result = telegram_api_call("banChatMember", p) elif "Mute" in action: perms = {"can_send_messages": False, "can_send_media_messages": False, "can_send_other_messages": False} p = {"chat_id": target_chat_id, "user_id": target_user_id, "permissions": perms} if until_date: p["until_date"] = until_date result = telegram_api_call("restrictChatMember", p) elif "Unban" in action: result = telegram_api_call("unbanChatMember", {"chat_id": target_chat_id, "user_id": target_user_id, "only_if_banned": False}) if result.get("ok"): st.success("✅ Sanction appliquée !") else: st.error(f"❌ Erreur Telegram : {result.get('description', 'Inconnue')}")