如何在Python主文件中获取Telegram处理器更新的状态变量？

你遇到的核心问题是状态变量无法跨线程/进程共享：

• 若分别启动main.py和telegram.py，二者是独立进程，拥有各自的内存空间，STATUS_FLAG在两个进程中是完全独立的副本，修改其中一个不会影响另一个。
• 若在main.py中导入telegram.py并启动bot，aiogram的executor.start_polling会阻塞主线程，导致schedule任务无法执行；即使放到多线程，未同步的全局变量可能因线程缓存问题，导致主线程无法读取到最新值。

方案一：整合到异步事件循环（推荐） #

将telegram bot和schedule任务整合到同一个异步环境中，用类封装状态保证安全，避免多进程/线程的共享问题。

重构telegram.py #

import logging
from aiogram import Bot, Dispatcher, types
from aiogram.utils.executor import start_polling

TG_TOKEN = "TOKEN"

# 用类封装状态，避免全局变量的混乱，同时保证线程/异步安全
class BotStatus:
    def __init__(self):
        self._running = False
    
    @property
    def is_running(self):
        return self._running
    
    def start(self):
        self._running = True
    
    def stop(self):
        self._running = False

# 全局状态实例
status = BotStatus()

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

# 初始化bot和调度器
bot = Bot(token=TG_TOKEN)
dp = Dispatcher(bot)

@dp.message_handler(commands=['start', 'stop', 'status', 'help'])
async def send_welcome(message: types.Message):
    cmd = message.text
    if cmd == "/start":
        if not status.is_running:
            status.start()
            await message.reply("I'm going to start bot")
        else:
            await message.reply("Bot is already running")
    elif cmd == "/stop":
        if status.is_running:
            status.stop()
            await message.reply("I'm going to stop bot")
        else:
            await message.reply("Bot is already stopped")
    elif cmd == "/status":
        print(status.is_running)
        await message.reply("Bot is running" if status.is_running else "Bot is stopped")
    elif cmd == "/help":
        help_text = "/start - Start the bot\n/stop - Stop the bot\n/status - Get bot status"
        await message.reply(help_text)

@dp.message_handler()
async def echo(message: types.Message):
    print(message.text)
    await message.answer(message.text)

def run_bot():
    # 启动bot polling（同步阻塞方法，需放单独线程）
    start_polling(dp, skip_updates=True)

编写main.py #

import asyncio
import schedule
from threading import Thread
from telegram import status, run_bot

# 你的核心机器人任务
def core_bot_task():
    if status.is_running:
        print("Main bot is executing task...")
        # 这里替换为你的机器人业务逻辑

# 将schedule同步任务包装为异步函数
async def run_schedule():
    while True:
        schedule.run_pending()
        await asyncio.sleep(1)

async def main():
    # 在后台线程启动telegram bot，避免阻塞异步事件循环
    bot_thread = Thread(target=run_bot, daemon=True)
    bot_thread.start()
    
    # 初始化schedule任务
    schedule.every(5).seconds.do(core_bot_task)
    
    # 启动异步调度循环
    await run_schedule()

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

方案二：多进程共享状态（适用于必须分开进程的场景） #

用multiprocessing.Value创建跨进程共享的状态变量，通过锁保证修改的原子性。

修改telegram.py #

import logging
from aiogram import Bot, Dispatcher, executor, types
from multiprocessing import Value

TG_TOKEN = "TOKEN"

# 创建跨进程共享的整数状态，'i'表示int类型，初始值0
STATUS_FLAG = Value('i', 0)

logging.basicConfig(level=logging.INFO)
bot = Bot(token=TG_TOKEN)
dp = Dispatcher(bot)

@dp.message_handler(commands=['start', 'stop', 'status', 'help'])
async def send_welcome(message: types.Message):
    cmd = message.text
    if cmd == "/start":
        with STATUS_FLAG.get_lock():
            if STATUS_FLAG.value == 0:
                STATUS_FLAG.value = 1
                await message.reply("I'm going to start bot")
            else:
                await message.reply("Bot is already running")
    elif cmd == "/stop":
        with STATUS_FLAG.get_lock():
            if STATUS_FLAG.value == 1:
                STATUS_FLAG.value = 0
                await message.reply("I'm going to stop bot")
            else:
                await message.reply("Bot is already stopped")
    elif cmd == "/status":
        with STATUS_FLAG.get_lock():
            current_status = STATUS_FLAG.value
        print(current_status)
        await message.reply("Bot is running" if current_status else "Bot is stopped")
    elif cmd == "/help":
        help_text = "/start - Start the bot\n/stop - Stop the bot\n/status - Get bot status"
        await message.reply(help_text)

@dp.message_handler()
async def echo(message: types.Message):
    print(message.text)
    await message.answer(message.text)

def run_bot():
    executor.start_polling(dp, skip_updates=True)

编写main.py #

import schedule
import time
from multiprocessing import Process
from telegram import STATUS_FLAG, run_bot

def core_bot_task():
    with STATUS_FLAG.get_lock():
        current_status = STATUS_FLAG.value
    if current_status == 1:
        print("Main bot is executing task...")
        # 替换为你的业务逻辑

if __name__ == "__main__":
    # 启动telegram bot进程
    bot_process = Process(target=run_bot)
    bot_process.start()
    
    # 初始化schedule任务
    schedule.every(5).seconds.do(core_bot_task)
    
    while True:
        schedule.run_pending()
        time.sleep(1)

内容的提问来源于stack exchange，提问作者Pedro Faria