Spaces:
Paused
Paused
| """ | |
| LINE Official Account Webhook Server | |
| ──────────────────────────────────── | |
| รับ event จาก LINE → ตรวจสอบ signature → ส่งต่อไปยัง HuggingFace Sentiment API | |
| """ | |
| import hashlib | |
| import hmac | |
| import base64 | |
| import logging | |
| from contextlib import asynccontextmanager | |
| from fastapi import FastAPI, Request, HTTPException, BackgroundTasks | |
| from fastapi.responses import JSONResponse | |
| from app.config import settings | |
| from app.models import LineWebhookPayload | |
| from app.queue import message_queue | |
| from app.sentiment import analyze_sentiment | |
| from app.database import db, init_db | |
| logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") | |
| logger = logging.getLogger(__name__) | |
| async def lifespan(app: FastAPI): | |
| """Startup / Shutdown events""" | |
| logger.info("🚀 Starting LINE Webhook Server...") | |
| await init_db() | |
| yield | |
| logger.info("🛑 Shutting down...") | |
| app = FastAPI( | |
| title="LINE Webhook Sentiment Server", | |
| version="1.0.0", | |
| lifespan=lifespan, | |
| ) | |
| # ─── Signature Verification ──────────────────────────────────────────────────── | |
| def verify_line_signature(body: bytes, signature: str) -> bool: | |
| """ | |
| ตรวจสอบ X-Line-Signature header จาก LINE | |
| ป้องกัน request ปลอมที่ไม่ได้มาจาก LINE Platform | |
| """ | |
| channel_secret = settings.LINE_CHANNEL_SECRET.encode("utf-8") | |
| hash_value = hmac.new(channel_secret, body, hashlib.sha256).digest() | |
| expected = base64.b64encode(hash_value).decode("utf-8") | |
| return hmac.compare_digest(expected, signature) | |
| # ─── Webhook Endpoint ────────────────────────────────────────────────────────── | |
| async def line_webhook(request: Request, background_tasks: BackgroundTasks): | |
| """ | |
| รับ webhook events จาก LINE Platform | |
| Flow: | |
| 1. ตรวจสอบ LINE Signature | |
| 2. Parse JSON payload | |
| 3. กรองเฉพาะ message events (text) | |
| 4. ส่งไปประมวลผล sentiment ใน background | |
| """ | |
| # 1. ดึง raw body สำหรับตรวจสอบ signature | |
| body = await request.body() | |
| signature = request.headers.get("X-Line-Signature", "") | |
| if not signature: | |
| logger.warning("⚠️ Missing X-Line-Signature header") | |
| raise HTTPException(status_code=400, detail="Missing signature") | |
| if not verify_line_signature(body, signature): | |
| logger.warning("⚠️ Invalid signature — request rejected") | |
| raise HTTPException(status_code=403, detail="Invalid signature") | |
| # 2. Parse payload | |
| try: | |
| payload = LineWebhookPayload.model_validate_json(body) | |
| except Exception as e: | |
| logger.error(f"❌ Failed to parse payload: {e}") | |
| raise HTTPException(status_code=400, detail="Invalid payload") | |
| # 3. Process events ใน background (ไม่ block LINE server) | |
| for event in payload.events: | |
| if event.type == "message" and event.message.type == "text": | |
| background_tasks.add_task(process_message_event, event) | |
| else: | |
| logger.debug(f"Skipped event type: {event.type}") | |
| # LINE ต้องการ HTTP 200 ทันที ไม่งั้น retry | |
| return JSONResponse(content={"status": "ok"}) | |
| # ─── Background Processing ───────────────────────────────────────────────────── | |
| async def process_message_event(event): | |
| """ | |
| ประมวลผลแต่ละ message event: | |
| 1. บันทึก raw conversation ลง DB | |
| 2. ส่งไป HuggingFace วิเคราะห์ sentiment | |
| 3. อัปเดต result กลับใน DB | |
| 4. Push ผ่าน queue ไปยัง Dashboard | |
| """ | |
| user_id = event.source.userId | |
| msg_text = event.message.text | |
| conv_id = event.replyToken or f"{user_id}_{event.timestamp}" | |
| logger.info(f"📨 New message | user={user_id} | text={msg_text[:50]}...") | |
| # Step 1 — บันทึก conversation | |
| conversation = await db.save_conversation( | |
| conv_id=conv_id, | |
| user_id=user_id, | |
| message=msg_text, | |
| timestamp=event.timestamp, | |
| source_type=event.source.type, | |
| ) | |
| # Step 2 — วิเคราะห์ sentiment ผ่าน HuggingFace | |
| sentiment_result = await analyze_sentiment(msg_text) | |
| logger.info( | |
| f"🧠 Sentiment | label={sentiment_result.label} " | |
| f"score={sentiment_result.score:.3f}" | |
| ) | |
| # Step 3 — อัปเดต sentiment ใน DB | |
| await db.update_sentiment( | |
| conv_id=conv_id, | |
| label=sentiment_result.label, | |
| score=sentiment_result.score, | |
| raw_scores=sentiment_result.raw_scores, | |
| ) | |
| # Step 4 — Push ไป queue → Dashboard รับแบบ real-time | |
| await message_queue.push({ | |
| "conv_id": conv_id, | |
| "user_id": user_id, | |
| "message": msg_text, | |
| "sentiment": sentiment_result.label, | |
| "score": sentiment_result.score, | |
| "timestamp": event.timestamp, | |
| }) | |
| # Alert ถ้า negative สูงเกิน threshold | |
| if ( | |
| sentiment_result.label == "negative" | |
| and sentiment_result.score >= settings.NEGATIVE_ALERT_THRESHOLD | |
| ): | |
| logger.warning( | |
| f"🚨 HIGH NEGATIVE ALERT | user={user_id} | score={sentiment_result.score:.3f}" | |
| ) | |
| await message_queue.push_alert({ | |
| "type": "high_negative", | |
| "conv_id": conv_id, | |
| "user_id": user_id, | |
| "score": sentiment_result.score, | |
| "message": msg_text, | |
| }) | |
| # ─── Health & Dashboard Endpoints ───────────────────────────────────────────── | |
| async def health_check(): | |
| """Health check สำหรับ load balancer / monitoring""" | |
| return {"status": "healthy", "version": app.version} | |
| async def get_conversations(limit: int = 50, offset: int = 0): | |
| """ | |
| ดึง conversation list พร้อม sentiment สำหรับ Lovable Dashboard | |
| """ | |
| data = await db.get_conversations(limit=limit, offset=offset) | |
| return {"conversations": data, "total": len(data)} | |
| async def get_sentiment_summary(): | |
| """ | |
| สรุป sentiment ภาพรวมสำหรับ stat cards บน Dashboard | |
| """ | |
| summary = await db.get_sentiment_summary() | |
| return summary | |