Update app.py

app.py

@@ -1,310 +1,660 @@
+from __future__ import annotations
+
+import copy
+import json
 import os
-import sys
+import re
 import shutil
 import subprocess
-import json
-import copy
-import re
-from collections import Counter  # 🚀 導入計數器進行多數決投票
-import gradio as gr
+import sys
+import tempfile
+import threading
+import time
+import uuid
+from collections import Counter
+from pathlib import Path
+from typing import Any
+
 import cv2
+import gradio as gr
 import pandas as pd
 
 # =======================================================
-# 🚀 核心修復：Gradio 官方 OpenAPI Schema 解析崩潰布林值補丁
+# Gradio 4.44.x OpenAPI schema compatibility patch
 # =======================================================
 try:
     import gradio_client.utils
-    orig_get_type = gradio_client.utils.get_type
-    orig_json_schema_to_python_type = gradio_client.utils._json_schema_to_python_type
 
-    def patched_get_type(schema, *args, **kwargs):
+    _orig_get_type = gradio_client.utils.get_type
+    _orig_json_schema_to_python_type = (
+        gradio_client.utils._json_schema_to_python_type
+    )
+
+    def _patched_get_type(schema: Any, *args: Any, **kwargs: Any) -> Any:
         if isinstance(schema, bool):
             return "Any"
-        return orig_get_type(schema, *args, **kwargs)
+        return _orig_get_type(schema, *args, **kwargs)
 
-    def patched_json_schema_to_python_type(schema, *args, **kwargs):
+    def _patched_json_schema_to_python_type(
+        schema: Any, *args: Any, **kwargs: Any
+    ) -> Any:
         if isinstance(schema, bool):
             return "Any"
-        return orig_json_schema_to_python_type(schema, *args, **kwargs)
+        return _orig_json_schema_to_python_type(schema, *args, **kwargs)
 
-    gradio_client.utils.get_type = patched_get_type
-    gradio_client.utils._json_schema_to_python_type = patched_json_schema_to_python_type
-    print("【系統提示】成功動態植入 Gradio 核心 Schema 容錯補丁，已免疫 bool 迭代錯誤。")
-except Exception as patch_e:
-    print(f"【系統提示】嘗試套用 Gradio 補丁時發生異常（若未崩潰可忽略）: {patch_e}")
-# =======================================================
-
-OMR_DIR = "OMRChecker"
-MOCK_DIR = os.path.abspath("mock_libs")
+    gradio_client.utils.get_type = _patched_get_type
+    gradio_client.utils._json_schema_to_python_type = (
+        _patched_json_schema_to_python_type
+    )
+    print("【系統提示】Gradio Schema 容錯補丁已套用。")
+except Exception as patch_error:
+    print(f"【系統提示】Gradio 補丁未套用：{patch_error}")
 
 # =======================================================
-# 在硬碟建立實體偽裝庫，並透過 PYTHONPATH 注入子程序 (驅逐 screeninfo 報錯)
+# Paths and repository settings
 # =======================================================
-os.makedirs(MOCK_DIR, exist_ok=True)
-with open(os.path.join(MOCK_DIR, "screeninfo.py"), "w", encoding="utf-8") as f:
-    f.write('''\
+ROOT_DIR = Path(__file__).resolve().parent
+OMR_DIR = ROOT_DIR / "OMRChecker"
+MOCK_DIR = ROOT_DIR / "mock_libs"
+GENERATED_DIR = ROOT_DIR / "generated"
+
+OMR_REPO_URL = "https://github.com/Udayraj123/OMRChecker.git"
+OMR_REPO_BRANCH = "master"
+
+# OMRChecker is CPU-heavy and writes multiple files. Serialize requests so
+# two students cannot overwrite each other's temporary outputs.
+PROCESS_LOCK = threading.Lock()
+REPO_LOCK = threading.Lock()
+
+ALLOWED_ANSWERS = {"A", "B", "C", "D"}
+
+SENSITIVITY_PROFILES = [
+    {
+        "name": "Level 1：標準掃描",
+        "shrink_w": 0,
+        "shrink_h": 0,
+        "levels_high": 0.90,
+    },
+    {
+        "name": "Level 2：增強靈敏度",
+        "shrink_w": 12,
+        "shrink_h": 8,
+        "levels_high": 0.75,
+    },
+    {
+        "name": "Level 3：淡筆跡模式",
+        "shrink_w": 20,
+        "shrink_h": 14,
+        "levels_high": 0.55,
+    },
+]
+
+
+def _prepare_headless_helpers() -> None:
+    """Create modules injected into the OMRChecker subprocess.
+
+    OMRChecker imports screeninfo and may call OpenCV GUI functions. Hugging
+    Face Spaces is headless, so both behaviours need safe fallbacks.
+    """
+
+    MOCK_DIR.mkdir(parents=True, exist_ok=True)
+
+    (MOCK_DIR / "screeninfo.py").write_text(
+        """
 class ScreenInfoError(Exception):
     pass
+
 class FakeMonitor:
     width = 1920
     height = 1080
+
+
 def get_monitors():
     return [FakeMonitor()]
-''')
+""".strip()
+        + "\n",
+        encoding="utf-8",
+    )
+
+    # Python imports sitecustomize automatically when it is available on
+    # PYTHONPATH. This prevents cv2.imshow/namedWindow from crashing in the
+    # headless Space even if a debug path is accidentally reached.
+    (MOCK_DIR / "sitecustomize.py").write_text(
+        """
+try:
+    import cv2
+
+    cv2.imshow = lambda *args, **kwargs: None
+    cv2.namedWindow = lambda *args, **kwargs: None
+    cv2.moveWindow = lambda *args, **kwargs: None
+    cv2.destroyAllWindows = lambda *args, **kwargs: None
+    cv2.getWindowProperty = lambda *args, **kwargs: 1.0
+    cv2.waitKey = lambda *args, **kwargs: ord('q')
+except Exception:
+    pass
+""".strip()
+        + "\n",
+        encoding="utf-8",
+    )
 
-# 1. 啟動時檢查並自動克隆 OMRChecker 官方專案
-if not os.path.exists(OMR_DIR):
-    print("正在從 GitHub 複製 OMRChecker 核心專案...")
-    subprocess.run(["git", "clone", "https://github.com/udayraj123/OMRChecker.git"], check=True)
-    subprocess.run([sys.executable, "-m", "pip", "install", "-r", f"{OMR_DIR}/requirements.txt"], check=True)
 
+_prepare_headless_helpers()
+GENERATED_DIR.mkdir(parents=True, exist_ok=True)
 
-def process_omr(image_file, template_content=None):
-    if image_file is None:
-        return None, None, "錯誤：請務必上傳答案卡圖片。"
 
-    # 🚀 偵測上傳圖片的實際解析度
+def _valid_omr_checkout(path: Path) -> bool:
+    return (
+        path.is_dir()
+        and (path / "main.py").is_file()
+        and (path / "src").is_dir()
+    )
+
+
+def ensure_omrchecker() -> None:
+    """Clone OMRChecker lazily and recover from incomplete checkouts.
+
+    Dependencies are intentionally *not* installed here. Hugging Face installs
+    requirements.txt during the build stage; runtime pip installation caused
+    the original Space crash.
+    """
+
+    if _valid_omr_checkout(OMR_DIR):
+        return
+
+    with REPO_LOCK:
+        if _valid_omr_checkout(OMR_DIR):
+            return
+
+        if OMR_DIR.exists():
+            shutil.rmtree(OMR_DIR, ignore_errors=True)
+
+        temporary_checkout = ROOT_DIR / f"OMRChecker.clone.{uuid.uuid4().hex}"
+
+        command = [
+            "git",
+            "clone",
+            "--depth",
+            "1",
+            "--branch",
+            OMR_REPO_BRANCH,
+            OMR_REPO_URL,
+            str(temporary_checkout),
+        ]
+
+        result = subprocess.run(
+            command,
+            cwd=ROOT_DIR,
+            capture_output=True,
+            text=True,
+            timeout=180,
+        )
+
+        if result.returncode != 0:
+            shutil.rmtree(temporary_checkout, ignore_errors=True)
+            raise RuntimeError(
+                "未能下載 OMRChecker 核心。\n"
+                + (result.stderr or result.stdout or "Git clone failed.")[-2000:]
+            )
+
+        if not _valid_omr_checkout(temporary_checkout):
+            shutil.rmtree(temporary_checkout, ignore_errors=True)
+            raise RuntimeError(
+                "OMRChecker 下載完成，但缺少 main.py 或 src 目錄。"
+            )
+
+        temporary_checkout.replace(OMR_DIR)
+        print("【系統提示】OMRChecker 核心已準備完成。")
+
+
+def _find_local_file(prefix: str, extensions: set[str]) -> Path | None:
+    for path in sorted(ROOT_DIR.iterdir()):
+        if (
+            path.is_file()
+            and path.name.lower().startswith(prefix.lower())
+            and path.suffix.lower() in extensions
+        ):
+            return path
+    return None
+
+
+def _load_template_content(template_content: str | None) -> tuple[dict[str, Any], Path]:
+    if template_content and str(template_content).strip():
+        try:
+            return json.loads(str(template_content)), ROOT_DIR
+        except json.JSONDecodeError as error:
+            raise ValueError(f"Template JSON 格式不正確：{error}") from error
+
+    template_path = _find_local_file("template", {".json"})
+    if template_path is None:
+        raise FileNotFoundError(
+            "找不到 template.json。請把 Template JSON 放在 Space 根目錄。"
+        )
+
     try:
-        img = cv2.imread(image_file)
-        if img is not None:
-            img_h, img_w, _ = img.shape
-            dimension_info = f"【圖片載入成功】解析度: {img_w} x {img_h} px\n"
-        else:
-            dimension_info = "【圖片載入失敗】無法讀取圖片。\n"
-    except Exception as e:
-        dimension_info = f"【圖片讀取錯誤】{str(e)}\n"
-
-    current_dir = os.path.dirname(__file__)
-
-    # 🚀 智能降級機制：若 API 或網頁端未傳入 JSON 配置，主動撈取本地目錄的 template*.json
-    if not template_content or str(template_content).strip() == "":
-        for file in os.listdir(current_dir):
-            if file.lower().startswith("template") and file.lower().endswith(".json"):
-                json_src = os.path.join(current_dir, file)
-                try:
-                    with open(json_src, "r", encoding="utf-8") as f:
-                        template_content = f.read()
-                    print(f"【系統提示】未偵測到傳入的 JSON 配置，已自動載入本地配置: {file}")
-                    break
-                except Exception:
-                    pass
-
-    if not template_content:
-        return None, None, f"{dimension_info}錯誤：找不到有效的 Template JSON 配置。\n請提供 JSON 內容或在後端存放 template.json"
+        return json.loads(template_path.read_text(encoding="utf-8")), template_path.parent
+    except json.JSONDecodeError as error:
+        raise ValueError(f"{template_path.name} JSON 格式不正確：{error}") from error
+
+
+def _expand_question_label(label: Any) -> list[int]:
+    text = str(label).strip()
+
+    range_match = re.fullmatch(r"q(\d+)\.\.(\d+)", text, flags=re.IGNORECASE)
+    if range_match:
+        start, end = map(int, range_match.groups())
+        step = 1 if end >= start else -1
+        return list(range(start, end + step, step))
+
+    single_match = re.fullmatch(r"q0*(\d+)", text, flags=re.IGNORECASE)
+    if single_match:
+        return [int(single_match.group(1))]
+
+    return []
+
+
+def _question_numbers_from_template(template: dict[str, Any]) -> list[int]:
+    numbers: set[int] = set()
+
+    for block in (template.get("fieldBlocks") or {}).values():
+        if not isinstance(block, dict):
+            continue
+        for label in block.get("fieldLabels") or []:
+            numbers.update(_expand_question_label(label))
+
+    if not numbers:
+        raise ValueError(
+            "Template JSON 沒有可辨識的 q1、q2 或 q1..100 題號。"
+        )
+
+    return sorted(numbers)
+
+
+def _normalize_answer(value: Any) -> str:
+    if value is None or (isinstance(value, float) and pd.isna(value)):
+        return "—"
+
+    answer = str(value).strip().upper()
+    return answer if answer in ALLOWED_ANSWERS else "—"
+
+
+def _apply_profile(template: dict[str, Any], profile: dict[str, Any]) -> dict[str, Any]:
+    current = copy.deepcopy(template)
+    base_width, base_height = current.get("bubbleDimensions", [60, 30])
+
+    current["bubbleDimensions"] = [
+        max(10, int(base_width) - int(profile["shrink_w"])),
+        max(10, int(base_height) - int(profile["shrink_h"])),
+    ]
+
+    for processor in current.get("preProcessors") or []:
+        if not isinstance(processor, dict):
+            continue
+        if processor.get("name") == "Levels":
+            processor.setdefault("options", {})["high"] = profile["levels_high"]
+
+    return current
+
+
+def _alignment_reference_names(template: dict[str, Any]) -> set[str]:
+    references: set[str] = set()
+
+    for processor in template.get("preProcessors") or []:
+        if not isinstance(processor, dict):
+            continue
+        options = processor.get("options") or {}
+        for key in ("reference", "relativePath"):
+            value = options.get(key)
+            if isinstance(value, str) and value.strip():
+                references.add(Path(value).name)
+
+    return references
+
+
+def _copy_alignment_assets(template: dict[str, Any], destination: Path) -> None:
+    reference_names = _alignment_reference_names(template)
+
+    # The provided DSE template uses template.jpg. Also support any explicit
+    # reference filename that exists in the Space root.
+    for reference_name in reference_names:
+        direct = ROOT_DIR / reference_name
+        source = direct if direct.is_file() else None
+
+        if source is None and reference_name.lower().startswith("template"):
+            source = _find_local_file("template", {".jpg", ".jpeg", ".png"})
+
+        if source is None:
+            raise FileNotFoundError(
+                f"Template 需要對齊參考圖 {reference_name}，但 Space 根目錄找不到它。"
+            )
+
+        shutil.copy2(source, destination / reference_name)
+
+
+def _write_headless_config(destination: Path) -> None:
+    config = {
+        "outputs": {
+            "show_image_level": 0,
+            "save_image_level": 3,
+            "colored_outputs_enabled": True,
+            "save_detections": True,
+        }
+    }
+    (destination / "config.json").write_text(
+        json.dumps(config, ensure_ascii=False, indent=2),
+        encoding="utf-8",
+    )
+
+
+def _find_output_file(output_dir: Path, suffixes: set[str], prefer: str = "") -> Path | None:
+    candidates = [
+        path
+        for path in output_dir.rglob("*")
+        if path.is_file() and path.suffix.lower() in suffixes
+    ]
+
+    if not candidates:
+        return None
+
+    if prefer:
+        preferred = [path for path in candidates if prefer.lower() in str(path).lower()]
+        if preferred:
+            candidates = preferred
+
+    return max(candidates, key=lambda path: (path.stat().st_mtime, path.stat().st_size))
+
+
+def _parse_csv_answers(csv_path: Path | None, question_numbers: list[int]) -> dict[int, str]:
+    answers = {number: "—" for number in question_numbers}
+
+    if csv_path is None or not csv_path.is_file():
+        return answers
 
     try:
-        base_template_dict = json.loads(template_content)
-    except Exception as e:
-        return None, None, f"{dimension_info}錯誤：Template JSON 格式不正確。\n{str(e)}"
-
-    base_bubble_w, base_bubble_h = base_template_dict.get("bubbleDimensions", [60, 30])
-
-    # =======================================================
-    # 🚀 定義多重採樣策略 (Multi-Sampling)
-    # =======================================================
-    sensitivity_profiles = [
-        {"name": "Level 1: 標���掃描 (嚴謹判斷防雜訊)", "shrink_w": 0, "shrink_h": 0, "levels_high": 0.9},
-        {"name": "Level 2: 增強靈敏度 (適應單橫線筆跡)", "shrink_w": 12, "shrink_h": 8, "levels_high": 0.75},
-        {"name": "Level 3: 極限靈敏度 (適應極淡筆跡)", "shrink_w": 20, "shrink_h": 14, "levels_high": 0.55}
+        dataframe = pd.read_csv(csv_path)
+    except Exception:
+        return answers
+
+    if dataframe.empty:
+        return answers
+
+    normalized_columns = {
+        str(column).strip().lower(): column for column in dataframe.columns
+    }
+
+    for number in question_numbers:
+        column = normalized_columns.get(f"q{number}")
+        if column is not None:
+            answers[number] = _normalize_answer(dataframe.iloc[0][column])
+
+    return answers
+
+
+def _parse_log_answers(text: str, question_numbers: list[int]) -> dict[int, str]:
+    allowed_numbers = set(question_numbers)
+    answers = {number: "—" for number in question_numbers}
+
+    patterns = [
+        r"(?i)\bq(?:uestion)?\s*0*(\d+)\s*[:=]\s*([A-D])\b",
+        r"(?i)\b0*(\d+)\s*[|:]\s*([A-D])\b",
     ]
 
-    job_dir = os.path.join(OMR_DIR, "inputs", "hf_space_job")
-    images_dir = os.path.join(job_dir, "images")
-    outputs_dir = os.path.join(OMR_DIR, "outputs")
-
-    all_scans_results = []
-    accumulated_logs = f"--- 🚀 開始執行多重採樣與決策融合 (Ensemble Voting) ---\n"
-
-    # 🚀 確保對齊用的 template.jpg 基準圖存在
-    template_img_found = False
-    for file in os.listdir(current_dir):
-        if file.lower().startswith("template") and file.lower().endswith((".jpg", ".jpeg", ".png")):
-            template_img_src = os.path.join(current_dir, file)
-            template_img_found = True
-            break
-
-    # 執行三輪完整掃描以收集樣本進行投票
-    for profile in sensitivity_profiles:
-        accumulated_logs += f"\n👉 執行採樣: [{profile['name']}]...\n"
-        
-        if os.path.exists(job_dir):
-            shutil.rmtree(job_dir)
-        os.makedirs(images_dir, exist_ok=True)
-        if os.path.exists(outputs_dir):
-            shutil.rmtree(outputs_dir)
-
-        # 複製當前處理的學生考卷
-        shutil.copy(image_file, os.path.join(images_dir, "sheet.jpg"))
-        
-        # 複製對齊用的基準圖
-        if template_img_found:
-            shutil.copy(template_img_src, os.path.join(job_dir, "template.jpg"))
-
-        # 動態微調氣泡大小與 Levels 閥值
-        current_template = copy.deepcopy(base_template_dict)
-        new_w = max(10, base_bubble_w - profile['shrink_w'])
-        new_h = max(10, base_bubble_h - profile['shrink_h'])
-        current_template["bubbleDimensions"] = [new_w, new_h]
-
-        if "preProcessors" in current_template:
-            for pp in current_template["preProcessors"]:
-                if pp.get("name") == "Levels" and "options" in pp:
-                    pp["options"]["high"] = profile["levels_high"]
-
-        # 寫入本次靈敏度的臨時配置
-        with open(os.path.join(job_dir, "template.json"), "w", encoding="utf-8") as f:
-            json.dump(current_template, f)
-
-        with open(os.path.join(OMR_DIR, "inputs", "template.json"), "w", encoding="utf-8") as f:
-            json.dump(current_template, f)
-
-        # 強迫啟用高等級視覺化劃記圈圈圖渲染 (show_image_level = 3)
-        config_path = os.path.join(OMR_DIR, "config.json")
-        if os.path.exists(config_path):
-            try:
-                with open(config_path, "r", encoding="utf-8") as f:
-                    config_data = json.load(f)
-                config_data["show_image_level"] = 3
-                with open(config_path, "w", encoding="utf-8") as f:
-                    json.dump(config_data, f, indent=4)
-            except Exception as e:
-                print(f"自動調整 config.json 失敗: {e}")
-
-        # 配置環境變數，注入外部 mock_libs 實體偽裝庫
-        current_env = os.environ.copy()
-        current_env["PYTHONPATH"] = MOCK_DIR + os.pathsep + current_env.get("PYTHONPATH", "")
-        
-        result = subprocess.run(
-            [sys.executable, "main.py", "-i", "inputs/hf_space_job"],
-            cwd=OMR_DIR, capture_output=True, text=True, env=current_env
-        )
+    for pattern in patterns:
+        for question_text, answer in re.findall(pattern, text or ""):
+            number = int(question_text)
+            if number in allowed_numbers:
+                answers[number] = answer.upper()
 
-        current_csv = None
-        current_img = None
-        if os.path.exists(outputs_dir):
-            for root, dirs, files in os.walk(outputs_dir):
-                for file in files:
-                    full_path = os.path.join(root, file)
-                    if file.endswith(".csv"):
-                        current_csv = full_path
-                    elif file.endswith((".jpg", ".jpeg", ".png")):
-                        if "checkedomrs" in root.lower():
-                            current_img = full_path
-                        elif current_img is None:
-                            current_img = full_path
-
-        curr_answers = {i: "—" for i in range(1, 37)}
-        full_log = (result.stdout or "") + "\n" + (result.stderr or "")
-
-        # 嘗試解析本輪 CSV 答案
-        has_csv_data = False
-        if current_csv and os.path.exists(current_csv):
-            try:
-                df = pd.read_csv(current_csv)
-                if not df.empty:
-                    norm_cols = {str(col).strip().lower(): col for col in df.columns}
-                    for i in range(1, 37):
-                        col_key = f"q{i}"
-                        if col_key in norm_cols:
-                            val = df.iloc[0][norm_cols[col_key]]
-                            if not pd.isna(val) and str(val).strip() != "":
-                                curr_answers[i] = str(val).strip().upper()
-                                has_csv_data = True
-            except:
-                pass
-
-        # 若無實體 CSV，降級啟動文本正則撈取日誌答案
-        if not has_csv_data:
-            matches = re.findall(r'(?i)(?:q(?:uestion)?|"q(?:uestion)?")\s*0*([1-9][0-9]*)\b[^A-E]*?\b([A-E])\b', full_log)
-            for q_str, ans in matches:
-                q_num = int(q_str)
-                if 1 <= q_num <= 36:
-                    curr_answers[q_num] = ans.upper()
-            
-            if list(curr_answers.values()).count("—") > 25:
-                for line in full_log.splitlines():
-                    t_matches = re.findall(r'\b0*([1-9][0-9]*)\b\s*[|:]\s*\b([A-E])\b', line)
-                    for q_str, ans in t_matches:
-                        q_num = int(q_str)
-                        if 1 <= q_num <= 36:
-                            curr_answers[q_num] = ans.upper()
-
-        valid_count = sum(1 for v in curr_answers.values() if v != "—")
-        all_scans_results.append({
-            "profile_name": profile["name"],
-            "answers": curr_answers,
-            "csv": current_csv,
-            "img": current_img,
-            "valid_count": valid_count
-        })
-        accumulated_logs += f"   ✅ 本輪辨識出有效題數: {valid_count}/36\n"
-
-    # =======================================================
-    # 🧠 啟動決策融合與投票機制 (Majority Voting)
-    # =======================================================
-    accumulated_logs += "\n--- 🧠 啟動決策融合 (Majority Voting) ---\n"
-    final_extracted_answers = {i: "—" for i in range(1, 37)}
-    
-    for i in range(1, 37):
-        votes = [scan["answers"][i] for scan in all_scans_results if scan["answers"][i] != "—"]
-        
-        if not votes:
-            final_extracted_answers[i] = "—"
-        else:
-            vote_counts = Counter(votes)
-            best_ans, count = vote_counts.most_common(1)[0]
-            final_extracted_answers[i] = best_ans
-            
-            if len(vote_counts) > 1:
-                accumulated_logs += f"   ⚠️ 發現分歧 [q{i:02d}]: 投票分佈 {dict(vote_counts)} 👉 系統決策為: {best_ans}\n"
-
-    final_valid_count = sum(1 for v in final_extracted_answers.values() if v != "—")
-    accumulated_logs += f"\n🎯 融合後最終有效題數: {final_valid_count}/36\n"
-
-    # =======================================================
-    # 🎯 產出最終排版 (融合答案摘要與核心 Log，精準對齊 3 個 API 輸出)
-    # =======================================================
-    summary_lines = [f"### 📊 學生選擇題最佳答案摘要 (經多重採樣與投票融合)\n```text"]
-    current_row = []
-    for i in range(1, 37):
-        current_row.append(f"q{i:02d}: {final_extracted_answers[i]}")
-        if len(current_row) == 4:
-            summary_lines.append("    |    ".join(current_row))
-            current_row = []
-    if current_row:
-        summary_lines.append("    |    ".join(current_row))
-    summary_lines.append("```\n💡 *本結果已經過三種不同靈敏度交叉比對與多數決演算法驗證，準確度已達最大化。*")
-    
-    answers_summary = "\n".join(summary_lines)
-    
-    # 🚀 將摘要文字置頂，與詳細底層日誌結合成單一文字區塊
-    final_combined_logs = f"{dimension_info}\n{answers_summary}\n\n========================================\n⚙️ 詳細後台決策與執行日誌 (Logs):\n========================================\n{accumulated_logs}"
-    
-    # 挑選單次表現最好、搜集題數最高的一組實體檔案回傳
-    best_single_scan = max(all_scans_results, key=lambda x: x["valid_count"])
-    
-    # 🎯 嚴格遵循 3 個元素的 tuple 回傳: [0]=CSV路徑, [1]=圖片路徑, [2]=文字日誌
-    return best_single_scan["csv"], best_single_scan["img"], final_combined_logs
+    return answers
+
+
+def _merge_answer_sources(primary: dict[int, str], fallback: dict[int, str]) -> dict[int, str]:
+    merged = dict(primary)
+    for number, answer in fallback.items():
+        if merged.get(number, "—") == "—" and answer != "—":
+            merged[number] = answer
+    return merged
+
+
+def _cleanup_old_generated_files(max_age_seconds: int = 7200) -> None:
+    now = time.time()
+    for path in GENERATED_DIR.iterdir():
+        try:
+            if now - path.stat().st_mtime > max_age_seconds:
+                if path.is_dir():
+                    shutil.rmtree(path, ignore_errors=True)
+                else:
+                    path.unlink(missing_ok=True)
+        except OSError:
+            pass
+
+
+def _format_answer_summary(final_answers: dict[int, str]) -> str:
+    lines = [
+        "### 📊 多重採樣投票結果",
+        "```text",
+    ]
+
+    row: list[str] = []
+    for number in sorted(final_answers):
+        row.append(f"q{number:02d}: {final_answers[number]}")
+        if len(row) == 4:
+            lines.append("    |    ".join(row))
+            row = []
+
+    if row:
+        lines.append("    |    ".join(row))
+
+    lines.extend(
+        [
+            "```",
+            "💡 請在前端逐題人工核對；投票結果不是免校對的正式答案。",
+        ]
+    )
+    return "\n".join(lines)
+
+
+def process_omr(image_file: str | None, template_content: str | None = None):
+    if image_file is None:
+        return None, None, "錯誤：請先上傳答案卡圖片。"
+
+    with PROCESS_LOCK:
+        request_root: Path | None = None
+
+        try:
+            ensure_omrchecker()
+            _cleanup_old_generated_files()
+
+            image = cv2.imread(str(image_file))
+            if image is None:
+                raise ValueError("OpenCV 無法讀取上傳圖片。")
+
+            image_height, image_width = image.shape[:2]
+            dimension_info = (
+                f"【圖片載入成功】解析度：{image_width} × {image_height} px"
+            )
+
+            base_template, _ = _load_template_content(template_content)
+            question_numbers = _question_numbers_from_template(base_template)
+            question_count = len(question_numbers)
+
+            request_id = uuid.uuid4().hex
+            request_root = Path(
+                tempfile.mkdtemp(prefix=f"hf_omr_{request_id}_", dir=str(ROOT_DIR))
+            )
+            result_dir = GENERATED_DIR / request_id
+            result_dir.mkdir(parents=True, exist_ok=True)
+
+            scans: list[dict[str, Any]] = []
+            logs = [
+                dimension_info,
+                f"【Template】題號範圍：q{question_numbers[0]}–q{question_numbers[-1]}，共 {question_count} 題",
+                "--- 開始三輪多重採樣 ---",
+            ]
+
+            for index, profile in enumerate(SENSITIVITY_PROFILES, start=1):
+                profile_dir = request_root / f"profile_{index}"
+                images_dir = profile_dir / "images"
+                output_dir = request_root / f"output_{index}"
+
+                images_dir.mkdir(parents=True, exist_ok=True)
+                output_dir.mkdir(parents=True, exist_ok=True)
+
+                current_template = _apply_profile(base_template, profile)
+                (profile_dir / "template.json").write_text(
+                    json.dumps(current_template, ensure_ascii=False, indent=2),
+                    encoding="utf-8",
+                )
+                _write_headless_config(profile_dir)
+                _copy_alignment_assets(current_template, profile_dir)
+
+                shutil.copy2(image_file, images_dir / "sheet.jpg")
+
+                environment = os.environ.copy()
+                environment["PYTHONPATH"] = (
+                    str(MOCK_DIR)
+                    + os.pathsep
+                    + environment.get("PYTHONPATH", "")
+                )
+                environment["OMR_CHECKER_CONTAINER"] = "true"
+                environment["MPLBACKEND"] = "Agg"
+
+                command = [
+                    sys.executable,
+                    "main.py",
+                    "-i",
+                    str(profile_dir),
+                    "-o",
+                    str(output_dir),
+                ]
+
+                result = subprocess.run(
+                    command,
+                    cwd=OMR_DIR,
+                    capture_output=True,
+                    text=True,
+                    env=environment,
+                    timeout=180,
+                )
+
+                full_log = (result.stdout or "") + "\n" + (result.stderr or "")
+                csv_path = _find_output_file(output_dir, {".csv"})
+                image_path = _find_output_file(
+                    output_dir,
+                    {".jpg", ".jpeg", ".png"},
+                    prefer="CheckedOMRs",
+                )
+
+                csv_answers = _parse_csv_answers(csv_path, question_numbers)
+                log_answers = _parse_log_answers(full_log, question_numbers)
+                answers = _merge_answer_sources(csv_answers, log_answers)
+                valid_count = sum(answer != "—" for answer in answers.values())
+
+                logs.append(
+                    f"{profile['name']}：returncode={result.returncode}，辨識 {valid_count}/{question_count} 題"
+                )
+
+                if result.returncode != 0:
+                    logs.append("本輪錯誤摘要：" + full_log[-1500:].replace("\n", " | "))
+
+                scans.append(
+                    {
+                        "profile": profile["name"],
+                        "answers": answers,
+                        "valid_count": valid_count,
+                        "csv": csv_path,
+                        "image": image_path,
+                    }
+                )
+
+            if not scans:
+                raise RuntimeError("沒有完成任何 OMR 掃描。")
+
+            final_answers: dict[int, str] = {}
+            for number in question_numbers:
+                votes = [
+                    scan["answers"][number]
+                    for scan in scans
+                    if scan["answers"][number] != "—"
+                ]
+
+                if not votes:
+                    final_answers[number] = "—"
+                    continue
+
+                vote_counts = Counter(votes)
+                final_answer = vote_counts.most_common(1)[0][0]
+                final_answers[number] = final_answer
+
+                if len(vote_counts) > 1:
+                    logs.append(
+                        f"q{number} 分歧：{dict(vote_counts)} → {final_answer}"
+                    )
+
+            final_valid_count = sum(answer != "—" for answer in final_answers.values())
+            logs.append(
+                f"融合後辨識：{final_valid_count}/{question_count} 題；空白 {question_count - final_valid_count} 題。"
+            )
+
+            # Return a CSV that matches the fused answers, rather than a raw
+            # CSV from only one sensitivity profile.
+            fused_csv = result_dir / "omr_fused_answers.csv"
+            pd.DataFrame(
+                [{f"q{number}": final_answers[number] for number in question_numbers}]
+            ).to_csv(fused_csv, index=False, encoding="utf-8-sig")
+
+            best_scan = max(scans, key=lambda item: item["valid_count"])
+            checked_image: Path | None = None
+            if best_scan["image"] and Path(best_scan["image"]).is_file():
+                source_image = Path(best_scan["image"])
+                checked_image = result_dir / (
+                    "checked_omr" + source_image.suffix.lower()
+                )
+                shutil.copy2(source_image, checked_image)
+
+            combined_log = (
+                _format_answer_summary(final_answers)
+                + "\n\n========================================\n"
+                + "⚙️ 執行摘要\n"
+                + "========================================\n"
+                + "\n".join(logs)
+            )
+
+            return (
+                str(fused_csv),
+                str(checked_image) if checked_image else None,
+                combined_log,
+            )
+
+        except subprocess.TimeoutExpired:
+            return None, None, "錯誤：OMRChecker 單輪處理超過 180 秒，已中止。"
+        except Exception as error:
+            return None, None, f"錯誤：{type(error).__name__}: {error}"
+        finally:
+            if request_root is not None:
+                shutil.rmtree(request_root, ignore_errors=True)
 
 
-# =======================================================
-# 建立符合 2 輸入、3 輸出規格的 Gradio 介面
-# =======================================================
 interface = gr.Interface(
     fn=process_omr,
     inputs=[
         gr.Image(type="filepath", label="1. 上傳 OMR 答案卡圖片 (JPG/PNG)"),
-        gr.Textbox(lines=6, label="2. [選填] Template JSON 配置 (網頁手動時可不填)", placeholder="留空時系統會自動套用雲端後端的 template.json")
+        gr.Textbox(
+            lines=6,
+            label="2. [選填] Template JSON 配置",
+            placeholder="留空時自動使用 Space 根目錄的 template.json",
+        ),
     ],
     outputs=[
-        gr.File(label="下載辨識結果 (CSV)"),              # 對應 API 輸出的 [0]
-        gr.Image(label="視覺化劃記檢視圖片"),              # 對應 API 輸出的 [1]
-        gr.Textbox(label="系統執行日誌 (Logs)", lines=15)    # 對應 API 輸出的 [2] (內含 4 欄答案摘要 + 投票軌跡)
+        gr.File(label="下載融合辨識結果 (CSV)"),
+        gr.Image(label="視覺化劃記檢視圖片"),
+        gr.Textbox(label="辨識答案與執行摘要", lines=18),
     ],
     title="DSE OMR 雲端辨識系統 API 後端",
-    description="本系統基於 OMRChecker 內核驅動。採用「多重採樣與多數決投票 (Ensemble Voting)」演算法，在背景以三種靈敏度掃描試卷，排除橡皮擦雜訊並拯救淡筆跡。完美對齊 2 輸入、3 輸出之自動化 API 規格。"
+    description=(
+        "以 OMRChecker 執行三輪靈敏度掃描及多數決。"
+        "正式提交前仍須由學生在前端人工核對。"
+    ),
 )
 
 if __name__ == "__main__":
-    interface.launch(server_name="0.0.0.0", server_port=7860)
+    interface.launch(server_name="0.0.0.0", server_port=7860)