Mortal-Rating/batchmortal/browser.py

import logging
import os
import queue
import threading
import time
from collections import deque

from seleniumbase import SB

REVIEW_BASE_URL = "https://mjai.ekyu.moe"
DEFAULT_REVIEW_LANGUAGE = "zh-CN"
DEFAULT_REVIEW_UI = "classic"

# Mirrors the reviewer form field `select[name="lang"]`.
# Supported values on mjai.ekyu.moe are:
#   zh-CN -> Simplified Chinese, page /zh-cn.html
#   en    -> English, page /
#   ja    -> Japanese, page /ja.html
#   ko    -> Korean, page /ko.html
REVIEW_LANGUAGE_URL_PATHS = {
    "zh-CN": "/zh-cn.html",
    "en": "/",
    "ja": "/ja.html",
    "ko": "/ko.html",
}
REVIEW_LANGUAGE_ALIASES = {
    "zh": "zh-CN",
    "zh-cn": "zh-CN",
    "zh_cn": "zh-CN",
    "cn": "zh-CN",
    "en-us": "en",
    "english": "en",
    "jp": "ja",
    "ja-jp": "ja",
    "japanese": "ja",
    "kr": "ko",
    "ko-kr": "ko",
    "korean": "ko",
}
INPUT_SELECTOR = 'input[name="log-url"]'
SUBMIT_SELECTOR = 'button[name="submitBtn"]'
FORM_SELECTOR = 'form[name="reviewForm"]'
TURNSTILE_RESPONSE_SELECTOR = 'input[name="cf-turnstile-response"]'
RESULT_SELECTOR = "details > dl"
REPORT_URL_FRAGMENT = "/report/"
BAD_MOVE_STRICT_LIMIT = 5
BAD_MOVE_LOOSE_LIMIT = 10

REVIEW_UI_ALIASES = {
    "classic": "classic",
    "killerducky": "killerducky",
    "killer-ducky": "killerducky",
    "kd": "killerducky",
}


def normalize_review_ui(review_ui):
    if review_ui is None:
        return DEFAULT_REVIEW_UI

    key = str(review_ui).strip().lower().replace("_", "-")
    normalized = REVIEW_UI_ALIASES.get(key)
    if normalized:
        return normalized

    raise ValueError(
        f"Unsupported review UI '{review_ui}'. Supported values: classic, killerducky"
    )


PROXY_DIRECT = "direct"


def build_browser_kwargs(headless: bool, proxy: str | None) -> dict:
    """Translate headless/proxy into ``SB()`` kwargs.

    A proxy value of ``"direct"`` (case-insensitive) forces Chrome to bypass
    every proxy — including the macOS system proxy that Clash/etc. configure —
    by passing ``--proxy-server=direct://``. This is required for mjai.ekyu.moe:
    Cloudflare Turnstile flags datacenter proxy exit IPs and refuses to
    auto-issue a token, so the browser must reach the site from the residential
    IP. SeleniumBase's ``proxy=`` param cannot express this because its
    ``validate_proxy_string`` rejects ``direct://`` and falls back to None,
    which leaves Chrome on the system proxy.
    """
    kwargs = {"uc": True, "headless": headless}
    if proxy and str(proxy).strip().lower() == PROXY_DIRECT:
        kwargs["chromium_arg"] = "--proxy-server=direct://"
    elif proxy:
        kwargs["proxy"] = proxy
    return kwargs


def parse_killerducky_metadata(data):
    """Convert KillerDucky's report JSON into the metadata shape used by results.py."""
    data = data if isinstance(data, dict) else {}
    review = data.get("review")
    review = review if isinstance(review, dict) else {}

    total_matches = review.get("total_matches")
    total_reviewed = review.get("total_reviewed")
    matches_total = ""
    if (
        isinstance(total_matches, (int, float))
        and isinstance(total_reviewed, (int, float))
        and total_reviewed > 0
    ):
        rate = 100 * total_matches / total_reviewed
        matches_total = f"{int(total_matches)}/{int(total_reviewed)} = {rate:.3f}%"

    rating = review.get("rating")
    formatted_rating = ""
    if isinstance(rating, (int, float)):
        formatted_rating = f"{rating * 100:.3f}"

    def text(value):
        return "" if value is None else str(value)

    return {
        "engine": text(data.get("engine")),
        "model tag": text(review.get("model_tag")),
        "rating": formatted_rating,
        "matches/total": matches_total,
        "temperature": text(review.get("temperature")),
        "game length": text(data.get("game_length")),
        "player id": text(data.get("player_id")),
        "review duration": text(data.get("review_time")),
    }


def parse_killerducky_bad_move_stats(
    data,
    strict_limit=BAD_MOVE_STRICT_LIMIT,
    loose_limit=BAD_MOVE_LOOSE_LIMIT,
):
    """Calculate bad-move rates from KillerDucky's structured Mortal decisions."""
    data = data if isinstance(data, dict) else {}
    review = data.get("review")
    review = review if isinstance(review, dict) else {}
    total_reviewed = review.get("total_reviewed")
    denominator = total_reviewed if isinstance(total_reviewed, int) and total_reviewed >= 0 else None

    strict_count = 0
    loose_count = 0
    mismatch_count = 0
    unparsed_count = 0

    kyokus = review.get("kyokus")
    for kyoku in kyokus if isinstance(kyokus, list) else []:
        if not isinstance(kyoku, dict):
            continue
        entries = kyoku.get("entries")
        for entry in entries if isinstance(entries, list) else []:
            if not isinstance(entry, dict) or entry.get("is_equal") is not False:
                continue

            mismatch_count += 1
            actual_index = entry.get("actual_index")
            details = entry.get("details")
            if (
                not isinstance(actual_index, int)
                or not isinstance(details, list)
                or actual_index < 0
                or actual_index >= len(details)
                or not isinstance(details[actual_index], dict)
            ):
                unparsed_count += 1
                continue

            probability = details[actual_index].get("prob")
            if not isinstance(probability, (int, float)):
                unparsed_count += 1
                continue

            probability_percent = probability * 100
            if probability_percent <= strict_limit:
                strict_count += 1
            if probability_percent <= loose_limit:
                loose_count += 1

    def format_rate(count):
        if denominator is None or denominator <= 0:
            return ""
        return f"{100 * count / denominator:.3f}%"

    return {
        "badMoveRate5": format_rate(strict_count),
        "badMoveCount5": str(strict_count),
        "badMoveRate10": format_rate(loose_count),
        "badMoveCount10": str(loose_count),
        "badMoveDenominator": "" if denominator is None else str(denominator),
        "badMoveOrderLossCount": str(mismatch_count),
        "badMoveUnparsedCount": str(unparsed_count),
    }


def normalize_review_language(language):
    if language is None:
        return DEFAULT_REVIEW_LANGUAGE

    text = str(language).strip()
    if not text:
        return DEFAULT_REVIEW_LANGUAGE
    if text in REVIEW_LANGUAGE_URL_PATHS:
        return text

    normalized = text.lower().replace("_", "-")
    if normalized in REVIEW_LANGUAGE_ALIASES:
        return REVIEW_LANGUAGE_ALIASES[normalized]

    supported = ", ".join(REVIEW_LANGUAGE_URL_PATHS)
    raise ValueError(f"Unsupported review language '{language}'. Supported values: {supported}")


def build_review_url(language):
    normalized_language = normalize_review_language(language)
    return f"{REVIEW_BASE_URL}{REVIEW_LANGUAGE_URL_PATHS[normalized_language]}"


class ReviewSubmissionCoordinator:
    def __init__(
        self,
        base_interval=6.0,
        max_interval=20.0,
        cooldown_seconds=30.0,
        failure_threshold=2,
    ):
        self.base_interval = base_interval
        self.current_interval = base_interval
        self.max_interval = max_interval
        self.cooldown_seconds = cooldown_seconds
        self.failure_threshold = failure_threshold
        self.condition = threading.Condition()
        self.active_uuid = None
        self.next_submit_time = 0.0
        self.cooldown_until = 0.0
        self.consecutive_failures = 0

    def wait_for_submit_slot(self, uuid):
        with self.condition:
            while True:
                now = time.monotonic()
                if self.active_uuid is None and now >= self.next_submit_time and now >= self.cooldown_until:
                    self.active_uuid = uuid
                    return

                wake_at = max(self.next_submit_time, self.cooldown_until)
                if self.active_uuid is not None:
                    timeout = 0.5
                else:
                    timeout = max(0.2, wake_at - now)
                self.condition.wait(timeout=min(timeout, 1.0))

    def release_submit_slot(self, uuid, token_wait_seconds=0.0):
        with self.condition:
            if self.active_uuid == uuid:
                self.active_uuid = None

            self._adapt_interval_from_token(token_wait_seconds)
            self.next_submit_time = max(time.monotonic(), self.next_submit_time) + self.current_interval
            self.condition.notify_all()

    def report_outcome(
        self,
        uuid,
        success,
        error_text="",
        token_wait_seconds=0.0,
        submit_wait_seconds=0.0,
        result_wait_seconds=0.0,
    ):
        del uuid
        with self.condition:
            if success:
                self.consecutive_failures = 0
                if token_wait_seconds <= 8 and submit_wait_seconds <= 8 and result_wait_seconds <= 18:
                    self.current_interval = max(self.base_interval, self.current_interval - 0.5)
                elif token_wait_seconds >= 6 or submit_wait_seconds >= 10 or result_wait_seconds >= 18:
                    self.current_interval = min(self.max_interval, self.current_interval + 1.5)
            else:
                normalized = (error_text or "").lower()
                if any(
                    marker in normalized
                    for marker in (
                        "captcha",
                        "turnstile",
                        "rate limit",
                        "stalled before token issuance",
                        "timed out waiting for turnstile token",
                        "review submission never left the form page",
                        "timed out waiting for review results",
                    )
                ):
                    self.consecutive_failures += 1
                    self.current_interval = min(self.max_interval, self.current_interval + 2.5)
                    if self.consecutive_failures >= self.failure_threshold:
                        self.cooldown_until = max(
                            self.cooldown_until,
                            time.monotonic() + self.cooldown_seconds,
                        )
                        logging.warning(
                            "[Throttle] Consecutive review failures detected. "
                            f"Cooling down submissions for {self.cooldown_seconds:.0f}s."
                        )
                else:
                    self.consecutive_failures = 0

            self.condition.notify_all()

    def _adapt_interval_from_token(self, token_wait_seconds):
        if token_wait_seconds <= 0:
            return
        if token_wait_seconds >= 20:
            self.current_interval = min(self.max_interval, self.current_interval + 2.0)
        elif token_wait_seconds >= 12:
            self.current_interval = min(self.max_interval, self.current_interval + 1.0)
        elif token_wait_seconds <= 6:
            self.current_interval = max(self.base_interval, self.current_interval - 0.25)


class BrowserAutomator:
    def __init__(
        self,
        headless=True,
        proxy=None,
        submission_coordinator=None,
        controlled_submission=True,
        review_language=DEFAULT_REVIEW_LANGUAGE,
        review_ui=DEFAULT_REVIEW_UI,
    ):
        self.headless = headless
        self.proxy = proxy
        self.review_language = normalize_review_language(review_language)
        self.review_ui = normalize_review_ui(review_ui)
        self.review_url = build_review_url(self.review_language)
        self.controlled_submission = controlled_submission
        if controlled_submission:
            self.submission_coordinator = submission_coordinator or ReviewSubmissionCoordinator()
        else:
            self.submission_coordinator = None

    def run_worker(self, task_queue, result_queue, max_retries=3):
        while True:
            if task_queue.empty():
                break

            try:
                with SB(**build_browser_kwargs(self.headless, self.proxy)) as sb:
                    tasks_processed = 0

                    while True:
                        try:
                            task = task_queue.get(timeout=3)
                        except queue.Empty:
                            return

                        result = None
                        fatal_error = False
                        try:
                            result = self.analyze_single(sb, task)
                        except Exception as exc:
                            err_str = str(exc).lower()
                            logging.error(f"{task.get('log_prefix', '[' + task['uuid'] + ']')} ERROR exception: {exc}")

                            if any(
                                marker in err_str
                                for marker in ("no such window", "closed", "invalid session", "disconnected")
                            ):
                                fatal_error = True

                            try:
                                error_screenshot = os.path.join(task["mode_dir"], f"{task['uuid']}_error.png")
                                sb.save_screenshot(error_screenshot)
                            except Exception:
                                pass

                        if result:
                            result_queue.put({"status": "success", "task": task, "result": result})
                        else:
                            task["retries"] = task.get("retries", 0) + 1
                            if task["retries"] <= max_retries:
                                logging.warning(
                                    f"{task.get('log_prefix', '[' + task['uuid'] + ']')} RETRY Analysis failed. "
                                    f"Retrying ({task['retries']}/{max_retries}) with a fresh page load."
                                )
                                task_queue.put(task)
                            else:
                                logging.error(
                                    f"{task.get('log_prefix', '[' + task['uuid'] + ']')} SKIP Analysis permanently failed "
                                    f"after {max_retries} retries."
                                )
                                result_queue.put({"status": "fail", "task": task})

                        task_queue.task_done()
                        tasks_processed += 1

                        if tasks_processed >= 10:
                            logging.info("  [MEMORY] Worker hit 10 tasks limit. Recycling browser to flush memory...")
                            break

                        if fatal_error:
                            logging.warning("  [RECOVER] Browser instance dead. Respawning...")
                            break

            except Exception as spawn_err:
                logging.error(f"  [FATAL] Browser spawn failed: {spawn_err}. Retrying in 5s...")
                time.sleep(5)

    def iter_alternating_windows(self, tasks, max_retries=3):
        pending = deque(tasks)

        with SB(**build_browser_kwargs(self.headless, self.proxy)) as sb:
            slots = [
                {
                    "name": "window-a",
                    "handle": sb.driver.current_window_handle,
                    "ready": False,
                },
                {
                    "name": "window-b",
                    "handle": None,
                    "ready": False,
                },
            ]

            slot_index = 0
            while pending:
                task = pending.popleft()
                slot = slots[slot_index]
                slot_index = (slot_index + 1) % len(slots)

                try:
                    started_at = time.perf_counter()
                    self._ensure_rotation_slot_ready(sb, slot, task.get("log_prefix", '[' + task["uuid"] + ']'))
                    result = self._analyze_loaded_form(
                        sb,
                        task,
                        started_at=started_at,
                        ready_message=f"{slot['name']} ready for assigned task",
                    )
                    yield {"status": "success", "task": task, "result": result}
                except Exception as exc:
                    failure_event = self._handle_rotation_failure(
                        sb,
                        slot,
                        task,
                        exc,
                        max_retries,
                        pending,
                    )
                    if failure_event:
                        yield failure_event
                finally:
                    slot["ready"] = False

    def iter_dual_window_pipeline(self, tasks, max_retries=3):
        pending = deque(tasks)

        with SB(**build_browser_kwargs(self.headless, self.proxy)) as sb:
            active_slot = {
                "name": "active",
                "handle": sb.driver.current_window_handle,
                "task": None,
                "prepared": False,
                "started_at": 0.0,
                "submitted_at": 0.0,
            }
            standby_slot = {
                "name": "standby",
                "handle": self._open_pipeline_tab(sb),
                "task": None,
                "prepared": False,
                "started_at": 0.0,
                "submitted_at": 0.0,
            }

            while pending or active_slot["task"] or standby_slot["task"]:
                if active_slot["task"] is None:
                    if standby_slot["task"] and standby_slot["prepared"]:
                        active_slot, standby_slot = standby_slot, active_slot
                    elif standby_slot["task"] and not standby_slot["prepared"]:
                        self._reset_pipeline_slot(standby_slot)
                        continue
                    elif pending:
                        task = pending.popleft()
                        failure_event = self._prepare_pipeline_slot(sb, active_slot, task, max_retries, pending)
                        if failure_event:
                            yield failure_event
                            continue
                    else:
                        break

                if active_slot["task"] is None:
                    break

                active_task = active_slot["task"]

                try:
                    token_wait_seconds = self._submit_pipeline_slot(sb, active_slot, active_task)

                    if standby_slot["task"] is None and pending:
                        next_task = pending.popleft()
                        failure_event = self._prepare_pipeline_slot(sb, standby_slot, next_task, max_retries, pending)
                        if failure_event:
                            yield failure_event

                    result = self._collect_pipeline_result(sb, active_slot, active_task)
                    yield {"status": "success", "task": active_task, "result": result}
                except Exception as exc:
                    failure_event = self._handle_pipeline_failure(
                        sb,
                        active_slot,
                        active_task,
                        exc,
                        max_retries,
                        pending,
                    )
                    if failure_event:
                        yield failure_event
                finally:
                    active_slot["task"] = None
                    active_slot["prepared"] = False
                    active_slot["started_at"] = 0.0
                    active_slot["submitted_at"] = 0.0

                if standby_slot["task"] and standby_slot["prepared"]:
                    active_slot, standby_slot = standby_slot, active_slot

    def analyze_single(self, sb, task):
        uuid = task["uuid"]
        log_prefix = task.get("log_prefix", f"[{uuid}]")
        started_at = time.perf_counter()

        logging.info(f"{log_prefix} Opening fresh review form")
        self._open_fresh_review_page(sb, log_prefix)
        return self._analyze_loaded_form(
            sb,
            task,
            started_at=started_at,
            ready_message="Form ready and waiting for submit slot",
        )

    def _analyze_loaded_form(self, sb, task, started_at, ready_message):
        paipu_url = task["paipu_url"]
        uuid = task["uuid"]
        log_prefix = task.get("log_prefix", f"[{uuid}]")
        model_tag = task["model_tag"]
        output_dir = task["mode_dir"]
        save_screenshot = task.get("save_screenshot", False)

        os.makedirs(output_dir, exist_ok=True)
        screenshot_path = os.path.join(output_dir, f"{uuid}.png")
        local_paipu_path = os.path.join(output_dir, f"{uuid}.html")
        save_local_paipu = task.get("save_local_paipu", False)
        token_wait_seconds = 0.0
        submit_wait_seconds = 0.0
        result_wait_seconds = 0.0
        submit_slot_held = False
        submit_slot_released = False

        self._populate_form(sb, paipu_url, model_tag)
        logging.info(f"{log_prefix} {ready_message}")

        try:
            if self.submission_coordinator is not None:
                self.submission_coordinator.wait_for_submit_slot(log_prefix)
                submit_slot_held = True
                logging.info(f"{log_prefix} Submit slot granted, starting Turnstile")
            else:
                logging.info(f"{log_prefix} Unthrottled mode, starting Turnstile")
            self._prepare_review_form(sb)

            token_started_at = time.perf_counter()
            self._acquire_turnstile_token(sb, task, log_prefix)
            token_wait_seconds = time.perf_counter() - token_started_at
            logging.info(f"{log_prefix} Turnstile token ready in {token_wait_seconds:.1f}s")

            submit_started_at = time.perf_counter()
            self._submit_review(sb, log_prefix)
            self._wait_for_submission_departure_or_error(sb, log_prefix, timeout=15)
            submit_wait_seconds = time.perf_counter() - submit_started_at
            if self.submission_coordinator is not None:
                self.submission_coordinator.release_submit_slot(log_prefix, token_wait_seconds=token_wait_seconds)
                submit_slot_released = True

            logging.info(f"{log_prefix} Review submitted, waiting for final result")
            result_started_at = time.perf_counter()
            self._wait_for_result_or_error(sb, log_prefix, timeout=45)
            result_wait_seconds = time.perf_counter() - result_started_at
            result_url = sb.get_current_url()
            metadata = self._extract_metadata(sb)
            bad_move_stats = {}
            if task.get("analyze_bad_move_rate", False):
                bad_move_stats = self._extract_bad_move_stats(sb, log_prefix)
            if self.submission_coordinator is not None:
                self.submission_coordinator.report_outcome(
                    log_prefix,
                    success=True,
                    token_wait_seconds=token_wait_seconds,
                    submit_wait_seconds=submit_wait_seconds,
                    result_wait_seconds=result_wait_seconds,
                )
            logging.info(
                f"{log_prefix} Result ready in {time.perf_counter() - started_at:.1f}s: {result_url}"
            )

            saved_screenshot_path = ""
            if save_screenshot:
                self._expand_metadata_panel(sb, log_prefix)
                sb.save_screenshot(screenshot_path)
                saved_screenshot_path = screenshot_path
                logging.info(f"{log_prefix} Screenshot saved to {screenshot_path}")

            saved_local_paipu_path = ""
            if save_local_paipu:
                saved_local_paipu_path = self._save_local_paipu(
                    sb,
                    local_paipu_path,
                    result_url,
                    log_prefix,
                )

            return {
                "resultUrl": result_url,
                "localPaipuPath": saved_local_paipu_path,
                "screenshotPath": saved_screenshot_path,
                "metadata": metadata,
                "badMoveStats": bad_move_stats,
            }
        except Exception as exc:
            if self.submission_coordinator is not None:
                self.submission_coordinator.report_outcome(
                    log_prefix,
                    success=False,
                    error_text=str(exc),
                    token_wait_seconds=token_wait_seconds,
                    submit_wait_seconds=submit_wait_seconds,
                    result_wait_seconds=result_wait_seconds,
                )
            raise
        finally:
            if self.submission_coordinator is not None and submit_slot_held and not submit_slot_released:
                self.submission_coordinator.release_submit_slot(
                    log_prefix,
                    token_wait_seconds=token_wait_seconds,
                )

    def _handle_rotation_failure(self, sb, slot, task, exc, max_retries, pending):
        err_str = str(exc).lower()
        logging.error(f"{task.get('log_prefix', '[' + task['uuid'] + ']')} ERROR exception: {exc}")

        try:
            self._switch_to_slot(sb, slot)
            error_screenshot = os.path.join(task["mode_dir"], f"{task['uuid']}_error.png")
            sb.save_screenshot(error_screenshot)
        except Exception:
            pass

        if any(marker in err_str for marker in ("no such window", "closed", "invalid session", "disconnected")):
            raise exc

        task["retries"] = task.get("retries", 0) + 1
        if task["retries"] <= max_retries:
            logging.warning(
                f"{task.get('log_prefix', '[' + task['uuid'] + ']')} RETRY Analysis failed. "
                f"Retrying ({task['retries']}/{max_retries}) on the next window turn."
            )
            pending.append(task)
            return None

        logging.error(
            f"{task.get('log_prefix', '[' + task['uuid'] + ']')} SKIP Analysis permanently failed after {max_retries} retries."
        )
        return {"status": "fail", "task": task}

    def _open_pipeline_window(self, sb):
        current_handle = sb.driver.current_window_handle
        try:
            sb.driver.switch_to.new_window("window")
            new_handle = sb.driver.current_window_handle
            sb.driver.switch_to.window(current_handle)
            return new_handle
        except Exception:
            existing_handles = set(sb.driver.window_handles)
            sb.execute_script("window.open('about:blank', '_blank');")

            deadline = time.time() + 5
            while time.time() < deadline:
                current_handles = set(sb.driver.window_handles)
                new_handles = current_handles - existing_handles
                if new_handles:
                    new_handle = new_handles.pop()
                    sb.driver.switch_to.window(current_handle)
                    return new_handle
                time.sleep(0.2)

        raise RuntimeError("Could not open standby browser window")

    def _prepare_pipeline_slot(self, sb, slot, task, max_retries, pending):
        try:
            self._switch_to_slot(sb, slot)
            self._prepare_task_in_current_tab(sb, task)
            self._refresh_slot_handle(sb, slot)
            slot["task"] = task
            slot["prepared"] = True
            slot["started_at"] = time.perf_counter()
            slot["submitted_at"] = 0.0
            logging.info(f"{task.get('log_prefix', '[' + task['uuid'] + ']')} Form prewarmed in {slot['name']} tab")
            return None
        except Exception as exc:
            return self._handle_pipeline_failure(sb, slot, task, exc, max_retries, pending)

    def _submit_pipeline_slot(self, sb, slot, task):
        self._switch_to_slot(sb, slot)
        uuid = task["uuid"]
        log_prefix = task.get("log_prefix", f"[{uuid}]")

        logging.info(f"{log_prefix} {slot['name']} tab entering Turnstile")
        token_started_at = time.perf_counter()
        self._acquire_turnstile_token(sb, task, log_prefix)
        token_wait_seconds = time.perf_counter() - token_started_at
        logging.info(f"{log_prefix} Turnstile token ready in {token_wait_seconds:.1f}s")

        self._submit_review(sb, log_prefix)
        self._wait_for_submission_departure_or_error(sb, log_prefix, timeout=15)
        self._refresh_slot_handle(sb, slot)
        slot["submitted_at"] = time.perf_counter()
        logging.info(f"{log_prefix} Review submitted from {slot['name']} tab")
        return token_wait_seconds

    def _collect_pipeline_result(self, sb, slot, task):
        self._switch_to_slot(sb, slot)
        uuid = task["uuid"]
        log_prefix = task.get("log_prefix", f"[{uuid}]")
        screenshot_path = os.path.join(task["mode_dir"], f"{uuid}.png")
        local_paipu_path = os.path.join(task["mode_dir"], f"{uuid}.html")
        save_screenshot = task.get("save_screenshot", False)
        save_local_paipu = task.get("save_local_paipu", False)

        logging.info(f"{log_prefix} Waiting for result page")
        self._wait_for_result_or_error(sb, log_prefix, timeout=45)
        result_url = sb.get_current_url()
        metadata = self._extract_metadata(sb)
        bad_move_stats = {}
        if task.get("analyze_bad_move_rate", False):
            bad_move_stats = self._extract_bad_move_stats(sb, log_prefix)

        total_elapsed = time.perf_counter() - slot["started_at"] if slot["started_at"] else 0.0
        logging.info(f"{log_prefix} Result ready in {total_elapsed:.1f}s: {result_url}")

        saved_screenshot_path = ""
        if save_screenshot:
            self._expand_metadata_panel(sb, log_prefix)
            sb.save_screenshot(screenshot_path)
            saved_screenshot_path = screenshot_path
            logging.info(f"{log_prefix} Screenshot saved to {screenshot_path}")

        saved_local_paipu_path = ""
        if save_local_paipu:
            saved_local_paipu_path = self._save_local_paipu(
                sb,
                local_paipu_path,
                result_url,
                log_prefix,
            )

        return {
            "resultUrl": result_url,
            "localPaipuPath": saved_local_paipu_path,
            "screenshotPath": saved_screenshot_path,
            "metadata": metadata,
            "badMoveStats": bad_move_stats,
        }

    def _handle_pipeline_failure(self, sb, slot, task, exc, max_retries, pending):
        err_str = str(exc).lower()
        logging.error(f"{task.get('log_prefix', '[' + task['uuid'] + ']')} ERROR exception: {exc}")

        try:
            self._switch_to_slot(sb, slot)
            error_screenshot = os.path.join(task["mode_dir"], f"{task['uuid']}_error.png")
            sb.save_screenshot(error_screenshot)
        except Exception:
            pass

        if any(marker in err_str for marker in ("no such window", "closed", "invalid session", "disconnected")):
            raise exc

        self._reset_pipeline_slot(slot)
        task["retries"] = task.get("retries", 0) + 1
        if task["retries"] <= max_retries:
            logging.warning(
                f"{task.get('log_prefix', '[' + task['uuid'] + ']')} RETRY Analysis failed. "
                f"Retrying ({task['retries']}/{max_retries}) with a fresh page load."
            )
            pending.append(task)
            return None

        logging.error(
            f"{task.get('log_prefix', '[' + task['uuid'] + ']')} SKIP Analysis permanently failed after {max_retries} retries."
        )
        return {"status": "fail", "task": task}

    def _reset_pipeline_slot(self, slot):
        slot["task"] = None
        slot["prepared"] = False
        slot["started_at"] = 0.0
        slot["submitted_at"] = 0.0

    def _switch_to_slot(self, sb, slot):
        handles = list(sb.driver.window_handles)
        if slot["handle"] not in handles:
            current_handle = None
            try:
                current_handle = sb.driver.current_window_handle
            except Exception:
                current_handle = None

            if current_handle in handles:
                slot["handle"] = current_handle
            elif handles:
                slot["handle"] = handles[-1]
            else:
                raise RuntimeError(f"{slot['name']} tab is no longer available")

        sb.driver.switch_to.window(slot["handle"])
        try:
            sb.wait_for_ready_state_complete()
        except Exception:
            pass

    def _refresh_slot_handle(self, sb, slot):
        try:
            slot["handle"] = sb.driver.current_window_handle
        except Exception:
            pass

    def _prepare_task_in_current_tab(self, sb, task):
        uuid = task["uuid"]
        log_prefix = task.get("log_prefix", f"[{uuid}]")
        logging.info(f"{log_prefix} Opening fresh review form")
        self._open_fresh_review_page(sb, log_prefix)
        self._populate_form(sb, task["paipu_url"], task["model_tag"])
        self._prepare_review_form(sb)

    def _prime_rotation_slot(self, sb, slot, label):
        if not slot.get("handle"):
            self._spawn_rotation_window(sb, slot, label)
        self._switch_to_slot(sb, slot)
        logging.info(f"{label} Refreshing {slot['name']} back to review form")
        self._open_fresh_review_page(sb, label)
        self._refresh_slot_handle(sb, slot)
        slot["ready"] = True

    def _ensure_rotation_slot_ready(self, sb, slot, label):
        if not slot.get("handle"):
            self._spawn_rotation_window(sb, slot, label)
        self._switch_to_slot(sb, slot)
        if not slot.get("ready", False) or not self._is_review_form_ready(sb):
            logging.info(f"{label} Preparing {slot['name']} for the next assigned task")
            self._prime_rotation_slot(sb, label=label, slot=slot)
            slot["ready"] = True
        self._refresh_slot_handle(sb, slot)

    def _is_review_form_ready(self, sb):
        try:
            return bool(
                sb.execute_script(
                    """
                    const input = document.querySelector(arguments[0]);
                    return !!(input && document.readyState !== 'loading');
                    """,
                    INPUT_SELECTOR,
                )
            )
        except Exception:
            return False

    def _spawn_rotation_window(self, sb, slot, label):
        current_handle = sb.driver.current_window_handle
        existing_handles = set(sb.driver.window_handles)
        logging.info(f"{label} Spawning {slot['name']} from the active review context")

        try:
            sb.execute_script("window.open(arguments[0], '_blank');", self.review_url)
        except Exception:
            sb.driver.switch_to.new_window("window")
            slot["handle"] = sb.driver.current_window_handle
            self._open_fresh_review_page(sb, label)
            self._refresh_slot_handle(sb, slot)
            sb.driver.switch_to.window(current_handle)
            return

        deadline = time.time() + 8
        while time.time() < deadline:
            current_handles = set(sb.driver.window_handles)
            new_handles = list(current_handles - existing_handles)
            if new_handles:
                slot["handle"] = new_handles[-1]
                self._switch_to_slot(sb, slot)
                logging.info(f"{label} {slot['name']} spawned successfully")
                self._refresh_slot_handle(sb, slot)
                try:
                    sb.driver.switch_to.window(current_handle)
                except Exception:
                    pass
                return
            time.sleep(0.2)

        raise RuntimeError(f"{label} Could not spawn {slot['name']}")

    def _open_fresh_review_page(self, sb, label):
        last_exc = None
        for attempt in range(2):
            current_url = ""
            try:
                current_url = sb.get_current_url()
            except Exception:
                current_url = ""

            try:
                if "mjai.ekyu.moe" in current_url:
                    sb.execute_script("window.location.replace(arguments[0]);", self.review_url)
                else:
                    sb.uc_open_with_reconnect(self.review_url, reconnect_time=4)
            except Exception:
                sb.open(self.review_url)

            try:
                sb.wait_for_ready_state_complete()
            except Exception:
                pass

            try:
                sb.wait_for_element(INPUT_SELECTOR, timeout=20)
                return
            except Exception as exc:
                last_exc = exc
                if attempt == 0:
                    logging.warning(f"{label} Review page not ready, retrying open once...")
                    try:
                        sb.execute_script("window.location.replace(arguments[0]);", self.review_url)
                    except Exception:
                        try:
                            sb.refresh()
                        except Exception:
                            pass
                    time.sleep(2)

        raise last_exc

    def _prepare_review_form(self, sb):
        sb.execute_script(
            """
            const submit = document.querySelector(arguments[0]);
            if (submit) {
              submit.classList.remove('is-loading');
              submit.disabled = false;
              submit.style.pointerEvents = '';
            }

            const form = document.querySelector(arguments[1]);
            if (form) {
              form.target = '_self';
            }
            """,
            SUBMIT_SELECTOR,
            FORM_SELECTOR,
        )

    def _populate_form(self, sb, paipu_url, model_tag):
        success = sb.execute_script(
            """
            const paipuUrl = arguments[0];
            const modelTag = arguments[1];
            const reviewLanguage = arguments[2];
            const reviewUi = arguments[3];

            const dispatch = (el) => {
              el.dispatchEvent(new Event('input', { bubbles: true }));
              el.dispatchEvent(new Event('change', { bubbles: true }));
            };

            const radio = document.querySelector('input[name="input-method"][value="log-url"]');
            if (radio && !radio.checked) {
              radio.click();
            }

            const input = document.querySelector('input[name="log-url"]');
            if (!input) {
              return false;
            }
            if (input.value !== paipuUrl) {
              input.value = paipuUrl;
              dispatch(input);
            }

            const setSelect = (selector, value) => {
              const el = document.querySelector(selector);
              if (!el || el.value === value) {
                return;
              }
              el.value = value;
              dispatch(el);
            };

            setSelect('select[name="engine"]', 'mortal');
            setSelect('select[name="mortal-model-tag"]', modelTag);
            setSelect('select[name="ui"]', reviewUi);
            setSelect('select[name="lang"]', reviewLanguage);

            const details = document.querySelector('details.details.mb-3');
            if (details) {
              details.open = true;
            }

            const showRating = document.querySelector('input[name="show-rating"]');
            if (showRating && !showRating.checked) {
              showRating.click();
            }

            const form = document.querySelector(arguments[4]);
            if (form) {
              form.target = '_self';
            }

            return true;
            """,
            paipu_url,
            model_tag,
            self.review_language,
            self.review_ui,
            FORM_SELECTOR,
        )

        if not success:
            raise RuntimeError("Could not populate review form")

    def _wait_for_turnstile_token(self, sb, log_prefix, timeout):
        """Pure poller: returns when cf-turnstile-response has a token.

        No GUI clicks, no widget reset, no page reload. Recovery is handled by
        the caller (_acquire_turnstile_token) so this method stays a simple
        timeout-bound poll.
        """
        deadline = time.time() + timeout
        while time.time() < deadline:
            state = self._read_review_state(sb)
            if state["token_length"] > 0:
                return

            if state["page_text"] and (
                "invalid captcha response" in state["page_text"]
                or "timeout-or-duplicate" in state["page_text"]
            ):
                raise RuntimeError(f"{log_prefix} Turnstile token was rejected before submission")

            time.sleep(0.5)

        raise RuntimeError(f"{log_prefix} Timed out waiting for Turnstile token")

    def _acquire_turnstile_token(
        self,
        sb,
        task,
        log_prefix,
        per_attempt_timeout: float = 12.0,
        max_reloads: int = 2,
    ):
        """Obtain a Turnstile token via headless UC-mode auto-pass.

        No GUI clicks and no OS permissions are needed. On stall we first try a
        cheap in-page JS reset of the Turnstile widget, then fall back to fully
        reloading the review form (up to ``max_reloads`` times) to give
        Cloudflare a fresh chance to auto-issue a token. Server-side rejections
        (invalid/timeout-or-duplicate) are propagated without retry.
        """
        js_reset_done = False
        reloads = 0

        while True:
            try:
                self._wait_for_turnstile_token(sb, log_prefix, per_attempt_timeout)
                return
            except RuntimeError as exc:
                msg = str(exc).lower()
                if "rejected" in msg or "rate" in msg:
                    raise

                if not js_reset_done:
                    js_reset_done = True
                    logging.info(f"{log_prefix} Turnstile token missing, resetting widget")
                    try:
                        self._recover_turnstile_widget(sb)
                    except Exception:
                        pass
                    continue

                if reloads >= max_reloads:
                    raise RuntimeError(
                        f"{log_prefix} Turnstile widget stalled before token issuance"
                    ) from exc

                reloads += 1
                logging.info(
                    f"{log_prefix} Turnstile token still missing, "
                    f"reloading form ({reloads}/{max_reloads})"
                )
                try:
                    self._open_fresh_review_page(sb, log_prefix)
                    self._populate_form(sb, task["paipu_url"], task["model_tag"])
                    self._prepare_review_form(sb)
                except Exception as reload_exc:
                    logging.warning(f"{log_prefix} Form reload failed: {reload_exc}")


    def _submit_review(self, sb, log_prefix):
        submitted = sb.execute_script(
            """
            const form = document.querySelector(arguments[0]);
            const submit = document.querySelector(arguments[1]);
            const token = document.querySelector(arguments[2]);
            if (!form || !submit) {
              return 'missing-form';
            }
            if (!token || !token.value) {
              return 'missing-token';
            }

            form.target = '_self';
            submit.disabled = false;
            submit.classList.remove('is-loading');
            submit.style.pointerEvents = '';

            if (typeof form.requestSubmit === 'function') {
              form.requestSubmit(submit);
            } else {
              submit.click();
            }
            return 'submitted';
            """,
            FORM_SELECTOR,
            SUBMIT_SELECTOR,
            TURNSTILE_RESPONSE_SELECTOR,
        )

        if submitted != "submitted":
            raise RuntimeError(f"{log_prefix} Review form submission failed before navigation: {submitted}")

        time.sleep(0.2)

    def _wait_for_submission_departure_or_error(self, sb, log_prefix, timeout):
        deadline = time.time() + timeout
        while time.time() < deadline:
            state = self._read_review_state(sb)
            current_url = state["url"]
            page_text = state["page_text"]

            if REPORT_URL_FRAGMENT in current_url and current_url != self.review_url:
                return

            if "invalid captcha response" in page_text or "timeout-or-duplicate" in page_text:
                raise RuntimeError(f"{log_prefix} Turnstile token was rejected")

            if "too many requests" in page_text or "rate limit" in page_text:
                raise RuntimeError(f"{log_prefix} Review site rate limited this request")

            time.sleep(0.5)

        raise RuntimeError(f"{log_prefix} Review submission never left the form page")

    def _extract_metadata(self, sb):
        killerducky_data = self._extract_killerducky_data(sb)
        if killerducky_data:
            return parse_killerducky_metadata(killerducky_data)
        if self.review_ui == "killerducky":
            raise RuntimeError("Could not extract KillerDucky report JSON")

        metadata = sb.execute_script(
            """
            const data = {};
            for (const dl of document.querySelectorAll('details > dl')) {
              const dts = dl.querySelectorAll('dt');
              const dds = dl.querySelectorAll('dd');
              const count = Math.min(dts.length, dds.length);
              for (let i = 0; i < count; i += 1) {
                data[dts[i].textContent.trim()] = dds[i].textContent.trim();
              }
            }
            return data;
            """
        )
        return metadata or {}

    def _extract_killerducky_data(self, sb, include_entries=False):
        try:
            data = sb.execute_script(
                """
                if (!(window.MM && window.MM.GS && window.MM.GS.fullData)) {
                  return null;
                }

                const source = window.MM.GS.fullData;
                const review = source.review || {};
                const includeEntries = arguments[0];
                return {
                  engine: source.engine,
                  game_length: source.game_length,
                  review_time: source.review_time,
                  player_id: source.player_id,
                  review: {
                    model_tag: review.model_tag,
                    rating: review.rating,
                    temperature: review.temperature,
                    total_matches: review.total_matches,
                    total_reviewed: review.total_reviewed,
                    kyokus: includeEntries && Array.isArray(review.kyokus)
                      ? review.kyokus.map((kyoku) => ({
                          entries: Array.isArray(kyoku.entries) ? kyoku.entries.map((entry) => {
                            const actualIndex = entry.actual_index;
                            const details = Array.isArray(entry.details)
                              ? entry.details.map((detail, index) => (
                                  index === actualIndex ? {prob: detail.prob} : null
                                ))
                              : null;
                            return {
                              is_equal: entry.is_equal,
                              actual_index: actualIndex,
                              details,
                            };
                          }) : [],
                        }))
                      : [],
                  },
                };
                """,
                include_entries,
            )
            return data if isinstance(data, dict) else None
        except Exception as exc:
            if self.review_ui == "killerducky":
                raise RuntimeError("Could not read KillerDucky report JSON") from exc
            return None

    def _save_local_paipu(self, sb, filepath, source_url, log_prefix):
        os.makedirs(os.path.dirname(filepath), exist_ok=True)
        html = sb.execute_script(
            """
            const sourceUrl = arguments[0];
            const clone = document.documentElement.cloneNode(true);
            const absolutize = (selector, attr) => {
              for (const el of clone.querySelectorAll(selector)) {
                const value = el.getAttribute(attr);
                if (!value || value.startsWith('#') || value.startsWith('data:') || value.startsWith('blob:')) {
                  continue;
                }
                try {
                  el.setAttribute(attr, new URL(value, document.baseURI).href);
                } catch (e) {
                }
              }
            };

            absolutize('a[href]', 'href');
            absolutize('link[href]', 'href');
            absolutize('script[src]', 'src');
            absolutize('img[src]', 'src');
            absolutize('iframe[src]', 'src');

            const meta = clone.ownerDocument.createElement('meta');
            meta.setAttribute('name', 'batchmortal-source-url');
            meta.setAttribute('content', sourceUrl);
            const head = clone.querySelector('head');
            if (head) {
              head.insertBefore(meta, head.firstChild);
            }

            return `<!doctype html>\\n<!-- Saved from ${sourceUrl} -->\\n${clone.outerHTML}`;
            """,
            source_url,
        )
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(html or "")
        logging.info(f"{log_prefix} Local paipu saved to {filepath}")
        return filepath

    def _extract_bad_move_stats(self, sb, log_prefix):
        try:
            killerducky_data = self._extract_killerducky_data(sb, include_entries=True)
            if killerducky_data:
                return parse_killerducky_bad_move_stats(killerducky_data)

            stats = sb.execute_script(
                """
                const strictLimit = arguments[0];
                const looseLimit = arguments[1];

                const parseFirstNumber = (text) => {
                  const match = String(text || '').replace(',', '.').match(/-?\\d+(?:\\.\\d+)?/);
                  return match ? Number.parseFloat(match[0]) : NaN;
                };

                const parseFirstInteger = (text) => {
                  const match = String(text || '').match(/\\d+/);
                  return match ? Number.parseInt(match[0], 10) : NaN;
                };

                const extractTotalChoices = () => {
                  for (const dl of document.querySelectorAll('details > dl')) {
                    const dts = dl.querySelectorAll('dt');
                    const dds = dl.querySelectorAll('dd');
                    const count = Math.min(dts.length, dds.length);
                    for (let i = 0; i < count; i += 1) {
                      const key = dts[i].textContent.trim().toLowerCase();
                      if (!key.includes('一致率') && !key.includes('match')) {
                        continue;
                      }
                      const match = dds[i].textContent.match(/\\d+\\s*\\/\\s*(\\d+)/);
                      if (match) {
                        return Number.parseInt(match[1], 10);
                      }
                    }
                  }
                  return null;
                };

                const extractChosenWeight = (orderLoss) => {
                  const chosenIndex = parseFirstInteger(orderLoss.textContent);
                  if (!Number.isFinite(chosenIndex) || chosenIndex < 1) {
                    return NaN;
                  }

                  let collapseEntry = orderLoss.closest('.collapse.entry');
                  if (!collapseEntry) {
                    const turnInfo = orderLoss.parentElement;
                    const summary = turnInfo ? turnInfo.parentElement : null;
                    collapseEntry = summary ? summary.parentElement : null;
                  }
                  if (!collapseEntry) {
                    return NaN;
                  }

                  const rows = collapseEntry.querySelectorAll('table tbody tr');
                  const chosenRow = rows[chosenIndex - 1];
                  if (!chosenRow) {
                    return NaN;
                  }

                  const cells = chosenRow.querySelectorAll('td, th');
                  const weightCell = cells.length ? cells[cells.length - 1] : chosenRow.lastElementChild;
                  return weightCell ? parseFirstNumber(weightCell.textContent) : NaN;
                };

                let countStrict = 0;
                let countLoose = 0;
                let unparsed = 0;
                const orderLosses = Array.from(document.getElementsByClassName('order-loss'));

                for (const orderLoss of orderLosses) {
                  const chosenWeight = extractChosenWeight(orderLoss);
                  if (!Number.isFinite(chosenWeight)) {
                    unparsed += 1;
                    continue;
                  }
                  if (chosenWeight <= strictLimit) {
                    countStrict += 1;
                  }
                  if (chosenWeight <= looseLimit) {
                    countLoose += 1;
                  }
                }

                const totalChoices = extractTotalChoices();
                const formatRate = (count) => (
                  Number.isFinite(totalChoices) && totalChoices > 0
                    ? `${(100 * count / totalChoices).toFixed(3)}%`
                    : ''
                );

                return {
                  badMoveRate5: formatRate(countStrict),
                  badMoveCount5: String(countStrict),
                  badMoveRate10: formatRate(countLoose),
                  badMoveCount10: String(countLoose),
                  badMoveDenominator: Number.isFinite(totalChoices) ? String(totalChoices) : '',
                  badMoveOrderLossCount: String(orderLosses.length),
                  badMoveUnparsedCount: String(unparsed),
                };
                """,
                BAD_MOVE_STRICT_LIMIT,
                BAD_MOVE_LOOSE_LIMIT,
            )
            return stats or {}
        except Exception as exc:
            logging.warning(f"{log_prefix} Could not extract bad move stats: {exc}")
            return {}

    def _wait_for_result_or_error(self, sb, log_prefix, timeout):
        deadline = time.time() + timeout
        while time.time() < deadline:
            if self.review_ui == "killerducky":
                try:
                    killerducky_ready = sb.execute_script(
                        """
                        return !!(
                          window.MM && window.MM.GS && window.MM.GS.fullData
                          && window.MM.GS.fullData.review
                          && document.querySelector('.about-metadata table')
                        );
                        """
                    )
                except Exception:
                    killerducky_ready = False
                if killerducky_ready:
                    return
            elif sb.is_element_present(RESULT_SELECTOR):
                return

            state = self._read_review_state(sb)
            page_text = state["page_text"]

            if "invalid captcha response" in page_text or "timeout-or-duplicate" in page_text:
                raise RuntimeError(f"{log_prefix} Turnstile token was rejected")

            if "too many requests" in page_text or "rate limit" in page_text:
                raise RuntimeError(f"{log_prefix} Review site rate limited this request")

            time.sleep(0.5)

        raise RuntimeError(f"{log_prefix} Timed out waiting for review results")

    def _read_review_state(self, sb):
        return sb.execute_script(
            """
            const submit = document.querySelector(arguments[0]);
            const token = document.querySelector(arguments[1]);
            return {
              url: window.location.href,
              token_length: token && token.value ? token.value.length : 0,
              page_text: document.body ? document.body.innerText.toLowerCase() : '',
              submit_disabled: submit ? !!submit.disabled : true,
              submit_busy: submit ? submit.classList.contains('is-loading') : false,
            };
            """,
            SUBMIT_SELECTOR,
            TURNSTILE_RESPONSE_SELECTOR,
        )

    def _expand_metadata_panel(self, sb, log_prefix):
        try:
            if self.review_ui == "killerducky":
                sb.execute_script(
                    """
                    const modal = document.getElementById('about-modal');
                    if (modal && !modal.open) {
                      modal.showModal();
                    }
                    """
                )
                time.sleep(0.5)
                return

            is_open = sb.execute_script(
                """
                const details = document.querySelector('body > details:nth-child(6)');
                return details ? details.open : false;
                """
            )
            if not is_open:
                sb.click("body > details:nth-child(6) > summary")
                time.sleep(0.5)
        except Exception as exc:
            logging.warning(f"{log_prefix} Could not expand metadata menu: {exc}")

    def _recover_turnstile_widget(self, sb):
        try:
            sb.execute_script(
                """
                const token = document.querySelector(arguments[0]);
                if (token) {
                  token.value = '';
                }

                const submit = document.querySelector(arguments[1]);
                if (submit) {
                  submit.disabled = false;
                  submit.classList.remove('is-loading');
                  submit.style.pointerEvents = '';
                }

                if (window.turnstile) {
                  const widgets = Array.from(document.querySelectorAll('.cf-turnstile'));
                  for (const widget of widgets) {
                    const widgetId = widget.getAttribute('data-widget-id');
                    try {
                      if (widgetId) {
                        window.turnstile.reset(widgetId);
                      } else {
                        window.turnstile.reset();
                      }
                    } catch (e) {
                    }
                  }
                }
                """,
                TURNSTILE_RESPONSE_SELECTOR,
                SUBMIT_SELECTOR,
            )
        except Exception:
            pass