148 lines
4.2 KiB
Python
Executable File
148 lines
4.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
GENERATE_SCRIPT = SCRIPT_DIR / "generate_table.py"
|
|
ENV_PATH = Path.home() / ".openclaw" / ".env"
|
|
|
|
|
|
class SendTableError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def load_env_file(path: Path) -> dict[str, str]:
|
|
values: dict[str, str] = {}
|
|
if not path.exists():
|
|
return values
|
|
|
|
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)
|
|
values[key.strip()] = value.strip().strip('"').strip("'")
|
|
return values
|
|
|
|
|
|
def resolve_credentials(bot_token: str | None, chat_id: str | None) -> tuple[str, str]:
|
|
env_values = load_env_file(ENV_PATH)
|
|
token = bot_token or os.environ.get("BOT_TOKEN") or env_values.get("BOT_TOKEN")
|
|
chat = chat_id or os.environ.get("CHAT_ID") or env_values.get("CHAT_ID")
|
|
|
|
if not token:
|
|
raise SendTableError("BOT_TOKEN is required (arg or ~/.openclaw/.env)")
|
|
if not chat:
|
|
raise SendTableError("CHAT_ID is required (arg or ~/.openclaw/.env)")
|
|
|
|
return token, chat
|
|
|
|
|
|
def parse_args(argv: list[str]) -> tuple[str, dict[str, Any], str | None, str | None]:
|
|
if len(argv) < 2:
|
|
raise SendTableError("Usage: ./send_table.py TITLE JSON [BOT_TOKEN] [CHAT_ID]")
|
|
|
|
title = argv[0]
|
|
raw_json = argv[1]
|
|
bot_token = argv[2] if len(argv) >= 3 else None
|
|
chat_id = argv[3] if len(argv) >= 4 else None
|
|
|
|
try:
|
|
payload = json.loads(raw_json)
|
|
except json.JSONDecodeError as exc:
|
|
raise SendTableError(f"Invalid JSON: {exc}") from exc
|
|
|
|
if not isinstance(payload, dict):
|
|
raise SendTableError("JSON payload must be an object")
|
|
|
|
payload["title"] = title
|
|
return title, payload, bot_token, chat_id
|
|
|
|
|
|
def generate_png(payload: dict[str, Any]) -> Path:
|
|
result = subprocess.run(
|
|
[sys.executable, str(GENERATE_SCRIPT)],
|
|
input=json.dumps(payload, ensure_ascii=False),
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
raise SendTableError(result.stderr.strip() or "generate_table.py failed")
|
|
|
|
output = result.stdout.strip()
|
|
if not output:
|
|
raise SendTableError("generate_table.py did not return output path")
|
|
|
|
path = Path(output)
|
|
if not path.exists():
|
|
raise SendTableError(f"Generated file not found: {path}")
|
|
|
|
return path
|
|
|
|
|
|
def send_photo(bot_token: str, chat_id: str, title: str, image_path: Path) -> int:
|
|
caption = title[:1024]
|
|
url = f"https://api.telegram.org/bot{bot_token}/sendPhoto"
|
|
command = [
|
|
"curl",
|
|
"--silent",
|
|
"--show-error",
|
|
"--fail",
|
|
"-X",
|
|
"POST",
|
|
url,
|
|
"-F",
|
|
f"chat_id={chat_id}",
|
|
"-F",
|
|
f"caption={caption}",
|
|
"-F",
|
|
f"photo=@{image_path}",
|
|
]
|
|
result = subprocess.run(command, text=True, capture_output=True, check=False)
|
|
|
|
if result.returncode != 0:
|
|
raise SendTableError(result.stderr.strip() or "curl sendPhoto failed")
|
|
|
|
try:
|
|
response = json.loads(result.stdout)
|
|
except json.JSONDecodeError as exc:
|
|
raise SendTableError(f"Telegram API returned invalid JSON: {exc}") from exc
|
|
|
|
if not response.get("ok"):
|
|
raise SendTableError(f"Telegram API error: {response}")
|
|
|
|
message_id = response.get("result", {}).get("message_id")
|
|
if message_id is None:
|
|
raise SendTableError("Telegram response missing message_id")
|
|
|
|
return int(message_id)
|
|
|
|
|
|
def main() -> int:
|
|
try:
|
|
title, payload, bot_token, chat_id = parse_args(sys.argv[1:])
|
|
token, chat = resolve_credentials(bot_token, chat_id)
|
|
image_path = generate_png(payload)
|
|
message_id = send_photo(token, chat, title, image_path)
|
|
except SendTableError as exc:
|
|
print(str(exc), file=sys.stderr)
|
|
return 1
|
|
except Exception as exc: # pragma: no cover - defensive CLI handling
|
|
print(f"Unexpected error: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
print(message_id)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|