如何修复 Pyrogram Telegram 机器人无法发送视频的问题

本文详解 pyrogram 机器人下载并发送 youtube 视频时“卡在最后一步”的常见原因，包括异步阻塞、文件路径错误、权限问题及缺乏异常追踪，并提供可立即验证的调试方案与健壮实现。

你的代码逻辑看似完整：下载视频 → 等待文件生成 → 调用 send_video 发送。但实际中 await client.send_video(...) 未执行或静默失败，根本原因往往不是配置错误，而是未捕获底层异常 + 同步/异步混用导致阻塞 + 文件路径与权限隐患。下面逐层分析并给出生产级修复方案。

? 1. 关键问题定位：必须启用详细异常追踪

原代码中 except Exception as e: 仅打印字符串，而 Pyrogram 的网络异常（如 FilePartMissing, Timeout, PeerIdInvalid, ChatWriteForbidden）常被吞掉。务必引入 traceback 获取完整堆栈：

import traceback

# 替换原 send_video 的 except 块：
except Exception as e:
    print(f"❌ Failed to send video:")
    traceback.print_exc()  # ← 关键！输出完整错误链
    return

运行后你极可能看到类似：

Magician

Figma插件，AI生成图标、图片和UX文案

• pyrogram.errors.exceptions.bad_request_400.FilePartMissing: The uploaded file is broken
• pyrogram.errors.exceptions.flood_420.FloodWait: A wait of X seconds is required
• OSError: [Errno 2] No such file or directory: 'video.mp4'

这些信息是调试的唯一入口。

⚠️ 2. 核心陷阱：同步下载阻塞异步事件循环

ys.download(...) 是同步阻塞调用，会冻结整个 Pyrogram 异步事件循环，导致后续 await client.send_video() 无法调度，甚至引发超时或死锁。PyTube 不支持原生异步下载，必须用 loop.run_in_executor 脱离主线程：

import asyncio
from concurrent.futures import ThreadPoolExecutor

# 创建线程池（全局复用）
executor = ThreadPoolExecutor(max_workers=4)

@app.on_message()
async def send_video(client, message):
    if not message.text or not message.text.startswith("https://"):
        await message.reply("⚠️ 请发送有效的 YouTube 链接")
        return

    URL = message.text.strip()
    yt = YouTube(URL)

    # ✅ 异步安全下载：委托给线程池
    try:
        ys = yt.streams.get_highest_resolution()
        video_path = "video.mp4"

        # 在线程中执行下载（避免阻塞事件循环）
        await asyncio.get_event_loop().run_in_executor(
            executor, 
            lambda: ys.download(filename=video_path)
        )
        print("✅ Video downloaded successfully")
    except Exception as e:
        print(f"❌ Download failed: {e}")
        await message.reply("下载失败，请检查链接或重试")
        return

    # ✅ 不再轮询等待：直接校验文件存在性
    if not os.path.exists(video_path) or os.path.getsize(video_path) == 0:
        await message.reply("⚠️ 下载完成但文件为空，请重试")
        return

    # ✅ 使用绝对路径避免路径歧义（尤其 Docker/服务器环境）
    abs_path = os.path.abspath(video_path)

    try:
        # 发送前可选：添加进度回调（提升用户体验）
        def progress(current, total):
            percent = (current / total) * 100
            if int(percent) % 20 == 0:  # 每20%上报一次
                asyncio.create_task(message.reply(f"? 上传中... {int(percent)}%"))

        await client.send_video(
            chat_id=message.chat.id,
            video=abs_path,
            duration=yt.length,
            caption=f"? {yt.title[:50]}...",
            progress=progress
        )
        print("✅ Video sent successfully")
        await message.reply("✅ 视频已发送！")
    except Exception as e:
        print(f"❌ Send failed:")
        traceback.print_exc()
        await message.reply("发送失败，请稍后重试或联系管理员")
    finally:
        # ✅ 清理临时文件（防止磁盘占满）
        if os.path.exists(abs_path):
            os.remove(abs_path)

?️ 3. 必须检查的硬性前提

• 文件大小限制：Telegram Bot API 对普通用户聊天中视频限制为 50 MB；若视频超限，send_video 会直接抛出 FileTooBig 异常。建议提前校验：

  # 在下载后加入
  size_mb = os.path.getsize(abs_path) / (1024 * 1024)
  if size_mb > 45:  # 预留缓冲
      await message.reply(f"⚠️ 视频过大 ({size_mb:.1f} MB)，Telegram 限制为 50 MB")
      return

• Bot 权限：确保 Bot 在目标群组/频道中具有 can_send_videos 权限（私聊默认允许）。
• API 限频：频繁调用可能触发 FloodWait。添加简单退避：

  except FloodWait as e:
      print(f"⏳ Flood wait for {e.value} seconds")
      await asyncio.sleep(e.value + 1)
      # 重试逻辑（需封装为递归或循环）

✅ 最终验证步骤

1. 用小视频测试（如答案中的 1.5MB 示例）确认基础流程通路；
2. 启用 traceback.print_exc() 后复现问题，根据报错精准定位；
3. 将 yt.streams.get_highest_resolution() 替换为 yt.streams.filter(progressive=True, file_extension='mp4').first() 提升兼容性；
4. 生产环境务必使用 logging 替代 print，并添加消息去重（防重复触发）。

? 总结：Pyrogram 机器人“不发送”几乎从不是配置问题，而是同步阻塞、异常静默、路径/权限疏漏三者叠加的结果。用 ThreadPoolExecutor 解耦下载、用 traceback 揭示真相、用绝对路径和大小校验筑牢防线——这才是稳定交付的关键。