mirror of
https://github.com/artemium428/tvsignals-to-tg.git
synced 2026-09-15 17:16:21 +00:00
114 lines
3.8 KiB
Python
114 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import ValidationError
|
|
|
|
from app.binance import fetch_klines, to_binance_interval, to_binance_symbol
|
|
from app.chart import render_setup_chart
|
|
from app.config import get_settings
|
|
from app.formatter import format_caption
|
|
from app.models import SignalPayload
|
|
from app.telegram import TelegramError, send_message, send_photo
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
app = FastAPI(title="TV Signals → Telegram", version="1.0.0")
|
|
|
|
|
|
def _parse_signal_body(raw: bytes) -> SignalPayload:
|
|
"""Parse JSON from raw body (works for application/json and text/plain)."""
|
|
try:
|
|
text = raw.decode("utf-8").strip()
|
|
except UnicodeDecodeError as exc:
|
|
raise HTTPException(status_code=422, detail="Body must be UTF-8 text") from exc
|
|
|
|
if not text:
|
|
raise HTTPException(status_code=422, detail="Empty body")
|
|
|
|
try:
|
|
data: Any = json.loads(text)
|
|
except json.JSONDecodeError as exc:
|
|
raise HTTPException(status_code=422, detail=f"Invalid JSON: {exc}") from exc
|
|
|
|
try:
|
|
return SignalPayload.model_validate(data)
|
|
except ValidationError as exc:
|
|
raise HTTPException(status_code=422, detail=json.loads(exc.json())) from exc
|
|
|
|
|
|
@app.get("/health")
|
|
async def health() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.post("/webhook")
|
|
async def webhook(request: Request) -> JSONResponse:
|
|
signal = _parse_signal_body(await request.body())
|
|
settings = get_settings()
|
|
caption = format_caption(signal)
|
|
logger.info(
|
|
"Signal received: %s %s seq=%s tf=%s",
|
|
signal.ticker,
|
|
signal.action.value,
|
|
signal.signal_sequence,
|
|
signal.visual_timeframe,
|
|
)
|
|
|
|
photo: bytes | None = None
|
|
chart_error: str | None = None
|
|
|
|
try:
|
|
symbol = to_binance_symbol(signal.ticker)
|
|
interval = to_binance_interval(signal.visual_timeframe)
|
|
df = await fetch_klines(symbol, interval)
|
|
photo = render_setup_chart(
|
|
df,
|
|
ticker=signal.ticker,
|
|
action=signal.action.value, # type: ignore[arg-type]
|
|
entry=signal.entry_price,
|
|
stop_loss=signal.stop_loss_price,
|
|
tp1=signal.take_profit_1_price,
|
|
tp2=signal.take_profit_2_price,
|
|
tp3=signal.take_profit_3_price,
|
|
timeframe=signal.visual_timeframe,
|
|
current_price=(
|
|
signal.current_price if signal.signal_sequence > 1 else None
|
|
),
|
|
signal_time=signal.signal_time,
|
|
)
|
|
except Exception as exc: # noqa: BLE001 — fallback to text-only post
|
|
chart_error = str(exc)
|
|
logger.exception("Chart generation failed, falling back to text-only: %s", exc)
|
|
|
|
try:
|
|
if photo is not None:
|
|
await send_photo(settings, photo=photo, caption=caption)
|
|
return JSONResponse(
|
|
{"ok": True, "delivered": "photo", "chart_error": None},
|
|
status_code=200,
|
|
)
|
|
|
|
await send_message(settings, text=caption)
|
|
return JSONResponse(
|
|
{
|
|
"ok": True,
|
|
"delivered": "text",
|
|
"chart_error": chart_error,
|
|
},
|
|
status_code=200,
|
|
)
|
|
except TelegramError as exc:
|
|
logger.exception("Telegram delivery failed: %s", exc)
|
|
raise HTTPException(status_code=502, detail=str(exc)) from exc
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.exception("Unexpected delivery error: %s", exc)
|
|
raise HTTPException(status_code=500, detail=str(exc)) from exc
|