Home Telegram Bot

from fastapi import FastAPI, File, HTTPException, UploadFile, status
from telegram import Bot

from model.fastapi_internal import Message, SystemMessage

app = FastAPI()
api_path_v1 = "/homebot/api/v1"

telegram_bot = Bot(token=settings["bot_token"])
admin_chat_id = settings["admin_chat_id"]


@app.post(f"{api_path_v1}/msg", status_code=status.HTTP_200_OK)
async def send_message(msg: Message):
    await telegram_bot.send_message(
        chat_id=admin_chat_id,
        text=f"{msg.type}:\n{msg.text}",
    )

    return "Message is sent successfully"


@app.post(f"{api_path_v1}/sys_msg", status_code=status.HTTP_200_OK)
async def send_system_message(msg: SystemMessage):
    await telegram_bot.send_message(
        chat_id=admin_chat_id,
        text=f"{msg.type}:\n{msg.text}",
    )

    return "Message is sent successfully"


@app.post(f"{api_path_v1}/media", status_code=status.HTTP_200_OK)
async def upload_media(file: UploadFile | None = File(...)):
    accepted_photo_types = [
        "image/png",
        "image/jpeg",
        "image/jpg",
        "image/heic",
        "image/heif",
        "image/heics",
    ]

    if file.content_type is None:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail="No file was uploaded",
        )

    if file.content_type not in accepted_photo_types:
        raise HTTPException(
            status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
            detail="Unsupported file type",
        )

    await telegram_bot.send_photo(
        chat_id=admin_chat_id,
        photo=file.file,
    )

    return "Photo was uploaded successfully"

from telegram import Bot, Chat, Update, User

telegram_bot = Bot(token=settings["bot_token"])


def check_full(user: User, chat: Chat) -> bool:
    texter = {
        "name": user.first_name,
        "username": user.username,
        "id": str(user.id),
    }

    for chat_id in settings["permitted_chat_list"]:
        if chat.id != int(chat_id):
            continue

        for permitted_user in settings["permitted_user_list"]:
            if texter == permitted_user:
                return True

    return False


async def check_user(update: Update) -> bool:
    user = update.effective_user
    chat = update.effective_chat

    access = check_full(user, chat)

    if access:
        log(f"Access granted\nUserID: {user.id}\nUserName: {user.username}\nChatID: {chat.id}")
        return True

    await update.message.reply_text(
        text="❌ Access denied. This bot does not operate for your account. ❌"
    )

    await telegram_bot.ban_chat_member(
        chat_id=chat.id,
        user_id=user.id,
    )

    await telegram_bot.send_message(
        chat_id=settings["admin_chat_id"],
        text=(
            f"Unknown user tried to access the Home Telegram Bot.\n\n"
            f"Name: {user.name}\n"
            f"Username: {user.username}\n"
            f"User ID: {user.id}\n"
            f"Chat ID: {chat.id}"
        ),
    )

    log(f"Access denied\nUserID: {user.id}\nUserName: {user.username}\nChatID: {chat.id}")

    return False

import asyncio

import uvicorn
from fastapi import FastAPI
from telegram import Bot
from telegram.ext import Application, CommandHandler

app = FastAPI()
telegram_bot = Bot(token=settings["bot_token"])


async def main() -> None:
    telegram_app = Application.builder().bot(telegram_bot).build()

    telegram_app.add_handler(CommandHandler("hi", say_hello))
    telegram_app.add_handler(CommandHandler("homeaddr", get_home_ip))
    telegram_app.add_handler(CommandHandler("unban", unban_user))

    config = uvicorn.Config(
        app=app,
        host=settings["api_url"],
        port=int(settings["api_port"]),
        use_colors=False,
        reload=False,
    )

    webserver = uvicorn.Server(config)

    async with telegram_app:
        await telegram_app.initialize()
        await telegram_app.start()
        await telegram_app.updater.start_polling(pool_timeout=10)

        await webserver.serve()

        await telegram_app.updater.stop()
        await telegram_app.stop()
        await telegram_app.shutdown()


if __name__ == "__main__":
    asyncio.run(main())