tvsignals-to-tg/app/main.py

from __future__ import annotations

import asyncio
import json
import logging
import secrets
from typing import Any

from fastapi import BackgroundTasks, 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 Settings, 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",
    docs_url=None,
    redoc_url=None,
    openapi_url=None,
)


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


async def _deliver_signal(
    settings: Settings,
    signal: SignalPayload,
    message_thread_id: int,
) -> None:
    """Fetch chart + post to Telegram after the webhook HTTP response is sent."""
    caption = format_caption(signal)
    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 = await asyncio.to_thread(
            render_setup_chart,
            df,
            ticker=signal.ticker,
            action=signal.action.value,
            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,
                message_thread_id=message_thread_id,
            )
            logger.info(
                "Delivered photo: %s %s seq=%s thread=%s",
                signal.ticker,
                signal.action.value,
                signal.signal_sequence,
                message_thread_id,
            )
            return

        await send_message(
            settings,
            text=caption,
            message_thread_id=message_thread_id,
        )
        logger.info(
            "Delivered text: %s %s seq=%s thread=%s chart_error=%s",
            signal.ticker,
            signal.action.value,
            signal.signal_sequence,
            message_thread_id,
            chart_error,
        )
    except TelegramError as exc:
        logger.exception("Telegram delivery failed: %s", exc)
    except Exception as exc:  # noqa: BLE001
        logger.exception("Unexpected delivery error: %s", exc)


@app.api_route("/health", methods=["GET", "HEAD"])
async def health() -> dict[str, str]:
    return {"status": "ok"}


async def _handle_webhook(
    token: str,
    request: Request,
    thread_id: int | None,
    background_tasks: BackgroundTasks,
) -> JSONResponse:
    settings = get_settings()
    if not secrets.compare_digest(token, settings.webhook_secret):
        raise HTTPException(status_code=404, detail="Not Found")

    message_thread_id = (
        thread_id if thread_id is not None else settings.telegram_message_thread_id
    )

    signal = _parse_signal_body(await request.body())
    logger.info(
        "Signal accepted: %s %s seq=%s tf=%s thread=%s",
        signal.ticker,
        signal.action.value,
        signal.signal_sequence,
        signal.visual_timeframe,
        message_thread_id,
    )

    background_tasks.add_task(
        _deliver_signal,
        settings,
        signal,
        message_thread_id,
    )
    return JSONResponse({"ok": True, "accepted": True}, status_code=200)


@app.post("/h/{token}")
async def webhook(
    token: str,
    request: Request,
    background_tasks: BackgroundTasks,
) -> JSONResponse:
    return await _handle_webhook(
        token, request, thread_id=None, background_tasks=background_tasks
    )


@app.post("/h/{token}/{thread_id}")
async def webhook_with_thread(
    token: str,
    thread_id: int,
    request: Request,
    background_tasks: BackgroundTasks,
) -> JSONResponse:
    if thread_id < 1:
        raise HTTPException(status_code=422, detail="thread_id must be >= 1")
    return await _handle_webhook(
        token, request, thread_id=thread_id, background_tasks=background_tasks
    )