# -*- coding: utf-8 -*- """ラーメンスープ評価AI — キュー処理付きAPIサーバー（VPS常駐用 / 標準ライブラリのみ）。 複数人から動画が同時に送られても、受付だけ即座に行い、推論は1件ずつ順番に処理する。 （同時に複数の推論を走らせるとメモリ8GB環境ではOOMの恐れがあるため、ワーカーは1本固定） 起動: python api_server.py # 127.0.0.1:8000 python api_server.py --host 0.0.0.0 # 直接公開（要ファイアウォール/認証） 認証（推奨）: 環境変数 RAMEN_API_TOKEN を設定すると、全 /api/* リクエストに ヘッダ X-API-Key: <トークン> を要求する。未設定なら認証なし （nginxのBasic認証やSSHトンネル運用を想定）。 APIの使い方（クライアント例）: # 1) 動画を投げる（即時に job_id が返る） curl -X POST "http://HOST:8000/api/jobs?name=123.MOV&slot=開店前&brix=12" \ -H "X-API-Key: $TOKEN" --data-binary @123.MOV -> {"job_id": "j0001", "status": "queued", "queue_position": 2} # 2) 結果をポーリング（status が done / error / 要確認 になるまで数秒ごと） curl "http://HOST:8000/api/jobs/j0001" -H "X-API-Key: $TOKEN" -> {"job_id":"j0001","status":"done","result":{ total_score, total_grade, ... }} # 3) 稼働確認 curl "http://HOST:8000/api/health" 必要環境: ffmpeg(PATH) / BytePlus APIキー(secrets/byteplus.key か ARK_API_KEY) / RAMEN_SUITE_DIR（未設定ならパッケージ同梱 upstream_suite を使用） """ from __future__ import annotations import argparse import hmac import json import os import queue import threading import time import traceback import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path HERE = Path(__file__).resolve().parent UPLOAD_DIR = HERE / "_uploads" JOBS_DIR = HERE / "_jobs" UPLOAD_DIR.mkdir(exist_ok=True) JOBS_DIR.mkdir(exist_ok=True) HISTORY_PATH = JOBS_DIR / "history.jsonl" # 処理済みジョブの永続ログ（1行1件） MAX_UPLOAD_BYTES = 500 * 1024 * 1024 # 動画1本の上限 500MB MAX_JOBS_IN_MEMORY = 500 # メモリ上に保持する直近ジョブ数 # 上流AIが書き出すフレーム/切り出し画像の置き場。削除しないとディスクが際限なく増えるため、 # ジョブ完了のたびに保持期間を過ぎたものを自動削除する（直近分は検証用に残す） CROPS_DIR = HERE.parent / "local_batch_api" / "resolver_crops" CROPS_KEEP_DAYS = 7 def _cleanup_old_crops() -> None: cutoff = time.time() - CROPS_KEEP_DAYS * 86400 try: for f in CROPS_DIR.iterdir(): if f.is_file() and f.stat().st_mtime < cutoff: f.unlink(missing_ok=True) except OSError: pass API_TOKEN = os.environ.get("RAMEN_API_TOKEN", "").strip() # ------------------------------------------------------------------ # ジョブ管理（受付キュー ＋ 単一ワーカーで逐次推論） # ------------------------------------------------------------------ _jobs: dict[str, dict] = {} _jobs_order: list[str] = [] _jobs_lock = threading.Lock() _task_q: "queue.Queue[str]" = queue.Queue() _id_counter = 0 _PIPE = None _PIPE_LOCK = threading.Lock() def get_pipeline(): """重いパイプラインは初回ジョブ時に一度だけ読み込む。""" global _PIPE with _PIPE_LOCK: if _PIPE is None: import pipeline_full _PIPE = pipeline_full.FullPipeline() return _PIPE def _new_job_id() -> str: global _id_counter _id_counter += 1 return f"j{_id_counter:04d}_{int(time.time())}" def submit_job(video_path: Path, name: str, brix: str, slot: str) -> dict: job_id = _new_job_id() job = { "job_id": job_id, "status": "queued", # queued -> processing -> done / error "video_name": name, "brix": brix, "slot": slot, "submitted_at": time.strftime("%Y-%m-%d %H:%M:%S"), "started_at": None, "finished_at": None, "result": None, "error": None, "_video_path": str(video_path), } with _jobs_lock: _jobs[job_id] = job _jobs_order.append(job_id) # 古いジョブをメモリから間引く（履歴はhistory.jsonlに残る）。 # 処理中は間引かない。queuedのまま間引く場合は動画ファイルも削除（ディスクリーク防止） while len(_jobs_order) > MAX_JOBS_IN_MEMORY: old_id = _jobs_order[0] old = _jobs.get(old_id) if old is not None and old.get("status") == "processing": break _jobs_order.pop(0) _jobs.pop(old_id, None) if old is not None and old.get("status") == "queued": try: Path(old["_video_path"]).unlink(missing_ok=True) except OSError: pass _task_q.put(job_id) return job def queue_position(job_id: str) -> int: """待ち順（0=処理中/次、1=1件待ち…）。done/errorは-1。""" with _jobs_lock: waiting = [j for j in _jobs_order if _jobs.get(j, {}).get("status") == "queued"] try: return waiting.index(job_id) + (1 if _current_processing() else 0) except ValueError: return -1 def _current_processing() -> str | None: with _jobs_lock: for j in reversed(_jobs_order): if _jobs.get(j, {}).get("status") == "processing": return j return None def public_view(job: dict, with_result: bool = True) -> dict: out = {k: v for k, v in job.items() if not k.startswith("_")} if not with_result: out.pop("result", None) if job["status"] == "queued": out["queue_position"] = queue_position(job["job_id"]) return out def _append_history(job: dict) -> None: try: with HISTORY_PATH.open("a", encoding="utf-8") as fh: fh.write(json.dumps(public_view(job), ensure_ascii=False) + "\n") except OSError: pass def worker_loop() -> None: """1本だけのワーカー。キューから取り出して順番に推論する。""" while True: job_id = _task_q.get() with _jobs_lock: job = _jobs.get(job_id) if job is None: continue job["status"] = "processing" job["started_at"] = time.strftime("%Y-%m-%d %H:%M:%S") video = Path(job["_video_path"]) try: pipe = get_pipeline() result = pipe.run(video, brix=job["brix"] or None, slot=job["slot"] or None) job["result"] = result job["status"] = "done" except Exception as e: traceback.print_exc() job["error"] = str(e) job["status"] = "error" finally: job["finished_at"] = time.strftime("%Y-%m-%d %H:%M:%S") try: video.unlink() # 動画は処理後に削除（ディスクを溜めない） except OSError: pass _cleanup_old_crops() # 古いフレーム/切り出し画像も掃除 _append_history(job) _task_q.task_done() # ------------------------------------------------------------------ # 画面（ジョブ投入 → ポーリング表示） # ------------------------------------------------------------------ SLOTS = ["開店前", "17時〜19時"] SLOT_OPTS = "".join(f'' for s in SLOTS) PAGE = """ ラーメンスープ評価AI

🍜 ラーメンスープ評価AI

動画は受付後に順番待ちキューへ入り、1件ずつ処理されます（他の人と同時でもOK）。

""" # ------------------------------------------------------------------ # HTTPハンドラ # ------------------------------------------------------------------ class Handler(BaseHTTPRequestHandler): def _send(self, code, body, ctype="text/html; charset=utf-8"): data = body.encode("utf-8") if isinstance(body, str) else body self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _json(self, code, obj): self._send(code, json.dumps(obj, ensure_ascii=False), "application/json; charset=utf-8") def _authorized(self) -> bool: """RAMEN_API_TOKEN が設定されている場合のみ X-API-Key を検査する。""" if not API_TOKEN: return True given = self.headers.get("X-API-Key", "") if not given: auth = self.headers.get("Authorization", "") if auth.startswith("Bearer "): given = auth[len("Bearer "):] return hmac.compare_digest(given.strip(), API_TOKEN) def do_GET(self): path = urllib.parse.urlparse(self.path).path if path in ("/", "/index.html"): self._send(200, PAGE.replace("__SLOT__", SLOT_OPTS)) return if path == "/api/health": with _jobs_lock: queued = sum(1 for j in _jobs.values() if j["status"] == "queued") processing = sum(1 for j in _jobs.values() if j["status"] == "processing") self._json(200, {"status": "ok", "queued": queued, "processing": processing, "model_loaded": _PIPE is not None, "auth_required": bool(API_TOKEN)}) return if path.startswith("/api/"): if not self._authorized(): self._json(401, {"error": "X-API-Key が不正です"}); return if path == "/api/jobs": with _jobs_lock: items = [public_view(_jobs[j], with_result=False) for j in _jobs_order[-50:]] self._json(200, {"jobs": items}) return if path.startswith("/api/jobs/"): job_id = path[len("/api/jobs/"):] with _jobs_lock: job = _jobs.get(job_id) if job is None: self._json(404, {"error": f"job {job_id} が見つかりません"}); return self._json(200, public_view(job)) return self._send(404, "not found") def do_POST(self): path = urllib.parse.urlparse(self.path).path if path != "/api/jobs": self._send(404, "not found"); return if not self._authorized(): self._json(401, {"error": "X-API-Key が不正です"}); return try: qs = urllib.parse.urlparse(self.path).query q = urllib.parse.parse_qs(qs) name = (q.get("name", ["upload.mov"])[0]) or "upload.mov" brix = (q.get("brix", [""])[0]).strip() slot = (q.get("slot", [""])[0]).strip() ext = Path(name).suffix or ".mov" n = int(self.headers.get("Content-Length", 0)) if n <= 0: self._json(400, {"error": "空のアップロードです"}); return if n > MAX_UPLOAD_BYTES: self._json(413, {"error": f"動画が大きすぎます（上限 {MAX_UPLOAD_BYTES // (1024*1024)}MB）"}); return # 受信してディスクへ（同時アップロード衝突回避のためスレッドID＋時刻） stamp = f"{threading.get_ident()}_{int(time.time()*1000)}" tmp = UPLOAD_DIR / f"upload_{stamp}{ext}" remaining = n with tmp.open("wb") as fh: while remaining > 0: chunk = self.rfile.read(min(1024 * 1024, remaining)) if not chunk: break fh.write(chunk) remaining -= len(chunk) if remaining > 0: # 途中切断: 不完全な動画をキューに入れない tmp.unlink(missing_ok=True) self._json(400, {"error": f"アップロードが途中で切れました（残り {remaining} bytes）"}) return job = submit_job(tmp, name=name, brix=brix, slot=slot) self._json(202, public_view(job, with_result=False)) except Exception as e: traceback.print_exc() self._json(500, {"error": str(e)}) def log_message(self, *a): pass def main(): ap = argparse.ArgumentParser() ap.add_argument("--port", type=int, default=8000) ap.add_argument("--host", default="127.0.0.1", help="待受アドレス。VPSでリバースプロキシ経由なら 127.0.0.1 のまま。" "直接公開する場合のみ 0.0.0.0（要ファイアウォール/API_TOKEN）。") args = ap.parse_args() # 起動時の環境チェック（不足があってもサーバーは起動するが、処理時に必ず失敗するため強く警告） try: from batch_test import preflight issues = preflight() for x in issues: print(("[環境NG] " if not x.startswith("[警告]") else "[警告] ") + x) if any(not x.startswith("[警告]") for x in issues): print("[環境NG] ↑ このままでは全ジョブがエラーになります。解消してから再起動してください。") elif not issues: print("事前チェック: すべてOK（ffmpeg・モデル・プロンプト・キー）") except Exception as e: print(f"事前チェックを実行できませんでした: {e}") threading.Thread(target=worker_loop, daemon=True, name="inference-worker").start() print(f"ラーメンスープ評価AI APIサーバー: http://{args.host}:{args.port}") print(f" 認証: {'X-API-Key 必須' if API_TOKEN else 'なし（RAMEN_API_TOKEN 未設定）'}") print(" POST /api/jobs?name=&brix=&slot= (動画バイナリ) -> job_id") print(" GET /api/jobs/ -> 状態/結果") print(" GET /api/health -> 稼働確認") ThreadingHTTPServer((args.host, args.port), Handler).serve_forever() if __name__ == "__main__": main()