FunBotDiscord.py

import discord
from urllib.parse import quote_plus
import json
import LeagueOfLegends
from hell_let_loose import HLL_View
from discord.ext import commands, tasks
import asyncio
import sqlite3
from datetime import datetime, timedelta
import re
import cv2
import numpy as np
import os
import aiohttp
import subprocess
import time
from typing import Optional

intents = discord.Intents.all()
f = open("botToken.txt", "r")
TOKEN = f.read()
bot = commands.Bot(command_prefix="/", intents=intents)
VALID_CLIP_EXTENSIONS = ['.mp4', '.webm', '.mov']
CLIP_CHANNEL_ID = 1326956425762570240
AUTHORIZED_USER_ID = 999736048596816014
MY_USER_ID = 262347377476632577
LEAGUE_USER_ID1 = 299335372859768852
LEAGUE_USER_ID2 = 368913296771776512
ANSI_RE = re.compile(r'\x1B\[[0-?]*[ -/]*[@-~]')
AUTHORIZED_USER_IDS = [262347377476632577,115560354012069897,341327371179261962,341327371179261962,154046751916032001]
MAX_LEN = 2000

def tail_text(s: str, max_len: int = MAX_LEN) -> str:
    if len(s) <= max_len:
        return s
    return s[-max_len:]

def setup_database():
    # Connect to SQLite database
    conn = sqlite3.connect('reminders.db')
    c = conn.cursor()

    # Create table if it does not exist
    c.execute('''CREATE TABLE IF NOT EXISTS reminders
                 (id INTEGER PRIMARY KEY, message TEXT, due_date DATETIME, channel_id INTEGER)''')
    conn.commit()
    return conn, c

conn, c = setup_database()

class Google(discord.ui.View):
    def __init__(self, query: str):
        super().__init__()
        # we need to quote the query string to make a valid url. Discord will raise an error if it isn't valid.
        query = quote_plus(query)
        url = f'https://www.google.com/search?q={query}'

        # Link buttons cannot be made with the decorator
        # Therefore we have to manually create one.
        # We add the quoted url to the button, and add the button to the view.
        self.add_item(discord.ui.Button(label='Click Here', url=url))

@bot.event
async def on_ready():
    print("Ready to go")
    try:
        synced = await bot.tree.sync()
        print(f"Synced {len(synced)} command(s)")
        print(f'Logged in as {bot.user}')

    except Exception as e:
        print(e)
    #check_due_reminders.start()  # Start the reminders background task

@bot.hybrid_command(name="ping")
async def hello(ctx: commands.Context):
    """Pings the Bot"""
    await ctx.send(f'pong') 
    
@bot.hybrid_command(name="google")
async def google(ctx: commands.Context, *, query: str):
    """Returns a google link for a query"""
    await ctx.send(f'Google Result for: `{query}`', view=Google(query))

@bot.hybrid_command(name="hll")
async def survey(ctx):
    view = HLL_View()
    await ctx.send(view=view)
    await view.wait()
    await ctx.send(view.answer1)

@bot.hybrid_command(name="hytale", description="Hytale control: inject/print/restart")
async def hytale(ctx, action: str, *, payload: Optional[str] = None):
    if ctx.author.id not in AUTHORIZED_USER_IDS:
        return await ctx.reply("You are not allowed to command Hytale server", ephemeral=True)
    action = action.lower()

    # Optional: restrict who can run it
    # if ctx.author.id not in {1234567890}:
    #     return await ctx.reply("Not authorized.", ephemeral=True)

    if action == "inject":
        if not payload or not payload.strip():
            return await ctx.reply("Usage: `/hytale inject <command>`", ephemeral=True)

        cmd = payload.strip()
        try:
            result = hytaleRunCommands(cmd)
            result = tail_text(result, 2000)
        except subprocess.CalledProcessError as e:
            return await ctx.reply(f"Failed to inject: `{e}`", ephemeral=True)

        await ctx.reply(f"Injected into screen: `{cmd}`", ephemeral=True)
        return await ctx.reply(f"the following is result: `{result}`", ephemeral=True)

    elif action == "print":
        try:
            out = hytalePrint()
            out = tail_text(out, 2000)
        except subprocess.CalledProcessError as e:
            return await ctx.reply(f"Failed to read log: `{e}`", ephemeral=True)
        # Discord message limit safety
        if len(out) > 1800:
            out = out[-1800:]
        return await ctx.reply(f"Last 100 log lines:\n```{out}```", ephemeral=True)

    elif action == "restart":
        # Example: inject a restart script command into the screen session
        try:
            hytaleAdminStuff()
        except subprocess.CalledProcessError as e:
            return await ctx.reply(f"Failed to restart: `{e}`", ephemeral=True)

        return await ctx.reply("Restart triggered.", ephemeral=True)

    else:
        return await ctx.reply("Invalid action. Use: `inject`, `print`, or `restart`.", ephemeral=True)

@bot.command(name="reminder")
async def reminder(ctx, *, time_and_message: str):
    try:
        message_pattern = r'"([^"]+)"'  # Extract message within quotes
        message_match = re.search(message_pattern, time_and_message)
        if not message_match:
            await ctx.send('Invalid format. Use `!reminder <time> "<message>"` with time units being day(s), hour(s), or minute(s).')
            return

        message = message_match.group(1)
        time_str = time_and_message.replace(f'"{message}"', '').strip()
        due_date = calculate_due_date(time_str)

        if due_date is None:
            await ctx.send('Invalid time format. Use units like day(s), hour(s), or minute(s).')
            return

        c.execute("INSERT INTO reminders (message, due_date, channel_id) VALUES (?, ?, ?)",
                  (message, due_date, ctx.channel.id))
        conn.commit()
        await ctx.send(f'Scheduled a reminder: "{message}" at {due_date}.')
    except Exception as e:
        await ctx.send(f'Error: {str(e)}')

def calculate_due_date(time_string):
    time_units = re.findall(r'(\d+)\s*(day|days|hour|hours|minute|minutes)', time_string)
    if not time_units:
        return None

    due_date = datetime.now()

    for amount, unit in time_units:
        amount = int(amount)
        if unit in ['day', 'days']:
            due_date += timedelta(days=amount)
        elif unit in ['hour', 'hours']:
            due_date += timedelta(hours=amount)
        elif unit in ['minute', 'minutes']:
            due_date += timedelta(minutes=amount)

    return due_date

@tasks.loop(minutes=1)
async def check_due_reminders():
    now = datetime.now()
    c.execute("SELECT * FROM reminders WHERE due_date <= ?", (now,))
    due_reminders = c.fetchall()
    for reminder in due_reminders:
        _, message, due_date, channel_id = reminder
        channel = bot.get_channel(channel_id)
        await send_reminder(channel, message)
        c.execute("DELETE FROM reminders WHERE id = ?", (reminder[0],))
    conn.commit()

async def send_reminder(channel, message):
    await channel.send(f'Reminder: {message}')

@bot.event
async def on_message(message):
    if message.author == bot.user:
        return

    if message.content.lower() == 'hello':
        await message.channel.send(f'Hello, {message.author.display_name}!')

    if message.content.lower() == 'hey jared':
        await message.channel.send('that guy sucks dont talk to him!')

    if message.content.lower().startswith("!hytale"):
        if "print" in message.content.lower():
            return hytalePrint()
        elif "authenticate" in message.content.lower():
            return hytaleAdminStuff(message.content.lower())
        elif "inject" in message.content.lower() and message.author.id in AUTHORIZED_USER_ID:
            return hytaleRunCommands(message.content.lower().split("inject", 1)[1].lstrip())

    # Check if the message contains attachments
    if message.attachments and message.channelID != CLIP_CHANNEL_ID:
        for attachment in message.attachments:
            # Check if the attachment is a valid clip (based on file extension)
            if any(attachment.filename.endswith(ext) for ext in VALID_CLIP_EXTENSIONS):
                # Forward the message content and original author info
                embed = discord.Embed(
                description=f"**{message.author.name}** shared a clip!\n{message.content}",
                color=discord.Color.blue())
        #await clip_channel.send(embed=embed, file=await attachment.to_file())

    if message.attachments and message.author.id in [AUTHORIZED_USER_ID]:
        for attachment in message.attachments:
            # Check if the attachment is a valid clip (based on file extension)
            if any(attachment.filename.lower().endswith(ext) for ext in
                   ['png']):  # Validate image format
                image = await download_image(attachment.url)
                if image is not None:
                    print('image at ' + attachment.url)
                    answer = LeagueOfLegends.find_best_match(image, "AllLeagueChampions")
                    hp_and_atk = get_value_from_json(answer)
                    answer = answer.replace('_', ' ')
                    answer += "\nhp = %s\natk = %s" % (hp_and_atk["hp"],hp_and_atk["atk"])
                    user = await bot.fetch_user(MY_USER_ID)
                    await user.send(answer)
                    user = await bot.fetch_user(LEAGUE_USER_ID1)
                    await user.send(answer)
                    user = await bot.fetch_user(LEAGUE_USER_ID2)
                    await user.send(answer)
                else:
                    await message.channel.send("Failed to process the image.")
    await bot.process_commands(message)

async def download_image(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            if resp.status == 200:
                image_bytes = await resp.read()
                image_array = np.asarray(bytearray(image_bytes), dtype=np.uint8)
                return cv2.imdecode(image_array, cv2.IMREAD_COLOR)  # Decode image for OpenCV
    return None

def clean_terminal_text(s: str) -> str:
    s = ANSI_RE.sub("", s)
    # Remove common control chars that show up as  and  etc.
    s = s.replace("\r", "")
    s = s.replace("\b", "")   # backspace
    s = s.replace("\x07", "") # bell
    # Remove any remaining C0 control chars except \n and \t
    s = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]", "", s)
    return s

def get_value_from_json(name: str):
    """
    Retrieves the value associated with a given name (key) from a JSON file.

    :param file_path: Path to the JSON file.
    :param name: The key whose value needs to be retrieved.
    :return: The value associated with the key, or None if key is not found.
    """
    try:
        with open("league_values.json", 'r', encoding='utf-8') as file:
            data = json.load(file)

        return data.get(name, None)  # Returns None if key is not found
    except FileNotFoundError:
        print(f"Error: The file was not found.")
    except json.JSONDecodeError:
        print("Error: The file is not a valid JSON.")
    except Exception as e:
        print(f"Unexpected error: {e}")

    return None
    # Don't forget to process commands if the bot has any



def hytaleAdminStuff(message):
    if "authenticate" in message:
        full_cmd = f'{"/auth login device"}\r'
        subprocess.run([
            "screen",
            "-S", str("Hytale"),
            "-p", str(0),
            "-X", "stuff", full_cmd
        ], check=True)
    time.sleep(2)
    log_path = "/opt/screenlogs.txt"
    authentication_code = subprocess.run(
        ["tail", "-n", "10", log_path],
        check=True,
        capture_output=True,
        text=True,
    )
    return clean_terminal_text(authentication_code.stdout).strip()
def hytalePrint():
    result = subprocess.run(
        ["tail", "-n", "10", "/opt/screenlogs.txt"],
        check=True,
        capture_output=True,
        text=True,
    )
    return clean_terminal_text(result.stdout).strip()


def hytaleRunCommands(commands):
    full_cmd = f'{commands}\r'
    subprocess.run([
        "screen",
        "-S", str("Hytale"),
        "-p", str(0),
        "-X", "stuff", full_cmd
    ], check=True)
    time.sleep(5)
    return hytalePrint()




bot.run(TOKEN)