app.py

from __future__ import annotations

import copy
import json
import itertools
import math
import os
import re
import shutil
import subprocess
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
import numpy as np
import fastapi
import starlette

print(
    "【系統版本】"
    f"gradio={gr.__version__}, "
    f"fastapi={fastapi.__version__}, "
    f"starlette={starlette.__version__}"
)

# =======================================================
# 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: Any, *args: Any, **kwargs: Any) -> Any:
        if isinstance(schema, bool):
            return "Any"
        return _orig_get_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)

    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}")

# =======================================================
# Paths and repository settings
# =======================================================
ROOT_DIR = Path(__file__).resolve().parent
OMR_DIR = ROOT_DIR / "OMRChecker"
MOCK_DIR = ROOT_DIR / "mock_libs"
GENERATED_DIR = ROOT_DIR / "generated"
MARKER_CONFIG_PATH = ROOT_DIR / "marker_config.json"
MARKER_REFERENCE_PATH = ROOT_DIR / "marker_reference.png"

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",
    )


_prepare_headless_helpers()
GENERATED_DIR.mkdir(parents=True, exist_ok=True)


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:
        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 _load_marker_config() -> dict[str, Any] | None:
    """Load the optional four-corner marker configuration.

    The Space can still run without marker_config.json, in which case the
    original OMRChecker FeatureBasedAlignment path is used as a fallback.
    """
    if not MARKER_CONFIG_PATH.is_file():
        return None

    try:
        data = json.loads(MARKER_CONFIG_PATH.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise ValueError(f"marker_config.json 無法讀取：{error}") from error

    dimensions = data.get("pageDimensions")
    markers = data.get("largeCornerMarkers")
    if (
        not isinstance(dimensions, list)
        or len(dimensions) != 2
        or not isinstance(markers, dict)
    ):
        raise ValueError("marker_config.json 缺少 pageDimensions 或 largeCornerMarkers。")

    required = {"top_left", "top_right", "bottom_left", "bottom_right"}
    if not required.issubset(markers):
        raise ValueError("marker_config.json 的四個大型定位點不完整。")

    return data


def _resize_for_detection(image: np.ndarray, max_dimension: int = 1800) -> tuple[np.ndarray, float]:
    """Downscale only for marker detection; return scale back to original."""
    height, width = image.shape[:2]
    longest = max(height, width)
    if longest <= max_dimension:
        return image.copy(), 1.0

    ratio = max_dimension / float(longest)
    resized = cv2.resize(
        image,
        (max(1, int(round(width * ratio))), max(1, int(round(height * ratio)))),
        interpolation=cv2.INTER_AREA,
    )
    return resized, 1.0 / ratio


class MarkerDetectionError(RuntimeError):
    """Marker failure carrying an optional candidate-debug image."""

    def __init__(self, message: str, debug_image: np.ndarray | None = None):
        super().__init__(message)
        self.debug_image = debug_image


def _hierarchy_max_depth(hierarchy_row: np.ndarray | None, index: int) -> int:
    if hierarchy_row is None:
        return 0

    def walk(node: int, visited: set[int]) -> int:
        if node < 0 or node in visited:
            return 0
        visited = set(visited)
        visited.add(node)
        child = int(hierarchy_row[node][2])
        best = 0
        while child >= 0:
            best = max(best, 1 + walk(child, visited))
            child = int(hierarchy_row[child][0])
        return best

    return walk(index, set())


def _square_marker_candidates(
    image: np.ndarray, marker_config: dict[str, Any]
) -> list[dict[str, Any]]:
    """Find nested square candidates and reject small five-question marks.

    V3 relies on the contour hierarchy of the five-layer marker. Text glyphs,
    answer fills and the smaller group markers should not survive the combined
    depth, size and aspect-ratio filters.
    """
    detection_image, scale_back = _resize_for_detection(image)
    gray = cv2.cvtColor(detection_image, cv2.COLOR_BGR2GRAY)
    gray = cv2.createCLAHE(clipLimit=2.2, tileGridSize=(8, 8)).apply(gray)
    blurred = cv2.GaussianBlur(gray, (5, 5), 0)
    _, binary = cv2.threshold(
        blurred, 0, 255, cv2.THRESH_BINARY_INV + cv2.THRESH_OTSU
    )

    contours, hierarchy = cv2.findContours(
        binary, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE
    )
    hierarchy_row = hierarchy[0] if hierarchy is not None else None
    height, width = gray.shape[:2]
    page_width, page_height = map(float, marker_config['pageDimensions'])
    outer_size = float(marker_config.get('largeMarkerOuterSize', 144))
    options = marker_config.get('detection') or {}
    min_depth = int(options.get('minNestedDepth', 3))
    side_range = options.get('candidateSideScaleRange', [0.38, 2.50])

    scale_x = width / page_width
    scale_y = height / page_height
    expected_side = outer_size * math.sqrt(max(scale_x * scale_y, 1e-9))
    min_side = max(12.0, expected_side * float(side_range[0]))
    max_side = expected_side * float(side_range[1])

    raw: list[dict[str, Any]] = []
    for index, contour in enumerate(contours):
        x, y, box_width, box_height = cv2.boundingRect(contour)
        side = (box_width + box_height) / 2.0
        if side < min_side or side > max_side:
            continue

        aspect = box_width / max(float(box_height), 1.0)
        if not 0.76 <= aspect <= 1.24:
            continue

        perimeter = cv2.arcLength(contour, True)
        if perimeter <= 0:
            continue
        polygon = cv2.approxPolyDP(contour, 0.035 * perimeter, True)
        if len(polygon) < 4 or len(polygon) > 8:
            continue

        depth = _hierarchy_max_depth(hierarchy_row, index)
        if depth < min_depth:
            continue

        area = float(cv2.contourArea(contour))
        fill_ratio = area / max(float(box_width * box_height), 1.0)
        if fill_ratio < 0.55:
            continue

        center_x = (x + box_width / 2.0) * scale_back
        center_y = (y + box_height / 2.0) * scale_back
        raw.append({
            'id': index,
            'center': (center_x, center_y),
            'side': side * scale_back,
            'area': area * scale_back * scale_back,
            'depth': depth,
            'aspect': aspect,
            'rect': (
                int(round(x * scale_back)),
                int(round(y * scale_back)),
                int(round(box_width * scale_back)),
                int(round(box_height * scale_back)),
            ),
        })

    # The same physical marker can create several nested candidate contours.
    # Keep the largest candidate around each center.
    raw.sort(key=lambda item: (-item['side'], -item['depth']))
    candidates: list[dict[str, Any]] = []
    for candidate in raw:
        cx, cy = candidate['center']
        duplicate = False
        for kept in candidates:
            kx, ky = kept['center']
            distance = math.hypot(cx - kx, cy - ky)
            if distance < 0.22 * max(candidate['side'], kept['side']):
                duplicate = True
                break
        if not duplicate:
            candidates.append(candidate)

    return candidates


def _marker_candidate_debug(
    image: np.ndarray, candidates: list[dict[str, Any]]
) -> np.ndarray:
    debug = image.copy()
    for candidate in candidates:
        x, y, w, h = candidate['rect']
        cv2.rectangle(debug, (x, y), (x + w, y + h), (255, 80, 0), 3)
        cv2.putText(
            debug,
            f"d{candidate['depth']} s{int(candidate['side'])}",
            (x, max(18, y - 7)),
            cv2.FONT_HERSHEY_SIMPLEX,
            0.52,
            (180, 40, 0),
            2,
            cv2.LINE_AA,
        )
    return debug


def _quadrilateral_geometry_score(
    points: np.ndarray,
    marker_sides: list[float],
    image_width: int,
    image_height: int,
    options: dict[str, Any],
) -> float | None:
    tl, tr, br, bl = points
    if not (tl[0] < tr[0] and bl[0] < br[0] and tl[1] < bl[1] and tr[1] < br[1]):
        return None
    if not cv2.isContourConvex(points.reshape(-1, 1, 2).astype(np.float32)):
        return None

    def distance(a: np.ndarray, b: np.ndarray) -> float:
        return float(np.linalg.norm(a - b))

    top = distance(tl, tr)
    bottom = distance(bl, br)
    left = distance(tl, bl)
    right = distance(tr, br)
    diag1 = distance(tl, br)
    diag2 = distance(tr, bl)
    polygon_area = abs(float(cv2.contourArea(points.reshape(-1, 1, 2))))
    area_ratio = polygon_area / max(float(image_width * image_height), 1.0)

    if area_ratio < float(options.get('minQuadrilateralAreaRatio', 0.20)):
        return None
    if min(top, bottom) < image_width * float(options.get('minHorizontalSpanRatio', 0.48)):
        return None
    if min(left, right) < image_height * float(options.get('minVerticalSpanRatio', 0.48)):
        return None

    def in_range(value: float, range_value: list[float]) -> bool:
        return float(range_value[0]) <= value <= float(range_value[1])

    width_ratio = top / max(bottom, 1.0)
    height_ratio = left / max(right, 1.0)
    diagonal_ratio = diag1 / max(diag2, 1.0)
    if not in_range(width_ratio, options.get('topBottomWidthRatioRange', [0.48, 1.85])):
        return None
    if not in_range(height_ratio, options.get('leftRightHeightRatioRange', [0.48, 1.85])):
        return None
    if not in_range(diagonal_ratio, options.get('diagonalRatioRange', [0.55, 1.80])):
        return None

    size_mean = float(np.mean(marker_sides))
    size_cv = float(np.std(marker_sides) / max(size_mean, 1e-6))
    if size_cv > float(options.get('maxMarkerSizeCv', 0.42)):
        return None

    # Lower is better. Log penalties treat reciprocal distortion symmetrically.
    return (
        abs(math.log(max(width_ratio, 1e-6)))
        + abs(math.log(max(height_ratio, 1e-6)))
        + 0.55 * abs(math.log(max(diagonal_ratio, 1e-6)))
        + 1.4 * size_cv
        - 0.20 * area_ratio
    )


def _corner_regions() -> dict[str, Any]:
    """Normalized safe-search regions for the four main markers."""
    return {
        'top_left': lambda nx, ny: 0.08 <= nx <= 0.48 and 0.18 <= ny <= 0.53,
        'top_right': lambda nx, ny: 0.68 <= nx <= 1.02 and 0.18 <= ny <= 0.53,
        'bottom_right': lambda nx, ny: 0.68 <= nx <= 1.02 and 0.72 <= ny <= 1.02,
        'bottom_left': lambda nx, ny: 0.08 <= nx <= 0.48 and 0.72 <= ny <= 1.02,
    }


def _build_corner_pools(
    candidates: list[dict[str, Any]],
    image_width: int,
    image_height: int,
    marker_config: dict[str, Any],
) -> dict[str, list[tuple[float, dict[str, Any]]]]:
    page_width, page_height = map(float, marker_config['pageDimensions'])
    expected_raw = marker_config['largeCornerMarkers']
    options = marker_config.get('detection') or {}
    max_per_corner = int(options.get('maxCandidatesPerCorner', 8))
    expected = {
        name: (point[0] / page_width, point[1] / page_height)
        for name, point in expected_raw.items()
    }
    regions = _corner_regions()
    expected_side = float(marker_config.get('largeMarkerOuterSize', 144)) * math.sqrt(
        max((image_width / page_width) * (image_height / page_height), 1e-9)
    )

    pools: dict[str, list[tuple[float, dict[str, Any]]]] = {}
    for marker_name in ('top_left', 'top_right', 'bottom_right', 'bottom_left'):
        ex, ey = expected[marker_name]
        scored: list[tuple[float, dict[str, Any]]] = []
        for candidate in candidates:
            cx, cy = candidate['center']
            nx, ny = cx / image_width, cy / image_height
            if not regions[marker_name](nx, ny):
                continue
            distance_score = math.hypot(nx - ex, ny - ey)
            size_score = abs(
                math.log(max(candidate['side'] / max(expected_side, 1e-6), 1e-6))
            )
            depth_bonus = -0.015 * max(0, int(candidate.get('depth', 3)) - 3)
            rescue_penalty = 0.025 if candidate.get('source') == 'local_rescue' else 0.0
            match_bonus = -0.035 * max(0.0, float(candidate.get('match_score', 0.0)) - 0.40)
            scored.append((
                distance_score + 0.20 * size_score + depth_bonus
                + rescue_penalty + match_bonus,
                candidate,
            ))
        scored.sort(key=lambda item: item[0])
        pools[marker_name] = scored[:max_per_corner]
    return pools


def _select_best_marker_set(
    pools: dict[str, list[tuple[float, dict[str, Any]]]],
    image_width: int,
    image_height: int,
    options: dict[str, Any],
) -> tuple[float, tuple[dict[str, Any], ...]] | None:
    ordered_names = ('top_left', 'top_right', 'bottom_right', 'bottom_left')
    if any(not pools.get(name) for name in ordered_names):
        return None

    best: tuple[float, tuple[dict[str, Any], ...]] | None = None
    for combination in itertools.product(*(pools[name] for name in ordered_names)):
        local_scores = [item[0] for item in combination]
        selected = tuple(item[1] for item in combination)
        if len({str(item['id']) for item in selected}) != 4:
            continue
        points = np.array([item['center'] for item in selected], dtype=np.float32)
        geometry_score = _quadrilateral_geometry_score(
            points,
            [float(item['side']) for item in selected],
            image_width,
            image_height,
            options,
        )
        if geometry_score is None:
            continue
        total_score = sum(local_scores) + geometry_score
        if best is None or total_score < best[0]:
            best = (total_score, selected)
    return best


def _load_marker_reference() -> np.ndarray:
    reference = cv2.imread(str(MARKER_REFERENCE_PATH), cv2.IMREAD_GRAYSCALE)
    if reference is None:
        raise MarkerDetectionError(
            'V3.1 局部補救需要 marker_reference.png，但專案根目錄找不到或無法讀取。'
        )
    return reference


def _local_template_rescue_candidate(
    image: np.ndarray,
    marker_name: str,
    marker_config: dict[str, Any],
    synthetic_id: str,
) -> dict[str, Any] | None:
    """Search one corner ROI using local contrast, multi-threshold and edge matching.

    This path is deliberately only used after the strict V3 contour detector
    cannot produce a valid four-marker set.  It handles the common case where
    a phone/hand shadow darkens the white rings and collapses contour hierarchy.
    """
    local_options = marker_config.get('localRescue') or {}
    page_width, page_height = map(float, marker_config['pageDimensions'])
    image_height, image_width = image.shape[:2]
    expected_point = marker_config['largeCornerMarkers'][marker_name]
    expected_x = expected_point[0] / page_width * image_width
    expected_y = expected_point[1] / page_height * image_height

    half_width = image_width * float(local_options.get('roiHalfWidthRatio', 0.22))
    half_height = image_height * float(local_options.get('roiHalfHeightRatio', 0.20))
    x0 = max(0, int(round(expected_x - half_width)))
    x1 = min(image_width, int(round(expected_x + half_width)))
    y0 = max(0, int(round(expected_y - half_height)))
    y1 = min(image_height, int(round(expected_y + half_height)))
    if x1 - x0 < 40 or y1 - y0 < 40:
        return None

    roi = image[y0:y1, x0:x1]
    gray = cv2.cvtColor(roi, cv2.COLOR_BGR2GRAY)
    gray = cv2.createCLAHE(clipLimit=2.5, tileGridSize=(8, 8)).apply(gray)
    gray = cv2.GaussianBlur(gray, (3, 3), 0)
    _, local_otsu = cv2.threshold(
        gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU
    )
    block_size = int(local_options.get('adaptiveBlockSize', 31))
    if block_size % 2 == 0:
        block_size += 1
    block_size = max(15, block_size)
    adaptive = cv2.adaptiveThreshold(
        gray,
        255,
        cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
        cv2.THRESH_BINARY,
        block_size,
        float(local_options.get('adaptiveC', 7)),
    )
    edges = cv2.Canny(gray, 50, 150)

    reference = _load_marker_reference()
    expected_side = float(marker_config.get('largeMarkerOuterSize', 144)) * math.sqrt(
        max((image_width / page_width) * (image_height / page_height), 1e-9)
    )
    scales = local_options.get(
        'templateMatchScales', [0.68, 0.78, 0.88, 0.98, 1.08, 1.18, 1.28]
    )
    angles = local_options.get('templateMatchAngles', [-10, 0, 10])
    threshold = float(local_options.get('templateMatchThreshold', 0.40))

    best: tuple[float, dict[str, Any]] | None = None
    for scale in scales:
        side = max(24, int(round(expected_side * float(scale))))
        if side >= roi.shape[0] or side >= roi.shape[1]:
            continue
        interpolation = cv2.INTER_AREA if side < reference.shape[0] else cv2.INTER_CUBIC
        resized_reference = cv2.resize(reference, (side, side), interpolation=interpolation)
        for angle in angles:
            matrix = cv2.getRotationMatrix2D((side / 2.0, side / 2.0), float(angle), 1.0)
            template = cv2.warpAffine(
                resized_reference,
                matrix,
                (side, side),
                flags=cv2.INTER_LINEAR,
                borderMode=cv2.BORDER_CONSTANT,
                borderValue=255,
            )
            if gray.shape[0] < side or gray.shape[1] < side:
                continue

            gray_result = cv2.matchTemplate(gray, template, cv2.TM_CCOEFF_NORMED)
            _, gray_score, _, location = cv2.minMaxLoc(gray_result)
            lx, ly = location

            def score_at(source: np.ndarray, target: np.ndarray) -> float:
                result = cv2.matchTemplate(source, target, cv2.TM_CCOEFF_NORMED)
                if ly >= result.shape[0] or lx >= result.shape[1]:
                    return -1.0
                return float(result[ly, lx])

            otsu_score = score_at(local_otsu, template)
            adaptive_score = score_at(adaptive, template)
            edge_template = cv2.Canny(template, 50, 150)
            edge_score = score_at(edges, edge_template)
            combined_score = (
                0.62 * float(gray_score)
                + 0.16 * otsu_score
                + 0.12 * adaptive_score
                + 0.10 * edge_score
            )
            if best is None or combined_score > best[0]:
                center_x = x0 + lx + side / 2.0
                center_y = y0 + ly + side / 2.0
                candidate = {
                    'id': synthetic_id,
                    'center': (center_x, center_y),
                    # Use the physically expected outer size for cross-corner
                    # size consistency; the matched inner/outer crop may be smaller.
                    'side': expected_side,
                    'area': expected_side * expected_side,
                    'depth': int(marker_config.get('largeMarkerNestedLevels', 5)),
                    'aspect': 1.0,
                    'rect': (
                        int(round(center_x - side / 2.0)),
                        int(round(center_y - side / 2.0)),
                        side,
                        side,
                    ),
                    'source': 'local_rescue',
                    'match_score': combined_score,
                    'gray_score': float(gray_score),
                    'otsu_score': otsu_score,
                    'adaptive_score': adaptive_score,
                    'edge_score': edge_score,
                }
                best = (combined_score, candidate)

    if best is None or best[0] < threshold:
        return None
    return best[1]


def _merge_marker_candidates(
    base: list[dict[str, Any]], additions: list[dict[str, Any]]
) -> list[dict[str, Any]]:
    merged = list(base)
    for candidate in additions:
        cx, cy = candidate['center']
        duplicate = False
        for kept in merged:
            kx, ky = kept['center']
            if math.hypot(cx - kx, cy - ky) < 0.28 * max(
                float(candidate['side']), float(kept['side'])
            ):
                duplicate = True
                break
        if not duplicate:
            merged.append(candidate)
    return merged


def _detect_large_corner_markers(
    image: np.ndarray, marker_config: dict[str, Any]
) -> tuple[np.ndarray, np.ndarray, bool]:
    """Detect V3 markers, then use V3.1 local rescue only when necessary."""
    candidates = _square_marker_candidates(image, marker_config)
    image_height, image_width = image.shape[:2]
    options = marker_config.get('detection') or {}

    pools = _build_corner_pools(candidates, image_width, image_height, marker_config)
    best = _select_best_marker_set(
        pools, image_width, image_height, options
    )
    rescue_used = False

    local_options = marker_config.get('localRescue') or {}
    rescue_enabled = bool(local_options.get('enabled', True))
    minimum_global = int(local_options.get('minimumGlobalCandidates', 2))
    if best is None and rescue_enabled and len(candidates) >= minimum_global:
        missing_names = [
            name for name in ('top_left', 'top_right', 'bottom_right', 'bottom_left')
            if not pools.get(name)
        ]
        # If each region has a candidate but the set fails geometry, retest all
        # four ROIs.  Otherwise only search the empty corner(s), limiting work.
        search_names = missing_names or [
            'top_left', 'top_right', 'bottom_right', 'bottom_left'
        ]
        max_rescue_corners = int(local_options.get('maxRescueCorners', 2))
        if len(search_names) <= max_rescue_corners or not missing_names:
            rescued: list[dict[str, Any]] = []
            for index, marker_name in enumerate(search_names):
                candidate = _local_template_rescue_candidate(
                    image,
                    marker_name,
                    marker_config,
                    synthetic_id=f"rescue-{marker_name}-{index}",
                )
                if candidate is not None:
                    rescued.append(candidate)
            if rescued:
                candidates = _merge_marker_candidates(candidates, rescued)
                pools = _build_corner_pools(
                    candidates, image_width, image_height, marker_config
                )
                best = _select_best_marker_set(
                    pools, image_width, image_height, options
                )
                rescue_used = best is not None and any(
                    item.get('source') == 'local_rescue' for item in best[1]
                )

    candidate_debug = _marker_candidate_debug(image, candidates)
    if best is None:
        empty_regions = [
            name for name in ('top_left', 'top_right', 'bottom_right', 'bottom_left')
            if not pools.get(name)
        ]
        if len(candidates) < 4:
            message = (
                f"只找到 {len(candidates)} 個符合 V3/V3.1 結構、尺寸或局部相似度的候選。"
            )
        elif empty_regions:
            message = "以下角點搜尋區沒有可靠標記：" + ", ".join(empty_regions)
        else:
            message = (
                '候選標記存在，但沒有任何四點組合通過面積、跨度、比例及尺寸一致性檢查。'
            )
        raise MarkerDetectionError(message, candidate_debug)

    selected = best[1]
    points = np.array([item['center'] for item in selected], dtype=np.float32)
    debug_image = candidate_debug
    labels = ('TL', 'TR', 'BR', 'BL')
    for label, candidate, (point_x, point_y) in zip(labels, selected, points):
        x, y, w, h = candidate['rect']
        color = (0, 215, 255) if candidate.get('source') == 'local_rescue' else (0, 255, 0)
        cv2.rectangle(debug_image, (x, y), (x + w, y + h), color, 6)
        position = (int(round(point_x)), int(round(point_y)))
        cv2.circle(debug_image, position, 18, color, 5)
        suffix = '*' if candidate.get('source') == 'local_rescue' else ''
        cv2.putText(
            debug_image,
            label + suffix,
            (position[0] + 18, position[1] - 18),
            cv2.FONT_HERSHEY_SIMPLEX,
            1.0,
            color,
            3,
            cv2.LINE_AA,
        )
    if rescue_used:
        cv2.putText(
            debug_image,
            'V3.1 LOCAL SHADOW RESCUE',
            (28, 48),
            cv2.FONT_HERSHEY_SIMPLEX,
            1.05,
            (0, 165, 255),
            3,
            cv2.LINE_AA,
        )

    return points, debug_image, rescue_used

def _warp_with_large_markers(
    image: np.ndarray, marker_config: dict[str, Any]
) -> tuple[np.ndarray, np.ndarray, bool]:
    source_points, debug_image, rescue_used = _detect_large_corner_markers(
        image, marker_config
    )
    page_width, page_height = map(int, marker_config['pageDimensions'])
    marker_points = marker_config['largeCornerMarkers']
    destination_points = np.array(
        [
            marker_points['top_left'],
            marker_points['top_right'],
            marker_points['bottom_right'],
            marker_points['bottom_left'],
        ],
        dtype=np.float32,
    )

    transform = cv2.getPerspectiveTransform(source_points, destination_points)
    warped = cv2.warpPerspective(
        image,
        transform,
        (page_width, page_height),
        flags=cv2.INTER_CUBIC,
        borderMode=cv2.BORDER_CONSTANT,
        borderValue=(255, 255, 255),
    )
    return warped, debug_image, rescue_used

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 _remove_feature_alignment(template: dict[str, Any]) -> dict[str, Any]:
    """Remove FeatureBasedAlignment after successful marker homography.

    Once the four large markers have already mapped the photo into the fixed
    3093 x 4374 template coordinate system, a second feature-based warp is both
    unnecessary and potentially harmful on a sheet containing many repeated
    A/B/C/D boxes and grid lines.
    """
    current = copy.deepcopy(template)
    current["preProcessors"] = [
        processor
        for processor in (current.get("preProcessors") or [])
        if not (
            isinstance(processor, dict)
            and processor.get("name") == "FeatureBasedAlignment"
        )
    ]
    return current


def _crop_answer_region(
    image: np.ndarray, marker_config: dict[str, Any], padding: int = 24
) -> np.ndarray:
    """Crop the corrected answer area for a clearer UI preview."""
    bounds = marker_config.get("answerRegionBounds") or {}
    height, width = image.shape[:2]
    left = max(0, int(bounds.get("left", 0)) - padding)
    top = max(0, int(bounds.get("top", 0)) - padding)
    right = min(width, int(bounds.get("right", width)) + padding)
    bottom = min(height, int(bounds.get("bottom", height)) + padding)
    if right <= left or bottom <= top:
        return image
    return image[top:bottom, left:right]


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:
    """Write a deliberately minimal OMRChecker config.

    OMRChecker's master branch has changed its JSON schema over time.  Keeping
    only the two long-standing image-level keys avoids schema failures caused
    by optional output keys that are present in some releases but absent in
    others.  If even this minimal config is rejected, the scan loop retries
    once without a local config.json so OMRChecker can use its own defaults.
    """
    config = {
        "outputs": {
            "show_image_level": 0,
            "save_image_level": 3,
        }
    }
    (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:
        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",
    ]

    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()

    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, 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)

            logs = [
                dimension_info,
                f"【Template】題號範圍：q{question_numbers[0]}–q{question_numbers[-1]}，共 {question_count} 題",
            ]

            # -------------------------------------------------------
            # Stage 1: marker-based perspective correction.
            # -------------------------------------------------------
            omr_input_path = Path(image_file)
            marker_debug_path: Path | None = None
            corrected_preview: Path | None = None
            marker_config = _load_marker_config()
            marker_warp_succeeded = False

            if marker_config is not None:
                try:
                    warped, marker_debug, rescue_used = _warp_with_large_markers(
                        image, marker_config
                    )
                    warped_full_path = result_dir / "01_perspective_corrected_full.jpg"
                    corrected_preview = result_dir / "02_corrected_answer_region.jpg"
                    marker_debug_path = result_dir / "00_detected_markers.jpg"

                    cv2.imwrite(
                        str(warped_full_path),
                        warped,
                        [int(cv2.IMWRITE_JPEG_QUALITY), 95],
                    )
                    cv2.imwrite(
                        str(corrected_preview),
                        _crop_answer_region(warped, marker_config),
                        [int(cv2.IMWRITE_JPEG_QUALITY), 95],
                    )
                    cv2.imwrite(
                        str(marker_debug_path),
                        marker_debug,
                        [int(cv2.IMWRITE_JPEG_QUALITY), 92],
                    )
                    omr_input_path = warped_full_path
                    marker_warp_succeeded = True
                    if rescue_used:
                        logs.append("【四角校正】V3.1 局部補救成功：陰影中的缺失角點已恢復，四點幾何檢查通過。")
                    else:
                        logs.append("【四角校正】V3 嚴格模式成功偵測四個大型定位點。")
                    logs.append("【四角校正】答案區已映射到固定座標。")
                    logs.append("【預覽說明】介面只顯示校正後答案區；上半頁不參與答案座標判定。")
                except Exception as marker_error:
                    fallback_path = result_dir / "01_original_marker_failure.jpg"
                    shutil.copy2(image_file, fallback_path)
                    corrected_preview = fallback_path
                    if isinstance(marker_error, MarkerDetectionError) and marker_error.debug_image is not None:
                        marker_debug_path = result_dir / "00_marker_candidates_failed.jpg"
                        cv2.imwrite(
                            str(marker_debug_path),
                            marker_error.debug_image,
                            [int(cv2.IMWRITE_JPEG_QUALITY), 92],
                        )

                    strict_mode = bool(marker_config.get("strictMarkerMode", True))
                    logs.append(
                        "【四角校正】失敗："
                        f"{type(marker_error).__name__}: {marker_error}"
                    )
                    if strict_mode:
                        failure_log = (
                            "### ❌ 四角定位未通過 V3.1 安全檢查\n\n"
                            "系統沒有把不可靠的校正圖交給 OMRChecker。請確保四個大型定位標記完整入鏡、"
                            "相片清晰，並避免手機影子完全遮蔽標記後重新拍攝。\n\n"
                            "========================================\n"
                            "⚙️ 執行摘要\n"
                            "========================================\n"
                            + "\n".join(logs)
                        )
                        return (
                            None,
                            str(marker_debug_path) if marker_debug_path else None,
                            str(corrected_preview) if corrected_preview else None,
                            None,
                            failure_log,
                        )

                    logs.append("【回退模式】strictMarkerMode=false，改用 FeatureBasedAlignment。")
            else:
                fallback_path = result_dir / "01_original_no_marker_config.jpg"
                shutil.copy2(image_file, fallback_path)
                corrected_preview = fallback_path
                logs.append("【四角校正】沒有 marker_config.json，使用原始對齊流程。")

            scans: list[dict[str, Any]] = []
            logs.append("--- 開始三輪多重採樣 ---")

            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)
                if marker_warp_succeeded:
                    current_template = _remove_feature_alignment(current_template)

                (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(omr_input_path, 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 "")

                # Runtime compatibility fallback: if the checked-out master
                # branch rejects our minimal config schema, remove config.json
                # and let OMRChecker load its own built-in defaults.
                if (
                    result.returncode != 0
                    and "Provided config JSON is Invalid" in full_log
                ):
                    (profile_dir / "config.json").unlink(missing_ok=True)
                    logs.append(
                        f"{profile['name']}：偵測到 OMRChecker config schema 差異，"
                        "已移除本地 config.json 並自動重試。"
                    )
                    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[-1800:].replace("\n", " | "))

                scans.append(
                    {
                        "profile": profile["name"],
                        "answers": answers,
                        "valid_count": valid_count,
                        "csv": csv_path,
                        "image": image_path,
                        "returncode": result.returncode,
                    }
                )

            successful_scans = [scan for scan in scans if scan["returncode"] == 0]
            if not successful_scans:
                failure_log = (
                    "### ❌ OMRChecker 三輪均未能執行\n\n"
                    "本次沒有產生答案 CSV；這不是 100 題全部留空，而是辨識核心未成功啟動。\n\n"
                    "========================================\n"
                    "⚙️ 執行摘要\n"
                    "========================================\n"
                    + "\n".join(logs)
                )
                return (
                    None,
                    str(marker_debug_path) if marker_debug_path else None,
                    str(corrected_preview) if corrected_preview else None,
                    None,
                    failure_log,
                )

            final_answers: dict[int, str] = {}
            for number in question_numbers:
                votes = [
                    scan["answers"][number]
                    for scan in successful_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} 題。"
            )

            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(successful_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(marker_debug_path) if marker_debug_path else None,
                str(corrected_preview) if corrected_preview else None,
                str(checked_image) if checked_image else None,
                combined_log,
            )

        except subprocess.TimeoutExpired:
            return None, None, None, None, "錯誤：OMRChecker 單輪處理超過 180 秒，已中止。"
        except Exception as error:
            return None, None, None, None, f"錯誤：{type(error).__name__}: {error}"
        finally:
            if request_root is not None:
                shutil.rmtree(request_root, ignore_errors=True)


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="留空時自動使用 Space 根目錄的 template.json",
        ),
    ],
    outputs=[
        gr.File(label="下載融合辨識結果 (CSV)"),
        gr.Image(label="四個大型定位點偵測結果"),
        gr.Image(label="透視校正後答案區"),
        gr.Image(label="OMRChecker 視覺化劃記檢視"),
        gr.Textbox(label="辨識答案與執行摘要", lines=22),
    ],
    title="DSE OMR V3.1 手機陰影局部補救辨識系統",
    description=(
        "先以 V3 嚴格模式偵測四個 144px 五層巢狀大型定位標記；"
        "若手機陰影令一至兩角輪廓失真，才啟動 V3.1 局部多閾值與模板補救。"
        "四點通過幾何安全檢查後會停用第二次 FeatureBasedAlignment，"
        "再以 OMRChecker 執行三輪靈敏度掃描及融合。"
        "定位未通過幾何安全檢查時會直接拒絕處理；正式提交前仍建議學生核對辨識結果。"
    ),
)

if __name__ == "__main__":
    interface.launch(
        server_name=os.getenv("GRADIO_SERVER_NAME", "0.0.0.0"),
        server_port=int(os.getenv("GRADIO_SERVER_PORT", "7860")),
        show_error=True,
    )