"""backend/api/webhook.py — Webhook inbound + Public REST API (S354).""" import os, asyncio from typing import Optional from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel, field_validator from .state import _get_mem_manager, _get_executor, _get_planner import logging _logger = logging.getLogger("api.webhook") try: from .telegram_notify import ( notify_task_start as _tg_start, notify_task_done as _tg_done, notify_task_error as _tg_error, ) except Exception: async def _tg_start(*_a, **_kw): pass # type: ignore[misc] async def _tg_done(*_a, **_kw): pass # type: ignore[misc] async def _tg_error(*_a, **_kw): pass # type: ignore[misc] router = APIRouter() # ── Telegram callback_query handler ────────────────────────────────────────── # Flusso: bottone "🚫 Rifiuta" su iPhone → Telegram → POST /api/telegram/callback # → scrivi su Supabase telegram_reject_proposals → Realtime → SPA rejectProposal() @router.post('/api/telegram/callback') async def telegram_callback(request: Request): """ Riceve callback_query da Telegram quando l'utente clicca un bottone inline. Autenticazione: X-Telegram-Bot-Api-Secret-Token header (impostato al set-webhook). Se callback_data = "reject_confirm:{key}" → upsert su Supabase + answerCallbackQuery. Risponde 200 immediato — Telegram richiede risposta entro 3s. """ import time as _t, json as _js _tg_secret = os.getenv('TELEGRAM_WEBHOOK_SECRET', '').strip() _req_secret = request.headers.get('x-telegram-bot-api-secret-token', '') if _tg_secret and _req_secret != _tg_secret: # Risponde 200 (Telegram non ritenta i 401) ma non processa return {'ok': True, 'ignored': 'bad_secret'} try: body = await request.json() except Exception: return {'ok': True, 'ignored': 'bad_json'} cbq = body.get('callback_query') or {} if not cbq: return {'ok': True, 'ignored': 'not_callback_query'} cbq_id = cbq.get('id', '') cb_data = cbq.get('data', '') cb_from = cbq.get('from', {}).get('first_name', '?') # Risponde a Telegram immediatamente (finestra 3s) _tg_token = os.getenv('TELEGRAM_BOT_TOKEN', '') if _tg_token and cbq_id: try: import httpx as _hx _tok_snap = _tg_token # cattura nel closure _id_snap = cbq_id async def _answer_cbq() -> None: async with _hx.AsyncClient(timeout=4.0) as _c: await _c.post( f'https://api.telegram.org/bot{_tok_snap}/answerCallbackQuery', json={'callback_query_id': _id_snap, 'text': '🚫 Rifiuto registrato'}, ) asyncio.create_task(_answer_cbq()).add_done_callback( lambda t: _logger.debug("[webhook] _answer_cbq exc: %s", t.exception()) if not t.cancelled() and t.exception() is not None else None ) except Exception as _exc: _logger.debug("[webhook] silenced %s", type(_exc).__name__) # noqa: BLE001 # Processa solo rifiuti espliciti (callback_data = "reject_confirm:{key}") if not cb_data.startswith('reject_confirm:'): return {'ok': True, 'action': 'noop', 'data': cb_data} confirm_key = cb_data[len('reject_confirm:'):] if not confirm_key: return {'ok': True, 'ignored': 'empty_key'} # Scrivi su Supabase (tabella telegram_reject_proposals) — Realtime notifica la SPA _sb_written = False try: from api.state import _sb as _supa if _supa: _supa.table('telegram_reject_proposals').upsert( { 'confirm_key': confirm_key, 'reason': f'rifiuto esplicito di {cb_from} via Telegram', 'ts': _t.time() * 1000, }, on_conflict='confirm_key', ).execute() _sb_written = True except Exception as _se: _logger.warning('TG-CALLBACK warn supabase: %s', _se) _logger.info('TG-CALLBACK reject_confirm:%r from=%r sb=%s', confirm_key, cb_from, _sb_written) return {'ok': True, 'action': 'reject_registered', 'key': confirm_key, 'sb': _sb_written} @router.post('/api/telegram/set-webhook') async def telegram_set_webhook(request: Request): """ Registra il webhook Telegram su Railway. Richiede: X-Admin-Token header + TELEGRAM_BOT_TOKEN env var. Chiama: POST https://api.telegram.org/bot{TOKEN}/setWebhook Il secret token è TELEGRAM_WEBHOOK_SECRET (generato casualmente se assente). """ from api.auth_guard import require_role, AuthRole, _get_token admin_tok = _get_token('ADMIN_TOKEN') req_admin = request.headers.get('x-admin-token', '') if admin_tok and req_admin != admin_tok: raise HTTPException(status_code=403, detail='Admin token required') _tg_token = os.getenv('TELEGRAM_BOT_TOKEN', '') if not _tg_token: raise HTTPException(status_code=503, detail='TELEGRAM_BOT_TOKEN non configurato') _railway_url = os.getenv('RAILWAY_PUBLIC_DOMAIN', '') or os.getenv('RAILWAY_URL', '') if not _railway_url: raise HTTPException(status_code=503, detail='RAILWAY_PUBLIC_DOMAIN non configurato') _wh_url = f'https://{_railway_url.lstrip("https://").rstrip("/")}/api/telegram/callback' _secret = os.getenv('TELEGRAM_WEBHOOK_SECRET', '') if not _secret: import secrets as _sec _secret = _sec.token_hex(24) # Non possiamo settare env var runtime, ma logghiamo per configurazione manuale _logger.critical('TG-WEBHOOK: genera TELEGRAM_WEBHOOK_SECRET=%r e aggiungilo a Railway env!', _secret) try: import httpx as _hx r = await _hx.AsyncClient().post( f'https://api.telegram.org/bot{_tg_token}/setWebhook', json={ 'url': _wh_url, 'secret_token': _secret, 'allowed_updates': ['callback_query'], }, timeout=10.0, ) data = r.json() except Exception as exc: raise HTTPException(status_code=502, detail=f'Telegram API error: {exc}') return {'ok': data.get('ok'), 'description': data.get('description'), 'webhook_url': _wh_url} class WebhookPayload(BaseModel): goal: str context: list[dict] = [] max_steps: int = 10 @field_validator('goal', mode='before') @classmethod def validate_goal(cls, v: object) -> str: if not isinstance(v, str) or not v.strip(): raise ValueError('goal must be a non-empty string') return v.strip() class PublicChatPayload(BaseModel): message: str conversation_id: Optional[str] = None max_steps: int = 8 @field_validator('message', mode='before') @classmethod def validate_message(cls, v: object) -> str: if not isinstance(v, str) or not v.strip(): raise ValueError('message must be a non-empty string') s = v.strip() if len(s) > 4000: raise ValueError('message exceeds 4000 chars') return s @field_validator('max_steps', mode='before') @classmethod def clamp_steps(cls, v: object) -> int: if not isinstance(v, int): return 8 return max(1, min(int(v), 20)) @router.post('/api/webhook/{webhook_token}') async def inbound_webhook(webhook_token: str, body: WebhookPayload): """ S289 — Webhook inbound per triggering proattivo dell'agente. Auth: WEBHOOK_TOKEN env var deve matchare il path token. """ _expected = os.getenv('WEBHOOK_TOKEN', '').strip() if not _expected: raise HTTPException(status_code=503, detail='Webhook non configurato: variabile WEBHOOK_TOKEN mancante nello Space') if webhook_token != _expected: raise HTTPException(status_code=401, detail='Unauthorized: token non valido') try: from agents.unified_loop import UnifiedAgentLoop from models.ai_client import AIClient client = AIClient() try: from agents.critic import Critic from agents.response_verifier import ResponseVerifier _critic = Critic(llm_client=client) _verifier = ResponseVerifier() except Exception: _critic = None _verifier = None loop = UnifiedAgentLoop( llm_client=client, critic=_critic, verifier=_verifier, memory=_get_mem_manager(), executor=_get_executor(), planner=_get_planner(), ) context_str = '\n'.join(m.get('content', '') for m in body.context) if body.context else '' _wh_task_id = f"wh_{webhook_token[:6]}_{int(__import__('time').time()*1000)%999983:x}" await _tg_start(_wh_task_id, body.goal) try: result = await asyncio.wait_for( loop.run(goal=body.goal, context=context_str, max_steps=body.max_steps, on_step=lambda _s: None), timeout=float(os.getenv('AGENT_STREAM_TIMEOUT', '120')), ) except asyncio.TimeoutError: asyncio.ensure_future(_tg_error(_wh_task_id, body.goal, "Timeout: task terminato dopo 120s")) raise HTTPException(status_code=504, detail='Agent timeout — prova un goal più semplice o aumenta AGENT_STREAM_TIMEOUT') except Exception as exc: asyncio.ensure_future(_tg_error(_wh_task_id, body.goal, str(exc)[:300])) raise HTTPException(status_code=500, detail=f'Errore agente: {exc}') _out = result.get('output', '') asyncio.ensure_future(_tg_done(_wh_task_id, body.goal, _out[:500])) return { 'ok': result.get('success', False), 'output': _out, 'engine': result.get('engine', 'unknown'), 'goal': body.goal, 'steps': len(result.get('steps', [])), } except (HTTPException, asyncio.TimeoutError): raise except Exception as exc: raise HTTPException(status_code=500, detail=f'Errore agente: {exc}') @router.post('/api/public/chat') async def public_chat(payload: PublicChatPayload, request: Request): """ S292 — API REST pubblica autenticata per integrazioni esterne. Auth: Authorization: Bearer """ _expected = os.getenv('PUBLIC_API_TOKEN', '').strip() if not _expected: raise HTTPException( status_code=503, detail='Public API non configurata: variabile PUBLIC_API_TOKEN mancante nello Space.', ) auth_header = request.headers.get('authorization', '') token = auth_header.removeprefix('Bearer ').strip() if not token or token != _expected: raise HTTPException(status_code=401, detail='Unauthorized: token non valido.') try: from agents.unified_loop import UnifiedAgentLoop from models.ai_client import AIClient client = AIClient() try: from agents.critic import Critic from agents.response_verifier import ResponseVerifier _critic = Critic(llm_client=client) _verifier = ResponseVerifier() except Exception: _critic = None _verifier = None loop = UnifiedAgentLoop( llm_client=client, critic=_critic, verifier=_verifier, memory=_get_mem_manager(), executor=_get_executor(), planner=_get_planner(), ) _pc_task_id = f"pc_{int(__import__('time').time()*1000)%999983:x}" await _tg_start(_pc_task_id, payload.message) try: result = await asyncio.wait_for( loop.run(goal=payload.message, context='', max_steps=payload.max_steps, on_step=lambda _s: None), timeout=float(os.getenv('AGENT_STREAM_TIMEOUT', '120')), ) except asyncio.TimeoutError: asyncio.ensure_future(_tg_error(_pc_task_id, payload.message, "Timeout dopo 120s")) raise HTTPException(status_code=504, detail='Timeout — goal troppo complesso o server occupato.') except Exception as exc: asyncio.ensure_future(_tg_error(_pc_task_id, payload.message, str(exc)[:300])) raise HTTPException(status_code=500, detail=f'Errore agente: {exc}') _out_pc = result.get('output', '') asyncio.ensure_future(_tg_done(_pc_task_id, payload.message, _out_pc[:500])) return { 'ok': result.get('success', False), 'response': _out_pc, 'engine': result.get('engine', 'unknown'), 'conversation_id': payload.conversation_id, 'steps': len(result.get('steps', [])), } except (HTTPException, asyncio.TimeoutError): raise except Exception as exc: raise HTTPException(status_code=500, detail=f'Errore agente: {exc}')