| from typing import Union, Optional, Dict, Any |
| from pathlib import Path |
|
|
| from fastapi import FastAPI, Request, HTTPException |
| from fastapi.responses import HTMLResponse, FileResponse, Response |
| from chat_simple import get_response |
| import uvicorn |
| import json |
| import logging |
| from logging.handlers import RotatingFileHandler |
|
|
| app = FastAPI() |
|
|
| |
| logger = logging.getLogger("access") |
| logger.setLevel(logging.INFO) |
| if not logger.handlers: |
| logs_dir = Path(__file__).with_name("logs") |
| logs_dir.mkdir(exist_ok=True) |
| handler = RotatingFileHandler(logs_dir / "access.log", maxBytes=1_000_000, backupCount=3, encoding="utf-8") |
| handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s")) |
| logger.addHandler(handler) |
|
|
|
|
| def get_client_ip(request: Request) -> str: |
| xff = request.headers.get("x-forwarded-for") |
| if xff: |
| return xff.split(",")[0].strip() |
| xri = request.headers.get("x-real-ip") |
| if xri: |
| return xri.strip() |
| return request.client.host if request.client else "unknown" |
|
|
|
|
| @app.get("/", response_class=HTMLResponse) |
| def serve_root() -> Response: |
| html_path = Path(__file__).with_name("chat_ui.html") |
| if not html_path.exists(): |
| raise HTTPException(status_code=404, detail="File not found") |
| return FileResponse(html_path) |
|
|
|
|
| @app.post("/chat") |
| async def chat(request: Request): |
| body = await request.json() |
| query = body.get("query") |
| config = body.get("config") |
|
|
| |
| meta = body.get("meta") or {} |
| lat = meta.get("lat") |
| lon = meta.get("lon") |
| user_agent = meta.get("userAgent") |
| platform = meta.get("platform") |
| language = meta.get("language") |
| timezone = meta.get("timezone") |
| screen = meta.get("screen") |
|
|
| client_ip = get_client_ip(request) |
| q_preview = (query or "")[:500] |
| logger.info( |
| "client_ip=%s lat=%s lon=%s ua=%s platform=%s lang=%s tz=%s screen=%s thread_id=%s query=%s", |
| client_ip, |
| lat, |
| lon, |
| (user_agent or "")[:200], |
| platform, |
| language, |
| timezone, |
| screen, |
| ((config or {}).get("configurable") or {}).get("thread_id"), |
| q_preview, |
| ) |
|
|
| response = get_response(query, config) |
| logger.info("client_ip=%s response_len=%s", client_ip, len(str(response) if response is not None else "")) |
| return {"response": response} |
|
|
| if __name__ == "__main__": |
| uvicorn.run("chat_simple_app:app", host="0.0.0.0", port=8000, reload=True) |
