# 檔名:recommend_api.py # 功能: # - 提供 /recommend API # - 內部呼叫 recommender.run_recommend_model() 來產生 clothe.json # - 自動把每次收到的 request 存成 json 檔,放在 saved_requests/ # - 自動把每次回傳的 clothe_json 存成 json 檔,放在 saved_clothes/ # - ✅ 自動判斷身形,印出身形,並把結果塞進 report 給推薦系統用 from fastapi import FastAPI from pydantic import BaseModel from typing import Optional, Dict, Any from pathlib import Path import json import time from recommender import run_recommend_model, ClotheJSON from body_type_classifier import classify_male_body_type, classify_female_body_type app = FastAPI(title="Recommendation Service") BASE_DIR = Path(__file__).resolve().parent LOG_REQ_DIR = BASE_DIR / "saved_requests" LOG_REQ_DIR.mkdir(exist_ok=True) LOG_RES_DIR = BASE_DIR / "saved_clothes" LOG_RES_DIR.mkdir(exist_ok=True) class WeatherInfo(BaseModel): condition: str temperature: float humidity: Optional[float] = None class RecommendRequest(BaseModel): user_id: Optional[str] = None report: dict weather: WeatherInfo DEFAULT_GENDER = "male" def extract_gender(report: Dict[str, Any]) -> Optional[str]: gender = ( report.get("gender") or report.get("sex") or report.get("Gender") or report.get("user_gender") ) if isinstance(gender, str): g = gender.lower() if g in ("male", "m", "boy", "man", "男", "男性"): return "male" if g in ("female", "f", "girl", "woman", "女", "女性"): return "female" if DEFAULT_GENDER is not None: print(f"[BodyType] report 中沒有 gender 欄位,暫時使用 DEFAULT_GENDER={DEFAULT_GENDER}") return DEFAULT_GENDER return None def extract_body_measurements(report: Dict[str, Any]) -> Optional[Dict[str, float]]: bm = report.get("body_measurements") if isinstance(bm, dict): return bm return None def attach_body_type(report: Dict[str, Any]) -> None: body_measurements = extract_body_measurements(report) gender = extract_gender(report) if not body_measurements or not gender: print( f"[BodyType] 無法判斷:缺少 body_measurements 或 gender," f"gender={gender}, has_body={bool(body_measurements)}" ) return try: if gender == "male": body_type = classify_male_body_type(body_measurements) else: body_type = classify_female_body_type(body_measurements) print(f"[BodyType] gender={gender} → body_type={body_type}") report["body_type"] = body_type report["body_gender"] = gender except Exception as e: print(f"[BodyType] 判斷身形時發生錯誤: {e}") @app.post("/recommend", response_model=ClotheJSON) def recommend(req: RecommendRequest) -> ClotheJSON: weather_dict = req.weather.dict() timestamp = int(time.time()) user_part = req.user_id if req.user_id else "anonymous" base_name = f"{user_part}_{timestamp}" report_dict: Dict[str, Any] = dict(req.report) attach_body_type(report_dict) req_path = LOG_REQ_DIR / f"request_{base_name}.json" with req_path.open("w", encoding="utf-8") as f: json.dump( { "user_id": req.user_id, "report": report_dict, "weather": weather_dict, }, f, ensure_ascii=False, indent=2, ) clothe_json = run_recommend_model(report_dict, weather_dict) res_path = LOG_RES_DIR / f"clothe_{base_name}.json" with res_path.open("w", encoding="utf-8") as f: json.dump( clothe_json, f, ensure_ascii=False, indent=2, ) return clothe_json