# -*- coding: utf-8 -*- """Folder batch processor for ramen soup video evaluation. Pipeline: input folder CSV/videos -> best frame -> BytePlus Responses API -> SQLite -> JSONL/CSV dataset. """ from __future__ import annotations import argparse import base64 import csv import hashlib import json import os import queue import shutil import sqlite3 import subprocess import sys import tempfile import threading import time from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any import cv2 import numpy as np import requests import yaml ROOT = Path(__file__).resolve().parents[1] POLICY_PATH = ROOT / "APIテスト" / "豚山スープ画像評価プロンプト方針.md" CONFIG_PATH = Path(__file__).resolve().with_name("config.yaml") DEFAULT_CONFIG_PATH = Path(__file__).resolve().with_name("config.example.yaml") VIDEO_EXTENSIONS = {".mp4", ".mov", ".m4v", ".avi", ".mkv"} @dataclass class InputItem: video_path: Path csv_path: Path | None row_index: int | None metadata: dict[str, Any] @dataclass class ExtractedItem: item: InputItem video_hash: str frame_path: Path frame_metrics: dict[str, Any] @dataclass class FailedItem: item: InputItem video_hash: str error: str class RateLimiter: def __init__(self, requests_per_minute: float): self.interval = 60.0 / requests_per_minute if requests_per_minute > 0 else 0.0 self.lock = threading.Lock() self.next_allowed_at = 0.0 def wait(self) -> None: if self.interval <= 0: return with self.lock: now = time.monotonic() sleep_for = max(0.0, self.next_allowed_at - now) self.next_allowed_at = max(now, self.next_allowed_at) + self.interval if sleep_for > 0: time.sleep(sleep_for) def main() -> int: parser = argparse.ArgumentParser(description="Process ramen soup videos with VLM API.") parser.add_argument("--input", required=True, help="Folder containing videos and optional CSV.") parser.add_argument("--config", default=str(CONFIG_PATH), help="YAML config path.") parser.add_argument("--dry-run", action="store_true", help="Extract frames and record rows without API calls.") parser.add_argument("--limit", type=int, default=0, help="Maximum number of videos to process.") parser.add_argument("--force", action="store_true", help="Reprocess videos already present in DB.") args = parser.parse_args() config = load_config(Path(args.config)) setup_acceleration(config) input_dir = Path(args.input).expanduser().resolve() if not input_dir.exists() or not input_dir.is_dir(): raise SystemExit(f"Input folder not found: {input_dir}") output_dir = resolve_path(config["paths"]["output_dir"]) frames_dir = output_dir / "frames" output_dir.mkdir(parents=True, exist_ok=True) frames_dir.mkdir(parents=True, exist_ok=True) db_path = resolve_path(config["paths"]["db_path"]) db_path.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(db_path, check_same_thread=False) init_db(conn) prompt = build_prompt(POLICY_PATH) items = discover_items(input_dir) if args.limit > 0: items = items[: args.limit] print(f"input={input_dir}") print(f"items={len(items)} db={db_path}") print(f"opencl={cv2.ocl.useOpenCL()}") density_map = resolve_densities(items, config) processed = run_pipeline( conn=conn, items=items, config=config, prompt=prompt, frames_dir=frames_dir, dry_run=args.dry_run, force=args.force, density_map=density_map, ) export_datasets(conn, output_dir) conn.close() print(f"done processed={processed} output={output_dir}") return 0 def resolve_densities(items: list[InputItem], config: dict[str, Any]) -> dict[str, dict[str, Any]]: """LLM送信前に各動画の濃度を確定する条件分岐。 CSV等に濃度があればそのまま、無ければ画像モデルで推定（スープが見えなければ判定不可）。 config["density"]["enabled"] が False の場合は何もしない（従来動作）。 戻り値は video_path(str) -> 解決結果dict。 """ density_cfg = config.get("density", {}) or {} if not density_cfg.get("enabled", False): return {} try: from density_resolver import DensityResolver, density_in_metadata except Exception as exc: # 重い依存が無い環境では従来動作にフォールバック print(f"WARN density resolver unavailable: {exc}") return {} # (1)前処理キャッシュ（prepare_densities.py 生成）があれば再計算しない cache: dict[str, dict[str, Any]] = {} cache_cfg = density_cfg.get("cache_path") if cache_cfg: cache = load_density_cache(resolve_path(str(cache_cfg))) if cache: print(f"density cache: {len(cache)} 件読み込み") out: dict[str, dict[str, Any]] = {} resolver = None for item in items: res = cache.get(item.video_path.stem) if res is None: if resolver is None: # キャッシュに無い動画だけ実計算（遅延生成） resolver = DensityResolver( conf_keep=float(density_cfg.get("conf_keep", 0.50)), crop_threshold=float(density_cfg.get("crop_threshold", 0.10)), frames=int(density_cfg.get("frames", 16)), multi_min=int(density_cfg.get("multi_min", 4)), vote_min=float(density_cfg.get("vote_min", 0.70)), high_mean_conf=float(density_cfg.get("high_mean_conf", 0.72)), device_name=str(density_cfg.get("device", "auto")), ) res = resolver.resolve(item.video_path, item.metadata) out[str(item.video_path)] = res tag = res["source"] if res.get("status") == "ok" else res.get("status") print(f"DENSITY {item.video_path.name}: {res.get('value')} ({tag}/{res.get('confidence')}) {res.get('note')}") return out def load_density_cache(cache_path: Path) -> dict[str, dict[str, Any]]: """prepare_densities.py が出力した濃度キャッシュ(JSONL: 1行1動画)を stem->res で読む。""" if not cache_path.exists(): return {} out: dict[str, dict[str, Any]] = {} with cache_path.open("r", encoding="utf-8") as fh: for line in fh: line = line.strip() if not line: continue try: rec = json.loads(line) except json.JSONDecodeError: continue if rec.get("stem"): out[str(rec["stem"])] = rec return out def density_input_note(res: dict[str, Any] | None) -> str: """解決済み濃度を、LLMへ渡す『別入力』テキストに整形する。 精度最優先のため、LLMに数値を渡すのは - CSV提供値 (source=provided) - 画像モデル推定のうち高信頼 (confidence=high) のみ。中信頼以下は誤りリスクがあるため数値を渡さず「判定不可」とする （弱い推定値はDBの density_resolution に保持され、後段分析で参照できる）。 """ if not res: return "" lines: list[str] = [] # 濃度（別入力） if res["source"] == "provided" and res.get("value"): lines.append(f"【別入力】スープ濃度(計測/人手ラベル): {res['value']}") elif res["source"] == "model" and res.get("confidence") == "high" and res.get("value"): lines.append(f"【別入力】スープ濃度(画像モデル推定 / 信頼度:高): {res['value']}") else: lines.append("【別入力】スープ濃度: 判定不可（画像から高信頼に確認できないため数値なし）") # 可読性レイヤー（濃度の有無と独立）。スープ表面が読めない映像は全体評価を下げさせる。 if res.get("readable") is False: lines.append( "【撮影品質】湯気または構図によりスープ表面が確認できません。" "濃度値が分かっていても液体本体の見た目は判断できないため、" "assessable=false / retake_recommended=true とし、confidence を下げてください。" ) elif res.get("assessable") is False: lines.append( "【撮影品質】スープ表面の視認性が低めです。確信が持てない項目は confidence を下げ、" "必要に応じて retake_recommended=true としてください。" ) return "\n".join(lines) def decide_skip_llm(res: dict[str, Any] | None, config: dict[str, Any]) -> tuple[bool, str]: """不可読動画をLLMに送る前にスキップすべきか判定する。 action: - skip_if_steamy (既定): 不可読 かつ 湯気が多い(steam_mean>=閾値)ときのみスキップ。 湯気が薄いのに不可読＝RF-DETRの構図未対応(OOD)の疑いがあるためLLMに送る。 - always_skip : 不可読なら必ずスキップ。 - always_send / off : スキップしない（従来動作）。 可読(readable=True)な動画は常にLLMへ。 """ density_cfg = config.get("density", {}) or {} action = str(density_cfg.get("unreadable_action", "skip_if_steamy")) if not res or action in ("always_send", "off"): return False, "" if res.get("readable"): return False, "" # ここに来る = 不可読 if action == "always_skip": return True, f"不可読(全スキップ設定) steam_mean={res.get('steam_mean')}" steam_thr = float(density_cfg.get("steam_skip_threshold", 0.80)) steam_mean = float(res.get("steam_mean", 1.0)) if steam_mean >= steam_thr: return True, f"不可読かつ湯気多 steam_mean={steam_mean:.2f}>={steam_thr}" return False, "" # 不可読だが湯気薄 → OOD構図の疑い、LLMで第二判断 def synthetic_unassessable(res: dict[str, Any] | None, reason: str) -> dict[str, Any]: """LLMをスキップした不可読動画のための、評価スキーマ準拠の判定不可レコード。""" res = res or {} provided = res.get("value") if res.get("source") == "provided" else None return { "assessable": False, "retake_recommended": True, "photo_quality": "不良", "visual_density": {"value": "判定不可", "confidence": "低"}, "water_level": {"state": "判定不可"}, "oil_emulsification": {"state": "判定不可"}, "boiling_heat_state": "判定不可", "image_condition_score": 0, "soup_density_input": provided if provided else "判定不可", "summary": "可読性レイヤーによりスープ表面が確認できずLLM評価を見送り（判定不可）。", "skip_reason": reason, "evaluation_source": "readability_layer", } def run_pipeline( conn: sqlite3.Connection, items: list[InputItem], config: dict[str, Any], prompt: str, frames_dir: Path, dry_run: bool, force: bool, density_map: dict[str, dict[str, Any]] | None = None, ) -> int: density_map = density_map or {} concurrency = config.get("concurrency", {}) frame_workers = max(1, int(concurrency.get("frame_workers", 2))) api_workers = 1 if dry_run else max(1, int(concurrency.get("api_workers", 1))) rpm = float(concurrency.get("requests_per_minute", 20)) input_queue: queue.Queue[InputItem | None] = queue.Queue() api_queue: queue.Queue[ExtractedItem | FailedItem | None] = queue.Queue(maxsize=api_workers * 2 + frame_workers) db_lock = threading.Lock() limiter = RateLimiter(rpm) counters = {"processed": 0} for item in items: input_queue.put(item) for _ in range(frame_workers): input_queue.put(None) def extractor() -> None: while True: item = input_queue.get() try: if item is None: return video_hash = file_sha256(item.video_path) with db_lock: skip = (not force) and already_done(conn, video_hash) if skip: print(f"SKIP already_done {item.video_path.name}") continue print(f"EXTRACT {item.video_path.name}") frame = extract_best_frame(item.video_path, config) frame_name = f"{item.video_path.stem}_{video_hash[:10]}.jpg" frame_path = frames_dir / frame_name write_jpeg(frame_path, frame["image"], int(config["processing"]["jpeg_quality"])) api_queue.put( ExtractedItem( item=item, video_hash=video_hash, frame_path=frame_path, frame_metrics={k: v for k, v in frame.items() if k != "image"}, ) ) except Exception as exc: fallback_hash = file_sha256(item.video_path) if item and item.video_path.exists() else "" api_queue.put(FailedItem(item=item, video_hash=fallback_hash, error=str(exc))) finally: input_queue.task_done() def api_sender() -> None: while True: work = api_queue.get() try: if work is None: return if isinstance(work, FailedItem): with db_lock: save_result( conn=conn, item=work.item, video_hash=work.video_hash, frame_path=Path(""), frame_metrics={}, request_payload={}, api_response=None, assessment=None, error=work.error, dry_run=dry_run, ) counters["processed"] += 1 print(f"ERR extract {work.item.video_path.name} {work.error}") continue density_res = density_map.get(str(work.item.video_path)) # 可読性レイヤー: 不可読かつ湯気が多い動画はLLMに送らず判定不可レコードを直接保存 skip, skip_reason = decide_skip_llm(density_res, config) if skip: synthetic = synthetic_unassessable(density_res, skip_reason) frame_metrics_skip = {**work.frame_metrics, "density_resolution": density_res} with db_lock: save_result( conn=conn, item=work.item, video_hash=work.video_hash, frame_path=work.frame_path, frame_metrics=frame_metrics_skip, request_payload={"skipped": True, "reason": skip_reason}, api_response=None, assessment=synthetic, error="", dry_run=dry_run, ) counters["processed"] += 1 print(f"SKIP-LLM {work.item.video_path.name} 判定不可 ({skip_reason})") continue llm_frame_path = work.frame_path if config.get("density", {}).get("use_best_frame_for_llm", True) and density_res: best_frame = density_res.get("best_frame_path") or "" if best_frame and Path(best_frame).exists(): llm_frame_path = Path(best_frame) request_payload = build_request_payload(config, prompt, llm_frame_path, density_input_note(density_res)) api_response: dict[str, Any] | None = None parsed: dict[str, Any] | None = None error = "" if dry_run: error = "dry_run: API call skipped" else: limiter.wait() try: api_response = call_byteplus(config, request_payload) parsed = parse_assessment(api_response) except Exception as exc: error = str(exc) frame_metrics_out = {**work.frame_metrics, "density_resolution": density_res} if density_res else work.frame_metrics with db_lock: save_result( conn=conn, item=work.item, video_hash=work.video_hash, frame_path=llm_frame_path, frame_metrics=frame_metrics_out, request_payload=request_payload, api_response=api_response, assessment=parsed, error=error, dry_run=dry_run, ) counters["processed"] += 1 status = "OK" if parsed else ("DRY" if dry_run else "ERR") print(f"{status} frame={work.frame_path.name} score={work.frame_metrics.get('selection_score', 0):.2f} {error}") finally: api_queue.task_done() extract_threads = [threading.Thread(target=extractor, daemon=True) for _ in range(frame_workers)] api_threads = [threading.Thread(target=api_sender, daemon=True) for _ in range(api_workers)] for thread in extract_threads + api_threads: thread.start() for thread in extract_threads: thread.join() for _ in range(api_workers): api_queue.put(None) for thread in api_threads: thread.join() return counters["processed"] def load_config(path: Path) -> dict[str, Any]: source = path if path.exists() else DEFAULT_CONFIG_PATH with source.open("r", encoding="utf-8") as fh: return yaml.safe_load(fh) def resolve_path(value: str) -> Path: path = Path(value) return path if path.is_absolute() else ROOT / path def setup_acceleration(config: dict[str, Any]) -> None: enabled = bool(config.get("processing", {}).get("use_opencl", True)) try: cv2.ocl.setUseOpenCL(enabled) cv2.setUseOptimized(True) except Exception: pass def build_prompt(policy_path: Path) -> str: policy = policy_path.read_text(encoding="utf-8") return ( "あなたはラーメン豚山のスープ状態を画像から評価するAI評価者です。\n" "以下の方針を固定の評価基準として厳守してください。\n\n" f"{policy}\n\n" "出力はJSONのみ。Markdown、コードブロック、前置き、補足説明は出力しないでください。" ) def discover_items(input_dir: Path) -> list[InputItem]: csv_files = sorted(input_dir.glob("*.csv")) videos = {p.name.lower(): p for p in input_dir.iterdir() if p.suffix.lower() in VIDEO_EXTENSIONS} items: list[InputItem] = [] if csv_files: for csv_path in csv_files: with csv_path.open("r", encoding="utf-8-sig", newline="") as fh: for idx, row in enumerate(csv.DictReader(fh)): video_name = (row.get("添付ファイル") or row.get("video") or "").strip() video_id = (row.get("ID") or row.get("id") or "").strip() if not video_name and not video_id: continue video_path = videos.get(video_name.lower()) or input_dir / video_name if not video_path.exists() and video_id: video_path = resolve_video_by_id(input_dir, video_id, videos) if video_path.exists(): items.append(InputItem(video_path=video_path, csv_path=csv_path, row_index=idx, metadata=dict(row))) else: print(f"WARN missing_video row={idx + 1} name={video_name}") else: for video_path in sorted(videos.values()): items.append(InputItem(video_path=video_path, csv_path=None, row_index=None, metadata={})) return items def resolve_video_by_id(input_dir: Path, video_id: str, videos: dict[str, Path]) -> Path: for extension in sorted(VIDEO_EXTENSIONS): candidate_name = f"{video_id}{extension}" video_path = videos.get(candidate_name.lower()) if video_path: return video_path return input_dir / f"{video_id}.MOV" def extract_best_frame(video_path: Path, config: dict[str, Any]) -> dict[str, Any]: cap = cv2.VideoCapture(str(video_path)) total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) or 0 fps = float(cap.get(cv2.CAP_PROP_FPS) or 30.0) if total <= 0: cap.release() raise RuntimeError(f"Video has no readable frames: {video_path}") strategy = str(config.get("processing", {}).get("extraction_strategy", "best_sample")) if strategy == "ffmpeg_middle": image, frame_no, fps = extract_frame_with_ffmpeg(video_path, config) cap.release() metrics = frame_metrics(image) return { "image": image, "frame_no": int(frame_no), "video_fps": fps, "selection_score": float(metrics["sharpness"] * float(config["processing"].get("sharpness_weight", 0.030))), **metrics, "extraction_strategy": strategy, } if strategy == "fast_first": warmup_frames = max(0, int(config["processing"].get("fast_warmup_frames", 12))) image = None frame_no = 0 for frame_no in range(min(total, warmup_frames + 1)): ok, current = cap.read() if not ok: break image = current cap.release() if image is None: raise RuntimeError(f"No readable initial frames: {video_path}") metrics = frame_metrics(image) return { "image": image, "frame_no": int(frame_no), "video_fps": fps, "selection_score": float(metrics["sharpness"] * float(config["processing"].get("sharpness_weight", 0.030))), **metrics, "extraction_strategy": strategy, } sample_fps = float(config["processing"]["sample_fps"]) step = max(1, int(round(fps / sample_fps))) candidates = list(range(0, total, step)) max_candidates = int(config["processing"]["max_candidates"]) if max_candidates > 0 and len(candidates) > max_candidates: indexes = np.linspace(0, len(candidates) - 1, max_candidates).astype(int) candidates = [candidates[i] for i in indexes] best: dict[str, Any] | None = None fallback_best: dict[str, Any] | None = None min_sharpness = float(config["processing"].get("min_sharpness", 80.0)) sharpness_weight = float(config["processing"].get("sharpness_weight", 0.030)) for frame_no in candidates: cap.set(cv2.CAP_PROP_POS_FRAMES, frame_no) ok, image = cap.read() if not ok: continue metrics = frame_metrics(image) center_penalty = abs(frame_no - (total / 2)) / max(total, 1) selection_score = ( metrics["sharpness"] * sharpness_weight + metrics["contrast"] * 1.8 - metrics["blowout_ratio"] * 120.0 - metrics["steam_haze_score"] * 90.0 - metrics["motion_blur_score"] * 80.0 - center_penalty * 8.0 ) candidate = { "image": image, "frame_no": int(frame_no), "video_fps": fps, "selection_score": float(selection_score), **metrics, } if fallback_best is None or candidate["selection_score"] > fallback_best["selection_score"]: fallback_best = candidate if metrics["sharpness"] < min_sharpness: continue if best is None or candidate["selection_score"] > best["selection_score"]: best = candidate cap.release() if best is None: best = fallback_best if best is None: raise RuntimeError(f"No readable sampled frames: {video_path}") return best def frame_metrics(image: np.ndarray) -> dict[str, float]: gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) sharpness = float(cv2.Laplacian(gray, cv2.CV_64F).var()) sobel_x = cv2.Sobel(gray, cv2.CV_64F, 1, 0, ksize=3) sobel_y = cv2.Sobel(gray, cv2.CV_64F, 0, 1, ksize=3) directional_gap = abs(float(sobel_x.var()) - float(sobel_y.var())) / max(float(sobel_x.var() + sobel_y.var()), 1.0) blowout_ratio = float((gray > 245).mean()) contrast = float(gray.std() / 255.0) bright_ratio = float((gray > 220).mean()) low_contrast = max(0.0, 0.22 - contrast) steam_haze_score = float(min(1.0, bright_ratio * 0.75 + low_contrast * 2.0)) motion_blur_score = float(min(1.0, max(0.0, (120.0 - sharpness) / 120.0) + directional_gap * 0.25)) return { "sharpness": sharpness, "motion_blur_score": motion_blur_score, "blowout_ratio": blowout_ratio, "contrast": contrast, "steam_haze_score": steam_haze_score, } def extract_frame_with_ffmpeg(video_path: Path, config: dict[str, Any]) -> tuple[np.ndarray, int, float]: ffmpeg = resolve_binary(config["processing"].get("ffmpeg_path"), "ffmpeg") ffprobe = resolve_binary(config["processing"].get("ffprobe_path"), "ffprobe") duration = probe_float(ffprobe, video_path, "duration") fps = probe_fps(ffprobe, video_path) position_ratio = float(config["processing"].get("ffmpeg_position_ratio", 0.35)) seek_seconds = max(0.0, duration * position_ratio) if duration > 0 else 0.0 with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp: tmp_path = Path(tmp.name) try: cmd = [ ffmpeg, "-hide_banner", "-loglevel", "error", "-ss", f"{seek_seconds:.3f}", "-i", str(video_path), "-frames:v", "1", "-q:v", "2", "-y", str(tmp_path), ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) if result.returncode != 0: raise RuntimeError(f"ffmpeg failed for {video_path.name}: {result.stderr.strip()}") data = np.frombuffer(tmp_path.read_bytes(), dtype=np.uint8) image = cv2.imdecode(data, cv2.IMREAD_COLOR) if image is None: raise RuntimeError(f"ffmpeg output is not a readable image: {video_path.name}") frame_no = int(round(seek_seconds * fps)) if fps > 0 else 0 return image, frame_no, fps finally: try: tmp_path.unlink() except FileNotFoundError: pass def resolve_binary(configured: Any, name: str) -> str: if configured: path = Path(str(configured)).expanduser() if path.exists(): return str(path) found = shutil.which(name) if found: return found raise RuntimeError(f"{name} was not found. Install ffmpeg or set processing.{name}_path.") def probe_float(ffprobe: str, video_path: Path, field: str) -> float: result = subprocess.run( [ ffprobe, "-v", "error", "-show_entries", f"format={field}", "-of", "default=noprint_wrappers=1:nokey=1", str(video_path), ], capture_output=True, text=True, timeout=30, ) if result.returncode != 0: return 0.0 try: return float(result.stdout.strip()) except ValueError: return 0.0 def probe_fps(ffprobe: str, video_path: Path) -> float: result = subprocess.run( [ ffprobe, "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=avg_frame_rate", "-of", "default=noprint_wrappers=1:nokey=1", str(video_path), ], capture_output=True, text=True, timeout=30, ) value = result.stdout.strip() if "/" in value: numerator, denominator = value.split("/", 1) try: return float(numerator) / max(float(denominator), 1.0) except ValueError: return 30.0 try: return float(value) except ValueError: return 30.0 def write_jpeg(path: Path, image: np.ndarray, quality: int) -> None: ok, encoded = cv2.imencode(".jpg", image, [int(cv2.IMWRITE_JPEG_QUALITY), quality]) if not ok: raise RuntimeError(f"Failed to encode JPEG: {path}") path.write_bytes(encoded.tobytes()) def build_request_payload(config: dict[str, Any], prompt: str, frame_path: Path, density_note: str = "") -> dict[str, Any]: data_url = image_data_url(frame_path) text = f"{prompt}\n\n{density_note}" if density_note else prompt return { "model": config["api"]["model"], "input": [ { "role": "user", "content": [ {"type": "input_image", "image_url": data_url}, {"type": "input_text", "text": text}, ], } ], } def call_byteplus(config: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]: api_key = read_api_key(config) max_retries = int(config["api"].get("max_retries", 4)) retry_base = float(config["api"].get("retry_base_seconds", 2.0)) last_error = "" for attempt in range(max_retries + 1): response = requests.post( config["api"]["endpoint"], headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, json=payload, timeout=int(config["api"]["timeout_seconds"]), ) try: data = response.json() except ValueError: data = {"raw_text": response.text} if response.status_code < 400: return data last_error = f"API error {response.status_code}: {json.dumps(data, ensure_ascii=False)}" if response.status_code not in {429, 500, 502, 503, 504} or attempt >= max_retries: break retry_after = response.headers.get("retry-after") wait_seconds = float(retry_after) if retry_after and retry_after.replace(".", "", 1).isdigit() else retry_base * (2**attempt) time.sleep(wait_seconds) raise RuntimeError(last_error) def read_api_key(config: dict[str, Any]) -> str: env_name = config["api"].get("key_env", "ARK_API_KEY") key = os.environ.get(env_name, "").strip() if key: return key key_file = resolve_path(config["api"].get("key_file", "secrets/byteplus.key")) if key_file.exists(): return key_file.read_text(encoding="utf-8").strip() raise RuntimeError(f"API key not found. Set {env_name} or create {key_file}") def image_data_url(path: Path) -> str: encoded = base64.b64encode(path.read_bytes()).decode("ascii") return f"data:image/jpeg;base64,{encoded}" def parse_assessment(api_response: dict[str, Any]) -> dict[str, Any] | None: text = extract_output_text(api_response) if not text: return None candidates = [text.strip(), strip_code_fence(text), extract_json_object(text)] for candidate in candidates: if not candidate: continue try: parsed = json.loads(candidate) return parsed if isinstance(parsed, dict) else None except json.JSONDecodeError: continue return None def extract_output_text(data: dict[str, Any]) -> str: if isinstance(data.get("output_text"), str): return data["output_text"].strip() chunks: list[str] = [] for item in data.get("output", []) or []: for content in item.get("content", []) or []: text = content.get("text") if isinstance(text, str): chunks.append(text) return "\n".join(chunks).strip() def strip_code_fence(text: str) -> str: value = text.strip() if value.startswith("```"): value = value.split("\n", 1)[-1] if value.endswith("```"): value = value.rsplit("```", 1)[0] return value.strip() def extract_json_object(text: str) -> str: start = text.find("{") end = text.rfind("}") return text[start : end + 1] if start >= 0 and end > start else "" def init_db(conn: sqlite3.Connection) -> None: conn.execute( """ CREATE TABLE IF NOT EXISTS evaluations ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL, input_video_path TEXT NOT NULL, video_sha256 TEXT NOT NULL, csv_path TEXT, csv_row_index INTEGER, metadata_json TEXT NOT NULL, frame_path TEXT NOT NULL, frame_metrics_json TEXT NOT NULL, request_json TEXT NOT NULL, response_json TEXT, assessment_json TEXT, error TEXT NOT NULL DEFAULT '', dry_run INTEGER NOT NULL DEFAULT 0 ) """ ) conn.execute("CREATE INDEX IF NOT EXISTS idx_evaluations_video_sha256 ON evaluations(video_sha256)") conn.commit() def already_done(conn: sqlite3.Connection, video_hash: str) -> bool: row = conn.execute( "SELECT 1 FROM evaluations WHERE video_sha256 = ? AND error = '' LIMIT 1", (video_hash,), ).fetchone() return row is not None def save_result( conn: sqlite3.Connection, item: InputItem, video_hash: str, frame_path: Path, frame_metrics: dict[str, Any], request_payload: dict[str, Any], api_response: dict[str, Any] | None, assessment: dict[str, Any] | None, error: str, dry_run: bool, ) -> None: conn.execute( """ INSERT INTO evaluations ( created_at, input_video_path, video_sha256, csv_path, csv_row_index, metadata_json, frame_path, frame_metrics_json, request_json, response_json, assessment_json, error, dry_run ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( datetime.now(timezone.utc).isoformat(), str(item.video_path), video_hash, str(item.csv_path) if item.csv_path else None, item.row_index, json.dumps(item.metadata, ensure_ascii=False), str(frame_path), json.dumps(frame_metrics, ensure_ascii=False), json.dumps(redact_request_image(request_payload), ensure_ascii=False), json.dumps(api_response, ensure_ascii=False) if api_response is not None else None, json.dumps(assessment, ensure_ascii=False) if assessment is not None else None, error, 1 if dry_run else 0, ), ) conn.commit() def redact_request_image(payload: dict[str, Any]) -> dict[str, Any]: clone = json.loads(json.dumps(payload, ensure_ascii=False)) for item in clone.get("input", []): for content in item.get("content", []): if content.get("type") == "input_image": content["image_url"] = "[data-url-redacted]" return clone def export_datasets(conn: sqlite3.Connection, output_dir: Path) -> None: rows = conn.execute( """ SELECT input_video_path, csv_path, csv_row_index, metadata_json, frame_path, frame_metrics_json, assessment_json FROM evaluations WHERE assessment_json IS NOT NULL AND error = '' ORDER BY id """ ).fetchall() jsonl_path = output_dir / "training_dataset.jsonl" csv_path = output_dir / "training_dataset.csv" with jsonl_path.open("w", encoding="utf-8") as out: for row in rows: record = dataset_record(row) out.write(json.dumps(record, ensure_ascii=False) + "\n") fields = [ "video_path", "frame_path", "store", "slot", "brix_raw", "human_soup_score", "human_total_score", "visual_density", "water_level", "oil_emulsification", "boiling_heat_state", "photo_quality", "image_condition_score", "summary", ] with csv_path.open("w", encoding="utf-8-sig", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=fields) writer.writeheader() for row in rows: record = dataset_record(row) writer.writerow(flatten_for_csv(record)) def dataset_record(row: tuple[Any, ...]) -> dict[str, Any]: video_path, csv_path, csv_row_index, metadata_json, frame_path, frame_metrics_json, assessment_json = row metadata = json.loads(metadata_json) assessment = json.loads(assessment_json) return { "video_path": video_path, "csv_path": csv_path, "csv_row_index": csv_row_index, "metadata": metadata, "frame_path": frame_path, "frame_metrics": json.loads(frame_metrics_json), "assessment": assessment, "labels": { "store": metadata.get("提出者") or metadata.get("store"), "slot": metadata.get("時間帯") or metadata.get("slot"), "brix_raw": metadata.get("スープ濃度") or metadata.get("brix_raw"), "human_soup_score": metadata.get("スープのみの点数") or metadata.get("soup"), "human_total_score": metadata.get("濃度込みの点数") or metadata.get("total"), "human_comment": metadata.get("評価備考") or metadata.get("comment"), "evaluator": metadata.get("通知するブロック長") or metadata.get("evaluator"), }, } def flatten_for_csv(record: dict[str, Any]) -> dict[str, Any]: assessment = record["assessment"] labels = record["labels"] return { "video_path": record["video_path"], "frame_path": record["frame_path"], "store": labels.get("store"), "slot": labels.get("slot"), "brix_raw": labels.get("brix_raw"), "human_soup_score": labels.get("human_soup_score"), "human_total_score": labels.get("human_total_score"), "visual_density": (assessment.get("visual_density") or {}).get("value"), "water_level": (assessment.get("water_level") or {}).get("state"), "oil_emulsification": (assessment.get("oil_emulsification") or {}).get("state"), "boiling_heat_state": assessment.get("boiling_heat_state"), "photo_quality": assessment.get("photo_quality"), "image_condition_score": assessment.get("image_condition_score"), "summary": assessment.get("summary"), } def file_sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() if __name__ == "__main__": sys.exit(main())