mirror of
https://github.com/artemium428/tvsignals-to-tg.git
synced 2026-09-15 17:16:21 +00:00
cdbda8fea3
Per-strategy Heryon accounts, reversal captions with previous-trade path on charts, and Telegram replies chained by ticker. Co-authored-by: Cursor <cursoragent@cursor.com>
82 lines
2.3 KiB
Python
82 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
import httpx
|
|
|
|
from app.config import Settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TELEGRAM_API = "https://api.telegram.org"
|
|
|
|
|
|
class TelegramError(Exception):
|
|
pass
|
|
|
|
|
|
def telegram_message_id(payload: dict) -> int | None:
|
|
result = payload.get("result")
|
|
if not isinstance(result, dict):
|
|
return None
|
|
raw = result.get("message_id")
|
|
return int(raw) if raw is not None else None
|
|
|
|
|
|
def _reply_fields(reply_to_message_id: int | None) -> dict[str, str]:
|
|
if reply_to_message_id is None:
|
|
return {}
|
|
return {
|
|
"reply_to_message_id": str(reply_to_message_id),
|
|
"allow_sending_without_reply": "true",
|
|
}
|
|
|
|
|
|
async def send_photo(
|
|
settings: Settings,
|
|
*,
|
|
photo: bytes,
|
|
caption: str,
|
|
message_thread_id: int,
|
|
filename: str = "setup.png",
|
|
reply_to_message_id: int | None = None,
|
|
) -> dict:
|
|
url = f"{TELEGRAM_API}/bot{settings.telegram_bot_token}/sendPhoto"
|
|
data = {
|
|
"chat_id": settings.telegram_chat_id,
|
|
"message_thread_id": str(message_thread_id),
|
|
"caption": caption,
|
|
"parse_mode": "HTML",
|
|
**_reply_fields(reply_to_message_id),
|
|
}
|
|
files = {"photo": (filename, photo, "image/png")}
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
response = await client.post(url, data=data, files=files)
|
|
payload = response.json()
|
|
if response.status_code >= 400 or not payload.get("ok"):
|
|
raise TelegramError(f"sendPhoto failed: {payload}")
|
|
return payload
|
|
|
|
|
|
async def send_message(
|
|
settings: Settings,
|
|
*,
|
|
text: str,
|
|
message_thread_id: int,
|
|
reply_to_message_id: int | None = None,
|
|
) -> dict:
|
|
url = f"{TELEGRAM_API}/bot{settings.telegram_bot_token}/sendMessage"
|
|
data = {
|
|
"chat_id": settings.telegram_chat_id,
|
|
"message_thread_id": str(message_thread_id),
|
|
"text": text,
|
|
"parse_mode": "HTML",
|
|
"disable_web_page_preview": True,
|
|
**_reply_fields(reply_to_message_id),
|
|
}
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
response = await client.post(url, data=data)
|
|
payload = response.json()
|
|
if response.status_code >= 400 or not payload.get("ok"):
|
|
raise TelegramError(f"sendMessage failed: {payload}")
|
|
return payload
|