外出先から自宅の Mac を操作する Telegram Bot を作ってみた

コピー

bot/
├── main.py           # エントリポイント・ハンドラ登録
├── config.py         # 環境変数・設定
├── handlers/
│   ├── dl_handler.py    # /dl コマンド
│   ├── help_handler.py # /help
│   ├── status_handler.py # /status
│   └── ...
└── commands/
    ├── dl_dispatch.py   # URL 判定とプラットフォーム振り分け
    ├── dl_youtube.py   # yt-dlp による YouTube 動画・音声
    ├── dl_twitter.py   # Twitter 動画
    ├── dl_common.py    # シグナルハンドラ・サブプロセス終了
    └── dl_state.py     # 実行中タスクの状態管理

コピー

def download(url, audio_only, live_stream) -> tuple[bool, str]:
    if dl_youtube.is_youtube_url(url):
        if audio_only:
            return dl_youtube.download_audio(url, live_stream=live_stream)
        return dl_youtube.download_video(url, live_stream=live_stream)
    if dl_twitter.is_twitter_url(url):
        return dl_twitter.download_video(url)
    return False, "対応していない URL です。"

コピー

# ブロッキング処理を別スレッドで実行
download_id = register_download(url, media_type, live_stream, user_id)
try:
    success, message = await asyncio.to_thread(
        download, url, audio_only, live_stream
    )
    await update.message.reply_text(message)
finally:
    unregister_download(download_id)

コピー

ydl_opts = {
    "format": "bestvideo+bestaudio/best",
    "merge_output_format": "mp4",
    "outtmpl": str(DL_OUTPUT_DIR / "%(id)s.%(ext)s"),
    "restrictfilenames": True,  # パストラバーサル対策
    "quiet": True,
    "no_warnings": True,
}
if live_stream:
    ydl_opts["live_from_start"] = True

コピー

# Ctrl+C 等で Bot 終了時にサブプロセスを確実に終了
def _terminate_active_subprocesses():
    for proc in _active_ytdlp_processes:
        if proc.poll() is None:
            proc.terminate()
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()

signal.signal(signal.SIGINT, _signal_handler)
atexit.register(_terminate_active_subprocesses)

コピー

# 許可するユーザー ID のリスト（カンマ区切り）
_raw_ids = os.getenv("ALLOWED_USER_IDS", "").strip()
ALLOWED_USER_IDS = [int(uid.strip()) for uid in _raw_ids.split(",") if uid.strip()]