PokemonCardSearch/code/serve_card_match.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
卡牌匹配 HTTP 服务（双区级联版 · 与入库框架同源）

用途：前端 curl 发一张图片 URL / 上传文件 → 算法端用「上半区召回 + 下半区重排」
      返回命中的 card_id + cards_master_v2 有用字段。

流程（与 scripts/cascade_match.py / 入库完全体同源）：
  下载/读图 → CardDetector 裁卡
           → letterbox 392 → 上半区[0:196] DINOv2-large → 全库点积召回 Top-K
           → 下半区[196:392] Layer3 → fusion=α·upper+(1-α)·lower 重排
           → card_id → 查 card_master_fields.csv → 返回

返回字段（/match 后台接口 2026-08-28 二次定稿，契约见 指令文档/8000后台接口字段解释文档.md）：
  识别输入 = front_image_url（拍摄图正面地址，树莓派直传什么就是什么，不限定 cos——MinIO 内网
             地址等只要 73 服务器可访问即可下载识别；back_image_url 不参与识别仅回显）；
  识别得出：card_type(YOLO card_seg_pn_v2——基于 v1 于2026-09-04微调(404张密集多卡人工标注并入), 多卡mAP50 0.826→0.986; 取值 'pokemon'/'nba') / card_id(dinov2-large 双区检索 Top-1) /
           attribute(YOLO yolo26s_det_attribute, 中文如'火'，检不出 null；JSON 字段名 2026-09-02 由 element 改为 attribute)
  透传回显：front_image_url / back_image_url
  matches：Top-K 候选（默认 5，top_k 可调，match_rate=round(fusion×100)），每项除 rank/card_id/
           match_rate 外均为 card_id 查 cards_master_v2 库字段：
           card_name_ch / pg_label / year / card_no(编号'042/222') / rarity_short(简称'RR'，缺则回退
           全称) / language / material_cn(库值优先如'精灵球闪'，缺则"1") / image_url(库版cos图)
  /match_fields（小程序C端 2026-09-02）：顶层裸数组，每项结构与 /match 完全一致；
             一张拍照图可能有多张卡，按阅读顺序（上→下、同行左→右）依次输出；
             C 端拍摄图落到 249 MinIO grading/capp_img_data，front_image_url 回真地址，
             back_image_url 恒为 null（C 端无背面）。未检出卡返回 []。

启动：
  python serve_card_match.py --port 8000

调用（后台 /match：front_image_url 为识别输入源，任意 73 可访问地址均可，不限 cos；image_url 仍兼容）：
  curl 'http://<服务器>:8000/match?front_image_url=http://192.168.77.80:9000/pokemon/xxx.jpg&back_image_url=http://192.168.77.80:9000/pokemon/yyy.jpg'
  curl -X POST http://<服务器>:8000/match -F "image=@C:/本机/某张卡.jpg"
  curl 'http://<服务器>:8000/match?image_url=/home/user/某张卡.jpg'
"""
import os
import sys
import csv
import argparse
import traceback

sys.stdout.reconfigure(encoding="utf-8")
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

import numpy as np
import config
from modules import image_downloader

# DualCascadeMatcher 放在 scripts/ 下，需要把 scripts 加进 path
_SCRIPTS = os.path.join(os.path.dirname(os.path.abspath(__file__)), "scripts")
if _SCRIPTS not in sys.path:
    sys.path.insert(0, _SCRIPTS)
from cascade_match import DualCascadeMatcher  # noqa: E402
from modules.attribute_detector import AttributeDetector  # noqa: E402


# ==================== 启动时一次性加载 ====================

print("[启动] 加载双区级联匹配器（上半区召回 + 下半区重排）...", flush=True)
MATCHER = DualCascadeMatcher(
    alpha=config.CASCADE_ALPHA,
    top_k_recall=config.CASCADE_TOP_K_RECALL,
    verbose=True,
)
print("[启动] 图库大小:", len(MATCHER.card_ids),
      "α=", MATCHER.alpha, "K_recall=", MATCHER.top_k_recall, flush=True)

print("[启动] 加载元素属性检测模型...", flush=True)
ELEM = AttributeDetector()

# cards_master_v2(PG hs_sync_data.cards_master_v2) 导出的字段表（2026-08-28 v2 版）：
#   rarity=全称(C端用) / rarity_short=简称(后台matches用) / material_cn=材质(空则接口定值1) /
#   img_url=库版cos地址(pokemon_masked_v2，后台matches[].image_url) /
#   img_url_legacy=旧ai_image地址(C端frontImageUrl沿用，保证C端数据不变)
CARD_FIELDS = {}
_FIELDS_CSV = os.path.join(config.DATA_DIR, "card_master_fields_v2.csv")
if not os.path.exists(_FIELDS_CSV):
    _FIELDS_CSV = os.path.join(config.DATA_DIR, "card_master_fields.csv")
if os.path.exists(_FIELDS_CSV):
    with open(_FIELDS_CSV, encoding="utf-8-sig") as _f:
        for _r in csv.DictReader(_f):
            CARD_FIELDS[_r["card_id"]] = {
                "card_name_ch":  _r.get("card_name_ch"),
                "pg_label":      _r.get("pg_label"),
                "year":          _r.get("year"),
                "card_no":       _r.get("card_no"),
                "rarity":        _r.get("rarity"),
                "rarity_short":  _r.get("rarity_short"),
                "language":      _r.get("language"),
                "material_cn":   _r.get("material_cn"),
                "img_url":       _r.get("img_url"),
                "img_url_legacy": _r.get("img_url_legacy") or _r.get("img_url"),
            }
    print("[启动] 卡牌字段表: %d 张 (%s)" % (len(CARD_FIELDS), os.path.basename(_FIELDS_CSV)), flush=True)
else:
    print("[启动] ⚠ 未找到 card_master_fields_v2.csv / card_master_fields.csv，将回退双区 gallery meta", flush=True)

# 双区 meta 索引（fallback）
_ID2META = {}
for _m in MATCHER.metas:
    cid = _m.get("card_id")
    if cid and cid not in _ID2META:
        _ID2META[cid] = _m

HISTORY_JSON_PATH = os.path.join(config.DATA_DIR, "match_results_history.json")
print("[启动] 就绪，等待请求", flush=True)


def _public_item(result, front_url=None, back_url=None):
    """把 match_image 内部结果组装成 /match 对外契约（单卡 dict）。"""
    full_match = result.get("match") or {}
    return {
        "card_type": result.get("card_type") or 1,
        "card_id": full_match.get("card_id"),
        "attribute": result.get("element"),
        "front_image_url": front_url,
        "back_image_url": back_url,
        "matches": [{
            "rank": it.get("rank"),
            "card_id": it.get("card_id"),
            "match_rate": int(max(0, min(100, round(float(it.get("fusion", 0)) * 100)))),
            "card_name_ch": it.get("card_name_ch"),
            "pg_label": it.get("pg_label"),
            "year": it.get("year"),
            "card_no": it.get("card_no"),
            "rarity_short": it.get("rarity_short") or it.get("rarity"),
            "language": it.get("language"),
            "material_cn": it.get("material_cn") or "1",
            "image_url": it.get("img_url"),
        } for it in (result.get("top_k") or [])],
    }


def _upload_capp_to_minio(local_path):
    """C 端拍摄图落到 249 MinIO grading/capp_img_data，返回对外 URL；失败返回 None。"""
    if not local_path or not os.path.exists(local_path):
        return None
    cfg = getattr(config, "CAPP_MINIO", None)
    if not cfg:
        return None
    try:
        from minio import Minio
        from datetime import datetime
        client = Minio(
            cfg["endpoint"],
            access_key=cfg["access_key"],
            secret_key=cfg["secret_key"],
            secure=bool(cfg.get("secure")),
        )
        ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S_%f")
        ext = os.path.splitext(local_path)[1].lower() or ".jpg"
        if ext not in (".jpg", ".jpeg", ".png", ".webp", ".bmp"):
            ext = ".jpg"
        object_name = "%s/capp_front_%s%s" % (cfg["prefix"], ts, ext)
        ctype = "image/jpeg" if ext in (".jpg", ".jpeg") else "image/%s" % ext.lstrip(".")
        client.fput_object(cfg["bucket"], object_name, local_path, content_type=ctype)
        return "%s/%s/%s" % (cfg["public_base"].rstrip("/"), cfg["bucket"], object_name)
    except Exception:
        traceback.print_exc()
        return None


def _card_fields(card_id):
    """优先 card_master_fields_v2.csv；否则回退双区 gallery meta。"""
    row = CARD_FIELDS.get(card_id)
    if row:
        return dict(row)
    m = _ID2META.get(card_id) or {}
    return {
        "card_name_ch":  m.get("card_name_ch"),
        "pg_label":      m.get("pg_label"),
        "year":          m.get("year"),
        "card_no":       m.get("card_no"),
        "rarity":        None,
        "rarity_short":  None,
        "language":      m.get("language"),
        "material_cn":   None,
        "img_url":       None,
        "img_url_legacy": None,
    }


def _append_history(result):
    import json
    from datetime import datetime
    history = []
    if os.path.exists(HISTORY_JSON_PATH):
        try:
            with open(HISTORY_JSON_PATH, "r", encoding="utf-8") as f:
                history = json.load(f)
            if not isinstance(history, list):
                history = []
        except Exception:
            history = []
    result = dict(result)
    result["timestamp"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    history.append(result)
    with open(HISTORY_JSON_PATH, "w", encoding="utf-8") as f:
        json.dump(history, f, ensure_ascii=False, indent=2)


def _resolve_local_path(src):
    """把 image_url / 上传路径解析为服务器本地文件路径。
    返回 (path, cleanup_path)：cleanup 非 None 表示本次临时文件，推理后删。"""
    if src.startswith(("http://", "https://")):
        img_path = image_downloader.download_card_image(src, config.QUERY_IMG_DIR)
        if not img_path:
            raise RuntimeError("图片下载失败（公网 URL 访问不到）: " + src)
        return img_path, img_path
    path = src[7:] if src.startswith("file://") else src
    if not os.path.isabs(path):
        path = os.path.join(config.QUERY_IMG_DIR, path)
    if not os.path.exists(path):
        raise RuntimeError("本地文件不存在（注意是服务器上的路径，不是本机路径）: " + path)
    return path, None


def _internal_from_query(result_q, element):
    """query()/query_all() 单卡结果 + element → 内部 result（历史日志 + _public_item 共用）。"""
    predicted = result_q["predicted_card_id"]
    fusion = float(result_q["fusion_score"])
    upper = float(result_q["upper_sim"])
    lower = float(result_q["lower_sim"])

    topk = []
    for rank, item in enumerate(result_q.get("top_k") or [], 1):
        cid = item["card_id"]
        row = _card_fields(cid)
        row["card_id"] = cid
        row["rank"] = rank
        row["similarity"] = round(float(item.get("fusion", 0)), 4)
        row["fusion"] = round(float(item.get("fusion", 0)), 4)
        row["upper_sim"] = round(float(item.get("upper", 0)), 4)
        row["lower_sim"] = round(float(item.get("lower", 0)), 4)
        topk.append(row)

    match_row = _card_fields(predicted)
    match_row["card_id"] = predicted
    match_row["similarity"] = round(fusion, 4)
    match_row["fusion"] = round(fusion, 4)
    match_row["upper_sim"] = round(upper, 4)
    match_row["lower_sim"] = round(lower, 4)
    match_row["above_threshold"] = fusion >= config.SIMILARITY_THRESHOLD

    return {
        "predicted_card_id": predicted,
        "similarity": round(fusion, 4),
        "fusion": round(fusion, 4),
        "upper_sim": round(upper, 4),
        "lower_sim": round(lower, 4),
        "above_threshold": fusion >= config.SIMILARITY_THRESHOLD,
        "card_type": result_q.get("card_type"),
        "element": element,
        "match": match_row,
        "top_k": topk,
        "model": "card_seg_v0904",
        "alpha": MATCHER.alpha,
        "top_k_recall": MATCHER.top_k_recall,
    }


def match_image_all(src, top_k=5):
    """多卡：YOLO 一次检出全部实例，按阅读顺序各自检索。未检出返回 []。
    返回 (items, img_path, dl_cleanup)；调用方负责上传 MinIO 后再删临时文件。"""
    img_path, dl_cleanup = _resolve_local_path(src)
    results_q = MATCHER.query_all(img_path, top_n=top_k)
    items = []
    for result_q in results_q:
        crop = result_q.get("crop")
        element = ELEM.predict(crop) if crop is not None else None
        result = _internal_from_query(result_q, element)
        try:
            _append_history(result)
        except Exception:
            pass
        items.append(result)
    return items, img_path, dl_cleanup


def match_image(src, top_k=5, cleanup_path=None):
    """图片(URL/服务器路径/上传临时文件) → 双区级联预测 card_id + 字段。
    language/angle 参数在双区级联入库链路里不作为检索前置（入库用召回后语种消歧），
    本 HTTP 服务保持与入库同源：直接双区检索。"""
    img_path, dl_cleanup = _resolve_local_path(src)
    to_delete = cleanup_path or dl_cleanup
    try:
        # DualCascadeMatcher 内部：裁卡 → letterbox → 上下半区特征 → 召回+重排
        # 它的 query() 读文件路径；我们已落到本地文件
        result_q = MATCHER.query(img_path, top_n=top_k, want_crop=True)
        if result_q is None:
            raise RuntimeError("未检出卡牌区域或特征提取失败")

        # 元素属性检测（用裁卡图跑 element 模型）
        crop = result_q.get("crop")
        element = ELEM.predict(crop) if crop is not None else None

        result = _internal_from_query(result_q, element)
        try:
            _append_history(result)
        except Exception:
            pass
        return result
    finally:
        if to_delete and os.path.exists(to_delete):
            try:
                os.remove(to_delete)
            except Exception:
                pass


def create_app():
    from flask import Flask, request, jsonify
    app = Flask(__name__)
    # 按插入顺序输出字段（/match 的字段顺序即指令文档示例顺序）+ 中文直出
    app.json.sort_keys = False
    app.json.ensure_ascii = False

    @app.get("/health")
    def health():
        return jsonify({
            "status": "ok",
            "gallery_size": len(MATCHER.card_ids),
            "model": "card_seg_v0904",
            "feature_dim": 1024,
            "alpha": MATCHER.alpha,
            "top_k_recall": MATCHER.top_k_recall,
        })

    def _resolve_src():
        """解析请求里的图片来源。返回 (src, cleanup, top_k, err)。
        err 非 None 时为 ({...}, http_code)。"""
        data = request.get_json(silent=True) or {}
        try:
            top_k = int(data.get("top_k", request.args.get("top_k", request.form.get("top_k", config.TOP_K))))
        except (TypeError, ValueError):
            top_k = config.TOP_K

        src = None
        cleanup = None
        up = request.files.get("image")
        if up and up.filename:
            import hashlib
            import re as _re
            blob = up.read()
            if not blob:
                return None, None, top_k, ({"error": "上传文件为空"}, 400)
            config.ensure_dirs()
            # 保留上传方原始文件名（如树莓派的 raspi_front_<时间戳>.jpg，全链路命名统一）；
            # 名字不合法（缺名/带路径/非法字符）时回退 md5 命名。请求图均为临时文件，推理后即删。
            raw = os.path.basename(up.filename.replace("\\", "/")).strip()
            if raw and _re.fullmatch(r"[A-Za-z0-9._\-（）()一-鿿]+", raw) and raw.lower().endswith(
                    (".jpg", ".jpeg", ".png", ".webp", ".bmp")):
                fname = raw
            else:
                fname = hashlib.md5(blob).hexdigest() + ".jpg"
            dst = os.path.join(config.QUERY_IMG_DIR, fname)
            with open(dst, "wb") as fo:
                fo.write(blob)
            src = dst
            cleanup = dst
        if src is None:
            # 后台识别源：拍摄图正面 cos 地址（树莓派直传 front_image_url）；
            # 兼容旧参数 image_url（小程序 /match_fields 仍用它，不带 front_image_url 不受影响）
            src = (data.get("front_image_url") or request.args.get("front_image_url")
                   or request.form.get("front_image_url")
                   or data.get("image_url") or request.args.get("image_url")
                   or request.form.get("image_url"))
        if not src:
            return None, None, top_k, ({"error": "缺少图片：请传 front_image_url（拍摄图正面）或 image_url，或上传 image 文件"}, 400)
        return src, cleanup, top_k, None

    def _resolve_and_run():
        src, cleanup, top_k, err = _resolve_src()
        if err:
            return None, err
        try:
            result = match_image(src, top_k=top_k, cleanup_path=cleanup)
            return result, None
        except Exception as e:
            traceback.print_exc()
            return None, ({"error": str(e)}, 500)

    @app.post("/match")
    @app.get("/match")
    def match():
        """后台接口（2026-08-28 二次定稿，字段契约见文件头 docstring；C 端 /match_fields 不在此列）。

        识别输入 = front_image_url（拍摄图正面 cos 地址，树莓派直传）；back_image_url 不参与
        识别、仅回显。识别得出 card_type/card_id/attribute，其余字段经 card_id 从
        cards_master_v2 导出的字段表查得。"""
        # front/back 参数三种来源都收（GET query / POST form / JSON body）
        data = request.get_json(silent=True) or {}
        front_url = (data.get("front_image_url") or request.args.get("front_image_url")
                     or request.form.get("front_image_url"))
        back_url = (data.get("back_image_url") or request.args.get("back_image_url")
                    or request.form.get("back_image_url"))
        result, err = _resolve_and_run()
        if err:
            return jsonify(err[0]), err[1]
        return jsonify(_public_item(result, front_url=front_url, back_url=back_url))

    @app.post("/match_fields")
    @app.get("/match_fields")
    def match_fields():
        """小程序接口（2026-09-02）：顶层裸数组，每项结构与 /match 完全一致。
        一张拍照图可能有多张卡，按阅读顺序依次输出；未检出返回 []。
        C 端拍摄图落到 249 MinIO grading/capp_img_data，front_image_url 回真地址，
        back_image_url 恒为 null。传 URL 入参时 front_image_url 原样回显（不再二次上传）。"""
        src, cleanup, top_k, err = _resolve_src()
        if err:
            return jsonify(err[0]), err[1]
        data = request.get_json(silent=True) or {}
        passed_url = (data.get("front_image_url") or request.args.get("front_image_url")
                      or request.form.get("front_image_url")
                      or data.get("image_url") or request.args.get("image_url")
                      or request.form.get("image_url"))
        is_http = isinstance(passed_url, str) and passed_url.startswith(("http://", "https://"))
        img_path = None
        dl_cleanup = None
        try:
            items, img_path, dl_cleanup = match_image_all(src, top_k=top_k)
            if is_http:
                front_url = passed_url
            else:
                front_url = _upload_capp_to_minio(img_path)
            resp = [_public_item(it, front_url=front_url, back_url=None) for it in items]
            return jsonify(resp)
        except Exception as e:
            traceback.print_exc()
            return jsonify({"error": str(e)}), 500
        finally:
            for p in (cleanup, dl_cleanup):
                if p and os.path.exists(p):
                    try:
                        os.remove(p)
                    except Exception:
                        pass

    return app


if __name__ == "__main__":
    p = argparse.ArgumentParser(description="卡牌匹配 HTTP 服务（双区级联）")
    p.add_argument("--host", default="0.0.0.0")
    p.add_argument("--port", type=int, default=8000)
    args = p.parse_args()
    create_app().run(host=args.host, port=args.port, threaded=True)