import discord
from [Link] import commands
import os
import asyncio
from [Link] import ThreadPoolExecutor
import aiofiles
import json
with open('[Link]') as f:
config = [Link](f)
TOKEN = config["token"]
rmk1337 = [Link](command_prefix="+", intents=[Link]())
DUMP_FOLDER = 'dbip'
CANAL_ID = int(config["allowed_channel_id"])
ROLE_ID = int(config["required_role_id"])
EMBED_COLOR = int(config["embed_color"], 16)
ENCODINGS = ['utf-8', 'latin-1', 'cp1252']
MAX_CONCURRENT_TASKS = 20
BLOCK_SIZE = 1024 * 1024 # 1 Mo
FILE_PROCESSING_TIMEOUT = 5
executor = ThreadPoolExecutor(max_workers=MAX_CONCURRENT_TASKS)
async def read_file_in_chunks(file_path, license, found_event):
for encoding in ENCODINGS:
try:
async with [Link](file_path, mode='r', encoding=encoding,
errors='ignore') as f:
while True:
content = await [Link](BLOCK_SIZE)
if not content:
break
if license in content:
return content
except Exception as e:
print(f"Erreur avec l'encodage {encoding} pour le fichier {file_path}:
{e}")
return None
async def process_file(file_path, license, results, found_event):
if found_event.is_set():
return
try:
content = await asyncio.wait_for(read_file_in_chunks(file_path, license,
found_event), timeout=FILE_PROCESSING_TIMEOUT)
if content:
associated_info = next(([Link]() for line in [Link]('\n') if
license in line), None)
if associated_info:
[Link](associated_info)
found_event.set()
except [Link]:
print(f"Le traitement du fichier {file_path} a dépassé le temps limite.")
except Exception as e:
print(f"Erreur lors du traitement du fichier {file_path}: {e}")
@[Link](name="vip", description="Rechercher des informations pour une
licence FiveM")
async def vip(interaction: [Link], license: str):
if not any([Link] == ROLE_ID for role in [Link]):
await [Link].send_message("Vous n'avez pas le bon rôle pour
utiliser cette commande.", ephemeral=True)
return
if [Link] != CANAL_ID:
await [Link].send_message("Veuillez faire la commande dans le
salon VIP.", ephemeral=True)
return
embed = [Link](title="VIP Recherche", description="Recherche en
cours...", color=EMBED_COLOR)
await [Link].send_message(embed=embed)
found_event = [Link]()
results = []
semaphore = [Link](MAX_CONCURRENT_TASKS)
async def sem_task(file_path):
async with semaphore:
await process_file(file_path, license, results, found_event)
files_to_process = [[Link](root, file)
for root, _, files in [Link](DUMP_FOLDER)
for file in files if [Link](('.txt', '.sql'))]
tasks = [sem_task(file_path) for file_path in files_to_process]
await [Link](*tasks, return_exceptions=True)
if results:
results_message = results[0]
embed = [Link](
title="Résultat de la Recherche",
description=results_message,
color=EMBED_COLOR
)
embed.set_footer(text=config["footer_text"])
try:
await [Link](embed=embed)
await [Link]("Les résultats ont été envoyés en
message privé.")
except [Link]:
await [Link]("Je ne peux pas vous envoyer de message
privé. Veuillez vérifier vos paramètres de confidentialité.")
except [Link] as e:
print(f"Erreur lors de l'envoi du message: {e}")
await [Link]("Erreur lors de l'envoi du message
privé. Veuillez réessayer plus tard.")
else:
embed = [Link](
title="Aucune Information Trouvée",
description="Aucune information trouvée pour cette licence.",
color=EMBED_COLOR
)
embed.set_footer(text=config["footer_text"])
await [Link](embed=embed)
@[Link]
async def on_ready():
print(f"{[Link]} est connecté et prêt à l'emploi. 🚀")
print("🌟 Bot développé par rmk1337 pour LookData 💻✨")
await rmk1337.change_presence(activity=[Link](name=config["bot_status"]))
await [Link]()
[Link](TOKEN)