PokemonCardSearch/code/serve_card_match_v2.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
卡牌匹配 HTTP 服务 v2（8020 生产接口 · 排队 + 缓存 + 多 worker 分卡）

定位：替代 8000 serve_card_match.py 的小程序多卡链路，正式上线版。
  - 同步等待 + 服务内部排队：请求入队（上限 V2_QUEUE_SIZE，满即 429+Retry-After），
    N 个推理 worker 消费（V2_WORKERS=0 时自动=可见 GPU 数，249 双 V100 即双 worker 分卡），
    handler 等结果带超时（V2_WAIT_TIMEOUT，超时 504，队头任务会被取消跳过）。
  - 内容 md5 缓存（LRU+TTL）：URL 入参同样按下载后的内容算 key，同一张图重复识别零推理；
    缓存值不含 image_url，返回前按本次请求盖章（URL/上传/同内容异源全部命中）。
  - 近距 mask 合并：modules/yolo_detector.detect_and_crop_all 内实现（8000 多卡路径共用），
    阈值 CARD_MERGE_DIST_PX（默认 50px，中心距，保留 conf 高者）。
  - attribute 不跑 YOLO element 模型：按候选 card_id 查字段表 CSV（列现为空 → null，
    业务侧把 PG cards_master_v2.attribute 填好后重导 CSV 即自动生效）。
  - 历史日志：data/match_v2_history.jsonl 逐行追加（弃 8000 全量重写单 JSON 的竞态写法）。

端点（响应契约 2026-09-03 定稿，顶层裸数组，每项）：
  {"card_type":"pokemon", "card_id":"0119993",
   "image_url":"http://192.168.77.249:9000/grading/capp_img_data/capp_front_<ts>.jpg",
   "matches":[{rank, card_id, match_rate, card_name_ch, pg_label, year, card_no,
               rarity_short, language, attribute, material_cn, image_url} × 5]}
  top_k 固定 5（入参被忽略）；未检出返回 []；入参缺失 400；队列满 429；推理超时 504；
  异常 500 {"error":...}。请求可带 refresh=1 击穿缓存强制重算（测试/强制刷新用）。

启动（249，CVD 掩码必须 PCI_BUS_ID + CVD 两变量一起写，GTX1060=PCI index0 禁用）：
  # 直跑（Flask 内置 server）
  CUDA_DEVICE_ORDER=PCI_BUS_ID CUDA_VISIBLE_DEVICES=1,2 setsid nohup \
    ~/miniconda3/envs/pytorch/bin/python serve_card_match_v2.py --port 8020 \
    </dev/null > logs/serve_8020.log 2>&1 &
  # gunicorn（推荐：worker 崩溃自愈 + 优雅退出；禁 --preload，CUDA 不能 fork）
  CUDA_DEVICE_ORDER=PCI_BUS_ID CUDA_VISIBLE_DEVICES=1,2 setsid nohup \
    ~/miniconda3/envs/pytorch/bin/gunicorn -w 1 --threads 8 -b 0.0.0.0:8020 \
    --timeout 300 --graceful-timeout 30 serve_card_match_v2:app \
    </dev/null > logs/serve_8020.log 2>&1 &

调用：
  curl -X POST http://192.168.77.249:8020/match_fields -F "image=@C:/某张卡.jpg"
  curl -X POST http://192.168.77.249:8020/match_fields -H 'Content-Type: application/json' \
       -d '{"image_url":"<图片URL>"}'
"""
import os
import sys
import csv
import json
import time
import queue
import signal
import argparse
import threading
import traceback
import copy

sys.stdout.reconfigure(encoding="utf-8")
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
_SCRIPTS = os.path.join(os.path.dirname(os.path.abspath(__file__)), "scripts")
if _SCRIPTS not in sys.path:
    sys.path.insert(0, _SCRIPTS)

import config
from modules.capp_upload import upload_to_capp_minio, save_upload_to_tmp, file_md5
from cascade_match import DualCascadeMatcher

TOP_K_FIXED = config.TOP_K                      # 固定 5
QUEUE_MAX = max(1, int(getattr(config, "V2_QUEUE_SIZE", 32)))
WAIT_TIMEOUT = float(getattr(config, "V2_WAIT_TIMEOUT", 60))
CACHE_MAX = max(1, int(getattr(config, "V2_CACHE_MAX", 256)))
CACHE_TTL = float(getattr(config, "V2_CACHE_TTL", 900))
MAX_CONTENT_MB = int(getattr(config, "V2_MAX_CONTENT_MB", 20))
HISTORY_JSONL = os.path.join(config.DATA_DIR, "match_v2_history.jsonl")

_T0 = time.time()
_SHUTDOWN = threading.Event()
_LOG_LOCK = threading.Lock()
_STATS_LOCK = threading.Lock()
_STATS = {"submitted": 0, "done": 0, "failed": 0, "queue_full": 0,
          "timeouts": 0, "cache_hits": 0, "cache_misses": 0}


# ==================== 启动时一次性加载（gunicorn 兼容：模块级执行） ====================
def _auto_workers():
    """worker 数：V2_WORKERS>0 用配置；0=可见 GPU 数（至少 1，无 GPU 走 CPU 单 worker）。"""
    w = int(getattr(config, "V2_WORKERS", 0) or 0)
    if w > 0:
        return w
    try:
        import torch
        return max(1, torch.cuda.device_count()) if torch.cuda.is_available() else 1
    except Exception:
        return 1


def _gpu_count():
    try:
        import torch
        return torch.cuda.device_count() if torch.cuda.is_available() else 0
    except Exception:
        return 0


N_WORKERS = _auto_workers()
_GPU_COUNT = _gpu_count()
print(f"[启动] 推理 worker 数 = {N_WORKERS}（可见 GPU = {_GPU_COUNT}，"
      f"CUDA_VISIBLE_DEVICES={os.environ.get('CUDA_VISIBLE_DEVICES', '<未设>')}）", flush=True)

# 每 worker 一套独立 matcher（YOLO 分割 + 双区 DINOv2 + 检索后端连接）：
# 多卡时各绑一张卡（torch.device 输入不会被 vendored ultralytics 改写 CVD），
# milvus 连接各自独立 alias，避免多线程共享默认连接。
MATCHERS = []
for _i in range(N_WORKERS):
    _dev = None
    if _GPU_COUNT:
        _dev = f"cuda:{_i % _GPU_COUNT}"
        import torch as _torch
        _dev = _torch.device(_dev)
    print(f"[启动] worker{_i} 初始化 matcher（device={_dev}）...", flush=True)
    MATCHERS.append(DualCascadeMatcher(
        alpha=config.CASCADE_ALPHA,
        top_k_recall=config.CASCADE_TOP_K_RECALL,
        device=_dev,
        detector_device=_dev,
        milvus_alias=f"v2w{_i}",
        verbose=True,
    ))

# 字段表（92,333 行） DictReader 兼容无 attribute 列的旧 CSV（.get → None）
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"]] = _r
    print(f"[启动] 卡牌字段表: {len(CARD_FIELDS)} 张 ({os.path.basename(_FIELDS_CSV)})", flush=True)
else:
    print("[启动] ⚠ 未找到 card_master_fields_v2.csv，matches 字段将回退 gallery meta", flush=True)

# gallery meta 兜底索引
_ID2META = {}
for _m in (MATCHERS[0].metas or []):
    _cid = (_m or {}).get("card_id")
    if _cid and _cid not in _ID2META:
        _ID2META[_cid] = _m

config.ensure_dirs()
os.makedirs(os.path.dirname(HISTORY_JSONL), exist_ok=True)


def _card_fields(card_id):
    """card_id → 字段表行（dict）；字段表缺失回退 gallery meta（空值输出 None）。"""
    row = CARD_FIELDS.get(card_id)
    if row:
        return 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,
            "attribute": None}


# ==================== 缓存（LRU + TTL，线程安全） ====================
class TTLCache:
    """key=内容 md5 → 响应（不含 image_url，返回前盖章）。命中搬 LRU 尾，过期即删。"""

    def __init__(self, max_size, ttl):
        self.max_size = max_size
        self.ttl = ttl
        self._d = {}          # key -> (expire_ts, value)
        self._lock = threading.Lock()

    def get(self, key):
        now = time.time()
        with self._lock:
            item = self._d.get(key)
            if item is None:
                return None
            expire, value = item
            if now >= expire:
                self._d.pop(key, None)
                return None
            self._d[key] = self._d.pop(key)   # move to end (py3.7+ dict 有序)
            return value

    def put(self, key, value):
        with self._lock:
            self._d[key] = (time.time() + self.ttl, value)
            while len(self._d) > self.max_size:
                self._d.pop(next(iter(self._d)))

    def info(self):
        with self._lock:
            return {"size": len(self._d), "max": self.max_size, "ttl": self.ttl}


_CACHE = TTLCache(CACHE_MAX, CACHE_TTL)


# ==================== 任务队列 + worker ====================
class _Job:
    __slots__ = ("src", "cleanup", "result", "error", "done", "cancelled",
                 "worker_id", "elapsed_ms", "md5")

    def __init__(self, src, cleanup, md5hex):
        self.src = src
        self.cleanup = cleanup          # 临时文件删除责任归 worker（429 未入队时 handler 自删）
        self.md5 = md5hex
        self.result = None
        self.error = None
        self.done = threading.Event()
        self.cancelled = False
        self.worker_id = None
        self.elapsed_ms = None


_QUEUE = queue.Queue(maxsize=QUEUE_MAX)


def _run_pipeline(matcher, src):
    """本地图路径 → 多卡识别 → 基础响应（不含本次请求的 image_url，盖章在 handler）。"""
    results = matcher.query_all(src, top_n=TOP_K_FIXED)
    items = []
    for r in results:
        predicted = r.get("predicted_card_id")
        matches = []
        for rank, t in enumerate(r.get("top_k") or [], 1):
            row = _card_fields(t.get("card_id"))
            matches.append({
                "rank": rank,
                "card_id": t.get("card_id"),
                "match_rate": int(max(0, min(100, round(float(t.get("fusion", 0)) * 100)))),
                "card_name_ch": row.get("card_name_ch"),
                "pg_label": row.get("pg_label"),
                "year": row.get("year"),
                "card_no": row.get("card_no"),
                "rarity_short": row.get("rarity_short") or row.get("rarity"),
                "language": row.get("language"),
                "attribute": row.get("attribute") or None,   # 列现为空 → null；填库重导后自动生效
                "material_cn": row.get("material_cn") or "1",
                "image_url": row.get("img_url") or row.get("img_url_legacy") or None,
            })
        items.append({
            "card_type": r.get("card_type"),
            "card_id": predicted,
            "image_url": None,
            "matches": matches,
        })
    return items


def _worker_loop(matcher, wid):
    while not _SHUTDOWN.is_set():
        try:
            job = _QUEUE.get(timeout=0.5)
        except queue.Empty:
            continue
        try:
            if job.cancelled:           # handler 已超时放弃：跳过推理，仅清理
                with _STATS_LOCK:
                    _STATS["done"] += 1
                return_present(job, skipped=True)
                continue
            t0 = time.perf_counter()
            try:
                job.result = _run_pipeline(matcher, job.src)
                with _STATS_LOCK:
                    _STATS["done"] += 1
            except Exception as e:
                job.error = e
                traceback.print_exc()
                with _STATS_LOCK:
                    _STATS["failed"] += 1
            finally:
                job.worker_id = wid
                job.elapsed_ms = int((time.perf_counter() - t0) * 1000)
                return_present(job)
        finally:
            _QUEUE.task_done()


def return_present(job, skipped=False):
    """收尾：置 done、写 jsonl 历史、删临时文件。"""
    job.done.set()
    rec = {
        "ts": time.strftime("%Y-%m-%d %H:%M:%S"),
        "worker": job.worker_id,
        "md5": job.md5,
        "skipped": bool(skipped),
        "elapsed_ms": job.elapsed_ms,
        "n_cards": len(job.result) if job.result is not None else None,
        "card_ids": [it.get("card_id") for it in (job.result or [])],
        "error": (str(job.error)[:300] if job.error else None),
    }
    try:
        with _LOG_LOCK:
            with open(HISTORY_JSONL, "a", encoding="utf-8") as f:
                f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    except Exception:
        pass
    if job.cleanup and os.path.exists(job.cleanup):
        try:
            os.remove(job.cleanup)
        except OSError:
            pass


def _start_workers():
    for i, m in enumerate(MATCHERS):
        t = threading.Thread(target=_worker_loop, args=(m, i),
                             name=f"v2-worker{i}", daemon=True)
        t.start()
        print(f"[启动] worker{i} 线程已启动", flush=True)


_start_workers()


# ==================== HTTP ====================
def _stamp(items, image_url):
    """把本次请求的 image_url 盖进每个 item（deepcopy 防跨请求共享）。"""
    out = copy.deepcopy(items)
    for it in out:
        it["image_url"] = image_url
    return out


def _download_unique(url):
    """URL 图片下载到唯一临时路径（worker 推理后删）。
    不用 image_downloader.download_card_image：它按 URL 命名且"存在即复用"，
    并发同 URL 会在同一文件上互踩（8020 队列并发场景必炸），且其断点续传语义
    被建库批量逻辑依赖不能改。"""
    import requests as _rq
    last_err = None
    for _ in range(3):
        try:
            resp = _rq.get(url, timeout=15)
            if resp.status_code == 200 and resp.content:
                name = os.path.basename(url.split("?")[0].rstrip("/")) or ""
                return save_upload_to_tmp(resp.content, name, config.QUERY_IMG_DIR)
            last_err = "HTTP %s" % resp.status_code
        except Exception as e:
            last_err = str(e)
    print(f"[match_fields] 下载失败 url={url[:120]}: {last_err}", flush=True)
    return None


def create_app():
    from flask import Flask, request, jsonify
    app = Flask(__name__)
    app.json.sort_keys = False
    app.json.ensure_ascii = False
    app.config["MAX_CONTENT_LENGTH"] = MAX_CONTENT_MB * 1024 * 1024

    @app.errorhandler(413)
    def _too_large(_e):
        return jsonify({"error": f"上传文件超过 {MAX_CONTENT_MB}MB 上限"}), 413

    @app.get("/health")
    def health():
        m0 = MATCHERS[0]
        with _STATS_LOCK:
            stats = dict(_STATS)
        draining = _SHUTDOWN.is_set()
        return jsonify({
            "status": "draining" if draining else "ok",
            "service": "serve_card_match_v2",
            "model": (os.path.basename(config.CARD_SEG_MODEL_PATH)
                      if os.path.isdir(config.CARD_SEG_MODEL_PATH) else
                      os.path.basename(os.path.dirname(  # 路径直指 best.pt 时取上两级目录名
                          os.path.dirname(config.CARD_SEG_MODEL_PATH)))),
            "backend": m0.backend,
            "gallery_size": len(m0.card_ids),
            "workers": N_WORKERS,
            "gpus": {"count": _GPU_COUNT,
                     "visible": os.environ.get("CUDA_VISIBLE_DEVICES", "")},
            "merge": {"dist_px": getattr(m0.detector, "merge_dist_px", None),
                      "min_iou": getattr(m0.detector, "merge_min_iou", None)},
            "queue": {"max": QUEUE_MAX, "depth": _QUEUE.qsize(),
                      "unfinished": _QUEUE.unfinished_tasks},
            "cache": dict(_CACHE.info(), hits=stats["cache_hits"], misses=stats["cache_misses"]),
            "stats": {k: v for k, v in stats.items()},
            "top_k_fixed": TOP_K_FIXED,
            "uptime_s": round(time.time() - _T0, 1),
        })

    @app.post("/match_fields")
    @app.get("/match_fields")
    def match_fields():
        """多卡识别（8020 主端点，2026-09-03 契约）。响应顶层裸数组，
        每项 {card_type, card_id, image_url, matches[5]}。"""
        data = request.get_json(silent=True) or {}
        _refresh = str(data.get("refresh", request.args.get("refresh",
                                request.form.get("refresh", "0")))).lower() in ("1", "true", "yes")

        # ---- 来源解析：multipart 上传 / front_image_url / image_url（query/form/json）----
        cleanup = None
        image_url = None
        up = request.files.get("image")
        if up and up.filename:
            blob = up.read()
            if not blob:
                return jsonify({"error": "上传文件为空"}), 400
            config.ensure_dirs()
            cleanup = save_upload_to_tmp(blob, up.filename, config.QUERY_IMG_DIR)
        if cleanup is None:
            src_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"))
            if not src_url:
                return jsonify({"error": "缺少图片：请上传 image 文件，或传 front_image_url / image_url"}), 400
            if not str(src_url).startswith(("http://", "https://")):
                return jsonify({"error": "image_url 仅支持 http(s) 地址（需服务器可达）: " + str(src_url)[:200]}), 400
            config.ensure_dirs()
            path = _download_unique(src_url)
            if not path:
                return jsonify({"error": "图片下载失败（服务器访问不到该 URL）: " + str(src_url)[:200]}), 400
            cleanup = path
            image_url = src_url    # URL 入参原样回显，不二次上传（与 8000 /match_fields 一致）

        try:
            md5hex = file_md5(cleanup)
        except OSError as e:
            return jsonify({"error": f"读取图片失败: {e}"}), 400

        # ---- 缓存查询（上传场景的 MinIO 上传放在命中判断之后，命中时仍需上传拿回显地址）----
        if not _refresh:
            cached = _CACHE.get(md5hex)
            if cached is not None:
                with _STATS_LOCK:
                    _STATS["cache_hits"] += 1
                if image_url is None:
                    image_url = upload_to_capp_minio(cleanup)
                    if cleanup and os.path.exists(cleanup):
                        try:
                            os.remove(cleanup)
                        except OSError:
                            pass
                        cleanup = None
                return jsonify(_stamp(cached, image_url))
            with _STATS_LOCK:
                _STATS["cache_misses"] += 1

        if image_url is None:
            image_url = upload_to_capp_minio(cleanup)   # 失败 None，不阻塞识别主流程

        # ---- 入队 + 等待 ----
        job = _Job(cleanup, cleanup, md5hex)
        try:
            _QUEUE.put_nowait(job)
        except queue.Full:
            if cleanup and os.path.exists(cleanup):
                try:
                    os.remove(cleanup)
                except OSError:
                    pass
            with _STATS_LOCK:
                _STATS["queue_full"] += 1
            resp = jsonify({"error": "服务繁忙（推理队列已满），请稍后重试"})
            resp.status_code = 429
            resp.headers["Retry-After"] = "5"
            return resp
        with _STATS_LOCK:
            _STATS["submitted"] += 1
        print(f"[match_fields] md5={md5hex[:12]} src={'upload' if request.files.get('image') else 'url'} "
              f"qdepth={_QUEUE.qsize()}", flush=True)

        if not job.done.wait(WAIT_TIMEOUT):
            job.cancelled = True        # 仍在队列则 worker 取到即跳过；在跑则结果丢弃
            with _STATS_LOCK:
                _STATS["timeouts"] += 1
            return jsonify({"error": f"推理超时（>{WAIT_TIMEOUT:g}s），请稍后重试"}), 504
        if job.error is not None:
            return jsonify({"error": str(job.error) or "inference error"}), 500

        _CACHE.put(md5hex, job.result)
        return jsonify(_stamp(job.result, image_url))

    return app


app = create_app()


if __name__ == "__main__":
    def _graceful(signum, _frame):
        """直跑模式的优雅退出：停止收新任务，最多等 10s 排空队列。
        gunicorn 模式不注册（由 gunicorn master 管 SIGTERM/graceful-timeout）。"""
        print(f"[关闭] 收到信号 {signum}，draining ...", flush=True)
        _SHUTDOWN.set()
        deadline = time.time() + 10
        while _QUEUE.unfinished_tasks > 0 and time.time() < deadline:
            time.sleep(0.1)
        os._exit(0)

    signal.signal(signal.SIGTERM, _graceful)
    signal.signal(signal.SIGINT, _graceful)

    p = argparse.ArgumentParser(description="卡牌匹配 HTTP 服务 v2（8020 生产）")
    p.add_argument("--host", default="0.0.0.0")
    p.add_argument("--port", type=int, default=8020)
    args = p.parse_args()
    app.run(host=args.host, port=args.port, threaded=True)