#!/usr/bin/env python3 """Small Telegram Bot API bridge for messages and files.""" from __future__ import annotations import argparse import json import mimetypes import os import sys import time import urllib.error import urllib.parse import urllib.request from pathlib import Path from typing import Any from uuid import uuid4 API_ROOT = "https://api.telegram.org" FILE_ROOT = "https://api.telegram.org/file" class TelegramError(RuntimeError): pass def load_dotenv(path: Path = Path(".env")) -> None: if not path.exists(): return for raw_line in path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip().strip('"').strip("'") os.environ.setdefault(key, value) def require_env(name: str) -> str: value = os.environ.get(name) if not value: raise TelegramError(f"Missing {name}. Add it to .env or export it.") return value def compact_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, separators=(",", ":")) class TelegramBot: def __init__(self, token: str, data_dir: Path, download_dir: Path) -> None: self.token = token self.data_dir = data_dir self.download_dir = download_dir self.state_path = data_dir / "state.json" self.inbox_path = data_dir / "inbox.jsonl" def api(self, method: str, payload: dict[str, Any] | None = None) -> dict[str, Any]: payload = payload or {} body = urllib.parse.urlencode(payload).encode("utf-8") request = urllib.request.Request( f"{API_ROOT}/bot{self.token}/{method}", data=body, method="POST", ) return self._read_json(request) def api_multipart( self, method: str, fields: dict[str, Any], file_field: str, file_path: Path, ) -> dict[str, Any]: boundary = f"----codex-telegram-{uuid4().hex}" body = bytearray() for key, value in fields.items(): body.extend(f"--{boundary}\r\n".encode("utf-8")) body.extend(f'Content-Disposition: form-data; name="{key}"\r\n\r\n'.encode("utf-8")) body.extend(str(value).encode("utf-8")) body.extend(b"\r\n") content_type = mimetypes.guess_type(file_path.name)[0] or "application/octet-stream" body.extend(f"--{boundary}\r\n".encode("utf-8")) body.extend( ( f'Content-Disposition: form-data; name="{file_field}"; ' f'filename="{file_path.name}"\r\n' f"Content-Type: {content_type}\r\n\r\n" ).encode("utf-8") ) body.extend(file_path.read_bytes()) body.extend(b"\r\n") body.extend(f"--{boundary}--\r\n".encode("utf-8")) request = urllib.request.Request( f"{API_ROOT}/bot{self.token}/{method}", data=bytes(body), method="POST", headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, ) return self._read_json(request) def _read_json(self, request: urllib.request.Request) -> dict[str, Any]: try: with urllib.request.urlopen(request, timeout=90) as response: data = json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") raise TelegramError(f"Telegram API HTTP {exc.code}: {detail}") from exc except urllib.error.URLError as exc: raise TelegramError(f"Telegram API request failed: {exc.reason}") from exc if not data.get("ok"): raise TelegramError(f"Telegram API error: {compact_json(data)}") return data def get_me(self) -> dict[str, Any]: return self.api("getMe")["result"] def get_updates(self, offset: int | None, timeout: int) -> list[dict[str, Any]]: payload: dict[str, Any] = { "timeout": timeout, "allowed_updates": json.dumps(["message"], separators=(",", ":")), } if offset is not None: payload["offset"] = offset return self.api("getUpdates", payload)["result"] def send_message(self, chat_id: str, text: str) -> dict[str, Any]: return self.api("sendMessage", {"chat_id": chat_id, "text": text})["result"] def send_document(self, chat_id: str, path: Path, caption: str | None = None) -> dict[str, Any]: if not path.is_file(): raise TelegramError(f"File not found: {path}") fields: dict[str, Any] = {"chat_id": chat_id} if caption: fields["caption"] = caption return self.api_multipart("sendDocument", fields, "document", path)["result"] def download_file(self, file_id: str, target_name: str | None = None) -> Path: file_info = self.api("getFile", {"file_id": file_id})["result"] file_path = file_info["file_path"] suffix_name = target_name or Path(file_path).name target = self.download_dir / suffix_name target.parent.mkdir(parents=True, exist_ok=True) request = urllib.request.Request(f"{FILE_ROOT}/bot{self.token}/{file_path}") try: with urllib.request.urlopen(request, timeout=90) as response: target.write_bytes(response.read()) except urllib.error.URLError as exc: raise TelegramError(f"Telegram file download failed: {exc.reason}") from exc return target def load_offset(self) -> int | None: if not self.state_path.exists(): return None data = json.loads(self.state_path.read_text(encoding="utf-8")) return data.get("offset") def save_offset(self, offset: int) -> None: self.data_dir.mkdir(parents=True, exist_ok=True) self.state_path.write_text( json.dumps({"offset": offset}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) def append_inbox(self, update: dict[str, Any]) -> None: self.data_dir.mkdir(parents=True, exist_ok=True) with self.inbox_path.open("a", encoding="utf-8") as stream: stream.write(compact_json(update) + "\n") def handle_update(self, update: dict[str, Any]) -> None: self.append_inbox(update) message = update.get("message") or {} chat = message.get("chat") or {} chat_id = chat.get("id") text = message.get("text") downloaded = self.download_message_files(message) if text == "/start" and chat_id is not None: self.send_message(str(chat_id), "Bot is connected. Send messages or files here.") summary = { "update_id": update.get("update_id"), "chat_id": chat_id, "from": (message.get("from") or {}).get("username"), "text": text, "downloaded": [str(path) for path in downloaded], } print(compact_json(summary), flush=True) def download_message_files(self, message: dict[str, Any]) -> list[Path]: downloaded: list[Path] = [] document = message.get("document") if document: downloaded.append(self.download_file(document["file_id"], document.get("file_name"))) for key in ("audio", "voice", "video", "animation", "sticker"): item = message.get(key) if item: downloaded.append(self.download_file(item["file_id"])) photos = message.get("photo") or [] if photos: largest = max(photos, key=lambda item: item.get("file_size", 0)) downloaded.append(self.download_file(largest["file_id"])) return downloaded def run(self, timeout: int, once: bool = False) -> None: offset = self.load_offset() print("Polling Telegram updates. Press Ctrl+C to stop.", flush=True) while True: updates = self.get_updates(offset=offset, timeout=timeout) for update in updates: self.handle_update(update) offset = int(update["update_id"]) + 1 self.save_offset(offset) if once: return if not updates: time.sleep(1) def resolve_chat_id(args: argparse.Namespace) -> str: chat_id = getattr(args, "chat_id", None) if chat_id: return chat_id default_chat_id = os.environ.get("TELEGRAM_DEFAULT_CHAT_ID") if default_chat_id: return default_chat_id raise TelegramError("Missing chat_id. Pass it explicitly or set TELEGRAM_DEFAULT_CHAT_ID.") def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Telegram Bot API bridge for @kjn3bot.") subparsers = parser.add_subparsers(dest="command", required=True) subparsers.add_parser("get-me", help="Print bot identity.") run_parser = subparsers.add_parser("run", help="Receive incoming messages and files.") run_parser.add_argument("--timeout", type=int, default=50, help="Long polling timeout in seconds.") run_parser.add_argument("--once", action="store_true", help="Fetch one batch and exit.") send_message = subparsers.add_parser("send-message", help="Send a text message.") send_message.add_argument("values", nargs="+", help="[chat_id] text") send_file = subparsers.add_parser("send-file", help="Send a file as a document.") send_file.add_argument("values", nargs="+", help="[chat_id] path") send_file.add_argument("--caption", help="Optional file caption.") return parser def normalize_send_message_args(args: argparse.Namespace) -> None: if len(args.values) == 1: args.chat_id = None args.text = args.values[0] else: args.chat_id = args.values[0] args.text = " ".join(args.values[1:]) def normalize_send_file_args(args: argparse.Namespace) -> None: if len(args.values) == 1: args.chat_id = None args.path = args.values[0] else: args.chat_id = args.values[0] args.path = args.values[1] def main() -> int: load_dotenv() parser = build_parser() args = parser.parse_args() try: bot = TelegramBot( token=require_env("TELEGRAM_BOT_TOKEN"), data_dir=Path(os.environ.get("TELEGRAM_DATA_DIR", "data")), download_dir=Path(os.environ.get("TELEGRAM_DOWNLOAD_DIR", "downloads")), ) if args.command == "get-me": print(json.dumps(bot.get_me(), ensure_ascii=False, indent=2)) return 0 if args.command == "run": bot.run(timeout=args.timeout, once=args.once) return 0 if args.command == "send-message": normalize_send_message_args(args) result = bot.send_message(resolve_chat_id(args), args.text) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if args.command == "send-file": normalize_send_file_args(args) result = bot.send_document(resolve_chat_id(args), Path(args.path), args.caption) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 except KeyboardInterrupt: return 130 except TelegramError as exc: print(f"error: {exc}", file=sys.stderr) return 1 parser.print_help() return 2 if __name__ == "__main__": raise SystemExit(main())