""" MetaRec FastAPI Application 提供HTTP API接口，调用核心服务层 """ from dotenv import load_dotenv, find_dotenv dotenv_path = find_dotenv() load_dotenv(dotenv_path) from pathlib import Path from contextlib import asynccontextmanager from fastapi import Depends, FastAPI, HTTPException, Query, Request, Response from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse, StreamingResponse from pydantic import BaseModel, ConfigDict, Field from typing import List, Optional, Dict, Any, Awaitable, Callable from datetime import datetime from client import ( LLM_API_KEY, LLM_BASE_URL, create_async_client, create_sync_azure_client, create_sync_client, describe_openai_compatible_config, get_openai_compatible_transport_config, ) import os import asyncio import json import time import logging import sys import socket from urllib.parse import urlparse import httpx # 配置日志系统 - 确保实时输出到控制台 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S', handlers=[ logging.StreamHandler(sys.stdout) # 输出到标准输出（控制台） ], force=True # 强制重新配置，覆盖之前的配置 ) # 设置 uvicorn 的日志级别 logging.getLogger("uvicorn").setLevel(logging.INFO) logging.getLogger("uvicorn.access").setLevel(logging.INFO) logger = logging.getLogger("metarec.api") # 导入核心服务 from service import MetaRecService from internal.debug.router import create_debug_router from internal.admin.router import create_admin_router from internal.feedback.router import create_feedback_router from business_models import AuthSessionPayload, UserRole from business_repositories import auth_repository, conversation_repository, profile_repository from profile_model import apply_profile_memory_from_preferences, assemble_domains, normalize_profile from preference_specs import build_domain_form def _admin_allowlist_emails() -> List[str]: raw = os.getenv("METAREC_ADMIN_EMAILS", "") return [part.strip() for part in raw.split(",") if part.strip()] async def _promote_admins_from_allowlist() -> None: """Promote registered users in METAREC_ADMIN_EMAILS to ADMIN on startup. Idempotent and non-fatal — a user must already exist (register first) to be promoted; unknown emails are simply skipped.""" if not os.getenv("DATABASE_URL"): return emails = _admin_allowlist_emails() if not emails: return try: promoted = await auth_repository.promote_admins(emails) logging.getLogger(__name__).info( "[startup] Admin allowlist processed: %d/%d promoted", promoted, len(emails) ) except Exception as exc: # pragma: no cover - startup best-effort logging.getLogger(__name__).warning("[startup] Admin promotion failed: %s", exc) async def _seed_admin_user_from_env() -> None: """Create-and-promote a full admin account on startup from SEED_ADMIN_EMAIL / SEED_ADMIN_PASSWORD. Unlike METAREC_ADMIN_EMAILS (which only *promotes* an already-registered user), this creates the account if it is missing — the only shell-free way to bootstrap an admin on hosts without exec access (e.g. Hugging Face Spaces). Idempotent (re-asserts the role on every boot, password left untouched if the user exists) and non-fatal.""" if not os.getenv("DATABASE_URL"): return email = (os.getenv("SEED_ADMIN_EMAIL") or "").strip() password = os.getenv("SEED_ADMIN_PASSWORD") or "" if not email or not password: return if len(password) < 8: logging.getLogger(__name__).warning( "[startup] SEED_ADMIN_PASSWORD must be >= 8 chars; skipping admin seed" ) return try: try: await auth_repository.register(email=email, password=password, display_name="Admin") created = True except ValueError as exc: if "already registered" not in str(exc): raise created = False await auth_repository.set_role_by_email(email, UserRole.ADMIN) logging.getLogger(__name__).info( "[startup] Admin seed %s: %s", "created" if created else "existed (role ensured)", email ) except Exception as exc: # pragma: no cover - startup best-effort logging.getLogger(__name__).warning("[startup] Admin seed failed: %s", exc) @asynccontextmanager async def _lifespan(_app: "FastAPI"): await _seed_admin_user_from_env() await _promote_admins_from_allowlist() yield app = FastAPI(title="MetaRec API", version="1.0.0", lifespan=_lifespan) # CORS configuration app.add_middleware( CORSMiddleware, allow_origins=[ "http://localhost:5173", "http://127.0.0.1:5173", ], allow_origin_regex=r"https://.*\.hf\.space", allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # create OpenAI clients async_client = create_async_client() llm_model = os.getenv('LLM_MODEL') logging.getLogger("metarec.llm").info( "OpenAI-compatible LLM config: %s", describe_openai_compatible_config(llm_model), ) try: sync_client = create_sync_azure_client() summary_model = os.getenv('AZURE_AGENT_SUMMARY_MODEL', 'o4-mini') planning_model = os.getenv('AZURE_AGENT_PLANNING_MODEL', 'gpt-4.1') except Exception as e: print('[Warning] Unable to create AzureOpenAI client, falling back to OpenAI client') sync_client = create_sync_client() summary_model = os.getenv('AGENT_SUMMARY_MODEL') planning_model = os.getenv('AGENT_PLANNING_MODEL') # ==================== 创建服务实例 ==================== # 这是全局服务实例，可以被所有路由使用 metarec_service = MetaRecService(async_client, sync_client, summary_model, planning_model, llm_model) # ==================== Auth helpers ==================== AUTH_COOKIE_NAME = os.getenv("METAREC_SESSION_COOKIE_NAME", auth_repository.cookie_name) AUTH_COOKIE_SECURE = os.getenv("METAREC_SESSION_COOKIE_SECURE", "false").lower() in {"1", "true", "yes", "on"} AUTH_SESSION_MAX_AGE_SECONDS = int(os.getenv("METAREC_SESSION_MAX_AGE_SECONDS", str(30 * 24 * 60 * 60))) def _set_session_cookie(response: Response, token: str) -> None: response.set_cookie( key=AUTH_COOKIE_NAME, value=token, max_age=AUTH_SESSION_MAX_AGE_SECONDS, httponly=True, secure=AUTH_COOKIE_SECURE, samesite="lax", path="/", ) def _clear_session_cookie(response: Response) -> None: response.delete_cookie(key=AUTH_COOKIE_NAME, path="/") async def get_optional_auth_session(request: Request) -> Optional[AuthSessionPayload]: return await auth_repository.session_from_token(request.cookies.get(AUTH_COOKIE_NAME)) async def require_auth_session(request: Request) -> AuthSessionPayload: session = await get_optional_auth_session(request) if session is None: raise HTTPException(status_code=401, detail="Authentication required") return session async def resolve_request_user_id(request: Request, provided_user_id: Optional[str] = None) -> str: session = await require_auth_session(request) if provided_user_id and provided_user_id != "default" and provided_user_id != session.user.id: raise HTTPException(status_code=403, detail="user_id does not match authenticated session") return session.user.id async def require_path_user(request: Request, user_id: str) -> AuthSessionPayload: session = await require_auth_session(request) if session.user.id != user_id: raise HTTPException(status_code=403, detail="user_id does not match authenticated session") return session async def require_admin_session(request: Request) -> AuthSessionPayload: """Require an authenticated user with the ADMIN role. Reused by the debug arena to gate privileged tooling on a real user's role.""" session = await require_auth_session(request) if session.user.role != UserRole.ADMIN: raise HTTPException(status_code=403, detail="Admin role required") return session # 挂载内部 debug 路由（可用性由 DEBUG_UI_ENABLED 控制；访问需要 ADMIN 角色） app.include_router(create_debug_router(lambda: metarec_service, require_admin_session)) # 挂载管理后台路由（仅按 ADMIN 角色控制，不受 DEBUG_UI_ENABLED 影响） app.include_router(create_admin_router(require_admin_session)) # 挂载用户反馈路由（需登录会话；游客在端点内被拒绝） app.include_router(create_feedback_router(require_auth_session)) def _merge_meaningful_preferences(existing: Dict[str, Any], incoming: Dict[str, Any]) -> Dict[str, Any]: merged = dict(existing or {}) default_preferences = metarec_service.get_default_preferences() for key, value in (incoming or {}).items(): if value is None: continue existing_value = merged.get(key) default_value = default_preferences.get(key) if isinstance(value, list): meaningful = [item for item in value if item not in (None, "", "any")] if meaningful or key not in merged: merged[key] = value continue if isinstance(value, dict): meaningful_dict = {k: v for k, v in value.items() if v is not None and v != ""} if not meaningful_dict: continue if value == default_value and existing_value not in (None, {}, default_value): continue merged[key] = {**(existing_value if isinstance(existing_value, dict) else {}), **meaningful_dict} continue if value != "any" or key not in merged: merged[key] = value return merged _PERSISTABLE_RESTAURANT_PREFERENCE_KEYS = { "restaurant_types", "flavor_profiles", "dining_purpose", "budget_range", "location", } def _restaurant_preference_subset(preferences: Dict[str, Any]) -> Dict[str, Any]: return { key: value for key, value in (preferences or {}).items() if key in _PERSISTABLE_RESTAURANT_PREFERENCE_KEYS } async def _persist_profile_preferences_from_result( user_id: str, preferences: Optional[Dict[str, Any]], *, update_persona_memory: bool = False, ) -> None: if not isinstance(preferences, dict) or not preferences: return profile = await profile_repository.get_user_profile(user_id) domains = assemble_domains(profile) domain = str(preferences.get("domain") or "restaurant").lower() request_scoped_keys = {"food_intent", "domain", "query"} if domain in {"restaurant", "multi_domain", "unknown"}: restaurant_preferences = _restaurant_preference_subset(preferences) if not restaurant_preferences: return domains["restaurant"] = _merge_meaningful_preferences( domains.get("restaurant", {}), restaurant_preferences, ) else: domain_preferences = { key: value for key, value in preferences.items() if key not in request_scoped_keys and value not in (None, "", [], {}) } if not domain_preferences: return domains[domain] = _merge_meaningful_preferences(domains.get(domain, {}), domain_preferences) restaurant_slice = domains.pop("restaurant", {}) metadata = profile.setdefault("metadata", {}) metadata["preferences"] = {} metadata["domains"] = domains profile["dining_habits"] = restaurant_slice if update_persona_memory: profile = apply_profile_memory_from_preferences( profile, preferences, source="confirmed_recommendation", evidence=str(preferences.get("query") or "")[:240], ) await profile_repository.save_user_profile(user_id, profile) # ==================== 静态文件服务配置 ==================== FRONTEND_DIST = (Path(__file__).parent.parent / 'frontend-dist').resolve() # 启动时检查静态文件目录 def check_frontend_dist(): """检查前端静态文件目录是否存在""" if os.path.exists(FRONTEND_DIST): print(f"[INFO] Frontend dist directory found: {FRONTEND_DIST}") index_path = os.path.join(FRONTEND_DIST, "index.html") if os.path.exists(index_path): print(f"[INFO] Frontend index.html found: {index_path}") else: print(f"[WARN] index.html not found in {FRONTEND_DIST}") # 列出目录内容 try: files = os.listdir(FRONTEND_DIST) print(f"[INFO] Frontend dist contents: {files[:10]}...") # 只显示前10个 except Exception as e: print(f"[WARN] Error listing frontend dist: {e}") else: print(f"[WARN] Frontend dist directory not found: {FRONTEND_DIST}") # 在应用启动时检查 check_frontend_dist() # ==================== API数据模型 ==================== # 这些模型用于API请求和响应，与服务层的模型分离 class StrictBaseModel(BaseModel): model_config = ConfigDict(extra="forbid") class AuthUserAPI(StrictBaseModel): id: str kind: str role: str email: Optional[str] = None display_name: Optional[str] = None status: str class AuthSessionAPI(StrictBaseModel): id: str user_id: str anonymous_device_id: Optional[str] = None status: str expires_at: str class AuthResponseAPI(StrictBaseModel): user: AuthUserAPI session: AuthSessionAPI class GuestLoginRequestAPI(StrictBaseModel): device_id: str class RegisterRequestAPI(StrictBaseModel): email: str password: str display_name: Optional[str] = None class LoginRequestAPI(StrictBaseModel): email: str password: str class ProcessMessageAPI(StrictBaseModel): role: str content: str class ProcessRequestAPI(StrictBaseModel): query: str user_id: str = "default" conversation_history: Optional[List[ProcessMessageAPI]] = None conversation_id: Optional[str] = None use_online_agent: bool = False source_message_id: Optional[str] = None parent_message_id: Optional[str] = None replay_from_message_id: Optional[str] = None branch_id: Optional[str] = None time_travel_mode: Optional[str] = None domain_lock: Optional[str] = None hitl_state: Optional[Dict[str, Any]] = Field( default=None, json_schema_extra={"additionalProperties": True}, ) class HealthResponseAPI(StrictBaseModel): status: str timestamp: str class ApiInfoResponseAPI(StrictBaseModel): message: str version: str class FrontendConfigResponseAPI(StrictBaseModel): googleMapsApiKey: str class RestaurantAPI(StrictBaseModel): id: str name: str address: Optional[str] = None area: Optional[str] = None cuisine: Optional[str] = None type: Optional[str] = None location: Optional[str] = None rating: Optional[float] = None reviews_count: Optional[int] = None price: Optional[str] = None price_per_person_sgd: Optional[str] = None distance_or_walk_time: Optional[str] = None open_hours_note: Optional[str] = None highlights: Optional[List[str]] = None flavor_match: Optional[List[str]] = None purpose_match: Optional[List[str]] = None why: Optional[str] = None reason: Optional[str] = None reference: Optional[str] = None sources: Optional[Dict[str, str]] = None phone: Optional[str] = None gps_coordinates: Optional[Dict[str, float]] = None class RecommendationItemAPI(StrictBaseModel): id: str domain: str title: str subtitle: Optional[str] = None description: Optional[str] = None image_url: Optional[str] = None url: Optional[str] = None rating: Optional[float] = None reviews_count: Optional[int] = None source: Optional[str] = None tags: List[str] = Field(default_factory=list) why: Optional[str] = None # NOTE: the internal RecommendationItem carries a ``raw`` upstream payload for # persistence/debug, but it is deliberately *not* exposed here — see # ``_client_safe_item`` and ``_persist_recommendation_result``. class ThinkingStepAPI(StrictBaseModel): step: str description: str status: str details: Optional[str] = None class ConfirmationQuickActionAPI(StrictBaseModel): id: str label: str value: str preference_patch: Dict[str, Any] = Field( default_factory=dict, json_schema_extra={"additionalProperties": True}, ) message: Optional[str] = None class ConfirmationRequestAPI(StrictBaseModel): message: str preferences: Dict[str, Any] = Field( default_factory=dict, json_schema_extra={"additionalProperties": True}, ) needs_confirmation: bool = True # Server-generated, request-time preference form for the resolved domain. preference_form: Optional[Dict[str, Any]] = Field( default=None, json_schema_extra={"additionalProperties": True}, ) quick_actions: Optional[List[ConfirmationQuickActionAPI]] = None class RecommendationResponseAPI(StrictBaseModel): restaurants: List[RestaurantAPI] items: List[RecommendationItemAPI] = Field(default_factory=list) thinking_steps: Optional[List[ThinkingStepAPI]] = None confirmation_request: Optional[ConfirmationRequestAPI] = None llm_reply: Optional[str] = None # GPT-4 的回复（用于普通对话） intent: Optional[str] = None # 意图类型 task_id: Optional[str] = None result_id: Optional[str] = None domain: Optional[str] = None time_travel: Optional[Dict[str, Any]] = Field( default=None, json_schema_extra={"additionalProperties": True}, ) hitl_state: Optional[Dict[str, Any]] = Field( default=None, json_schema_extra={"additionalProperties": True}, ) metadata: Optional[Dict[str, Any]] = Field( default=None, json_schema_extra={"additionalProperties": True}, ) preferences: Optional[Dict[str, Any]] = Field( default=None, json_schema_extra={"additionalProperties": True}, ) # 提取的偏好设置（当 intent 为 "query" 时） class TaskStatusAPI(StrictBaseModel): task_id: str status: str # "processing", "completed", "error", "cancelled" progress: int # 0-100 message: str result: Optional[RecommendationResponseAPI] = None error: Optional[str] = None metadata: Optional[Dict[str, Any]] = Field( default=None, json_schema_extra={"additionalProperties": True}, ) class BudgetRangeInputAPI(StrictBaseModel): min: Optional[int] = 20 max: Optional[int] = 60 currency: str = "SGD" per: str = "person" class UpdatePreferencesRequestAPI(StrictBaseModel): user_id: str = "default" restaurantTypes: List[str] = ["any"] flavorProfiles: List[str] = ["any"] diningPurpose: str = "any" budgetRange: BudgetRangeInputAPI = BudgetRangeInputAPI() location: str = "any" class PreferencesResponseAPI(StrictBaseModel): preferences: Dict[str, Any] = Field( default_factory=dict, json_schema_extra={"additionalProperties": True}, ) class UpdatePreferencesResponseAPI(StrictBaseModel): message: str preferences: Dict[str, Any] = Field( default_factory=dict, json_schema_extra={"additionalProperties": True}, ) class UserPreferencesResponseAPI(StrictBaseModel): user_id: str preferences: Dict[str, Any] = Field( default_factory=dict, json_schema_extra={"additionalProperties": True}, ) class GenericSuccessResponseAPI(StrictBaseModel): success: bool message: str def _mask_debug_value(value: Optional[str]) -> Optional[str]: if value is None: return None if len(value) <= 10: return "***" return f"{value[:4]}...{value[-4:]}" def _debug_exception(exc: Exception) -> Dict[str, Any]: cause = getattr(exc, "__cause__", None) context = getattr(exc, "__context__", None) data: Dict[str, Any] = { "type": type(exc).__name__, "repr": repr(exc), } status_code = getattr(exc, "status_code", None) if status_code is not None: data["status_code"] = status_code body = getattr(exc, "body", None) if body is not None: data["body"] = repr(body)[:500] if cause is not None: data["cause"] = { "type": type(cause).__name__, "repr": repr(cause), } if context is not None and context is not cause: data["context"] = { "type": type(context).__name__, "repr": repr(context), } return data async def _debug_httpx_get(url: str, headers: Dict[str, str], trust_env: bool) -> Dict[str, Any]: try: async with httpx.AsyncClient(timeout=10, trust_env=trust_env) as client: response = await client.get(url, headers=headers) return { "ok": True, "status_code": response.status_code, "body_prefix": response.text[:160], } except Exception as exc: return { "ok": False, "error": _debug_exception(exc), } async def _debug_sdk_models() -> Dict[str, Any]: try: response = await async_client.models.list() return { "ok": True, "model_count": len(getattr(response, "data", []) or []), } except Exception as exc: return { "ok": False, "error": _debug_exception(exc), } async def _debug_sdk_chat() -> Dict[str, Any]: model = llm_model or os.getenv("LLM_MODEL") if not model: return { "ok": False, "error": {"type": "ConfigError", "repr": "LLM_MODEL is not configured"}, } try: response = await async_client.chat.completions.create( model=model, messages=[{"role": "user", "content": "Reply with only: ok"}], temperature=0, max_tokens=16, ) content = response.choices[0].message.content if response.choices else "" return { "ok": True, "content_prefix": (content or "")[:80], } except Exception as exc: return { "ok": False, "error": _debug_exception(exc), } # ==================== API路由 ==================== def _auth_response(payload: AuthSessionPayload) -> Dict[str, Any]: return { "user": { "id": payload.user.id, "kind": payload.user.kind, "role": payload.user.role.value, "email": payload.user.email, "display_name": payload.user.display_name, "status": payload.user.status, }, "session": { "id": payload.session.id, "user_id": payload.session.user_id, "anonymous_device_id": payload.session.anonymous_device_id, "status": payload.session.status, "expires_at": payload.session.expires_at.isoformat(), }, } @app.post("/api/auth/guest", response_model=AuthResponseAPI) async def guest_login(payload: GuestLoginRequestAPI, request: Request, response: Response): try: auth = await auth_repository.get_or_create_guest( device_id=payload.device_id, user_agent=request.headers.get("user-agent"), ) _set_session_cookie(response, auth.token) return _auth_response(auth) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) @app.post("/api/auth/register", response_model=AuthResponseAPI) async def register(payload: RegisterRequestAPI, request: Request, response: Response): try: existing_auth = await get_optional_auth_session(request) existing_guest_user_id = ( existing_auth.user.id if existing_auth is not None and existing_auth.user.kind == "guest" else None ) auth = await auth_repository.register( email=payload.email, password=payload.password, display_name=payload.display_name, existing_guest_user_id=existing_guest_user_id, ) _set_session_cookie(response, auth.token) return _auth_response(auth) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) @app.post("/api/auth/login", response_model=AuthResponseAPI) async def login(payload: LoginRequestAPI, response: Response): try: auth = await auth_repository.login(email=payload.email, password=payload.password) _set_session_cookie(response, auth.token) return _auth_response(auth) except ValueError as exc: raise HTTPException(status_code=401, detail=str(exc)) @app.post("/api/auth/logout", response_model=GenericSuccessResponseAPI) async def logout(request: Request, response: Response): await auth_repository.revoke_token(request.cookies.get(AUTH_COOKIE_NAME)) _clear_session_cookie(response) return {"success": True, "message": "Logged out"} @app.get("/api/auth/session", response_model=AuthResponseAPI) async def auth_session(auth: AuthSessionPayload = Depends(require_auth_session)): return _auth_response(auth) @app.get("/api", response_model=ApiInfoResponseAPI) async def api_root(): """ 返回API信息 Returns: API基本信息 """ return {"message": "MetaRec API is running!", "version": "1.0.0"} @app.get("/health", response_model=HealthResponseAPI) async def health_check(): """ 健康检查 Returns: 服务健康状态 """ return {"status": "healthy", "timestamp": datetime.now().isoformat()} @app.get("/api/debug/llm-connection") async def debug_llm_connection(_auth: AuthSessionPayload = Depends(require_admin_session)): """ Diagnose LLM connectivity from inside the running backend process. Admin-only: it discloses LLM transport config and issues a live probe request, so it must not be reachable unauthenticated. The response is still redacted and does not expose API keys. """ transport = get_openai_compatible_transport_config() parsed = urlparse(LLM_BASE_URL) host = parsed.hostname or "" dns_result: Dict[str, Any] try: addresses = socket.getaddrinfo(host, parsed.port or 443, type=socket.SOCK_STREAM) dns_result = { "ok": True, "addresses": sorted({item[4][0] for item in addresses})[:8], } except Exception as exc: dns_result = { "ok": False, "error": _debug_exception(exc), } proxy_env_names = [ "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "all_proxy", "no_proxy", "SSL_CERT_FILE", "SSL_CERT_DIR", "REQUESTS_CA_BUNDLE", "CURL_CA_BUNDLE", ] env_snapshot = { name: _mask_debug_value(os.getenv(name)) for name in proxy_env_names if os.getenv(name) is not None } headers = {"Authorization": f"Bearer {LLM_API_KEY}"} if LLM_API_KEY else {} models_url = f"{LLM_BASE_URL.rstrip('/')}/models" return { "config": { "base_url": LLM_BASE_URL, "models_url": models_url, "chat_url_expected": f"{LLM_BASE_URL.rstrip('/')}/chat/completions", "model": llm_model or os.getenv("LLM_MODEL"), "api_key_configured": bool(LLM_API_KEY), "transport": transport, }, "env": env_snapshot, "dns": dns_result, "httpx_models_trust_env_true": await _debug_httpx_get(models_url, headers, True), "httpx_models_trust_env_false": await _debug_httpx_get(models_url, headers, False), "sdk_models_current_client": await _debug_sdk_models(), "sdk_chat_current_client": await _debug_sdk_chat(), } @app.get("/api/config", response_model=FrontendConfigResponseAPI) async def get_config(): """ 获取前端配置信息（包括 Google Maps API Key） Returns: 配置信息 """ google_maps_api_key = os.getenv("VITE_GOOGLE_MAPS_API_KEY", "") return { "googleMapsApiKey": google_maps_api_key } @app.post("/api/process", response_model=RecommendationResponseAPI) async def process_user_request(query_data: ProcessRequestAPI, request: Request): """ 处理用户请求的统一接口 融合了 LLM 意图识别、偏好提取、确认流程 这个接口会自动处理： - 使用 LLM 进行意图识别和生成回复 - 如果是推荐餐厅请求：触发推荐流程 - 如果是普通对话：返回 LLM 的回复 Args: query_data: {"query": "用户查询", "user_id": "用户ID（可选）", "conversation_history": "对话历史（可选）"} Returns: 根据处理结果返回不同的响应： - 如果是 LLM 回复：返回 llm_reply 字段 - 如果是确认请求：返回确认请求对象 - 如果是任务创建：返回任务ID - 如果是修改请求：返回修改提示 """ try: query = query_data.query user_id = await resolve_request_user_id(request, query_data.user_id) conversation_history = query_data.conversation_history if conversation_history is not None: conversation_history = [msg.model_dump() for msg in conversation_history] conversation_id = query_data.conversation_id use_online_agent = query_data.use_online_agent replay_from_message_id = query_data.replay_from_message_id branch_id = query_data.branch_id time_travel_mode = query_data.time_travel_mode domain_lock = query_data.domain_lock hitl_state = query_data.hitl_state # 添加日志，确认参数接收 print(f"[API] Received request - use_online_agent: {use_online_agent} (type: {type(use_online_agent)})") if ( conversation_id and replay_from_message_id and (time_travel_mode is None or time_travel_mode == "linear_regenerate") ): try: await conversation_repository.mark_messages_superseded_after( user_id, conversation_id, replay_from_message_id, branch_id, ) except Exception as e: print(f"Warning: Failed to mark superseded messages: {e}") # 调用异步处理函数（使用 LLM 进行意图识别） result = await metarec_service.handle_user_request_async( query, user_id, conversation_history, conversation_id, use_online_agent, message_id=query_data.source_message_id, branch_id=branch_id, timeline_cursor=replay_from_message_id or query_data.parent_message_id, domain_lock=domain_lock, hitl_state=hitl_state, ) time_travel_payload = None if replay_from_message_id or branch_id or time_travel_mode: time_travel_payload = { "mode": time_travel_mode or "linear_regenerate", "replay_from_message_id": replay_from_message_id, "branch_id": branch_id, "source_message_id": query_data.source_message_id, "parent_message_id": query_data.parent_message_id, } # 如果响应包含 preferences 且有 conversation_id，更新 conversation 的 preferences # （food_intent 为请求级，不写入会话基线，避免上一句的菜品粘连到下一次请求） if result.get("preferences") and conversation_id: try: result_preferences = result["preferences"] persistable_preferences = _restaurant_preference_subset(result_preferences) if persistable_preferences: await conversation_repository.update_conversation_preferences(user_id, conversation_id, persistable_preferences) except Exception as e: print(f"Warning: Failed to update conversation preferences: {e}") if result.get("preferences"): try: await _persist_profile_preferences_from_result( user_id, result["preferences"], update_persona_memory=result.get("type") == "task_created", ) except Exception as e: print(f"Warning: Failed to update profile preferences: {e}") # 根据处理结果类型返回不同的响应 if result["type"] == "llm_reply": # LLM 的普通对话回复 # 如果是confirm no的情况（intent为confirmation_no或chat且有preferences），确保返回preferences intent = result.get("intent", "chat") preferences = result.get("preferences") return RecommendationResponseAPI( restaurants=[], thinking_steps=None, confirmation_request=None, llm_reply=result.get("llm_reply", ""), intent=intent, domain=result.get("domain"), time_travel=time_travel_payload, hitl_state=result.get("hitl_state"), metadata=client_safe_metadata(result.get("metadata")), preferences=preferences ) elif result["type"] == "task_created": # 任务已创建，返回任务ID和thinking step task_id = result["task_id"] metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {} metadata = client_safe_metadata({**metadata, "task_id": task_id}) return RecommendationResponseAPI( restaurants=[], thinking_steps=[ThinkingStepAPI( step="start_processing", description="Starting recommendation process...", status="thinking", details=f"Task ID: {task_id}" )], confirmation_request=None, task_id=task_id, domain=result.get("domain"), time_travel=time_travel_payload, hitl_state=result.get("hitl_state"), metadata=metadata, preferences=result.get("preferences") ) elif result["type"] == "confirmation": # 需要确认，返回确认请求 confirmation = result["confirmation_request"] # 确保返回intent信息（如果有） intent = result.get("intent") # 安全地转换 confirmation 对象，确保 preferences 中的列表被正确处理 confirmation_dict = confirmation.dict() # 确保 preferences 中的列表被正确复制（避免引用问题） if "preferences" in confirmation_dict: preferences = confirmation_dict["preferences"] if isinstance(preferences, dict): # 深拷贝 preferences 字典，确保列表被正确复制 import copy confirmation_dict["preferences"] = copy.deepcopy(preferences) return RecommendationResponseAPI( restaurants=[], thinking_steps=None, confirmation_request=ConfirmationRequestAPI(**confirmation_dict), intent=intent, domain=result.get("domain"), time_travel=time_travel_payload, hitl_state=result.get("hitl_state"), metadata=client_safe_metadata(result.get("metadata")), preferences=result.get("preferences") ) else: # modify_request # 需要修改，返回修改提示 return RecommendationResponseAPI( restaurants=[], thinking_steps=None, confirmation_request=ConfirmationRequestAPI( message=result["message"], preferences=result.get("preferences", {}), needs_confirmation=True ), domain=result.get("domain"), time_travel=time_travel_payload, hitl_state=result.get("hitl_state"), metadata=client_safe_metadata(result.get("metadata")), preferences=result.get("preferences") ) except HTTPException: raise except Exception: logger.exception("process_user_request failed") raise HTTPException(status_code=500, detail="Error processing request") # Server-side-only diagnostic keys carried in result/task metadata: raw third-party # tool outputs, the LLM tool-plan, and internal tool names. The frontend never reads # them; stripping at the API boundary keeps them out of responses AND out of the # conversation message metadata the client persists from a result. The durable result # store keeps full detail for admin/debug. _INTERNAL_METADATA_KEYS = ("executions", "plan_calls", "selected_tools", "skipped_tools", "progress_events") def client_safe_metadata(metadata: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]: """Return a copy of result/task metadata with server-only diagnostic blobs removed, recursing into the nested task-projection ``result_metadata``.""" if not isinstance(metadata, dict): return metadata cleaned = {key: value for key, value in metadata.items() if key not in _INTERNAL_METADATA_KEYS} nested = cleaned.get("result_metadata") if isinstance(nested, dict): cleaned["result_metadata"] = client_safe_metadata(nested) return cleaned def _client_safe_item(item: Any) -> Dict[str, Any]: """Project a recommendation item to its client-facing fields, dropping the ``raw`` upstream provider payload. ``raw`` is retained server-side in the recommendation_results store for persistence/debug but must never ship to the client (it can carry unbounded, unsanitized third-party data).""" if hasattr(item, "model_dump"): data = item.model_dump() elif hasattr(item, "dict"): data = item.dict() else: try: data = dict(item) except (TypeError, ValueError): data = {} if not isinstance(data, dict): data = {} data.pop("raw", None) return data def _build_task_status_api(task_status: Dict[str, Any], task_id: str) -> TaskStatusAPI: """Project a raw task-status dict (in-memory or persisted) into the public TaskStatusAPI shape. Shared by the polling endpoint and the SSE stream so both serve byte-identical frames.""" result_api = None if task_status.get("result"): result = task_status["result"] if hasattr(result, "model_dump"): result_data = result.model_dump() elif hasattr(result, "dict"): result_data = result.dict() else: result_data = result if isinstance(result, dict) else {} restaurants_data = result_data.get("restaurants", []) items_data = result_data.get("items", []) thinking_steps_data = result_data.get("thinking_steps") metadata = result_data.get("metadata") if isinstance(result_data.get("metadata"), dict) else {} metadata = client_safe_metadata(metadata) or {} status_metadata = task_status.get("metadata") if isinstance(task_status.get("metadata"), dict) else {} result_task_id = ( result_data.get("task_id") or metadata.get("task_id") or status_metadata.get("task_id") or task_status.get("task_id") or task_id ) result_id = result_data.get("result_id") or metadata.get("result_id") or status_metadata.get("result_id") if result_task_id and not metadata.get("task_id"): metadata["task_id"] = result_task_id if result_id and not metadata.get("result_id"): metadata["result_id"] = result_id result_api = RecommendationResponseAPI( restaurants=[ RestaurantAPI(**(r.dict() if hasattr(r, "dict") else r)) for r in restaurants_data ], items=[ RecommendationItemAPI(**_client_safe_item(item)) for item in items_data ], thinking_steps=[ ThinkingStepAPI(**(s.dict() if hasattr(s, "dict") else s)) for s in thinking_steps_data ] if thinking_steps_data else None, confirmation_request=None, task_id=result_task_id, result_id=result_id, domain=metadata.get("domain"), metadata=metadata or None, preferences=metadata.get("preferences"), ) return TaskStatusAPI( task_id=task_status.get("task_id", task_id), status=task_status.get("status", "unknown"), progress=task_status.get("progress", 0), message=task_status.get("message", ""), result=result_api, error=task_status.get("error"), metadata=client_safe_metadata(task_status.get("metadata")) if isinstance(task_status.get("metadata"), dict) else None, ) async def sse_task_status_frames( fetch_status: Callable[[], Awaitable[Optional[Dict[str, Any]]]], task_id: str, *, is_disconnected: Optional[Callable[[], Awaitable[bool]]] = None, interval: float = 0.4, not_found_timeout: float = 10.0, max_duration: float = 300.0, now: Callable[[], float] = time.monotonic, sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, ): """Yield Server-Sent-Events frames for a task's status until it settles. `fetch_status` returns the current public status dict (or None if the task isn't visible yet). A frame is emitted only when the serialized status changes, so an idle task produces no traffic. The generator stops after the task reaches a terminal state, the client disconnects, the task never appears within `not_found_timeout`, or `max_duration` elapses (safety cap). The clock and sleep are injectable so the logic is unit-testable without real delays.""" start = now() last_serialized: Optional[str] = None # Open the stream immediately so the client's connection settles fast. yield ": connected\n\n" while True: if is_disconnected is not None and await is_disconnected(): return if now() - start > max_duration: return status = await fetch_status() if status is None: if now() - start > not_found_timeout: # Surface as a terminal error frame (not an SSE `error` event) so # the client treats it as settled and stops instead of reconnecting. yield "data: " + json.dumps( { "task_id": task_id, "status": "error", "progress": 0, "message": "Task not found", "error": "Task not found", "result": None, "metadata": None, } ) + "\n\n" return await sleep(interval) continue serialized = json.dumps(status, default=str, sort_keys=True) if serialized != last_serialized: last_serialized = serialized yield "data: " + json.dumps(status, default=str) + "\n\n" if status.get("status") in {"completed", "error", "cancelled"}: return await sleep(interval) @app.get("/api/status/{task_id}", response_model=TaskStatusAPI) async def get_task_status( request: Request, task_id: str, user_id: Optional[str] = None, conversation_id: Optional[str] = None ): """ 获取任务状态 前端通过轮询此接口获取任务进度和最终结果（SSE 不可用时的回退路径） Args: task_id: 任务ID user_id: 用户ID（可选，提供后更精确查找） conversation_id: 会话ID（可选，提供后更精确查找） Returns: 任务状态信息，包括： - status: "processing" | "completed" | "error" | "cancelled" - progress: 0-100的进度值 - message: 当前状态消息 - result: 推荐结果（任务完成时） - error: 错误信息（任务失败时） """ if not user_id or not conversation_id: raise HTTPException( status_code=400, detail="user_id and conversation_id are required for scoped task status", ) user_id = await resolve_request_user_id(request, user_id) task_status = await metarec_service.get_task_status_async(task_id, user_id, conversation_id) if not task_status: raise HTTPException(status_code=404, detail="Task not found") return _build_task_status_api(task_status, task_id) @app.get( "/api/status/{task_id}/stream", response_class=StreamingResponse, responses={200: {"content": {"text/event-stream": {}}}}, ) async def stream_task_status( request: Request, task_id: str, user_id: Optional[str] = None, conversation_id: Optional[str] = None, ): """Stream task progress as Server-Sent Events. Replaces client-side 1s polling: the browser opens a single EventSource and the server pushes a status frame whenever the task projection changes, ending on completion/error. Reads the same projection the polling endpoint serves (no graph changes), so /api/status remains a working fallback when SSE can't get through (proxies, missing EventSource).""" if not user_id or not conversation_id: raise HTTPException( status_code=400, detail="user_id and conversation_id are required for scoped task status", ) resolved_user_id = await resolve_request_user_id(request, user_id) interval = float(os.getenv("METAREC_SSE_POLL_INTERVAL", "0.4")) not_found_timeout = float(os.getenv("METAREC_SSE_NOT_FOUND_TIMEOUT", "10")) max_duration = float(os.getenv("METAREC_SSE_MAX_DURATION", "300")) async def fetch_status() -> Optional[Dict[str, Any]]: task_status = await metarec_service.get_task_status_async( task_id, resolved_user_id, conversation_id ) if not task_status: return None return _build_task_status_api(task_status, task_id).model_dump(mode="json") return StreamingResponse( sse_task_status_frames( fetch_status, task_id, is_disconnected=request.is_disconnected, interval=interval, not_found_timeout=not_found_timeout, max_duration=max_duration, ), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", # Disable proxy/nginx response buffering so frames flush immediately. "X-Accel-Buffering": "no", }, ) @app.get("/api/tasks/{task_id}/result") async def get_task_result( request: Request, task_id: str, user_id: Optional[str] = None, conversation_id: Optional[str] = None, ): """Resolve the durable recommendation persisted for a Task ID. Reads from recommendation_results (the canonical, queryable source of truth), scoped to the owning user/conversation — used by the conversation side card and the /Debug testing arena to fetch a result by Task ID without re-deriving it from conversation messages. """ if not user_id or not conversation_id: raise HTTPException( status_code=400, detail="user_id and conversation_id are required for scoped task result lookup", ) user_id = await resolve_request_user_id(request, user_id) repository = getattr(metarec_service, "result_repository", None) if repository is None: raise HTTPException(status_code=503, detail="Result store is not available") payload = await repository.load_by_task(user_id, conversation_id, task_id) if payload is None: raise HTTPException(status_code=404, detail="No stored result for this task") if isinstance(payload, dict): payload = _client_safe_result_payload(payload) # Legacy rows nested a full duplicate of the result under ``result``; # sanitize it the same way (new rows store one canonical copy only). inner_result = payload.get("result") if isinstance(inner_result, dict): payload["result"] = _client_safe_result_payload(inner_result) return payload def _client_safe_result_payload(payload: Dict[str, Any]) -> Dict[str, Any]: """Client-facing projection of a stored recommendation payload: internal diagnostic metadata removed and every item's ``raw`` provider blob stripped (same rules the task-status endpoints apply via ``_client_safe_item``).""" cleaned = dict(payload) if isinstance(cleaned.get("metadata"), dict): cleaned["metadata"] = client_safe_metadata(cleaned["metadata"]) if isinstance(cleaned.get("items"), list): cleaned["items"] = [_client_safe_item(item) for item in cleaned["items"]] return cleaned @app.post("/api/update-preferences", response_model=UpdatePreferencesResponseAPI) async def update_preferences_endpoint(preferences_data: UpdatePreferencesRequestAPI, request: Request): """ 更新用户偏好设置 Args: preferences_data: 包含用户偏好的字典，格式： { "user_id": "用户ID（可选，默认'default'）", "restaurantTypes": ["casual", "fine-dining"], "flavorProfiles": ["spicy", "savory"], "diningPurpose": "friends", "budgetRange": {"min": 20, "max": 60, "currency": "SGD", "per": "person"}, "location": "Chinatown" } Returns: 更新后的偏好设置 """ try: user_id = await resolve_request_user_id(request, preferences_data.user_id) # 验证和标准化偏好数据 processed_preferences = { "restaurant_types": preferences_data.restaurantTypes, "flavor_profiles": preferences_data.flavorProfiles, "dining_purpose": preferences_data.diningPurpose, "budget_range": preferences_data.budgetRange.model_dump(), "location": preferences_data.location } profile = await profile_repository.get_user_profile(user_id) domains = assemble_domains(profile) domains["restaurant"] = _merge_meaningful_preferences(domains.get("restaurant", {}), processed_preferences) restaurant_slice = domains.pop("restaurant", {}) metadata = profile.setdefault("metadata", {}) metadata["preferences"] = {} metadata["domains"] = domains profile["dining_habits"] = restaurant_slice await profile_repository.save_user_profile(user_id, profile) updated_prefs = processed_preferences return { "message": "Preferences updated successfully", "preferences": updated_prefs } except HTTPException: raise except Exception: logger.exception("update_preferences failed") raise HTTPException(status_code=500, detail="Error updating preferences") @app.get("/api/user-preferences/{user_id}", response_model=UserPreferencesResponseAPI) async def get_user_preferences_endpoint(user_id: str, request: Request): """ 获取用户当前的偏好设置 Args: user_id: 用户ID Returns: 用户偏好设置，包括： - user_id: 用户ID - preferences: 偏好设置字典 """ try: await require_path_user(request, user_id) profile = await profile_repository.get_user_profile(user_id) domains = assemble_domains(profile) preferences = _merge_meaningful_preferences( metarec_service.get_default_preferences(), domains.get("restaurant", {}), ) return { "user_id": user_id, "preferences": preferences } except HTTPException: raise except Exception: logger.exception("get_user_preferences failed") raise HTTPException(status_code=500, detail="Error getting user preferences") # ==================== 三层用户画像 API ==================== class UserProfileAPI(StrictBaseModel): """Three-layer user profile: generic core (demographics + cross-domain constraints), an NL taste persona, and per-domain structured slices.""" user_id: str demographics: Dict[str, Any] = Field(default_factory=dict) constraints: Dict[str, Any] = Field(default_factory=dict) taste_persona: str = "" domains: Dict[str, Dict[str, Any]] = Field(default_factory=dict) class UserProfileUpdateAPI(StrictBaseModel): demographics: Dict[str, Any] = Field(default_factory=dict) constraints: Dict[str, Any] = Field(default_factory=dict) taste_persona: str = "" domains: Dict[str, Dict[str, Any]] = Field(default_factory=dict) def _profile_to_api(user_id: str, profile: Dict[str, Any]) -> UserProfileAPI: normalized = normalize_profile(profile) return UserProfileAPI( user_id=user_id, demographics=normalized.get("demographics", {}), constraints=normalized.get("constraints", {}), taste_persona=normalized.get("taste_persona", ""), domains=normalized.get("domains", {}), ) @app.get("/api/user-profile/{user_id}", response_model=UserProfileAPI) async def get_user_profile_endpoint(user_id: str, request: Request): """Return the three-layer profile for editing/fusion.""" try: await require_path_user(request, user_id) profile = await profile_repository.get_user_profile(user_id) return _profile_to_api(user_id, profile) except HTTPException: raise except Exception: logger.exception("get_user_profile failed") raise HTTPException(status_code=500, detail="Error getting user profile") @app.put("/api/user-profile/{user_id}", response_model=UserProfileAPI) async def update_user_profile_endpoint(user_id: str, payload: UserProfileUpdateAPI, request: Request): """Persist the three-layer profile. The restaurant slice maps onto the legacy ``dining_habits`` column; other domains and the persona/constraints live in ``metadata`` — keeping existing restaurant data backward compatible.""" try: await require_path_user(request, user_id) domains = dict(payload.domains or {}) restaurant_slice = domains.pop("restaurant", {}) or {} physical = { "user_id": user_id, "demographics": payload.demographics or {}, "dining_habits": restaurant_slice, "metadata": { "taste_persona": payload.taste_persona or "", "constraints": payload.constraints or {}, "domains": domains, "preferences": {}, }, } await profile_repository.save_user_profile(user_id, physical) refreshed = await profile_repository.get_user_profile(user_id) return _profile_to_api(user_id, refreshed) except HTTPException: raise except Exception: logger.exception("update_user_profile failed") raise HTTPException(status_code=500, detail="Error updating user profile") # ==================== 请求时偏好表单生成 API ==================== class PreferenceFieldAPI(StrictBaseModel): key: str label: str type: str options: List[str] = Field(default_factory=list) required: bool = False placeholder: str = "" value: Optional[Any] = None class DomainPreferenceFormAPI(StrictBaseModel): domain: str fields: List[PreferenceFieldAPI] = Field(default_factory=list) missing_required: List[str] = Field(default_factory=list) complete: bool = True @app.get("/api/domains/{domain}/preference-form", response_model=DomainPreferenceFormAPI) async def get_domain_preference_form(domain: str, request: Request): """Generate the (server-driven) preference form for a domain at request time. The frontend renders it generically; adding a domain's form is a data change.""" try: await resolve_request_user_id(request, None) return build_domain_form(domain) except HTTPException: raise except Exception: logger.exception("get_domain_preference_form failed") raise HTTPException(status_code=500, detail="Error building preference form") # ==================== 对话历史API ==================== class ConversationSummary(StrictBaseModel): """对话摘要（用于列表）""" id: str title: str model: str last_message: str timestamp: str updated_at: str message_count: int class MessageData(StrictBaseModel): """消息数据""" id: Optional[str] = None role: str content: str timestamp: Optional[str] = None branch_id: Optional[str] = None parent_message_id: Optional[str] = None fork_from_message_id: Optional[str] = None revision_of_message_id: Optional[str] = None metadata: Optional[Dict[str, Any]] = Field( default=None, json_schema_extra={"additionalProperties": True}, ) class BranchData(StrictBaseModel): """Conversation branch metadata.""" id: str parent_branch_id: Optional[str] = None fork_from_message_id: Optional[str] = None root_message_id: Optional[str] = None head_message_id: Optional[str] = None title: Optional[str] = None created_at: str updated_at: str class ConversationData(StrictBaseModel): """完整对话数据""" id: str user_id: str title: str model: str last_message: str timestamp: str updated_at: str active_branch_id: Optional[str] = "branch-main" branch_selection_state: Dict[str, str] = Field(default_factory=dict) branches: Dict[str, BranchData] = Field(default_factory=dict) messages: List[MessageData] preferences: Dict[str, Any] = Field( default_factory=dict, json_schema_extra={"additionalProperties": True}, ) # Internal conversation memory (e.g. the rolling context summary). Accepted so # the loaded conversation validates, but excluded from the API response — it is # server-side state, not part of the client contract. metadata: Dict[str, Any] = Field(default_factory=dict, exclude=True) class CreateConversationRequest(StrictBaseModel): """创建对话请求""" title: Optional[str] = None model: str = "Auto" class UpdateConversationRequest(StrictBaseModel): """更新对话请求""" title: Optional[str] = None model: Optional[str] = None class AddMessageRequest(StrictBaseModel): """添加消息请求""" role: str content: str metadata: Optional[Dict[str, Any]] = Field( default=None, json_schema_extra={"additionalProperties": True}, ) class SetActiveBranchRequest(StrictBaseModel): branch_id: str source_message_id: Optional[str] = None @app.get("/api/conversations/{user_id}", response_model=List[ConversationSummary]) async def get_all_conversations( user_id: str, request: Request, limit: int = Query(default=50, ge=1, le=100), offset: int = Query(default=0, ge=0), ): """ 获取用户的所有对话列表 Args: user_id: 用户ID Returns: 对话摘要列表 """ try: await require_path_user(request, user_id) conversations = await conversation_repository.get_all_conversations(user_id, limit=limit, offset=offset) return conversations except HTTPException: raise except Exception: logger.exception("get_all_conversations failed") raise HTTPException(status_code=500, detail="Error getting conversations") @app.get("/api/conversations/{user_id}/{conversation_id}", response_model=ConversationData) async def get_conversation(user_id: str, conversation_id: str, request: Request): """ 获取单个对话的完整信息（包含所有消息） Args: user_id: 用户ID conversation_id: 对话ID Returns: 完整的对话数据 """ try: await require_path_user(request, user_id) conversation = await conversation_repository.get_full_conversation(user_id, conversation_id) if not conversation: raise HTTPException(status_code=404, detail="Conversation not found") return conversation except HTTPException: raise except Exception: logger.exception("get_conversation failed") raise HTTPException(status_code=500, detail="Error getting conversation") @app.post("/api/conversations/{user_id}", response_model=ConversationData) async def create_conversation(user_id: str, request_data: CreateConversationRequest, request: Request): """ 创建新对话 Args: user_id: 用户ID request: 创建对话请求 Returns: 创建的对话数据 """ try: await require_path_user(request, user_id) conversation = await conversation_repository.create_conversation( user_id=user_id, title=request_data.title, model=request_data.model ) return conversation except HTTPException: raise except Exception: logger.exception("create_conversation failed") raise HTTPException(status_code=500, detail="Error creating conversation") @app.put("/api/conversations/{user_id}/{conversation_id}", response_model=ConversationData) async def update_conversation( user_id: str, conversation_id: str, request_data: UpdateConversationRequest, request: Request, ): """ 更新对话信息（如标题、模型等） Args: user_id: 用户ID conversation_id: 对话ID request: 更新请求 Returns: 更新后的对话数据 """ try: await require_path_user(request, user_id) updates = {} if request_data.title is not None: updates["title"] = request_data.title if request_data.model is not None: updates["model"] = request_data.model if not updates: raise HTTPException(status_code=400, detail="No fields to update") success = await conversation_repository.update_conversation(user_id, conversation_id, updates) if not success: raise HTTPException(status_code=404, detail="Conversation not found") conversation = await conversation_repository.get_full_conversation(user_id, conversation_id) return conversation except HTTPException: raise except Exception: logger.exception("update_conversation failed") raise HTTPException(status_code=500, detail="Error updating conversation") @app.post("/api/conversations/{user_id}/{conversation_id}/messages", response_model=GenericSuccessResponseAPI) async def add_message( user_id: str, conversation_id: str, request_data: AddMessageRequest, request: Request, ): """ 向对话添加消息 Args: user_id: 用户ID conversation_id: 对话ID request: 添加消息请求 Returns: 成功状态 """ try: await require_path_user(request, user_id) if request_data.role not in ["user", "assistant"]: raise HTTPException(status_code=400, detail="Role must be 'user' or 'assistant'") success = await conversation_repository.add_message( user_id=user_id, conversation_id=conversation_id, role=request_data.role, content=request_data.content, metadata=request_data.metadata ) if not success: raise HTTPException(status_code=404, detail="Conversation not found") return {"success": True, "message": "Message added successfully"} except HTTPException: raise except Exception: logger.exception("add_message failed") raise HTTPException(status_code=500, detail="Error adding message") @app.put("/api/conversations/{user_id}/{conversation_id}/active-branch", response_model=ConversationData) async def set_active_branch( user_id: str, conversation_id: str, request_data: SetActiveBranchRequest, request: Request, ): """ Switch the active visible branch for a conversation. """ try: await require_path_user(request, user_id) success = await conversation_repository.set_active_branch( user_id, conversation_id, request_data.branch_id, request_data.source_message_id, ) if not success: raise HTTPException(status_code=404, detail="Conversation or branch not found") conversation = await conversation_repository.get_full_conversation(user_id, conversation_id) if not conversation: raise HTTPException(status_code=404, detail="Conversation not found") return conversation except HTTPException: raise except Exception: logger.exception("set_active_branch failed") raise HTTPException(status_code=500, detail="Error setting active branch") @app.delete("/api/conversations/{user_id}/{conversation_id}", response_model=GenericSuccessResponseAPI) async def delete_conversation(user_id: str, conversation_id: str, request: Request): """ 删除对话 Args: user_id: 用户ID conversation_id: 对话ID Returns: 成功状态 """ try: await require_path_user(request, user_id) success = await conversation_repository.delete_conversation(user_id, conversation_id) if not success: raise HTTPException(status_code=404, detail="Conversation not found") try: cancel_summary = await metarec_service.cancel_conversation_tasks_async(user_id, conversation_id) except Exception: logger.exception("delete_conversation task cancellation failed") cancel_summary = {"cancelled": 0} cancelled = int(cancel_summary.get("cancelled") or 0) if cancelled: message = f"Conversation deleted successfully; cancelled {cancelled} running recommendation task(s)." else: message = "Conversation deleted successfully" return {"success": True, "message": message} except HTTPException: raise except Exception: logger.exception("delete_conversation failed") raise HTTPException(status_code=500, detail="Error deleting conversation") @app.get("/api/conversations/{user_id}/{conversation_id}/preferences", response_model=PreferencesResponseAPI) async def get_conversation_preferences(user_id: str, conversation_id: str, request: Request): """ 获取对话的偏好设置（优先从内存缓存获取） Args: user_id: 用户ID conversation_id: 对话ID Returns: 偏好设置字典 """ try: await require_path_user(request, user_id) preferences = await conversation_repository.get_conversation_preferences(user_id, conversation_id) if preferences is None: raise HTTPException(status_code=404, detail="Conversation not found") return {"preferences": preferences} except HTTPException: raise except Exception: logger.exception("get_conversation_preferences failed") raise HTTPException(status_code=500, detail="Error getting conversation preferences") @app.put("/api/conversations/{user_id}/{conversation_id}/preferences", response_model=PreferencesResponseAPI) async def update_conversation_preferences( user_id: str, conversation_id: str, preferences_data: Dict[str, object], request: Request, ): """ 更新对话的偏好设置（同时更新内存缓存和持久化层） Args: user_id: 用户ID conversation_id: 对话ID preferences_data: 偏好设置字典 Returns: 更新后的偏好设置（从内存缓存返回） """ try: await require_path_user(request, user_id) success = await conversation_repository.update_conversation_preferences(user_id, conversation_id, preferences_data) if not success: raise HTTPException(status_code=404, detail="Conversation not found") updated_preferences = await conversation_repository.get_conversation_preferences(user_id, conversation_id) if updated_preferences is None: raise HTTPException(status_code=404, detail="Conversation not found") return {"preferences": updated_preferences} except HTTPException: raise except Exception: logger.exception("update_conversation_preferences failed") raise HTTPException(status_code=500, detail="Error updating conversation preferences") # ==================== 静态文件服务（在所有 API 路由之后）==================== # 挂载静态资源目录 if os.path.exists(FRONTEND_DIST): assets_dir = os.path.join(FRONTEND_DIST, "assets") if os.path.exists(assets_dir): app.mount("/assets", StaticFiles(directory=assets_dir), name="assets") @app.get("/", include_in_schema=False) async def serve_root(): """服务根路径的前端应用""" index_path = os.path.join(FRONTEND_DIST, "index.html") if os.path.exists(index_path): return FileResponse(index_path) return {"message": "MetaRec API", "docs": "/docs"} @app.get("/{full_path:path}", include_in_schema=False) async def serve_spa(full_path: str): """SPA fallback - 所有未匹配的路由返回 index.html""" # 检查是否是静态文件 file_path = FRONTEND_DIST.joinpath(full_path).resolve() # 1. Prevent escaping FRONTEND_DIST directory using path traversal i.e. '../' which would otherwise allow user to access arbitrary files on the filesystem # TODO: consider logging this to track malicious users? if not file_path.is_relative_to(FRONTEND_DIST): raise HTTPException(status_code=403, detail="Forbidden") # 2. check that the requested file exists if file_path.is_file(): # checks for existence of file and that the file is a regular file return FileResponse(file_path) # 3. fallback to index page index_path = os.path.join(FRONTEND_DIST, "index.html") if os.path.exists(index_path): return FileResponse(index_path) # 如果没有前端文件，返回 404 raise HTTPException(status_code=404, detail="Not found") # ==================== 启动配置 ==================== if __name__ == "__main__": import uvicorn # 使用环境变量PORT，默认8000（本地开发） # Hugging Face Spaces 可以设置 PORT=7860 port = int(os.getenv("PORT", 8000)) print(f"🚀 Starting MetaRec API server on http://0.0.0.0:{port}") print(f"📖 API docs available at http://localhost:{port}/docs") print(f"🌐 Frontend should be available at http://localhost:{port}/") print(f"📝 Logging level: INFO - All print() messages will be displayed") # 配置 uvicorn 日志，确保实时输出 log_config = { "version": 1, "disable_existing_loggers": False, "formatters": { "default": { "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s", "datefmt": "%Y-%m-%d %H:%M:%S", }, }, "handlers": { "default": { "formatter": "default", "class": "logging.StreamHandler", "stream": "ext://sys.stdout", }, }, "root": { "level": "INFO", "handlers": ["default"], }, "loggers": { "uvicorn": {"level": "INFO"}, "uvicorn.error": {"level": "INFO"}, "uvicorn.access": {"level": "INFO"}, }, } uvicorn.run( app, host="0.0.0.0", port=port, log_config=log_config, log_level="info" )