# -*- coding: utf-8 -*-
"""ラーメンスープ評価 — 推論エンジン（利用システム / 自己完結・推論専用）

学習機能は持たない。AI開発システムが学習・保存した models/ を読み込み、
入力（濃度brix・時間帯slot・動画由来のLLM画像評価）から
スコア／グレード／要確認 を出力するだけのモジュール。

公開API:
    load_models(model_dir)         -> 学習済みモデル読込
    score_record(record, models)   -> 単件推論（dict in -> dict out）
    score_batch(raw_df, models)    -> バッチ推論（DataFrame in -> DataFrame out）
    grade(score)                   -> A/B/C/D グレード

入力レコードのキー（動画は上流で下記LLM評価項目に変換済みである前提）:
    brix_raw                : 濃度数値（実測 or 濃度推定AI）
    slot                    : 時間帯（開店前 / 17時〜19時 等）
    visual_density / water_level / oil_emulsification /
    boiling_heat_state / photo_quality / image_condition_score
                            : 動画→最良フレーム→LLM画像評価AI の出力
    readable（任意）        : 上流RF-DETRスープ可読性(bool)。Falseで「要確認」
"""
from __future__ import annotations

import re
from pathlib import Path

import numpy as np
import pandas as pd

# ====================================================================
# 定数・スコアマッピング（学習時と完全一致させること）
# ====================================================================
UNASSESSABLE = "判定不可"

WATER_SCORE = {"水位良好": 1.0, "水位不足": 0.0, UNASSESSABLE: 0.0}
OIL_SCORE = {
    "油感と乳化の状態が良い": 1.0, "乳化が進んでいる": 0.85,
    "油感が強い": 0.45, "乳化しすぎ": 0.35, UNASSESSABLE: 0.0,
}
BOILING_SCORE = {"良い": 1.0, "弱い": 0.45, "強すぎる": 0.35, UNASSESSABLE: 0.0}
PHOTO_SCORE = {
    "良い": 1.0, "湯気が多い": 0.45, "白飛び": 0.35,
    "暗い": 0.35, "角度不良": 0.35, "不良": 0.0, UNASSESSABLE: 0.0,
}

# 学習済みモデルの target → 出力名
TARGETS = ("human_total_score", "human_soup_score")


# ====================================================================
# グレード定義
# ====================================================================
def grade(score) -> np.ndarray:
    values = np.asarray(score, dtype=float)
    out = np.full(values.shape, "D", dtype="<U1")
    out[values >= 21] = "C"
    out[values >= 60] = "B"
    out[values >= 80] = "A"
    return out


# ====================================================================
# 補助
# ====================================================================
def _video_id(path: object) -> str:
    name = Path(str(path)).stem
    m = re.search(r"\d+", name)
    return m.group(0) if m else name


def _to_density(value: object) -> float:
    try:
        return float(value)
    except Exception:
        return np.nan


def _map_score(series: pd.Series, mapping: dict) -> pd.Series:
    return series.fillna(UNASSESSABLE).astype(str).str.strip().map(mapping).fillna(0.0)


# ====================================================================
# 特徴量生成（学習時 prepare と同一。推論なので人間点数は不要）
# ====================================================================
def prepare(raw: pd.DataFrame) -> pd.DataFrame:
    raw = raw.copy()
    df = pd.DataFrame()
    if "video_path" in raw:
        df["sample_id"] = raw["video_path"].map(_video_id)
    else:
        df["sample_id"] = pd.Series(range(len(raw)))
    df["video_path"] = raw.get("video_path")
    df["frame_path"] = raw.get("frame_path")
    df["store"] = raw.get("store", pd.Series(["unknown"] * len(raw))).fillna("unknown").astype(str)
    df["slot"] = raw.get("slot", pd.Series(["unknown"] * len(raw))).fillna("unknown").astype(str)
    df["brix"] = pd.to_numeric(raw.get("brix_raw"), errors="coerce")

    for col in ["visual_density", "water_level", "oil_emulsification", "boiling_heat_state", "photo_quality"]:
        df[col] = raw.get(col, pd.Series([UNASSESSABLE] * len(raw))).fillna(UNASSESSABLE).astype(str).str.strip()

    df["image_condition_score"] = pd.to_numeric(raw.get("image_condition_score"), errors="coerce").fillna(0)
    df["visual_density_num"] = df["visual_density"].map(_to_density)
    df["visual_density_assessable"] = df["visual_density_num"].notna().astype(int)
    df["assessable"] = (
        (df["visual_density"] != UNASSESSABLE)
        & (df["water_level"] != UNASSESSABLE)
        & (df["oil_emulsification"] != UNASSESSABLE)
        & (df["boiling_heat_state"] != UNASSESSABLE)
    ).astype(int)
    df["retake_recommended"] = ((df["photo_quality"] != "良い") | (df["assessable"] == 0)).astype(int)
    df["density_brix_abs_diff"] = (df["visual_density_num"] - df["brix"]).abs()
    df["density_brix_signed_diff"] = df["visual_density_num"] - df["brix"]
    df["density_below_brix"] = (df["visual_density_num"] < df["brix"]).fillna(False).astype(int)

    df["water_score"] = _map_score(df["water_level"], WATER_SCORE)
    df["oil_score"] = _map_score(df["oil_emulsification"], OIL_SCORE)
    df["boiling_score"] = _map_score(df["boiling_heat_state"], BOILING_SCORE)
    df["photo_score"] = _map_score(df["photo_quality"], PHOTO_SCORE)

    df["density_match_score"] = (1.0 - (df["density_brix_abs_diff"].fillna(4).clip(0, 4) / 4.0)).clip(0, 1)
    df["slot_is_opening"] = df["slot"].str.contains("開店", na=False).astype(int)
    df["slot_is_evening"] = df["slot"].str.contains("17|18|19|夜|夕", regex=True, na=False).astype(int)
    df["low_quality_flag"] = (df["image_condition_score"] <= 3).astype(int)
    df["unassessable_axis_count"] = (
        (df[["visual_density", "water_level", "oil_emulsification", "boiling_heat_state", "photo_quality"]] == UNASSESSABLE)
        .sum(axis=1).astype(int)
    )
    return df


# ====================================================================
# モデル読込・推論
# ====================================================================
def load_models(model_dir: str | Path) -> dict:
    import joblib
    model_dir = Path(model_dir)
    return {t: joblib.load(model_dir / f"{t}.pkl") for t in TARGETS}


def _predict_scores(feat: pd.DataFrame, models: dict) -> dict:
    preds = {}
    for target, bundle in models.items():
        X = feat[bundle["numeric"] + bundle["categorical"]]
        preds[target] = np.clip(bundle["pipeline"].predict(X), 0, 100)
    return preds


def score_batch(raw: pd.DataFrame, models: dict) -> pd.DataFrame:
    """バッチ推論。2段ゲート（① readable可読性 / ② assessable）で要確認を分岐。"""
    feat = prepare(raw)
    preds = _predict_scores(feat, models)
    if "readable" in raw.columns:
        readable = raw["readable"].fillna(True).astype(bool).to_numpy()
    else:
        readable = np.ones(len(feat), dtype=bool)
    llm_ok = (feat["assessable"] == 1).to_numpy()
    scoreable = llm_ok & readable
    out = pd.DataFrame({
        "sample_id": feat["sample_id"],
        "video_path": feat["video_path"],
        "frame_path": feat["frame_path"],
        "assessable": feat["assessable"],
        "readable": readable,
        "retake_recommended": feat["retake_recommended"],
    })
    out["status"] = np.where(scoreable, "scored", "要確認")
    out["reason"] = np.where(
        ~readable, "readable=False（スープ表面が抽出/確認できない）→撮り直し/職人確認へ",
        np.where(~llm_ok, "assessable=0（LLMが画像評価不能）→撮り直し/職人確認へ", ""))
    for target in TARGETS:
        col = "total" if target == "human_total_score" else "soup"
        s = preds[target].astype(float)
        out[f"{col}_score"] = np.where(scoreable, np.round(s, 1), np.nan)
        out[f"{col}_grade"] = np.where(scoreable, grade(s), "")
    return out


def score_record(record: dict, models: dict) -> dict:
    """単件推論。dict in -> dict out。"""
    res = score_batch(pd.DataFrame([record]), models).iloc[0].to_dict()
    for k, v in list(res.items()):
        if isinstance(v, float) and np.isnan(v):
            res[k] = None
    return res