开发Notcoin Telegram机器人中间件：新矿池通知功能失效求助

问题描述 #

我正在开发一个介于Notcoin迷你应用与Telegram频道之间的机器人中间件，功能是当notcoin_bot出现新矿池时，自动向自建频道发送通知。我写了main.py和bot.py，但测试后只有/start命令的问候回复正常，新矿池检测和通知逻辑完全没生效，求帮忙排查问题。

代码展示 #

main.py #

from dotenv import load_dotenv
import asyncio
import logging
import bot

load_dotenv()

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(bot.start())

bot.py #

import asyncio
import aiohttp
from os import environ
from aiogram import Bot, Dispatcher, types
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.filters import CommandStart
from aiogram.types import Message
from bs4 import BeautifulSoup


dp = Dispatcher()

additional_message = "Join now to start earning!"
channel_id = ""
sent_announcements = []
check_interval = 60 * 60

#Function triggered by the "/start" command
async def start():
    global bot
    #Create a Bot instance using the Telegram API token from environment variables
    bot = Bot(
        environ.get("TOKEN"),
        default=DefaultBotProperties(parse_mode=ParseMode.HTML),
)

    await bot.delete_webhook(drop_pending_updates=True)
    await dp.start_polling(bot)

@dp.message(CommandStart())
#Function handling messages with the "/start" command
async def getUserByID(message: Message):
    await message.answer(f"Hello, Ваш ID: {message.from_user.id}")

#Function to fetch and parse pool data
async def fetch_and_parse_pools():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://t.me/notcoin_bot") as response:
            soup = BeautifulSoup(await response.text(), "html.parser")

    pools = soup.find_all("div", class_="pool")
    for pool in pools:
        pool_name = pool.find("h3").text
        pool_reward = pool.find("span", class_="reward").text
    
        if pool_name not in sent_announcements:
            sent_announcements.append(pool_name)

            announcement_text = f"**Новый пул:** {pool_name} | **Награда:** {pool_reward} в   час.\n{additional_message}"
            await bot.send_message(channel_id, announcement_text)


#Function to schedule pool checking (improved version)
async def scheduled(delay):
    await asyncio.sleep(delay)
    loop = asyncio.get_event_loop()
    loop.create_task(fetch_and_parse_pools())

    loop = asyncio.get_event_loop()
    loop.create_task(scheduled(check_interval))

问题排查与修复方案 #

1. 定时任务未启动 #

原代码中scheduled函数从未被调用，start函数只启动了消息轮询，没有触发矿池检测的定时逻辑。需要在start函数中添加定时任务的启动代码。

2. 频道ID未配置 #

channel_id是空字符串，必须填写正确的频道ID（格式如-100xxxxxxxxx），否则机器人无法向频道发送消息。

3. 网页解析逻辑错误 #

直接请求https://t.me/notcoin_bot获取的是Telegram网页版频道页面，页面中根本不存在class="pool"的元素。Notcoin的矿池数据属于迷你应用内部内容，网页端无法抓取，需要改用Notcoin官方提供的API接口（需确认服务条款），或模拟迷你应用的请求逻辑来获取数据。

4. 异步任务写法不规范 #

原scheduled函数使用get_event_loop的方式不够合理，且没有异常捕获机制，任务执行失败会静默终止。建议改用asyncio.create_task，并添加异常捕获确保任务稳定运行。

5. 消息格式不匹配 #

Bot初始化使用的是ParseMode.HTML，但消息内容用了Markdown的**加粗语法，格式不会生效，甚至可能导致消息发送失败。需要统一格式：要么将Bot的解析模式改为ParseMode.MARKDOWN_V2，要么将加粗语法替换为HTML的<b>标签。

修复后的bot.py示例 #

import asyncio
import aiohttp
from os import environ
from aiogram import Bot, Dispatcher, types
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.filters import CommandStart
from aiogram.types import Message
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

dp = Dispatcher()

additional_message = "Join now to start earning!"
# 填写正确的频道ID
channel_id = "-1001234567890"
sent_announcements = []
check_interval = 60 * 60

async def start():
    global bot
    bot = Bot(
        environ.get("TOKEN"),
        default=DefaultBotProperties(parse_mode=ParseMode.MARKDOWN_V2),
    )

    await bot.delete_webhook(drop_pending_updates=True)
    # 启动定时任务
    asyncio.create_task(scheduled(check_interval))
    # 启动消息轮询
    await dp.start_polling(bot)

@dp.message(CommandStart())
async def getUserByID(message: Message):
    await message.answer(f"Hello, Ваш ID: {message.from_user.id}")

async def fetch_and_parse_pools():
    try:
        # 注意：此处为示例，实际需替换为Notcoin官方API或合法的数据获取方式
        async with aiohttp.ClientSession() as session:
            # 假设使用Notcoin的矿池API接口
            async with session.get("https://api.notcoin.example/pools") as response:
                pools_data = await response.json()

        for pool in pools_data:
            pool_name = pool["name"]
            pool_reward = pool["reward_per_hour"]
        
            if pool_name not in sent_announcements:
                sent_announcements.append(pool_name)
                # 转义Markdown特殊字符避免格式错误
                escaped_name = pool_name.replace("*", "\\*").replace("_", "\\_")
                escaped_reward = str(pool_reward).replace("*", "\\*")
                announcement_text = f"**Новый пул:** {escaped_name} | **Награда:** {escaped_reward} в час.\n{additional_message}"
                await bot.send_message(channel_id, announcement_text)
                logger.info(f"发送新矿池通知: {pool_name}")
    except Exception as e:
        logger.error(f"获取矿池数据失败: {str(e)}")

async def scheduled(interval):
    while True:
        await asyncio.sleep(interval)
        asyncio.create_task(fetch_and_parse_pools())

内容的提问来源于stack exchange，提问作者Anton2000 Pavlukhin