Live-streaming/app/bettergi_current_party.py

"""BetterGI adapter helpers for reading the active party preset name."""

from __future__ import annotations

import copy
import json
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any

try:
    from .bettergi_daily import DailyAutomationError, JsonUpdate
except ImportError:
    from bettergi_daily import DailyAutomationError, JsonUpdate


CURRENT_PARTY_SCRIPT_NAME = "LiveCurrentParty"
CURRENT_PARTY_GROUP_NAME = "直播系统读取当前队伍"
CURRENT_PARTY_STATUS_FILE = "status.json"
_REQUEST_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
_INVALID_BGI_NAME = re.compile(r'[<>:"/\\|?*\x00-\x1f]')


class CurrentPartyReadError(DailyAutomationError):
    """Raised when the managed current-party reader cannot complete safely."""


@dataclass(frozen=True)
class PreparedCurrentPartyRead:
    request_id: str
    group_name: str
    status_path: Path
    updates: tuple[JsonUpdate, ...]


@dataclass(frozen=True)
class CurrentPartyReadResult:
    party_name: str
    candidates: tuple[str, ...] = ()


def _read_json_object(path: Path, label: str) -> dict[str, Any]:
    if not path.is_file():
        raise CurrentPartyReadError(f"未找到{label}: {path}")
    try:
        data = json.loads(path.read_text(encoding="utf-8-sig"))
    except (OSError, json.JSONDecodeError) as exc:
        raise CurrentPartyReadError(f"读取{label}失败: {exc}") from exc
    if not isinstance(data, dict):
        raise CurrentPartyReadError(f"{label}必须是 JSON 对象: {path}")
    return data


def _safe_group_name(value: str) -> str:
    name = str(value or "").strip()
    if not name:
        raise CurrentPartyReadError("当前队伍读取配置组名称不能为空")
    if name in {".", ".."} or _INVALID_BGI_NAME.search(name):
        raise CurrentPartyReadError(f"当前队伍读取配置组名称包含非法字符: {name}")
    return name


def _next_group_index(group_dir: Path) -> int:
    indexes: list[int] = []
    for path in group_dir.glob("*.json"):
        try:
            data = json.loads(path.read_text(encoding="utf-8-sig"))
            value = data.get("index") if isinstance(data, dict) else None
            if isinstance(value, int):
                indexes.append(value)
        except (OSError, json.JSONDecodeError):
            continue
    return max(indexes, default=0) + 1


def _load_group_template(group_dir: Path, managed_path: Path) -> dict[str, Any]:
    if managed_path.is_file():
        return _read_json_object(managed_path, "托管当前队伍读取配置组")

    for name in ("切换队伍", "修改队员", "每日委托"):
        candidate = group_dir / f"{name}.json"
        if candidate.is_file():
            template = _read_json_object(candidate, f"配置组“{name}”")
            template["index"] = _next_group_index(group_dir)
            return template
    raise CurrentPartyReadError(
        "无法生成当前队伍读取配置组：请先在 BGI 创建“切换队伍”“修改队员”或“每日委托”中的任意一个配置组"
    )


def prepare_current_party_read(
    work_dir: str | Path,
    request_id: str,
    *,
    group_name: str = CURRENT_PARTY_GROUP_NAME,
) -> PreparedCurrentPartyRead:
    work_path = Path(work_dir)
    if not work_path.is_dir():
        raise CurrentPartyReadError(f"BetterGI 工作目录不存在: {work_path}")
    request = str(request_id or "").strip()
    if not _REQUEST_ID_PATTERN.fullmatch(request):
        raise CurrentPartyReadError("当前队伍读取请求 ID 格式无效")
    managed_name = _safe_group_name(group_name)

    group_dir = work_path / "User" / "ScriptGroup"
    if not group_dir.is_dir():
        raise CurrentPartyReadError(f"BGI 配置组目录不存在: {group_dir}")
    managed_path = group_dir / f"{managed_name}.json"
    managed = copy.deepcopy(_load_group_template(group_dir, managed_path))
    managed["name"] = managed_name
    managed["projects"] = [
        {
            "name": "读取当前队伍名称",
            "folderName": CURRENT_PARTY_SCRIPT_NAME,
            "jsScriptSettingsObject": {"requestId": request},
            "index": 1,
            "type": "Javascript",
            "status": "Enabled",
            "schedule": "Daily",
            "runNum": 1,
            "allowJsNotification": True,
            "allowJsHTTPHash": "",
        }
    ]

    status_path = (
        work_path
        / "User"
        / "JsScript"
        / CURRENT_PARTY_SCRIPT_NAME
        / CURRENT_PARTY_STATUS_FILE
    )
    update = JsonUpdate(managed_path, managed, "生成当前队伍读取配置组")
    return PreparedCurrentPartyRead(
        request_id=request,
        group_name=managed_name,
        status_path=status_path,
        updates=(update,),
    )


def clear_current_party_status(status_path: str | Path) -> None:
    path = Path(status_path)
    try:
        path.unlink(missing_ok=True)
    except OSError as exc:
        raise CurrentPartyReadError(f"清理当前队伍读取状态失败: {exc}") from exc


def read_current_party_status(
    status_path: str | Path,
    request_id: str,
) -> CurrentPartyReadResult | None:
    path = Path(status_path)
    if not path.is_file():
        return None
    try:
        data = json.loads(path.read_text(encoding="utf-8-sig"))
    except (OSError, json.JSONDecodeError):
        return None
    if not isinstance(data, dict) or str(data.get("request_id") or "") != request_id:
        return None

    state = str(data.get("state") or "").strip().casefold()
    if state in {"", "running"}:
        return None
    candidates = tuple(
        str(value).strip()
        for value in data.get("candidates", [])
        if str(value).strip()
    ) if isinstance(data.get("candidates"), list) else ()
    if state == "success":
        party_name = str(data.get("party_name") or "").strip()
        if not party_name:
            raise CurrentPartyReadError("当前队伍读取脚本返回成功，但队伍名称为空")
        return CurrentPartyReadResult(party_name=party_name, candidates=candidates)
    if state == "error":
        message = str(data.get("message") or "读取当前队伍失败").strip()
        raise CurrentPartyReadError(message)
    raise CurrentPartyReadError(f"当前队伍读取脚本返回未知状态: {state}")