Spaces:
Runtime error
Runtime error
File size: 3,483 Bytes
4a8701f 67bedca 0475441 4a8701f c9b0131 4a8701f c9b0131 4a8701f c9b0131 67bedca 61aaa78 67bedca 4a8701f 67bedca 4a8701f 67bedca c9b0131 4a8701f c9b0131 229bf89 61aaa78 4a8701f 0475441 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | import asyncio
import logging
import yaml
import os
import re
import html
import subprocess
from aiogram import Bot, Dispatcher, types, F
from aiogram.types import InlineQueryResultArticle, InputTextMessageContent
from aiogram.filters import CommandStart
from aiohttp import web
from config import BOT_TOKEN
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
async def health_check(request):
return web.Response(text="Bot is running!")
def load_snippets():
if not os.path.exists("snippets.yaml"):
return {}
with open("snippets.yaml", "r", encoding="utf-8") as f:
try:
return yaml.safe_load(f) or {}
except:
return {}
snippets = load_snippets()
def safe_format(text):
text = html.escape(text)
text = re.sub(r'\*\*(.*?)\*\*', r'<b>\1</b>', text)
text = re.sub(r'\*(.*?)\*', r'<i>\1</i>', text)
return text
bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()
@dp.message(CommandStart())
async def cmd_start(message: types.Message):
try:
bot_info = await bot.get_me()
await message.answer(f"Hello! I am an inline bot. Type @{bot_info.username} in any chat to see my snippets.")
except:
await message.answer("Hello! I am an inline bot.")
@dp.inline_query()
async def inline_query_handler(query: types.InlineQuery):
query_text = query.query.lower().strip()
results = []
for key, data in snippets.items():
title = data.get("title", key)
content = data.get("content", "")
if not query_text or query_text in key.lower() or query_text in title.lower() or query_text in content.lower():
description = content.split("\n")[0] if content else "No content"
if len(description) > 50:
description = description[:47] + "..."
results.append(
InlineQueryResultArticle(
id=key, title=title, description=description,
input_message_content=InputTextMessageContent(message_text=safe_format(content), parse_mode="HTML"),
)
)
await query.answer(results[:50], cache_time=1)
async def main():
app = web.Application()
app.router.add_get("/", health_check)
runner = web.AppRunner(app)
await runner.setup()
await web.TCPSite(runner, "0.0.0.0", 7860).start()
logger.info("Attempting to connect to Telegram...")
for attempt in range(1, 11):
try:
bot_info = await bot.get_me()
logger.info(f"Connected! Bot: @{bot_info.username}")
await dp.start_polling(bot)
break
except Exception as e:
if attempt == 10:
logger.error("Final connection attempt failed.")
return
logger.warning(f"Attempt {attempt} failed, retrying in 5s...")
await asyncio.sleep(5)
if __name__ == "__main__":
import sys
if "--worker" in sys.argv or os.name != 'nt' or os.environ.get("DOCKER_ENV"):
try:
asyncio.run(main())
except (KeyboardInterrupt, SystemExit):
logger.info("Bot stopped!")
else:
try:
subprocess.Popen([sys.executable, "manager.py"], creationflags=subprocess.CREATE_NEW_CONSOLE)
except:
asyncio.run(main())
|