userbot/main.py

#  Moon-Userbot - telegram userbot
#  Copyright (C) 2020-present Moon Userbot Organization
#
#  This program is free software: you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation, either version 3 of the License, or
#  (at your option) any later version.

#  This program is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.

#  You should have received a copy of the GNU General Public License
#  along with this program.  If not, see <https://www.gnu.org/licenses/>.

# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "pip",
#     "pyrofork",
#     "tgcrypto",
#     "wheel",
#     "gunicorn",
#     "flask",
#     "humanize",
#     "pygments",
#     "pymongo",
#     "psutil",
#     "Pillow>=10.3.0",
#     "click",
#     "dnspython",
#     "requests",
#     "environs",
#     "GitPython",
#     "beautifulsoup4",
#     "aiohttp",
#     "aiofiles",
#     "pySmartDL",
# ]
# ///
import contextlib
import logging
import os
import platform
import sqlite3
import subprocess

import requests
from pyrogram import Client, errors, idle
from pyrogram.enums.parse_mode import ParseMode
from pyrogram.raw.functions.account import DeleteAccount, GetAuthorizations
from utils import config
from utils.db import db
from utils.misc import userbot_version
from utils.module import ModuleManager
from utils.rentry import rentry_cleanup_job
from utils.scripts import restart

SCRIPT_PATH = os.path.dirname(os.path.realpath(__file__))
if os.getcwd() != SCRIPT_PATH:
    os.chdir(SCRIPT_PATH)

common_params = {
    'api_id': config.api_id,
    'api_hash': config.api_hash,
    'hide_password': True,
    'workdir': SCRIPT_PATH,
    'app_version': userbot_version,
    'device_model': 'mUserbot',
    'system_version': platform.version() + ' ' + platform.machine(),
    'sleep_threshold': 30,
    'test_mode': config.test_server,
    'parse_mode': ParseMode.HTML,
}

if config.STRINGSESSION:
    common_params['session_string'] = config.STRINGSESSION

app = Client('my_account', **common_params)


def load_missing_modules():
    all_modules = db.get('custom.modules', 'allModules', [])
    if not all_modules:
        return

    custom_modules_path = f'{SCRIPT_PATH}/modules/custom_modules'
    os.makedirs(custom_modules_path, exist_ok=True)

    try:
        resp = requests.get(
            'https://raw.githubusercontent.com/The-MoonTg-project/custom_modules/main/full.txt',
            timeout=10,
        )
        if not resp.ok:
            logging.error(
                'Failed to fetch custom modules list: HTTP %s',
                resp.status_code,
            )
            return
        f = resp.text
    except Exception as e:
        logging.error('Failed to fetch custom modules list: %s', e)
        return
    modules_dict = {line.split('/')[-1].split()[0]: line.strip() for line in f.splitlines()}

    for module_name in all_modules:
        module_path = f'{custom_modules_path}/{module_name}.py'
        if not os.path.exists(module_path) and module_name in modules_dict:
            url = f'https://raw.githubusercontent.com/The-MoonTg-project/custom_modules/main/{modules_dict[module_name]}.py'
            resp = requests.get(url)
            if resp.ok:
                with open(module_path, 'wb') as f:
                    f.write(resp.content)
                logging.info('Loaded missing module: %s', module_name)
            else:
                logging.warning('Failed to load module: %s', module_name)


async def main():
    logging.basicConfig(
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        handlers=[logging.FileHandler('moonlogs.txt'), logging.StreamHandler()],
        level=logging.INFO,
    )
    DeleteAccount.__new__ = None

    try:
        await app.start()
    except sqlite3.OperationalError as e:
        if str(e) == 'database is locked' and os.name == 'posix':
            logging.warning('Session file is locked. Trying to kill blocking process...')
            subprocess.run(['fuser', '-k', 'my_account.session'], check=True)
            restart()
        raise
    except (errors.NotAcceptable, errors.Unauthorized) as e:
        logging.error(
            '%s: %s\nMoving session file to my_account.session-old...',
            e.__class__.__name__,
            e,
        )
        os.rename('./my_account.session', './my_account.session-old')
        restart()

    load_missing_modules()
    module_manager = ModuleManager.get_instance()
    await module_manager.load_modules(app)

    if info := db.get('core.updater', 'restart_info'):
        text = {
            'restart': '<b>Restart completed!</b>',
            'update': '<b>Update process completed!</b>',
        }[info['type']]
        with contextlib.suppress(errors.RPCError):
            await app.edit_message_text(info['chat_id'], info['message_id'], text)
        db.remove('core.updater', 'restart_info')

    # required for sessionkiller module
    if db.get('core.sessionkiller', 'enabled', False):
        db.set(
            'core.sessionkiller',
            'auths_hashes',
            [auth.hash for auth in (await app.invoke(GetAuthorizations())).authorizations],
        )

    logging.info('Moon-Userbot started!')

    app.loop.create_task(rentry_cleanup_job())

    await idle()

    await app.stop()


if __name__ == '__main__':
    app.run(main())