144 lines
5.4 KiB
Python
144 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Send a test message via Telegram Bot API.
|
|
|
|
Usage:
|
|
python send_message.py --token "TOKEN" --chat-id "CHAT_ID" --text "Hello!"
|
|
python send_message.py --chat-id "CHAT_ID" --text "Hello!" # Uses env var
|
|
python send_message.py --chat-id "CHAT_ID" --photo "https://example.com/img.jpg"
|
|
python send_message.py --chat-id "CHAT_ID" --document "/path/to/file.pdf"
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from urllib.request import urlopen, Request
|
|
from urllib.error import HTTPError
|
|
|
|
|
|
def _mask_token(token: str) -> str:
|
|
"""Return a masked version of the token for safe logging."""
|
|
if not token or len(token) < 12:
|
|
return "***masked***"
|
|
return f"{token[:8]}...masked"
|
|
|
|
|
|
def api_call(token: str, method: str, data: dict) -> dict:
|
|
"""Make a Telegram Bot API call."""
|
|
url = f"https://api.telegram.org/bot{token}/{method}"
|
|
payload = json.dumps(data).encode("utf-8")
|
|
req = Request(url, data=payload, headers={"Content-Type": "application/json"})
|
|
|
|
try:
|
|
with urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read().decode())
|
|
except HTTPError as e:
|
|
error_body = json.loads(e.read().decode())
|
|
return error_body
|
|
|
|
|
|
def send_text(token: str, chat_id: str, text: str, parse_mode: str = None,
|
|
silent: bool = False) -> dict:
|
|
"""Send a text message."""
|
|
data = {"chat_id": chat_id, "text": text}
|
|
if parse_mode:
|
|
data["parse_mode"] = parse_mode
|
|
if silent:
|
|
data["disable_notification"] = True
|
|
return api_call(token, "sendMessage", data)
|
|
|
|
|
|
def send_photo(token: str, chat_id: str, photo: str, caption: str = None) -> dict:
|
|
"""Send a photo by URL or file_id."""
|
|
data = {"chat_id": chat_id, "photo": photo}
|
|
if caption:
|
|
data["caption"] = caption
|
|
return api_call(token, "sendPhoto", data)
|
|
|
|
|
|
def send_document_url(token: str, chat_id: str, document: str, caption: str = None) -> dict:
|
|
"""Send a document by URL."""
|
|
data = {"chat_id": chat_id, "document": document}
|
|
if caption:
|
|
data["caption"] = caption
|
|
return api_call(token, "sendDocument", data)
|
|
|
|
|
|
def send_location(token: str, chat_id: str, lat: float, lon: float) -> dict:
|
|
"""Send a location."""
|
|
return api_call(token, "sendLocation", {
|
|
"chat_id": chat_id,
|
|
"latitude": lat,
|
|
"longitude": lon
|
|
})
|
|
|
|
|
|
def send_poll(token: str, chat_id: str, question: str, options: list) -> dict:
|
|
"""Send a poll."""
|
|
return api_call(token, "sendPoll", {
|
|
"chat_id": chat_id,
|
|
"question": question,
|
|
"options": [{"text": opt} for opt in options]
|
|
})
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Send Telegram message")
|
|
parser.add_argument("--token", type=str, help="Bot token (or TELEGRAM_BOT_TOKEN env)")
|
|
parser.add_argument("--chat-id", type=str, required=True, help="Target chat ID")
|
|
parser.add_argument("--text", type=str, help="Text message to send")
|
|
parser.add_argument("--parse-mode", type=str, choices=["HTML", "MarkdownV2", "Markdown"],
|
|
help="Parse mode for text formatting")
|
|
parser.add_argument("--photo", type=str, help="Photo URL to send")
|
|
parser.add_argument("--document", type=str, help="Document URL to send")
|
|
parser.add_argument("--caption", type=str, help="Caption for photo/document")
|
|
parser.add_argument("--location", type=str, help="Location as 'lat,lon'")
|
|
parser.add_argument("--poll", type=str, help="Poll question")
|
|
parser.add_argument("--poll-options", type=str, nargs="+", help="Poll options")
|
|
parser.add_argument("--silent", action="store_true", help="Send without notification")
|
|
args = parser.parse_args()
|
|
|
|
token = args.token or os.environ.get("TELEGRAM_BOT_TOKEN")
|
|
if not token:
|
|
print("ERROR: Provide --token or set TELEGRAM_BOT_TOKEN environment variable")
|
|
sys.exit(1)
|
|
|
|
masked = _mask_token(token)
|
|
result = None
|
|
|
|
if args.text:
|
|
print(f"Sending text to {args.chat_id}...")
|
|
result = send_text(token, args.chat_id, args.text, args.parse_mode, args.silent)
|
|
elif args.photo:
|
|
print(f"Sending photo to {args.chat_id}...")
|
|
result = send_photo(token, args.chat_id, args.photo, args.caption)
|
|
elif args.document:
|
|
print(f"Sending document to {args.chat_id}...")
|
|
result = send_document_url(token, args.chat_id, args.document, args.caption)
|
|
elif args.location:
|
|
lat, lon = map(float, args.location.split(","))
|
|
print(f"Sending location to {args.chat_id}...")
|
|
result = send_location(token, args.chat_id, lat, lon)
|
|
elif args.poll and args.poll_options:
|
|
print(f"Sending poll to {args.chat_id}...")
|
|
result = send_poll(token, args.chat_id, args.poll, args.poll_options)
|
|
else:
|
|
print("ERROR: Provide --text, --photo, --document, --location, or --poll")
|
|
sys.exit(1)
|
|
|
|
if result and result.get("ok"):
|
|
msg = result["result"]
|
|
print(f"OK - Message sent! ID: {msg.get('message_id')}")
|
|
print(f" Chat: {msg.get('chat', {}).get('title', msg.get('chat', {}).get('first_name', 'N/A'))}")
|
|
print(f" Date: {msg.get('date')}")
|
|
else:
|
|
# Mask token in error output to prevent credential leakage
|
|
safe_output = json.dumps(result, indent=2, ensure_ascii=False).replace(token, masked)
|
|
print(f"FAIL - {safe_output}")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|