"""Entrypoint. Serves the Telegram webhook alongside a one-page Gradio UI. Two platform constraints shape the structure here, both learned the hard way: * **Gradio must own the server.** ZeroGPU (the only free hardware for personal accounts) instruments Gradio's ``launch()`` path to discover ``@spaces.GPU`` functions. Serving the same app with ``uvicorn.run()`` bypasses that hook and the Space dies with "No @spaces.GPU function detected during startup" -- even when such functions plainly exist at module level. So on a Space we launch Gradio and graft the FastAPI routes onto ``demo.app``. * **The webhook answers immediately** and does the real work in a background task. Reading a statement can take a minute; holding the webhook open that long makes Telegram time out and redeliver, importing the file twice. The routes live on an ``APIRouter`` so they can be attached either to our own FastAPI app (tests, self-hosting via the Dockerfile) or to Gradio's (on a Space), with the shared state reached through ``request.app.state.rt`` rather than a module global. """ from __future__ import annotations import asyncio import logging import os import threading from contextlib import asynccontextmanager from dataclasses import dataclass from typing import Optional import gradio as gr from fastapi import APIRouter, BackgroundTasks, FastAPI, Header, Request from fastapi.concurrency import run_in_threadpool from fastapi.responses import JSONResponse from finbot import __version__, local_llm from finbot.advice import Advisor from finbot.analytics import today_in from finbot.backend import make_llm from finbot.config import get_settings from finbot.dashboard import build_ui from finbot.features import default_features from finbot.features.base import Attachment, BotContext, Services from finbot.ingest import Ingestor from finbot.limits import RateLimiter from finbot.registry import FeatureRegistry from finbot.store import open_store from finbot.telegram import ( MAX_DOWNLOAD_BYTES, ParsedUpdate, SeenUpdates, TelegramClient, parse_update, redact, ) logging.basicConfig( level=os.environ.get("LOG_LEVEL", "INFO"), format="%(asctime)s %(levelname)-8s %(name)s: %(message)s", ) # httpx logs the full request URL at INFO, and the Telegram Bot API carries the # bot token in the URL *path* (/bot/getMe). On a public Space the # container logs are readable by anyone, so leaving this at INFO publishes the # token. Silence it; our own call sites log what matters, redacted. for noisy in ("httpx", "httpcore"): logging.getLogger(noisy).setLevel(logging.WARNING) log = logging.getLogger("finbot") WEBHOOK_PATH = "/telegram/webhook" # ---------------------------------------------------------------- runtime @dataclass class Runtime: """Everything a request needs, built once per process.""" services: Services telegram: TelegramClient seen: SeenUpdates limiter: RateLimiter #: Set by build_runtime so the Gradio UI, which is constructed at import time, #: can reach the services once they exist. _runtime: Optional["Runtime"] = None def build_runtime() -> Runtime: """Synchronous construction, so it works from any startup path.""" global _runtime settings = get_settings() registry = FeatureRegistry() llm = make_llm(settings) services = Services( settings=settings, store=open_store(settings), llm=llm, ingestor=Ingestor(llm, settings.default_currency), advisor=Advisor(llm), registry=registry, ) registry.register_all(default_features()) for gap in settings.missing_required(): log.warning("Not configured: %s", gap) if settings.open_access: log.warning( "OPEN ACCESS: anyone on Telegram may use this bot. Ledgers are " "isolated per user; uploads are capped at %s per user per day.", settings.uploads_per_user_per_day or "unlimited", ) _runtime = Runtime( services=services, telegram=TelegramClient(settings.telegram_bot_token), seen=SeenUpdates(), limiter=RateLimiter(settings.uploads_per_user_per_day), ) return _runtime def space_webhook_url() -> str: """Public URL of this Space, if we are running on one.""" if host := os.environ.get("SPACE_HOST"): return f"https://{host.rstrip('/')}{WEBHOOK_PATH}" return "" async def bootstrap_telegram(rt: Runtime) -> None: """Identify the bot, publish its command menu and register the webhook. Uses a throwaway client so this can run on a different event loop from the one that will later serve requests -- an httpx client is bound to the loop that created it, and reusing one across loops fails at the worst time. """ settings = rt.services.settings if not settings.telegram_bot_token: log.warning("TELEGRAM_BOT_TOKEN unset; running without a bot connection.") return client = TelegramClient(settings.telegram_bot_token) try: me = await client.get_me() username = me.get("username", "") rt.telegram.username = username # so /cmd@Bot mentions resolve log.info("Connected to Telegram as @%s", username) await client.set_my_commands(rt.services.registry.telegram_commands()) if not settings.telegram_webhook_secret: log.warning("TELEGRAM_WEBHOOK_SECRET unset; webhook not registered.") elif url := space_webhook_url(): await client.set_webhook(url, settings.telegram_webhook_secret) log.info("Webhook registered at %s", url) else: log.info("No SPACE_HOST; register manually via POST /admin/setup") except Exception as exc: # A Telegram hiccup must not stop the app booting: a live container # that reports the problem beats a crash loop. Redacted because an # httpx error carries the token-bearing URL. log.error( "Telegram bootstrap failed: %s", redact(exc, settings.telegram_bot_token) ) finally: await client.close() async def shutdown_runtime(rt: Runtime) -> None: log.info("Shutting down; flushing ledger to the Hub.") await rt.telegram.close() rt.services.llm.close() await run_in_threadpool(rt.services.store.close) def get_runtime(request: Request) -> Runtime: return request.app.state.rt # ----------------------------------------------------------------- routes router = APIRouter() @router.get("/health") async def health() -> JSONResponse: """Liveness only. No ledger or configuration detail -- this may be public.""" return JSONResponse({"status": "ok", "version": __version__}) @router.get("/admin/diagnostics") async def diagnostics( request: Request, x_admin_secret: Optional[str] = Header(default=None), ) -> JSONResponse: """Operational detail, gated on the webhook secret.""" rt = get_runtime(request) settings = rt.services.settings if ( not settings.telegram_webhook_secret or x_admin_secret != settings.telegram_webhook_secret ): return JSONResponse({"ok": False, "error": "forbidden"}, status_code=403) return JSONResponse( { "ok": True, "version": __version__, "llm_enabled": rt.services.llm.enabled, "llm": rt.services.llm.describe(), "sync_enabled": settings.sync_enabled, "sync_status": rt.services.store.sync_status, "rows": rt.services.store.count(), "missing": settings.missing_required(), } ) @router.post(WEBHOOK_PATH) async def telegram_webhook( request: Request, background: BackgroundTasks, x_telegram_bot_api_secret_token: Optional[str] = Header(default=None), ) -> JSONResponse: rt = get_runtime(request) settings = rt.services.settings expected = settings.telegram_webhook_secret if not expected or x_telegram_bot_api_secret_token != expected: log.warning("Rejected webhook call with a bad secret token.") return JSONResponse({"ok": False}, status_code=403) try: update = await request.json() except Exception: return JSONResponse({"ok": False, "error": "bad json"}, status_code=400) parsed = parse_update(update, rt.telegram.username) if parsed is None: return JSONResponse({"ok": True, "skipped": "unsupported update"}) if not rt.seen.add_if_new(parsed.update_id): log.info("Ignoring redelivered update %s", parsed.update_id) return JSONResponse({"ok": True, "skipped": "duplicate"}) # Answer now, work later: Telegram redelivers if we hold this open. background.add_task(handle_update, rt, parsed) return JSONResponse({"ok": True}) @router.post("/admin/setup") async def admin_setup( request: Request, x_admin_secret: Optional[str] = Header(default=None), ) -> JSONResponse: """Register the webhook manually (local dev behind a tunnel, or no SPACE_HOST). Authorised with the same webhook secret, so there is no extra credential. """ rt = get_runtime(request) settings = rt.services.settings if ( not settings.telegram_webhook_secret or x_admin_secret != settings.telegram_webhook_secret ): return JSONResponse({"ok": False, "error": "forbidden"}, status_code=403) body = {} try: body = await request.json() except Exception: pass url = body.get("url") or space_webhook_url() if not url: return JSONResponse( {"ok": False, "error": "no url given and SPACE_HOST is unset"}, status_code=400, ) await rt.telegram.set_webhook(url, settings.telegram_webhook_secret) await rt.telegram.set_my_commands(rt.services.registry.telegram_commands()) info = await rt.telegram.get_webhook_info() return JSONResponse({"ok": True, "webhook": info}) # ------------------------------------------------------------ update handling async def handle_update(rt: Runtime, parsed: ParsedUpdate) -> None: services = rt.services telegram = rt.telegram settings = services.settings if not settings.is_authorised(parsed.user_id, parsed.username): log.warning( "Unauthorised access attempt from %s (@%s)", parsed.user_id, parsed.username or "no-handle", ) who = ( f"@{parsed.username}" if parsed.username else f"id {parsed.user_id} (you have no @username set)" ) await telegram.send_message( parsed.chat_id, "This is a private bot and you're not on its allowlist.\n\n" f"If it's yours, add {who} to " "ALLOWED_TELEGRAM_USERNAMES in the Space secrets " "and restart it.", ) return # Refresh the handle registry on every message, so a rename is picked up # without anything else needing to change. await run_in_threadpool( services.store.record_user, parsed.user_id, parsed.username, parsed.first_name ) try: attachment = None if parsed.has_file: if parsed.file_size > MAX_DOWNLOAD_BYTES: await telegram.send_message( parsed.chat_id, "That file is over Telegram's 20 MB bot limit. " "Try exporting a narrower date range.", ) return # Reading a file occupies the GPU (or spends remote credits), so # cap it per user. Typing an expense is free and never metered. if not rt.limiter.check(parsed.user_id): await telegram.send_message( parsed.chat_id, "You've hit today's limit for reading receipts and " "statements. It resets in 24 hours — you can still log " "expenses by typing them.", ) return await telegram.send_typing(parsed.chat_id) data, file_path = await telegram.download(parsed.file_id) attachment = Attachment( data=data, filename=parsed.file_name or file_path.rsplit("/", 1)[-1], mime_type=parsed.mime_type, kind=parsed.file_kind, caption=parsed.caption, ) ctx = BotContext( user_id=parsed.user_id, chat_id=parsed.chat_id, services=services, today=today_in(settings.timezone), username=parsed.username, text=parsed.text or parsed.caption, command=parsed.command, args=parsed.args, attachment=attachment, ) if attachment is not None or parsed.command in {"advice", "report"}: await telegram.send_typing(parsed.chat_id) reply = await run_in_threadpool(services.registry.dispatch, ctx) if reply is None: if not parsed.command and ctx.text.strip(): await telegram.send_message( parsed.chat_id, "I couldn't find an amount in that. Try " "250 coffee, or /help.", ) return if reply.document is not None: filename, payload = reply.document await telegram.send_document( parsed.chat_id, filename, payload, caption=reply.text ) return await telegram.send_message(parsed.chat_id, reply.text, reply.parse_mode) except Exception: log.exception("Failed handling update %s", parsed.update_id) try: await telegram.send_message( parsed.chat_id, "Something went wrong on my side. The details are in the Space logs.", ) except Exception: pass # ----------------------------------------------------------------- gradio UI def _get_services() -> Optional[Services]: """Late-bound accessor for the dashboard. The UI object is built at import time, before any runtime exists, so it reaches for services through this rather than capturing them. """ return _runtime.services if _runtime is not None else None demo = build_ui(_get_services, local_llm.warmup) # ------------------------------------------------------------------- wiring @asynccontextmanager async def lifespan(fastapi_app: FastAPI): """Startup path for our own FastAPI app (tests, and the Dockerfile).""" rt = build_runtime() fastapi_app.state.rt = rt await bootstrap_telegram(rt) try: yield finally: await shutdown_runtime(rt) app = FastAPI(title="coco-finbot", version=__version__, lifespan=lifespan) app.include_router(router) app = gr.mount_gradio_app(app, demo, path="/") _launch_lock = threading.Lock() def serve_via_gradio() -> None: """Startup path for a Gradio Space, where Gradio must own the server.""" with _launch_lock: rt = build_runtime() # On a Space, let Gradio use its own env-driven defaults (GRADIO_SERVER_PORT # and the Space root path). Forcing a port breaks the frontend asset URLs # behind HF's proxy: the CSS/JS 404, no JavaScript runs, and the page never # leaves its initial server-rendered state. That looks exactly like "the # dashboard shows nothing" -- demo.load never fires, so the sign-in token in # the query string is never read. launch_kwargs: dict = { "prevent_thread_lock": True, "show_api": False, # Spaces sets GRADIO_SSR_MODE=1, which switches the frontend to a # Node-rendered SvelteKit build served from /_app/immutable/*. Launching # ourselves never starts that SSR server, so every CSS/JS asset 404s, # no JavaScript runs, and the page freezes on its server-rendered shell. # Client-side rendering serves the assets straight out of the installed # gradio package and needs no Node. "ssr_mode": False, } if port := os.environ.get("APP_PORT"): # Self-hosting / local dev, where nothing sets GRADIO_SERVER_PORT. launch_kwargs["server_name"] = "0.0.0.0" launch_kwargs["server_port"] = int(port) demo.launch(**launch_kwargs) # returns so we can attach routes # demo.app only exists once launched. demo.app.state.rt = rt demo.app.include_router(router) log.info("Attached %d API routes to the Gradio app", len(router.routes)) # Runs on its own loop; bootstrap_telegram uses a throwaway client so this # does not poison the server loop's connections. asyncio.run(bootstrap_telegram(rt)) demo.block_thread() if __name__ == "__main__": serve_via_gradio()