Lydia Reverse Engineered

bot.py

#  pip install coffeehouse
from coffeehouse import LydiaAI
from coffeehouse.api import API
from coffeehouse.exception import CoffeeHouseError

#  for lydia-chan
import random

#  Enable Logging
import logging
logging.basicConfig(level=logging.INFO)

#  Environment Variables
from os import environ

#  pip install pyrogram
from pyrogram import (
    Client,
    idle,
    filters,
    types
)

# async
import asyncio

# async loop
loop = asyncio.get_event_loop()

# database
# db contains: {chat_id: bool}
# sessions contains: {session: expire_date}
db = {}
sessions = {}


# Configs
# Coffeehouse API Key
if environ.get("CF_API") is None:
    CF_API = input("Enter your Intellivoid Coffeehouse API Key: ")
else:
    CF_API = environ.get("CF_API")

CoffeeHouseAPI = API(CF_API)
lydia = LydiaAI(CoffeeHouseAPI)
EXPIRE = ''


# Bot Token
if environ.get("BOT_TOKEN") is None:
    BOT_TOKEN = input("Enter your Telegram Bot Token: ")
else:
    BOT_TOKEN = environ.get("BOT_TOKEN")

BOT_ID = 0


#  initial
async def initialize():
    global BOT_ID
    global EXPIRE
    await miku.start()
    BOT_ID = (await miku.get_me()).id
    ses = lydia.create_session()
    sessions[str(ses.expires)] = str(ses.id)
    EXPIRE = str(ses.expires)


#  Pyrogram Client
miku = Client(
    ":memory:",
    api_id=6, # Official Android Telegram APP API_KEY (DO NOT USE ON A USER ACCOUNT OR YOU WILL BE BANNED)
    api_hash="eb06d4abfb49dc3eeb1aeb98ae0f581e", # Official Android Telegram APP API_HASH (DO NOT USE ON A USER ACCOUNT OR YOU WILL BE BANNED)
    bot_token=BOT_TOKEN,
)

#  Main Bot
async def main():
    #  Calling Intialize function
    await initialize()
    #  Reply to Incoming Messages
    @miku.on_message(
        ~filters.edited &
        (
            filters.group | filters.private
        ),
        group=1,
    )
    async def _(_, message: types.Message):
        if not await check_incoming(message):
            return
        query = message.text
        await miku.send_chat_action(message.chat.id, action='typing')
        try:
            rep = lydia.think_thought(sessions[EXPIRE], query)
        except CoffeeHouseError as e:
            if str(e) == "The session is no longer available":
                ses = lydia.create_session()
                sessions[EXPIRE] = ses.id
                rep = lydia.think_thought(sessions[EXPIRE], query)
            else:
                await message.reply(f"Something went terribly wrong: `{e}`\nReport to @DeprecatedChat")
        try:
            if db[message.chat.id]:
                await message.reply(owofiy(rep))
            else:
                await message.reply(rep)
        except KeyError:
            db[message.chat.id] = False
            if db[message.chat.id]:
                await message.reply(owofiy(rep))
            else:
                await message.reply(rep)

    #  Start Command
    @miku.on_message(
        filters.command("start")
    )
    async def _(_, message:types.Message):
        await message.reply(
            f"Hi {message.from_user.mention}, I am a Chatbot Powered by Intellivoid Coffeehouse"
        )
    
    #  Uwufy Switch
    @miku.on_message(
        filters.command("uwufy")
    )
    async def _(_, message: types.Message):
        try:
            await uwufy(message.chat.id, message.command[1])
        except IndexError:
            try:
                await message.reply(f"Uwufy Settings for this chat: `{db[message.chat.id]}`")
            except KeyError:
                db[message.chat.id] = False
                await message.reply(f"Uwufy Settings for this chat: `{db[message.chat.id]}`")
            return
        if db[message.chat.id]:
            await message.reply("UwU senpai.., ~")
        else:
            await message.reply("UwUfy is off")

    #  Bot IDLE
    await idle()


#  Check incoming message
async def check_incoming(message: types.Message) -> bool:
    if message.text.startswith("/"):
        return False
    if message.chat.type == 'private':
        return True
    if 'miku' in message.text.split():
        return True
    if message.reply_to_message:
        return message.reply_to_message.from_user.id == BOT_ID


#  Lydia-Chan, CC: @Netkas
def owofiy(msg: str):
    # Make it lowercase
    msg = msg.lower()
    # Fuck up the commas and periods
    commas = [",", ",,", ",,,", ",,,,", ",,,,,"]
    periods = ["..", ".,.,", "..,..,", ",,", ",,,,", "...", ".....", ",.."]
    questions = ["?", "??", "?????", "?!?!?", "??!?!?!??!"]
    msg = msg.replace(",", random.choice(commas))
    msg = msg.replace(".", random.choice(periods))
    msg = msg.replace(".", random.choice(questions))
    # Fuck up the ascii
    smiles = ["( ´ ω ` )", "(≧◡≦)", "(╯✧▽✧)╯"]
    sadness = ["｡ﾟ(TヮT)ﾟ｡", "o(TヘTo)", "o(〒﹏〒)o", "(╯︵╰,)", "( ；∀；)", "(´；ω；｀)","(╯︵╰,)","(｡•́︿•̀｡)","( ≧Д≦)","(ᗒᗩᗕ)","(-̩̩̩-̩̩̩-̩̩̩-̩̩̩-̩̩̩___-̩̩̩-̩̩̩-̩̩̩-̩̩̩-̩̩̩)","(╥﹏╥)","(个_个)","(´°̥̥̥̥̥̥̥̥ω°̥̥̥̥̥̥̥̥｀)","(⊙_◎)","(｡ﾉω＼｡)","｡:ﾟ(;´∩;)ﾟ:｡","(´ . .̫ . )","(´ . .̫ . `)"]
    msg = msg.replace(":(", random.choice(sadness))
    msg = msg.replace(":)", random.choice(smiles))
    # Fuck up the words
    brothers = ["onichan", "onichann", "oni-chan", "oniiichannnn"]
    sisters = ["onee-chan", "onneechann", "onechan", "oneeechaaannnnn"]
    dad = ["otuo-san", "outosann", "otuosaaann", "otuosaaannn"]
    msg = msg.replace("brother", random.choice(brothers))
    msg = msg.replace("you", "senpai")
    msg = msg.replace("love", "wanna fuk")
    msg = msg.replace("brown", "the color of shit")
    msg = msg.replace("red", "blood color")
    msg = msg.replace("anime", "the purpose to my life")
    msg = msg.replace("mother", "gay")
    msg = msg.replace("sister", random.choice(sisters))
    msg = msg.replace("dad", random.choice(dad))
    msg = msg.replace("awoooooo", "bakaaaaaa")
    msg = msg.replace("awooooo", "bakaaaaa")
    msg = msg.replace("awoooo", "bakaaaa")
    msg = msg.replace("awooo", "bakaaa")
    msg = msg.replace("awoo", "bakaa")
    msg = msg.replace("awo", "baka")
    msg = msg.replace("idiot", "baka")
    # Fuck up the spelling
    msg = msg.replace("ove", "uv")
    msg = msg.replace("l", "w")
    msg = msg.replace("r", "w")
    return msg + random.choice(smiles)


#  Lydia-Chan func
async def uwufy(chat_id: int, check: str):
    db[chat_id] = check.lower() == "on"


if __name__ == '__main__':
    loop.run_until_complete(main())