Spaces:
Paused
Paused
| """Channel Post Bot - entry point. Run: python bot.py""" | |
| from __future__ import annotations | |
| import asyncio | |
| import logging | |
| import os | |
| from urllib.parse import unquote, urlsplit | |
| from telethon import TelegramClient, functions, types | |
| from telethon.sessions import StringSession | |
| from app.common import Ctx | |
| from app.config import Config, load_config | |
| from app.db import Database | |
| from app.errorlog import ErrorReporter, TelegramLogHandler | |
| from app.expiry import Expirer | |
| from app.handlers import ( | |
| autopost, | |
| backup, | |
| basic, | |
| broadcast, | |
| channels, | |
| checkbtn, | |
| create, | |
| masslinks, | |
| posts, | |
| replace, | |
| repost, | |
| shift, | |
| sync, | |
| ) | |
| from app.health import start_health_server | |
| from app.syncer import Syncer | |
| from app.userbot import Userbot | |
| log = logging.getLogger("bot") | |
| COMMANDS = [ | |
| ("new", "Create a post"), | |
| ("posts", "My posts and drafts"), | |
| ("autopost", "Post a series of episode download buttons"), | |
| ("formatting", "How to format text: bold, links, quotes ... (HTML / Markdown)"), | |
| ("addchannel", "Register a channel"), | |
| ("channels", "List channels"), | |
| ("broadcast", "Send one message to all channels, optionally timed (owner)"), | |
| ("replace", "Replace a post's media and caption (owner)"), | |
| ("swap", "Swap a username in links (owner)"), | |
| ("handleswap", "Swap a @username written in posts (owner)"), | |
| ("repost", "Copy a whole channel in order (owner)"), | |
| ("shift", "Copy posts from one channel to another (owner)"), | |
| ("sync", "Compare channels with My posts now (owner)"), | |
| ("undo", "Undo the last replace (owner)"), | |
| ("userbot", "Status of the helper account that deletes and edits what the bot can't (owner)"), | |
| ("export", "Backup as JSON (owner)"), | |
| ("import", "Restore a backup made by /export (owner)"), | |
| ("admins", "List admins (owner)"), | |
| ("addadmin", "Add an admin by numeric ID (owner)"), | |
| ("removeadmin", "Remove an admin by numeric ID (owner)"), | |
| ("cancel", "Cancel what I'm doing"), | |
| ("help", "Show help"), | |
| ] | |
| def build_client(cfg: Config) -> TelegramClient: | |
| if cfg.session_string: | |
| session = StringSession(cfg.session_string) | |
| else: | |
| os.makedirs(os.path.dirname(cfg.session_path) or ".", exist_ok=True) | |
| session = cfg.session_path | |
| client = TelegramClient(session, cfg.api_id, cfg.api_hash) | |
| client.parse_mode = "html" | |
| return client | |
| def register_all(ctx: Ctx) -> None: | |
| for module in (basic, backup, channels, create, posts, masslinks, autopost, broadcast, checkbtn, replace, repost, shift, sync): | |
| module.register(ctx) | |
| async def set_commands(client: TelegramClient) -> None: | |
| try: | |
| await client( | |
| functions.bots.SetBotCommandsRequest( | |
| scope=types.BotCommandScopeDefault(), | |
| lang_code="", | |
| commands=[types.BotCommand(command=c, description=d) for c, d in COMMANDS], | |
| ) | |
| ) | |
| except Exception as e: # cosmetic only | |
| log.warning("could not set the command menu: %s", e) | |
| def secrets_of(cfg: Config) -> list: | |
| """Texts that must never appear in the error log.""" | |
| out = [cfg.bot_token, cfg.api_hash, cfg.session_string, cfg.userbot_session] | |
| try: | |
| parts = urlsplit(cfg.database_url) | |
| if parts.password: | |
| out += [parts.password, unquote(parts.password)] | |
| except Exception: | |
| pass | |
| return [s for s in out if s] | |
| async def main() -> None: | |
| cfg = load_config() | |
| logging.basicConfig(level=cfg.log_level, format="%(asctime)s %(levelname)s %(name)s: %(message)s") | |
| logging.getLogger("telethon").setLevel(logging.WARNING) | |
| client = build_client(cfg) | |
| reporter = ErrorReporter(client, cfg.error_log_chat_id, bot_token=cfg.bot_token, secrets=secrets_of(cfg)) | |
| reporter.start() | |
| logging.getLogger().addHandler(TelegramLogHandler(reporter)) | |
| asyncio.get_running_loop().set_exception_handler(reporter.loop_exception_handler) | |
| db = None | |
| ctx = None | |
| userbot = Userbot(client, cfg) | |
| try: | |
| db = Database(cfg.database_url, cfg.db_pool) | |
| await db.init() | |
| ctx = Ctx(cfg=cfg, db=db, client=client, userbot=userbot, reporter=reporter) | |
| ctx.admins = await db.list_admins() | |
| ctx.expirer = Expirer(ctx) | |
| ctx.syncer = Syncer(ctx) | |
| register_all(ctx) | |
| await client.start(bot_token=cfg.bot_token) | |
| me = await client.get_me() | |
| log.info("Logged in as @%s (owners: %s)", me.username, ", ".join(map(str, sorted(cfg.owners)))) | |
| await set_commands(client) | |
| if cfg.port: | |
| await start_health_server(cfg.port) | |
| problem = await reporter.check() | |
| if reporter.enabled: | |
| if problem: | |
| log.warning("errors cannot be sent to chat %s yet: %s", cfg.error_log_chat_id, problem) | |
| else: | |
| log.info("errors are sent to chat %s", cfg.error_log_chat_id) | |
| else: | |
| log.info("error log chat: off") | |
| if userbot.enabled: | |
| await userbot.connect() # never fatal: the userbot is only needed for old posts | |
| ctx.expirer.start() # deletes timed broadcasts when their time is up | |
| ctx.syncer.start() # compares the channels with My posts now and then (and notices changes made meanwhile) | |
| await client.run_until_disconnected() | |
| except Exception as e: | |
| await reporter.report_fatal("The bot stopped because of an unhandled exception", e) | |
| raise | |
| finally: | |
| if ctx is not None and ctx.syncer is not None: | |
| await ctx.syncer.stop() | |
| if ctx is not None and ctx.expirer is not None: | |
| await ctx.expirer.stop() | |
| await userbot.close() | |
| if db is not None: | |
| await db.close() | |
| await reporter.stop() | |
| if __name__ == "__main__": | |
| try: | |
| asyncio.run(main()) | |
| except KeyboardInterrupt: | |
| pass |