1#!/usr/bin/env python3
2"""Manage Telegram bot public identity and account-level settings via Bot API."""
3 
4from __future__ import annotations
5 
6import argparse
7import json
8import mimetypes
9import os
10import sys
11import tempfile
12import urllib.error
13import urllib.request
14import uuid
15from pathlib import Path
16from typing import Any
17 
18DEFAULT_TOKEN_ENV_VARS = [
19    "TELEGRAM_BOT_TOKEN",
20    "BOT_TOKEN",
21    "TG_BOT_TOKEN",
22]
23 
24DEFAULT_TOKEN_FILE_ENV_VARS = [
25    "TELEGRAM_BOT_TOKEN_FILE",
26    "BOT_TOKEN_FILE",
27    "TG_BOT_TOKEN_FILE",
28    "OPENCLAW_TELEGRAM_TOKEN_FILE",
29]
30 
31DEFAULT_TOKEN_FILES = [
32    "~/.config/telegram-bot-settings/bot.token",
33    "~/.config/telegram-bot-settings/telegram-bot.token",
34    "~/.config/telegram-bot-settings/telegram_bot_token",
35    "~/.secrets/telegram-bot-settings/bot.token",
36    "~/.secrets/telegram_bot_token",
37    "./secrets/telegram-bot.token",
38    "./secrets/telegram_bot_token",
39    "~/.openclaw/credentials/telegram-default.token",
40]
41 
42DEFAULT_DOTENV_FILES = [
43    os.environ.get("TELEGRAM_BOT_SETTINGS_ENV"),
44    "./secrets/telegram-bot-settings.env",
45    "./secrets/telegram.env",
46    "./.env.local",
47    "./.env",
48    "~/.config/telegram-bot-settings/.env",
49]
50 
51 
52class BotAPIError(RuntimeError):
53    pass
54 
55 
56def eprint(*args: Any) -> None:
57    print(*args, file=sys.stderr)
58 
59 
60def read_dotenv_value(path: Path, keys: list[str]) -> str | None:
61    try:
62        for raw_line in path.read_text(encoding="utf-8").splitlines():
63            line = raw_line.strip()
64            if not line or line.startswith("#"):
65                continue
66            if line.startswith("export "):
67                line = line[len("export ") :].strip()
68            if "=" not in line:
69                continue
70            name, value = line.split("=", 1)
71            name = name.strip()
72            if name not in keys:
73                continue
74            token = value.strip().strip("\"'").strip()
75            if token:
76                return token
77    except OSError:
78        return None
79    return None
80 
81 
82def read_token(args: argparse.Namespace) -> str:
83    if getattr(args, "token", None):
84        return args.token.strip()
85 
86    for env_name in DEFAULT_TOKEN_ENV_VARS:
87        env_token = os.environ.get(env_name)
88        if env_token:
89            return env_token.strip()
90 
91    candidate_paths: list[str] = []
92    if getattr(args, "token_file", None):
93        candidate_paths.append(args.token_file)
94    candidate_paths.extend(os.environ.get(env_name) for env_name in DEFAULT_TOKEN_FILE_ENV_VARS)
95    candidate_paths.extend(DEFAULT_TOKEN_FILES)
96 
97    for raw_path in candidate_paths:
98        if not raw_path:
99            continue
100        path = Path(raw_path).expanduser()
101        if path.exists():
102            token = path.read_text(encoding="utf-8").strip()
103            if token:
104                return token
105 
106    dotenv_keys = DEFAULT_TOKEN_ENV_VARS + DEFAULT_TOKEN_FILE_ENV_VARS
107    for raw_path in DEFAULT_DOTENV_FILES:
108        if not raw_path:
109            continue
110        path = Path(raw_path).expanduser()
111        if not path.exists():
112            continue
113        token = read_dotenv_value(path, DEFAULT_TOKEN_ENV_VARS)
114        if token:
115            return token
116        token_file = read_dotenv_value(path, DEFAULT_TOKEN_FILE_ENV_VARS)
117        if token_file:
118            candidate = Path(token_file).expanduser()
119            if candidate.exists():
120                value = candidate.read_text(encoding="utf-8").strip()
121                if value:
122                    return value
123 
124    raise SystemExit(
125        "Could not find a Telegram bot token. Pass --token, --token-file, set TELEGRAM_BOT_TOKEN/BOT_TOKEN/TG_BOT_TOKEN, or store it in a supported secret file or dotenv file."
126    )
127 
128 
129def load_jsonish(value: str | None) -> Any:
130    if value is None:
131        return None
132    raw = value
133    if raw.startswith("@"):
134        raw = Path(raw[1:]).expanduser().read_text(encoding="utf-8")
135    return json.loads(raw)
136 
137 
138def normalize_form_value(value: Any) -> str:
139    if isinstance(value, (dict, list)):
140        return json.dumps(value, ensure_ascii=False)
141    if isinstance(value, bool):
142        return "true" if value else "false"
143    if value is None:
144        return ""
145    return str(value)
146 
147 
148class TelegramBotAPI:
149    def __init__(self, token: str, timeout: int = 60) -> None:
150        self.token = token
151        self.timeout = timeout
152        self.base_url = f"https://api.telegram.org/bot{token}"
153 
154    def call(self, method: str, data: dict[str, Any] | None = None, files: dict[str, Path] | None = None) -> Any:
155        url = f"{self.base_url}/{method}"
156        data = {k: v for k, v in (data or {}).items() if v is not None}
157 
158        if files:
159            boundary = "----telegrambotsettings" + uuid.uuid4().hex
160            body = self._build_multipart_body(boundary, data, files)
161            request = urllib.request.Request(
162                url,
163                data=body,
164                headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
165            )
166        elif data:
167            body = json.dumps(data, ensure_ascii=False).encode("utf-8")
168            request = urllib.request.Request(
169                url,
170                data=body,
171                headers={"Content-Type": "application/json; charset=utf-8"},
172            )
173        else:
174            request = urllib.request.Request(url)
175 
176        try:
177            with urllib.request.urlopen(request, timeout=self.timeout) as response:
178                payload = json.load(response)
179        except urllib.error.HTTPError as exc:
180            payload = self._decode_error_payload(exc)
181            description = payload.get("description") or str(exc)
182            raise BotAPIError(f"{method} failed: {description}") from exc
183        except urllib.error.URLError as exc:
184            raise BotAPIError(f"{method} failed: {exc}") from exc
185 
186        if not payload.get("ok"):
187            description = payload.get("description") or "Unknown Telegram Bot API error"
188            raise BotAPIError(f"{method} failed: {description}")
189        return payload.get("result")
190 
191    @staticmethod
192    def _decode_error_payload(exc: urllib.error.HTTPError) -> dict[str, Any]:
193        try:
194            body = exc.read().decode("utf-8", "replace")
195            return json.loads(body)
196        except Exception:
197            return {"ok": False, "description": str(exc)}
198 
199    @staticmethod
200    def _build_multipart_body(boundary: str, data: dict[str, Any], files: dict[str, Path]) -> bytes:
201        crlf = "\r\n"
202        parts: list[bytes] = []
203 
204        for name, value in data.items():
205            parts.append(
206                (
207                    f"--{boundary}{crlf}"
208                    f'Content-Disposition: form-data; name="{name}"{crlf}{crlf}'
209                    f"{normalize_form_value(value)}{crlf}"
210                ).encode("utf-8")
211            )
212 
213        for field_name, path in files.items():
214            content_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream"
215            parts.append(
216                (
217                    f"--{boundary}{crlf}"
218                    f'Content-Disposition: form-data; name="{field_name}"; filename="{path.name}"{crlf}'
219                    f"Content-Type: {content_type}{crlf}{crlf}"
220                ).encode("utf-8")
221            )
222            parts.append(path.read_bytes())
223            parts.append(crlf.encode("utf-8"))
224 
225        parts.append(f"--{boundary}--{crlf}".encode("utf-8"))
226        return b"".join(parts)
227 
228 
229def print_json(payload: Any) -> None:
230    print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True))
231 
232 
233def maybe_language_code(args: argparse.Namespace) -> dict[str, Any]:
234    return {"language_code": args.language_code} if getattr(args, "language_code", None) else {}
235 
236 
237def maybe_scope(args: argparse.Namespace) -> dict[str, Any]:
238    data: dict[str, Any] = {}
239    if getattr(args, "scope", None):
240        data["scope"] = load_jsonish(args.scope)
241    if getattr(args, "language_code", None):
242        data["language_code"] = args.language_code
243    return data
244 
245 
246def ensure_static_photo(path: Path, keep_temp: bool = False) -> tuple[Path, Path | None]:
247    if path.suffix.lower() in {".jpg", ".jpeg"}:
248        return path, None
249 
250    try:
251        from PIL import Image
252    except ImportError as exc:
253        raise SystemExit(
254            "Static profile photos must be JPG/JPEG. Install Pillow to auto-convert or pre-convert the file."
255        ) from exc
256 
257    image = Image.open(path).convert("RGB")
258    temp = tempfile.NamedTemporaryFile(prefix="telegram-bot-photo-", suffix=".jpg", delete=False)
259    temp_path = Path(temp.name)
260    temp.close()
261    image.save(temp_path, format="JPEG", quality=95)
262    return temp_path, temp_path if not keep_temp else None
263 
264 
265def command_status(api: TelegramBotAPI, args: argparse.Namespace) -> None:
266    summary = {
267        "me": api.call("getMe"),
268        "name": api.call("getMyName", maybe_language_code(args)),
269        "description": api.call("getMyDescription", maybe_language_code(args)),
270        "short_description": api.call("getMyShortDescription", maybe_language_code(args)),
271        "commands": api.call("getMyCommands", maybe_scope(args)),
272        "menu_button": api.call("getChatMenuButton"),
273        "default_admin_rights": {
274            "groups": api.call("getMyDefaultAdministratorRights"),
275            "channels": api.call("getMyDefaultAdministratorRights", {"for_channels": True}),
276        },
277    }
278    print_json(summary)
279 
280 
281def command_get_name(api: TelegramBotAPI, args: argparse.Namespace) -> None:
282    print_json(api.call("getMyName", maybe_language_code(args)))
283 
284 
285def command_set_name(api: TelegramBotAPI, args: argparse.Namespace) -> None:
286    data = {"name": args.text, **maybe_language_code(args)}
287    print_json(api.call("setMyName", data))
288 
289 
290def command_get_description(api: TelegramBotAPI, args: argparse.Namespace) -> None:
291    print_json(api.call("getMyDescription", maybe_language_code(args)))
292 
293 
294def command_set_description(api: TelegramBotAPI, args: argparse.Namespace) -> None:
295    data = {"description": args.text, **maybe_language_code(args)}
296    print_json(api.call("setMyDescription", data))
297 
298 
299def command_get_short_description(api: TelegramBotAPI, args: argparse.Namespace) -> None:
300    print_json(api.call("getMyShortDescription", maybe_language_code(args)))
301 
302 
303def command_set_short_description(api: TelegramBotAPI, args: argparse.Namespace) -> None:
304    data = {"short_description": args.text, **maybe_language_code(args)}
305    print_json(api.call("setMyShortDescription", data))
306 
307 
308def command_get_commands(api: TelegramBotAPI, args: argparse.Namespace) -> None:
309    print_json(api.call("getMyCommands", maybe_scope(args)))
310 
311 
312def command_set_commands(api: TelegramBotAPI, args: argparse.Namespace) -> None:
313    data = {"commands": load_jsonish(args.commands), **maybe_scope(args)}
314    print_json(api.call("setMyCommands", data))
315 
316 
317def command_delete_commands(api: TelegramBotAPI, args: argparse.Namespace) -> None:
318    print_json(api.call("deleteMyCommands", maybe_scope(args)))
319 
320 
321def command_get_menu_button(api: TelegramBotAPI, args: argparse.Namespace) -> None:
322    data = {"chat_id": args.chat_id} if args.chat_id is not None else None
323    print_json(api.call("getChatMenuButton", data))
324 
325 
326def command_set_menu_button(api: TelegramBotAPI, args: argparse.Namespace) -> None:
327    data: dict[str, Any] = {"menu_button": load_jsonish(args.button)}
328    if args.chat_id is not None:
329        data["chat_id"] = args.chat_id
330    print_json(api.call("setChatMenuButton", data))
331 
332 
333def command_get_default_admin_rights(api: TelegramBotAPI, args: argparse.Namespace) -> None:
334    data = {"for_channels": True} if args.for_channels else None
335    print_json(api.call("getMyDefaultAdministratorRights", data))
336 
337 
338def command_set_default_admin_rights(api: TelegramBotAPI, args: argparse.Namespace) -> None:
339    data: dict[str, Any] = {}
340    if args.rights:
341        data["rights"] = load_jsonish(args.rights)
342    if args.for_channels:
343        data["for_channels"] = True
344    print_json(api.call("setMyDefaultAdministratorRights", data))
345 
346 
347def command_set_photo(api: TelegramBotAPI, args: argparse.Namespace) -> None:
348    path = Path(args.path).expanduser().resolve()
349    if not path.exists():
350        raise SystemExit(f"File does not exist: {path}")
351 
352    cleanup_path: Path | None = None
353    try:
354        if args.animated:
355            data: dict[str, Any] = {
356                "photo": {
357                    "type": "animated",
358                    "animation": "attach://animation_file",
359                }
360            }
361            if args.main_frame_timestamp is not None:
362                data["photo"]["main_frame_timestamp"] = args.main_frame_timestamp
363            result = api.call(
364                "setMyProfilePhoto",
365                data=data,
366                files={"animation_file": path},
367            )
368        else:
369            upload_path, cleanup_candidate = ensure_static_photo(path, keep_temp=args.keep_temp)
370            cleanup_path = cleanup_candidate
371            data = {
372                "photo": {
373                    "type": "static",
374                    "photo": "attach://photo_file",
375                }
376            }
377            result = api.call(
378                "setMyProfilePhoto",
379                data=data,
380                files={"photo_file": upload_path},
381            )
382        print_json(result)
383    finally:
384        if cleanup_path and cleanup_path.exists():
385            cleanup_path.unlink()
386 
387 
388def command_remove_photo(api: TelegramBotAPI, args: argparse.Namespace) -> None:
389    print_json(api.call("removeMyProfilePhoto"))
390 
391 
392def command_raw(api: TelegramBotAPI, args: argparse.Namespace) -> None:
393    data = load_jsonish(args.data) if args.data else None
394    files: dict[str, Path] = {}
395    for file_arg in args.file or []:
396        if "=" not in file_arg:
397            raise SystemExit(f"Invalid --file value '{file_arg}'. Use field_name=/path/to/file")
398        field_name, raw_path = file_arg.split("=", 1)
399        path = Path(raw_path).expanduser().resolve()
400        if not path.exists():
401            raise SystemExit(f"File does not exist: {path}")
402        files[field_name] = path
403    print_json(api.call(args.method, data=data, files=files or None))
404 
405 
406def add_common_auth_flags(parser: argparse.ArgumentParser) -> None:
407    parser.add_argument("--token", help="Telegram bot token. Overrides discovered env vars and secret files.")
408    parser.add_argument(
409        "--token-file",
410        help="Path to a file containing the Telegram bot token. Overrides discovered token-file env vars and default secret files.",
411    )
412 
413 
414def build_parser() -> argparse.ArgumentParser:
415    parser = argparse.ArgumentParser(
416        description="Manage Telegram bot public identity and account-level settings via Bot API."
417    )
418    add_common_auth_flags(parser)
419    subparsers = parser.add_subparsers(dest="command", required=True)
420 
421    status = subparsers.add_parser("status", help="Fetch a combined snapshot of common bot settings.")
422    add_common_auth_flags(status)
423    status.add_argument("--language-code", help="Optional IETF language code for localized values.")
424    status.add_argument("--scope", help="Optional JSON or @file for getMyCommands scope.")
425    status.set_defaults(func=command_status)
426 
427    get_name = subparsers.add_parser("get-name", help="Read the current bot name.")
428    add_common_auth_flags(get_name)
429    get_name.add_argument("--language-code", help="Optional IETF language code.")
430    get_name.set_defaults(func=command_get_name)
431 
432    set_name = subparsers.add_parser("set-name", help="Set the bot name.")
433    add_common_auth_flags(set_name)
434    set_name.add_argument("--text", required=True, help="New bot name.")
435    set_name.add_argument("--language-code", help="Optional IETF language code.")
436    set_name.set_defaults(func=command_set_name)
437 
438    get_description = subparsers.add_parser("get-description", help="Read the current bot description.")
439    add_common_auth_flags(get_description)
440    get_desc