recommend_api.py

# 檔名：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