From f623959fd4da7885d917b036b226d0801e6ddba4 Mon Sep 17 00:00:00 2001 From: ddaodan <731882332@qq.com> Date: Sat, 15 Aug 2026 16:49:17 +0800 Subject: [PATCH] Update Live-streaming code (auto-daily features) --- README.md | 34 +- app/bettergi_current_party.py | 181 +++++ app/bettergi_daily.py | 828 ++++++++++++++++++++ app/danmu_queue.py | 459 +++++++++-- config/config.json | 41 +- config/domain_aliases.json | 7 + docs/自动每日.md | 179 +++++ frontend/admin/src/main.js | 19 +- integrations/LiveCurrentParty/main.js | 154 ++++ integrations/LiveCurrentParty/manifest.json | 17 + integrations/LiveCurrentParty/settings.json | 8 + run.bat | 7 +- scripts/build.ps1 | 5 +- tests/test_bettergi_current_party.py | 102 +++ tests/test_bettergi_daily.py | 503 ++++++++++++ tests/test_bgi_log_format.py | 44 +- tests/test_daily_commands.py | 353 +++++++++ web/admin/assets/index-BdYZGJMQ.js | 636 +++++++++++++++ web/admin/index.html | 2 +- 19 files changed, 3517 insertions(+), 62 deletions(-) create mode 100644 app/bettergi_current_party.py create mode 100644 app/bettergi_daily.py create mode 100644 config/domain_aliases.json create mode 100644 docs/自动每日.md create mode 100644 integrations/LiveCurrentParty/main.js create mode 100644 integrations/LiveCurrentParty/manifest.json create mode 100644 integrations/LiveCurrentParty/settings.json create mode 100644 tests/test_bettergi_current_party.py create mode 100644 tests/test_bettergi_daily.py create mode 100644 tests/test_daily_commands.py create mode 100644 web/admin/assets/index-BdYZGJMQ.js diff --git a/README.md b/README.md index d4390c5..a04767c 100644 --- a/README.md +++ b/README.md @@ -1,19 +1,20 @@ # BetterGI 直播联动 -监听 B 站直播间弹幕，管理观众排队、积分、扫码上号、BetterGI 配置组执行、点歌和 TTS 播报。 +监听 B 站直播间弹幕，管理观众排队、积分、扫码上号、BetterGI 配置组与自动每日一条龙、队伍管理、点歌和 TTS 播报。 ## 环境 -- **Python**: conda 环境 `Live-streaming`（`E:\Programs\Anaconda3\envs\Live-streaming\python.exe`） +- **Python**: 项目虚拟环境 `.venv`（Python 3.11） - **Node.js**: 构建前端用 -- **BetterGI**: 自动化脚本引擎，路径在 `config/config.json` 中配置 +- **BetterGI**: `0.63.0+`，路径在 `config/config.json` 中配置 ```powershell # 安装 Python 依赖 -E:\Programs\Anaconda3\envs\Live-streaming\python.exe -m pip install -r requirements.txt +py -3.11 -m venv .venv +.venv\Scripts\python.exe -m pip install -r requirements.txt # 下载 TTS 模型（约 1.2GB，仅需一次） -E:\Programs\Anaconda3\envs\Live-streaming\python.exe -c "from huggingface_hub import snapshot_download; snapshot_download('Qwen/Qwen3-TTS-12Hz-0.6B-Base')" +.venv\Scripts\python.exe -c "from huggingface_hub import snapshot_download; snapshot_download('Qwen/Qwen3-TTS-12Hz-0.6B-Base')" # 安装前端依赖并构建 cd frontend\admin @@ -69,6 +70,20 @@ npm run build | `admin_uids` | 管理员 UID 列表（一级用户，可用重置等特权指令） | | `log_level` | 日志级别：`DEBUG` / `INFO` / `WARNING` | +### daily — 自动每日 + +| 字段 | 默认值 | 说明 | +|------|--------|------| +| `one_dragon_template` | 默认配置 | BGI 一条龙模板名称 | +| `managed_one_dragon_name` | 直播系统自动每日 | 每次覆盖生成并启动的托管配置名称 | +| `ley_line_craft_resin_before` | true | 地脉模式执行前是否合成树脂 | +| `commission_use_current_party` | true | 委托前读取游戏当前队伍名并写入 AutoCommissionNova 的战斗与元素采集队伍配置 | +| `current_party_read_timeout_sec` | 45 | 当前队伍 OCR 读取超时秒数 | + +自动每日流程为“领取邮件 → 合成树脂 → 可选其他任务 → 领取尘歌壶奖励 → 领取每日奖励”。秘境俗称维护在 `config/domain_aliases.json`，修改后无需重启。 + +完整的 BGI 模板、战斗策略和三个配置组配置方法见 [自动每日与队伍管理](docs/自动每日.md)。 + ### broadcast — 弹幕播报与 TTS | 字段 | 默认值 | 说明 | @@ -81,7 +96,7 @@ npm run build ### commands — 内置指令别名 -11 个内置指令，每个可单独启用/禁用、自定义别名： +14 个内置指令，每个可单独启用/禁用、自定义别名和允许角色： | key | 默认别名 | 功能 | |-----|----------|------| @@ -91,6 +106,9 @@ npm run build | `confirm_yes` | 是 | 确认账号正确 | | `confirm_no` | 不是 | 确认账号不正确，重新扫码 | | `run` | 执行, 跑, 开始 | 执行配置组（需带参数） | +| `daily` | 自动每日 | 启动托管的一条龙每日任务，可选秘境、地脉或委托模式 | +| `switch_party` | 切换队伍, 更换队伍 | 修改并执行配置组“切换队伍” | +| `edit_party` | 修改队员, 更换队员 | 校验四名角色后修改并执行配置组“修改队员” | | `leave` | 退出 | 退出队列 | | `reset` | 重置 | 一级用户重启原神和 BGI | | `points` | 积分 | 查询积分 | @@ -162,7 +180,7 @@ npm run build # 构建到 web/admin/ ```powershell # 编译检查 -E:\Programs\Anaconda3\envs\Live-streaming\python.exe -m py_compile app/danmu_queue.py +.venv\Scripts\python.exe -m py_compile app/danmu_queue.py ``` 核心文件： @@ -170,6 +188,8 @@ E:\Programs\Anaconda3\envs\Live-streaming\python.exe -m py_compile app/danmu_que |------|------| | `app/main.py` | 入口，启动 Web 服务和子模块 | | `app/danmu_queue.py` | 弹幕监听、指令处理、队列管理、BGI 控制、TTS、Web API | +| `app/bettergi_daily.py` | 自动每日配置生成、秘境别名和队伍参数更新 | +| `app/bettergi_current_party.py` | 当前队伍读取托管配置组、状态协议和防串读校验 | | `app/music_monitor.py` | SMTC 音乐监听与点歌调度 | | `app/core/runtime_paths.py` | 运行时路径解析 | diff --git a/app/bettergi_current_party.py b/app/bettergi_current_party.py new file mode 100644 index 0000000..4c6d887 --- /dev/null +++ b/app/bettergi_current_party.py @@ -0,0 +1,181 @@ +"""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}") diff --git a/app/bettergi_daily.py b/app/bettergi_daily.py new file mode 100644 index 0000000..ae62008 --- /dev/null +++ b/app/bettergi_daily.py @@ -0,0 +1,828 @@ +"""BetterGI managed configuration helpers for daily and party commands.""" + +from __future__ import annotations + +import copy +import difflib +import json +import os +import re +import tempfile +import uuid +from dataclasses import dataclass, replace +from functools import lru_cache +from pathlib import Path +from typing import Any, Iterable + + +class DailyAutomationError(ValueError): + """Raised when a managed BetterGI command cannot be prepared safely.""" + + +DAILY_MODE_NONE = "none" +DAILY_MODE_DOMAIN = "domain" +DAILY_MODE_LEY_LINE = "ley_line" +DAILY_MODE_COMMISSION = "commission" + +LEY_LINE_COUNTRIES = ("蒙德", "璃月", "稻妻", "须弥", "枫丹", "纳塔", "挪德卡莱") +LEY_LINE_TYPE_ALIASES = { + "经验": "启示之花", + "经验花": "启示之花", + "蓝花": "启示之花", + "启示": "启示之花", + "启示之花": "启示之花", + "摩拉": "藏金之花", + "摩拉花": "藏金之花", + "金币": "藏金之花", + "金币花": "藏金之花", + "黄花": "藏金之花", + "藏金": "藏金之花", + "藏金之花": "藏金之花", +} + +DAILY_TASK_NAMES = { + "mail": "领取邮件", + "craft_resin": "合成树脂", + "domain": "自动秘境", + "ley_line": "自动地脉花", + "commission": "每日委托", + "serenitea": "领取尘歌壶奖励", + "daily_reward": "领取每日奖励", +} + +_INVALID_BGI_NAME = re.compile(r'[<>:"/\\|?*\x00-\x1f]') +_NAME_NORMALIZE = re.compile(r"[\s\-_—－·.。,:：，、/|]+") +_MEMBER_SEPARATOR = re.compile(r"[\s,，、/|]+") + + +@dataclass(frozen=True) +class DailyRequest: + mode: str = DAILY_MODE_NONE + domain_name: str = "" + ley_line_type: str = "" + ley_line_country: str = "" + + @property + def task_name(self) -> str: + if self.mode == DAILY_MODE_DOMAIN: + return f"自动每日（秘境：{self.domain_name}）" + if self.mode == DAILY_MODE_LEY_LINE: + return f"自动每日（地脉：{self.ley_line_type}/{self.ley_line_country}）" + if self.mode == DAILY_MODE_COMMISSION: + return "自动每日（委托）" + return "自动每日" + + @property + def summary(self) -> str: + if self.mode == DAILY_MODE_DOMAIN: + return f"秘境 {self.domain_name}" + if self.mode == DAILY_MODE_LEY_LINE: + return f"地脉 {self.ley_line_type} {self.ley_line_country}" + if self.mode == DAILY_MODE_COMMISSION: + return "每日委托" + return "跳过其他任务" + + +@dataclass(frozen=True) +class JsonUpdate: + path: Path + data: dict[str, Any] + reason: str + + +@dataclass(frozen=True) +class PreparedDailyRun: + request: DailyRequest + config_name: str + updates: tuple[JsonUpdate, ...] + requires_current_party: bool = False + + @property + def task_name(self) -> str: + return self.request.task_name + + +def _normalize_name(value: str) -> str: + return _NAME_NORMALIZE.sub("", str(value or "").strip().casefold()) + + +def _safe_bgi_name(value: str, label: str) -> str: + name = str(value or "").strip() + if not name: + raise DailyAutomationError(f"{label}不能为空") + if name in {".", ".."} or _INVALID_BGI_NAME.search(name): + raise DailyAutomationError(f"{label}包含非法文件名字符: {name}") + return name + + +def _read_json_object(path: Path, label: str) -> dict[str, Any]: + if not path.exists(): + raise DailyAutomationError(f"未找到{label}: {path}") + try: + data = json.loads(path.read_text(encoding="utf-8-sig")) + except (OSError, json.JSONDecodeError) as exc: + raise DailyAutomationError(f"读取{label}失败: {exc}") from exc + if not isinstance(data, dict): + raise DailyAutomationError(f"{label}必须是 JSON 对象: {path}") + return data + + +def parse_daily_request(argument: str) -> DailyRequest: + text = str(argument or "").strip() + if not text: + return DailyRequest() + + parts = text.split() + mode = parts[0] + if mode == "秘境": + domain_name = " ".join(parts[1:]).strip() + if not domain_name: + raise DailyAutomationError("秘境模式需要指定秘境，例如：自动每日 秘境 风本") + return DailyRequest(mode=DAILY_MODE_DOMAIN, domain_name=domain_name) + + if mode in {"地脉", "地脉花"}: + if len(parts) != 3: + raise DailyAutomationError("地脉模式格式：自动每日 地脉 <经验|摩拉> <国家>") + type_name = LEY_LINE_TYPE_ALIASES.get(_normalize_name(parts[1])) + if not type_name: + raise DailyAutomationError("地脉花类型仅支持经验或摩拉") + country = parts[2].strip() + if country not in LEY_LINE_COUNTRIES: + raise DailyAutomationError( + f"不支持的地脉国家'{country}'，可用：{'、'.join(LEY_LINE_COUNTRIES)}" + ) + return DailyRequest( + mode=DAILY_MODE_LEY_LINE, + ley_line_type=type_name, + ley_line_country=country, + ) + + if mode in {"委托", "每日委托"}: + if len(parts) != 1: + raise DailyAutomationError("委托模式不接受额外参数，格式：自动每日 委托") + return DailyRequest(mode=DAILY_MODE_COMMISSION) + + raise DailyAutomationError( + "每日模式仅支持：秘境、地脉、委托；不指定模式时直接发送“自动每日”" + ) + + +class DomainAliasResolver: + def __init__(self, alias_path: Path, bettergi_work_dir: Path): + self.alias_path = Path(alias_path) + self.bettergi_work_dir = Path(bettergi_work_dir) + + def _available_domains(self) -> list[str]: + settings_path = ( + self.bettergi_work_dir + / "User" + / "JsScript" + / "AutoDomain" + / "settings.json" + ) + if not settings_path.exists(): + raise DailyAutomationError( + f"未找到 BGI 自动秘境设置文件，请安装或更新 AutoDomain 脚本: {settings_path}" + ) + try: + settings = json.loads(settings_path.read_text(encoding="utf-8-sig")) + except (OSError, json.JSONDecodeError) as exc: + raise DailyAutomationError(f"读取 BGI 秘境列表失败: {exc}") from exc + if not isinstance(settings, list): + raise DailyAutomationError("BGI AutoDomain/settings.json 格式无效") + for item in settings: + if isinstance(item, dict) and item.get("name") == "domainName": + options = item.get("options") + if isinstance(options, list): + domains = [str(value).strip() for value in options if str(value).strip()] + if domains: + return domains + raise DailyAutomationError("BGI AutoDomain/settings.json 中没有秘境名称列表") + + def resolve(self, raw_name: str) -> str: + available = self._available_domains() + available_set = set(available) + alias_data = _read_json_object(self.alias_path, "秘境俗称文件") + lookup: dict[str, str] = {} + display_terms: dict[str, str] = {} + + def add(term: str, canonical: str) -> None: + normalized = _normalize_name(term) + if not normalized: + return + previous = lookup.get(normalized) + if previous and previous != canonical: + raise DailyAutomationError( + f"秘境俗称'{term}'同时指向'{previous}'和'{canonical}'" + ) + lookup[normalized] = canonical + display_terms[normalized] = str(term).strip() + + for canonical in available: + add(canonical, canonical) + + for canonical, aliases in alias_data.items(): + if str(canonical).startswith("_"): + continue + canonical_name = str(canonical).strip() + if canonical_name not in available_set: + raise DailyAutomationError( + f"秘境俗称文件中的正式名称不受当前 BGI 支持: {canonical_name}" + ) + if not isinstance(aliases, list): + raise DailyAutomationError(f"秘境'{canonical_name}'的俗称必须是数组") + add(canonical_name, canonical_name) + for alias in aliases: + add(str(alias), canonical_name) + + normalized_input = _normalize_name(raw_name) + resolved = lookup.get(normalized_input) + if resolved: + return resolved + + matches = difflib.get_close_matches(normalized_input, list(lookup), n=3, cutoff=0.45) + if matches: + suggestions = [] + for match in matches: + canonical = lookup[match] + display = display_terms.get(match, canonical) + suggestion = canonical if display == canonical else f"{display}({canonical})" + if suggestion not in suggestions: + suggestions.append(suggestion) + raise DailyAutomationError( + f"未知秘境'{raw_name}'，可能是：{'、'.join(suggestions)}" + ) + raise DailyAutomationError(f"未知秘境'{raw_name}'，请检查 config/domain_aliases.json") + + +def _validate_strategy(work_dir: Path, strategy_name: str, label: str) -> None: + strategy = str(strategy_name or "").strip() + if not strategy: + raise DailyAutomationError(f"{label}未配置战斗策略") + auto_fight_dir = work_dir / "User" / "AutoFight" + if strategy == "根据队伍自动选择": + if not auto_fight_dir.is_dir(): + raise DailyAutomationError(f"{label}战斗策略目录不存在: {auto_fight_dir}") + return + json_path = auto_fight_dir / f"{strategy}.json" + txt_path = auto_fight_dir / f"{strategy}.txt" + if not json_path.is_file() and not txt_path.is_file(): + raise DailyAutomationError(f"{label}战斗策略文件不存在: {strategy}") + + +def _template_task_id_candidates(template: dict[str, Any]) -> dict[str, list[str]]: + enabled = template.get("TaskEnabledList") + order = template.get("TaskOrder") + definitions = template.get("TaskDefinitions") + if not isinstance(enabled, dict): + raise DailyAutomationError("一条龙模板缺少 TaskEnabledList 对象") + if order is None: + order = [] + if not isinstance(order, list): + raise DailyAutomationError("一条龙模板缺少 TaskOrder 数组") + if definitions is None: + definitions = {} + if not isinstance(definitions, dict): + # BetterGI 为每个一条龙配置独立生成任务 ID，只能按任务名复用模板 ID。 + raise DailyAutomationError("一条龙模板的 TaskDefinitions 必须是对象") + + ordered_ids: list[str] = [] + for raw_id in [*order, *definitions.keys(), *enabled.keys()]: + task_id = str(raw_id or "").strip() + if task_id and task_id not in ordered_ids: + ordered_ids.append(task_id) + + candidates: dict[str, list[str]] = {} + old_format = not definitions + for task_id in ordered_ids: + raw_name = task_id if old_format else definitions.get(task_id) + task_name = str(raw_name or "").strip() + if not task_name: + continue + candidates.setdefault(task_name, []).append(task_id) + return candidates + + +def _build_task_entries( + template: dict[str, Any], + task_keys: list[str], +) -> list[tuple[str, str]]: + candidates = _template_task_id_candidates(template) + reserved_ids = { + task_id + for ids in candidates.values() + for task_id in ids + } + used_ids: set[str] = set() + entries: list[tuple[str, str]] = [] + for task_key in task_keys: + task_name = DAILY_TASK_NAMES[task_key] + task_id = next( + (candidate for candidate in candidates.get(task_name, []) if candidate not in used_ids), + "", + ) + while not task_id: + candidate = str(uuid.uuid4()) + if candidate not in reserved_ids and candidate not in used_ids: + task_id = candidate + used_ids.add(task_id) + reserved_ids.add(task_id) + entries.append((task_id, task_name)) + return entries + + +def _build_one_dragon_config( + template: dict[str, Any], + request: DailyRequest, + managed_name: str, + ley_line_craft_resin_before: bool, +) -> dict[str, Any]: + config = copy.deepcopy(template) + task_keys = ["mail"] + if request.mode != DAILY_MODE_LEY_LINE or ley_line_craft_resin_before: + task_keys.append("craft_resin") + if request.mode == DAILY_MODE_DOMAIN: + task_keys.append("domain") + elif request.mode == DAILY_MODE_LEY_LINE: + task_keys.append("ley_line") + elif request.mode == DAILY_MODE_COMMISSION: + task_keys.append("commission") + task_keys.extend(["serenitea", "daily_reward"]) + + task_entries = _build_task_entries(template, task_keys) + task_order = [task_id for task_id, _ in task_entries] + config["TaskEnabledList"] = {task_id: True for task_id in task_order} + config["TaskOrder"] = task_order + config["TaskDefinitions"] = dict(task_entries) + config["Name"] = managed_name + config["NextTaskId"] = "" + config["CompletionAction"] = "无" + + if request.mode == DAILY_MODE_DOMAIN: + config["WeeklyDomainEnabled"] = False + config["DomainName"] = request.domain_name + + if request.mode == DAILY_MODE_LEY_LINE: + config["LeyLineOneDragonMode"] = True + config["LeyLineResinExhaustionMode"] = True + config["LeyLineOpenModeCountMin"] = False + config["LeyLineRunCount"] = 1 + for day in ( + "Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday" + ): + config[f"LeyLineRun{day}"] = True + config[f"LeyLine{day}Type"] = request.ley_line_type + config[f"LeyLine{day}Country"] = request.ley_line_country + + return config + + +def _validate_commission_group( + work_path: Path, + *, + use_current_party: bool, +) -> bool: + group_path = work_path / "User" / "ScriptGroup" / "每日委托.json" + group = _read_json_object(group_path, "每日委托配置组") + projects = group.get("projects") + if not isinstance(projects, list): + raise DailyAutomationError("BGI 配置组“每日委托”缺少 projects 数组") + enabled_projects = [ + project + for project in projects + if isinstance(project, dict) + and str(project.get("status", "Enabled")).casefold() != "disabled" + ] + if not enabled_projects: + raise DailyAutomationError("BGI 配置组“每日委托”没有启用的可执行项目") + + uses_auto_commission_nova = any( + str(project.get("folderName") or "") == "AutoCommissionNova" + for project in enabled_projects + ) + if not uses_auto_commission_nova: + return False + + user_config_path = ( + work_path + / "User" + / "JsScript" + / "AutoCommissionNova" + / "Data" + / "user-config.json" + ) + if not user_config_path.is_file(): + raise DailyAutomationError( + "AutoCommissionNova 尚未完成首次配置：缺少 Data/user-config.json；" + "请先在 BGI 中手动运行脚本并保存用户配置和战斗策略" + ) + + user_config = _read_json_object(user_config_path, "AutoCommissionNova 用户配置") + party = user_config.get("party") + global_party = party.get("global") if isinstance(party, dict) else None + if not isinstance(global_party, dict): + raise DailyAutomationError("AutoCommissionNova 用户配置缺少 party.global 对象") + + missing = [] + if not use_current_party: + if not str(global_party.get("battleTeamName") or "").strip(): + missing.append("战斗队伍") + if not str(global_party.get("elementTeamName") or "").strip(): + missing.append("元素采集队伍") + if missing: + raise DailyAutomationError( + f"AutoCommissionNova 首次配置不完整，缺少：{'、'.join(missing)}" + ) + strategy_name = str( + global_party.get("battleStrategy") or "根据队伍自动选择" + ).strip() + _validate_strategy(work_path, strategy_name, "每日委托") + return bool(use_current_party) + + +def prepare_daily_run( + work_dir: str | Path, + alias_path: str | Path, + argument: str, + *, + template_name: str, + managed_name: str, + ley_line_craft_resin_before: bool, + commission_use_current_party: bool = False, +) -> PreparedDailyRun: + work_path = Path(work_dir) + if not work_path.is_dir(): + raise DailyAutomationError(f"BetterGI 工作目录不存在: {work_path}") + template = _safe_bgi_name(template_name, "一条龙模板名称") + managed = _safe_bgi_name(managed_name, "托管一条龙名称") + if template == managed: + raise DailyAutomationError("一条龙模板名称不能与托管配置名称相同") + + request = parse_daily_request(argument) + updates: list[JsonUpdate] = [] + requires_current_party = False + user_config_path = work_path / "User" / "config.json" + + if request.mode == DAILY_MODE_DOMAIN: + resolved = DomainAliasResolver(Path(alias_path), work_path).resolve(request.domain_name) + request = replace(request, domain_name=resolved) + user_config = _read_json_object(user_config_path, "BGI User/config.json") + auto_fight = user_config.get("autoFightConfig") + auto_domain = user_config.get("autoDomainConfig") + if not isinstance(auto_fight, dict) or not isinstance(auto_domain, dict): + raise DailyAutomationError("当前 BGI 缺少自动战斗或自动秘境配置，请升级到 0.63.0+") + _validate_strategy(work_path, auto_fight.get("strategyName", ""), "自动秘境") + corrected = copy.deepcopy(user_config) + corrected["autoDomainConfig"]["specifyResinUse"] = False + if corrected != user_config: + updates.append(JsonUpdate(user_config_path, corrected, "关闭自动秘境指定树脂次数")) + + elif request.mode == DAILY_MODE_LEY_LINE: + user_config = _read_json_object(user_config_path, "BGI User/config.json") + auto_fight = user_config.get("autoFightConfig") + ley_line = user_config.get("autoLeyLineOutcropConfig") + if not isinstance(auto_fight, dict) or not isinstance(ley_line, dict): + raise DailyAutomationError("当前 BGI 缺少自动战斗或自动地脉花配置，请升级到 0.63.0+") + fight_config = ley_line.get("fightConfig") + strategy_name = "" + if isinstance(fight_config, dict): + strategy_name = str(fight_config.get("strategyName") or "").strip() + if not strategy_name: + strategy_name = str(auto_fight.get("strategyName") or "").strip() + _validate_strategy(work_path, strategy_name, "自动地脉花") + if ley_line.get("friendshipTeam") and not ley_line.get("team"): + raise DailyAutomationError("BGI 自动地脉花配置了好感队，但未配置战斗队伍") + corrected = copy.deepcopy(user_config) + corrected["autoLeyLineOutcropConfig"]["isGoToSynthesizer"] = False + if corrected != user_config: + updates.append(JsonUpdate(user_config_path, corrected, "关闭地脉花内部合成树脂")) + + elif request.mode == DAILY_MODE_COMMISSION: + requires_current_party = _validate_commission_group( + work_path, + use_current_party=bool(commission_use_current_party), + ) + + template_path = work_path / "User" / "OneDragon" / f"{template}.json" + template_config = _read_json_object(template_path, "一条龙模板") + managed_config = _build_one_dragon_config( + template_config, + request, + managed, + bool(ley_line_craft_resin_before), + ) + managed_path = work_path / "User" / "OneDragon" / f"{managed}.json" + updates.append(JsonUpdate(managed_path, managed_config, "生成直播自动每日一条龙")) + return PreparedDailyRun( + request=request, + config_name=managed, + updates=tuple(updates), + requires_current_party=requires_current_party, + ) + + +def prepare_commission_current_party_update( + work_dir: str | Path, + party_name: str, +) -> JsonUpdate: + work_path = Path(work_dir) + name = str(party_name or "").strip() + if not name: + raise DailyAutomationError("当前队伍名称为空") + if len(name) > 20 or any(ord(char) < 32 for char in name): + raise DailyAutomationError(f"当前队伍名称格式无效: {name}") + + user_config_path = ( + work_path + / "User" + / "JsScript" + / "AutoCommissionNova" + / "Data" + / "user-config.json" + ) + user_config = _read_json_object(user_config_path, "AutoCommissionNova 用户配置") + corrected = copy.deepcopy(user_config) + party = corrected.get("party") + global_party = party.get("global") if isinstance(party, dict) else None + if not isinstance(global_party, dict): + raise DailyAutomationError("AutoCommissionNova 用户配置缺少 party.global 对象") + global_party["battleTeamName"] = name + global_party["elementTeamName"] = name + return JsonUpdate( + user_config_path, + corrected, + f"将 AutoCommissionNova 战斗及元素采集队伍更新为当前队伍“{name}”", + ) + + +def _prepare_script_group_update( + work_dir: str | Path, + group_name: str, + folder_name: str, + settings_patch: dict[str, Any], +) -> JsonUpdate: + work_path = Path(work_dir) + safe_group = _safe_bgi_name(group_name, "配置组名称") + group_path = work_path / "User" / "ScriptGroup" / f"{safe_group}.json" + group = _read_json_object(group_path, f"配置组“{safe_group}”") + projects = group.get("projects") + if not isinstance(projects, list): + raise DailyAutomationError(f"配置组“{safe_group}”缺少 projects 数组") + matches = [ + project + for project in projects + if isinstance(project, dict) + and str(project.get("folderName") or "") == folder_name + and str(project.get("status", "Enabled")).casefold() != "disabled" + ] + if not matches: + raise DailyAutomationError( + f"配置组“{safe_group}”中没有启用的 {folder_name} JavaScript 项目" + ) + if len(matches) > 1: + raise DailyAutomationError( + f"配置组“{safe_group}”包含多个启用的 {folder_name} 项目，请只保留一个" + ) + settings = matches[0].get("jsScriptSettingsObject") + if not isinstance(settings, dict): + settings = {} + matches[0]["jsScriptSettingsObject"] = settings + settings.update(settings_patch) + return JsonUpdate(group_path, group, f"更新配置组“{safe_group}”参数") + + +def prepare_switch_party_update(work_dir: str | Path, party_name: str) -> JsonUpdate: + party = str(party_name or "").strip() + if not party: + raise DailyAutomationError("队伍名称不能为空") + return _prepare_script_group_update( + work_dir, + "切换队伍", + "AcceleratedEditionSwitchParty", + {"partyName": party}, + ) + + +def _add_character_lookup( + lookup: dict[str, str], + ambiguous: set[str], + raw_name: str, + canonical: str, +) -> None: + key = _normalize_name(raw_name) + if not key or key in ambiguous: + return + previous = lookup.get(key) + if previous and previous != canonical: + lookup.pop(key, None) + ambiguous.add(key) + return + lookup[key] = canonical + + +def _character_lookups_from_settings( + data: Any, +) -> tuple[dict[str, str], dict[str, str]] | None: + if not isinstance(data, list): + return None + + position_options: dict[str, list[str]] = {} + for item in data: + if not isinstance(item, dict): + continue + name = str(item.get("name") or "") + if name not in {"position1", "position2", "position3", "position4"}: + continue + options = item.get("options") + if isinstance(options, list): + position_options[name] = [ + str(option).strip() + for option in options + if str(option).strip() + ] + if len(position_options) != 4 or not position_options.get("position1"): + return None + + full_lookup: dict[str, str] = {} + simple_lookup: dict[str, str] = {} + ambiguous_full: set[str] = set() + ambiguous_simple: set[str] = set() + for option in position_options["position1"]: + _add_character_lookup(full_lookup, ambiguous_full, option, option) + simple_name = option.rsplit("-", 1)[-1].strip() + _add_character_lookup(simple_lookup, ambiguous_simple, simple_name, option) + if not simple_lookup: + return None + return full_lookup, simple_lookup + + +def _character_lookups_from_combat_avatar( + data: Any, +) -> tuple[dict[str, str], dict[str, str]] | None: + if not isinstance(data, list): + return None + + full_lookup: dict[str, str] = {} + simple_lookup: dict[str, str] = {} + ambiguous_full: set[str] = set() + ambiguous_simple: set[str] = set() + for item in data: + if not isinstance(item, dict): + continue + canonical = str(item.get("name") or "").strip() + if not canonical: + continue + terms = [canonical] + aliases = item.get("alias") + if isinstance(aliases, list): + terms.extend(str(alias).strip() for alias in aliases if str(alias).strip()) + for term in terms: + _add_character_lookup(full_lookup, ambiguous_full, term, canonical) + _add_character_lookup(simple_lookup, ambiguous_simple, term, canonical) + if not simple_lookup: + return None + return full_lookup, simple_lookup + + +def _load_character_options(work_dir: Path) -> tuple[dict[str, str], dict[str, str]]: + script_dir = work_dir / "User" / "JsScript" / "AutoSwitchRoles" + settings_path = script_dir / "settings.json" + avatar_path = script_dir / "combat_avatar.json" + failures: list[str] = [] + + if settings_path.exists(): + try: + settings_data = json.loads(settings_path.read_text(encoding="utf-8-sig")) + except (OSError, json.JSONDecodeError) as exc: + failures.append(f"settings.json 读取失败: {exc}") + else: + lookups = _character_lookups_from_settings(settings_data) + if lookups: + return lookups + failures.append("settings.json 未提供四个队员位置的 options") + else: + failures.append("缺少 settings.json") + + if avatar_path.exists(): + try: + avatar_data = json.loads(avatar_path.read_text(encoding="utf-8-sig")) + except (OSError, json.JSONDecodeError) as exc: + failures.append(f"combat_avatar.json 读取失败: {exc}") + else: + lookups = _character_lookups_from_combat_avatar(avatar_data) + if lookups: + return lookups + failures.append("combat_avatar.json 中没有可用角色") + else: + failures.append("缺少 combat_avatar.json") + + raise DailyAutomationError( + "AutoSwitchRoles 角色数据不可用，请安装或更新“配对界面切换角色”脚本: " + + "；".join(failures) + ) + + +def _resolve_member_token( + token: str, + full_lookup: dict[str, str], + simple_lookup: dict[str, str], +) -> str: + key = _normalize_name(token) + resolved = full_lookup.get(key) or simple_lookup.get(key) + if not resolved: + raise DailyAutomationError(f"未知或有歧义的角色名称: {token}") + return resolved + + +def _split_contiguous_members(text: str, simple_lookup: dict[str, str]) -> list[str]: + normalized = _normalize_name(text) + candidates = sorted(simple_lookup, key=len, reverse=True) + + @lru_cache(maxsize=None) + def walk(offset: int, slots: int) -> tuple[tuple[str, ...], ...]: + if slots == 4: + return ((),) if offset == len(normalized) else () + if offset >= len(normalized): + return () + results: list[tuple[str, ...]] = [] + for candidate in candidates: + if not normalized.startswith(candidate, offset): + continue + for remainder in walk(offset + len(candidate), slots + 1): + results.append((candidate, *remainder)) + if len(results) >= 2: + return tuple(results) + return tuple(results) + + segmentations = walk(0, 0) + if not segmentations: + raise DailyAutomationError("队员必须是4人，请使用空格、逗号、顿号或斜杠分隔") + if len(segmentations) > 1: + raise DailyAutomationError("连续角色名存在多种拆分方式，请使用空格分隔四名角色") + return [simple_lookup[key] for key in segmentations[0]] + + +def resolve_party_members(work_dir: str | Path, argument: str) -> tuple[list[str], list[str]]: + text = str(argument or "").strip() + if not text: + raise DailyAutomationError("队员必须是4人") + full_lookup, simple_lookup = _load_character_options(Path(work_dir)) + parts = [part for part in _MEMBER_SEPARATOR.split(text) if part] + if len(parts) == 4: + resolved = [ + _resolve_member_token(part, full_lookup, simple_lookup) + for part in parts + ] + else: + resolved = _split_contiguous_members(text, simple_lookup) + if len(resolved) != 4: + raise DailyAutomationError("队员必须是4人") + if len(set(resolved)) != 4: + raise DailyAutomationError("四名队员不能重复") + display_names = [value.rsplit("-", 1)[-1] for value in resolved] + return resolved, display_names + + +def prepare_edit_party_update( + work_dir: str | Path, + argument: str, +) -> tuple[JsonUpdate, tuple[str, ...]]: + resolved, display_names = resolve_party_members(work_dir, argument) + update = _prepare_script_group_update( + work_dir, + "修改队员", + "AutoSwitchRoles", + {f"position{index + 1}": value for index, value in enumerate(resolved)}, + ) + return update, tuple(display_names) + + +def _write_json_atomic(path: Path, data: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, temp_name = tempfile.mkstemp( + prefix=f".{path.name}.", + suffix=".tmp", + dir=str(path.parent), + ) + temp_path = Path(temp_name) + try: + with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle: + json.dump(data, handle, ensure_ascii=False, indent=2) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temp_path, path) + finally: + if temp_path.exists(): + temp_path.unlink() + + +def apply_json_updates(updates: Iterable[JsonUpdate]) -> None: + seen: set[Path] = set() + for update in updates: + path = Path(update.path) + resolved = path.resolve() + if resolved in seen: + raise DailyAutomationError(f"同一配置文件被重复更新: {path}") + seen.add(resolved) + _write_json_atomic(path, update.data) diff --git a/app/danmu_queue.py b/app/danmu_queue.py index e5ad7d6..8c68e30 100644 --- a/app/danmu_queue.py +++ b/app/danmu_queue.py @@ -15,6 +15,9 @@ BetterGI 弹幕排队系统 上号 - 队首触发扫码上号配置组 执行 <组名> - 已确认账号的队首触发配置组(如: 执行 泡泡桔) 跑 <组名> - 兼容旧指令，等同于执行 + 自动每日 [模式] - 启动托管的一条龙每日任务 + 切换队伍 <名称> - 执行“切换队伍”配置组 + 修改队员 <四人> - 执行“修改队员”配置组 退出 - 退出排队队列(三级队首不可用) 重置 - 一级用户重启原神和BetterGI 积分 - 查询自己的积分 @@ -105,6 +108,42 @@ if str(APP_DIR) not in sys.path: from admin_auth import AdminAuthManager, SESSION_COOKIE_NAME from admin_events import AdminEventBus +if __package__: + from .bettergi_current_party import ( + CURRENT_PARTY_SCRIPT_NAME, + CurrentPartyReadError, + PreparedCurrentPartyRead, + clear_current_party_status, + prepare_current_party_read, + read_current_party_status, + ) + from .bettergi_daily import ( + DailyAutomationError, + JsonUpdate, + apply_json_updates, + prepare_commission_current_party_update, + prepare_daily_run, + prepare_edit_party_update, + prepare_switch_party_update, + ) +else: + from bettergi_current_party import ( + CURRENT_PARTY_SCRIPT_NAME, + CurrentPartyReadError, + PreparedCurrentPartyRead, + clear_current_party_status, + prepare_current_party_read, + read_current_party_status, + ) + from bettergi_daily import ( + DailyAutomationError, + JsonUpdate, + apply_json_updates, + prepare_commission_current_party_update, + prepare_daily_run, + prepare_edit_party_update, + prepare_switch_party_update, + ) from bilibili_cookie_refresh import BilibiliCookieRefresher, BilibiliCredentialStore, BilibiliQrLogin from faster_qwen_worker import FasterQwenWorkerClient from mpv_player import MpvPlayer @@ -199,6 +238,9 @@ COMMAND_ALLOWED_ROLE_DEFAULTS = { "confirm_yes": [ROLE_SUPER_ADMIN, ROLE_PENDING_OPERATOR], "confirm_no": [ROLE_SUPER_ADMIN, ROLE_PENDING_OPERATOR], "run": [ROLE_SUPER_ADMIN, ROLE_ACTIVE_OPERATOR], + "daily": [ROLE_SUPER_ADMIN, ROLE_ACTIVE_OPERATOR], + "switch_party": [ROLE_SUPER_ADMIN, ROLE_ACTIVE_OPERATOR], + "edit_party": [ROLE_SUPER_ADMIN, ROLE_ACTIVE_OPERATOR], "leave": [ROLE_SUPER_ADMIN, ROLE_ACTIVE_OPERATOR, ROLE_VIEWER], "reset": [ROLE_SUPER_ADMIN], "points": ALL_DANMU_ROLES, @@ -406,6 +448,13 @@ class Config: self.data["bilibili"].setdefault("cookie_auto_refresh_enabled", True) self.data["bilibili"].setdefault("cookie_check_interval_hours", 6) self.data.setdefault("bettergi", {}) + self.data.setdefault("daily", {}) + daily_cfg = self.data["daily"] + daily_cfg.setdefault("one_dragon_template", "默认配置") + daily_cfg.setdefault("managed_one_dragon_name", "直播系统自动每日") + daily_cfg.setdefault("ley_line_craft_resin_before", True) + daily_cfg.setdefault("commission_use_current_party", True) + daily_cfg.setdefault("current_party_read_timeout_sec", 45) self.data.setdefault("global", {}) self.data["global"].pop("admin_uidsText", None) self.data.setdefault("queue", {}) @@ -574,6 +623,9 @@ class Config: "confirm_yes": {"enabled": True, "aliases": ["是"]}, "confirm_no": {"enabled": True, "aliases": ["不是"]}, "run": {"enabled": True, "aliases": ["执行", "跑", "开始"]}, + "daily": {"enabled": True, "aliases": ["自动每日"]}, + "switch_party": {"enabled": True, "aliases": ["切换队伍", "更换队伍"]}, + "edit_party": {"enabled": True, "aliases": ["修改队员", "更换队员"]}, "leave": {"enabled": True, "aliases": ["退出", "退出排队", "取消排队"]}, "reset": {"enabled": True, "aliases": ["重置"]}, "points": {"enabled": True, "aliases": ["积分"]}, @@ -629,6 +681,10 @@ class Config: wd = self.data.get("bettergi", {}).get("work_dir", "") return wd if wd else str(Path(self.bettergi_exe).parent) + @property + def daily_cfg(self) -> dict: + return self.data.get("daily", {}) + @property def default_group(self) -> str: return self.data.get("queue", {}).get("default_group", "薄荷") @@ -3050,10 +3106,11 @@ class Broadcaster: # ============== BetterGI 日志监控 ============== class BgiLogMonitor: - """实时监控BetterGI日志，检测配置组完成。""" + """实时监控 BetterGI 日志，检测配置组或一条龙完成。""" # 完成关键字: 配置组 "组名" 执行结束 FINISH_PATTERN = "执行结束" + ONE_DRAGON_FINISH_PATTERN = "一条龙和配置组任务结束" LOG_HEADER_PATTERN = re.compile( r"^\[(?P