3 Commits
29 changed files with 4040 additions and 134 deletions
+6
View File
@@ -2,6 +2,7 @@
__pycache__/ __pycache__/
*.py[cod] *.py[cod]
.venv/ .venv/
.venv-tts/
runtime/ runtime/
config/.venv_python_path.txt config/.venv_python_path.txt
@@ -21,6 +22,10 @@ data/queue_state.json
data/song_requests.json data/song_requests.json
data/tts_state.json data/tts_state.json
data/bilibili_credentials.json data/bilibili_credentials.json
data/admin_audit.log
data/admin_auth.json
data/statistics.sqlite3*
data/users.json
web/music_cover.jpg web/music_cover.jpg
# IDE # IDE
@@ -30,3 +35,4 @@ web/music_cover.jpg
# Large binaries (kept local only) # Large binaries (kept local only)
vendor/mpv/mpv.exe vendor/mpv/mpv.exe
vendor/mpv/mpv.7z vendor/mpv/mpv.7z
vendor/tts-model/
+27 -7
View File
@@ -1,19 +1,20 @@
# BetterGI 直播联动 # 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**: 构建前端用 - **Node.js**: 构建前端用
- **BetterGI**: 自动化脚本引擎,路径在 `config/config.json` 中配置 - **BetterGI**: `0.63.0+`,路径在 `config/config.json` 中配置
```powershell ```powershell
# 安装 Python 依赖 # 安装 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,仅需一次) # 下载 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 cd frontend\admin
@@ -69,6 +70,20 @@ npm run build
| `admin_uids` | 管理员 UID 列表(一级用户,可用重置等特权指令) | | `admin_uids` | 管理员 UID 列表(一级用户,可用重置等特权指令) |
| `log_level` | 日志级别:`DEBUG` / `INFO` / `WARNING` | | `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 ### broadcast — 弹幕播报与 TTS
| 字段 | 默认值 | 说明 | | 字段 | 默认值 | 说明 |
@@ -81,7 +96,7 @@ npm run build
### commands — 内置指令别名 ### commands — 内置指令别名
11 个内置指令,每个可单独启用/禁用、自定义别名: 14 个内置指令,每个可单独启用/禁用、自定义别名和允许角色
| key | 默认别名 | 功能 | | key | 默认别名 | 功能 |
|-----|----------|------| |-----|----------|------|
@@ -91,6 +106,9 @@ npm run build
| `confirm_yes` | 是 | 确认账号正确 | | `confirm_yes` | 是 | 确认账号正确 |
| `confirm_no` | 不是 | 确认账号不正确,重新扫码 | | `confirm_no` | 不是 | 确认账号不正确,重新扫码 |
| `run` | 执行, 跑, 开始 | 执行配置组(需带参数) | | `run` | 执行, 跑, 开始 | 执行配置组(需带参数) |
| `daily` | 自动每日, 每日, 日常, 做每日, 做日常 | 启动托管的一条龙每日任务,可选秘境、地脉或委托模式 |
| `switch_party` | 切换队伍, 更换队伍, 换队伍, 换队 | 修改并执行配置组“切换队伍” |
| `edit_party` | 修改队员, 更换队员, 换队员, 换角色 | 校验四名角色后修改并执行配置组“修改队员” |
| `leave` | 退出 | 退出队列 | | `leave` | 退出 | 退出队列 |
| `reset` | 重置 | 一级用户重启原神和 BGI | | `reset` | 重置 | 一级用户重启原神和 BGI |
| `points` | 积分 | 查询积分 | | `points` | 积分 | 查询积分 |
@@ -162,7 +180,7 @@ npm run build # 构建到 web/admin/
```powershell ```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/main.py` | 入口,启动 Web 服务和子模块 |
| `app/danmu_queue.py` | 弹幕监听、指令处理、队列管理、BGI 控制、TTS、Web API | | `app/danmu_queue.py` | 弹幕监听、指令处理、队列管理、BGI 控制、TTS、Web API |
| `app/bettergi_daily.py` | 自动每日配置生成、秘境别名和队伍参数更新 |
| `app/bettergi_current_party.py` | 当前队伍读取托管配置组、状态协议和防串读校验 |
| `app/music_monitor.py` | SMTC 音乐监听与点歌调度 | | `app/music_monitor.py` | SMTC 音乐监听与点歌调度 |
| `app/core/runtime_paths.py` | 运行时路径解析 | | `app/core/runtime_paths.py` | 运行时路径解析 |
+181
View File
@@ -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}")
+828
View File
@@ -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)
+793 -76
View File
File diff suppressed because it is too large Load Diff
+103 -9
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import io import io
import multiprocessing import multiprocessing
import os import os
import sys
import threading import threading
import time import time
import traceback import traceback
@@ -13,13 +14,14 @@ from typing import Any, Callable
MAX_NEW_TOKENS = 384 MAX_NEW_TOKENS = 384
DEFAULT_SYNTHESIS_TIMEOUT_SECONDS = 120.0 DEFAULT_SYNTHESIS_TIMEOUT_SECONDS = 120.0
DEFAULT_STARTUP_TIMEOUT_SECONDS = 480.0 DEFAULT_STARTUP_TIMEOUT_SECONDS = 900.0
# 启动失败后再次拉起 worker 的最小间隔:避免"启动超时→立即重启→再超时"的死循环 # 启动失败后再次拉起 worker 的最小间隔:避免"启动超时→立即重启→再超时"的死循环
# 在系统高负载时持续加载 torch/CUDA,进一步加剧卡顿。 # 在系统高负载时持续加载 torch/CUDA,进一步加剧卡顿。
STARTUP_FAILURE_BACKOFF_SECONDS = 60.0 STARTUP_FAILURE_BACKOFF_SECONDS = 60.0
DEFAULT_CPU_THREADS = 4 DEFAULT_CPU_THREADS = 4
DEFAULT_CPU_AFFINITY_COUNT = 8 DEFAULT_CPU_AFFINITY_COUNT = 8
DEFAULT_PROCESS_PRIORITY = "below_normal" DEFAULT_PROCESS_PRIORITY = "below_normal"
STARTUP_PROGRESS_POLL_SECONDS = 1.0
class FasterQwenWorkerError(RuntimeError): class FasterQwenWorkerError(RuntimeError):
@@ -122,15 +124,26 @@ def _apply_worker_process_limits(settings: dict[str, Any]) -> dict[str, Any]:
} }
def _load_runtime(settings: dict[str, Any]) -> dict[str, Any]: def _load_runtime(
settings: dict[str, Any],
progress: Callable[[str], None] | None = None,
) -> dict[str, Any]:
def report(stage: str) -> None:
if progress:
progress(stage)
device = str(settings.get("device") or "cuda") device = str(settings.get("device") or "cuda")
if device == "cpu" and "CUDA_VISIBLE_DEVICES" not in os.environ: if device == "cpu" and "CUDA_VISIBLE_DEVICES" not in os.environ:
os.environ["CUDA_VISIBLE_DEVICES"] = "" os.environ["CUDA_VISIBLE_DEVICES"] = ""
cpu_threads = _configure_worker_environment(settings) cpu_threads = _configure_worker_environment(settings)
report("torch_import_started")
import torch import torch
report("torch_import_complete")
_configure_torch_threads(torch, cpu_threads) _configure_torch_threads(torch, cpu_threads)
report("tts_library_import_started")
from faster_qwen3_tts import FasterQwen3TTS from faster_qwen3_tts import FasterQwen3TTS
report("tts_library_import_complete")
if device == "cpu": if device == "cpu":
torch.cuda.is_available = lambda: False torch.cuda.is_available = lambda: False
@@ -138,10 +151,12 @@ def _load_runtime(settings: dict[str, Any]) -> dict[str, Any]:
load_kwargs: dict[str, Any] = {} load_kwargs: dict[str, Any] = {}
if device == "cpu": if device == "cpu":
load_kwargs["device"] = "cpu" load_kwargs["device"] = "cpu"
report("from_pretrained_started")
model = FasterQwen3TTS.from_pretrained( model = FasterQwen3TTS.from_pretrained(
str(settings.get("model_name_or_path") or "Qwen/Qwen3-TTS-12Hz-0.6B-Base"), str(settings.get("model_name_or_path") or "Qwen/Qwen3-TTS-12Hz-0.6B-Base"),
**load_kwargs, **load_kwargs,
) )
report("from_pretrained_complete")
# voice_clone_prompt 不再预计算:当前 faster_qwen3_tts 的 FasterQwen3TTS 没有 # voice_clone_prompt 不再预计算:当前 faster_qwen3_tts 的 FasterQwen3TTS 没有
# create_voice_clone_prompt 方法,预计算只会失败。改为每次合成时在 # create_voice_clone_prompt 方法,预计算只会失败。改为每次合成时在
@@ -193,21 +208,63 @@ def _synthesize_wav(runtime: dict[str, Any], text: str) -> bytes:
def faster_qwen_worker_main(connection: Connection, settings: dict[str, Any]) -> None: def faster_qwen_worker_main(connection: Connection, settings: dict[str, Any]) -> None:
def report(stage: str, **details: Any) -> None:
try: try:
connection.send({
"type": "progress",
"stage": stage,
"pid": os.getpid(),
**details,
})
except Exception:
pass
try:
report("process_started", executable=sys.executable, prefix=sys.prefix)
cpu_threads = _configure_worker_environment(settings) cpu_threads = _configure_worker_environment(settings)
process_limits = _apply_worker_process_limits(settings) report("environment_ready", cpu_threads=cpu_threads)
load_started = time.monotonic() load_started = time.monotonic()
runtime = _load_runtime(settings) report("model_loading", model_name=str(settings.get("model_name_or_path") or ""))
runtime = _load_runtime(settings, progress=report)
load_ms = int((time.monotonic() - load_started) * 1000) load_ms = int((time.monotonic() - load_started) * 1000)
report("model_loaded", load_ms=load_ms)
warmup_started = time.monotonic() warmup_started = time.monotonic()
graph_warmup_ms = 0
report("warmup_started")
warmup = getattr(runtime["model"], "warmup", None)
if callable(warmup):
graph_warmup_started = time.monotonic()
warmup()
graph_warmup_ms = int((time.monotonic() - graph_warmup_started) * 1000)
report("graph_warmup_complete", graph_warmup_ms=graph_warmup_ms)
# model.warmup() 只捕获模型内部 CUDA Graph,不会初始化参考音频和
# voice-clone 生成路径。若把这一步留到首条播报,后台会在近一分钟内
# 看起来毫无响应,因此在降低进程优先级之前完成一次真实短句合成。
voice_clone_warmup_started = time.monotonic()
report("voice_clone_warmup_started")
_synthesize_wav(runtime, "系统启动") _synthesize_wav(runtime, "系统启动")
voice_clone_warmup_ms = int((time.monotonic() - voice_clone_warmup_started) * 1000)
report("voice_clone_warmup_complete", voice_clone_warmup_ms=voice_clone_warmup_ms)
warmup_ms = int((time.monotonic() - warmup_started) * 1000) warmup_ms = int((time.monotonic() - warmup_started) * 1000)
report(
"warmup_complete",
warmup_ms=warmup_ms,
graph_warmup_ms=graph_warmup_ms,
voice_clone_warmup_ms=voice_clone_warmup_ms,
)
# 加载和 CUDA Graph 捕获需要完整 CPU 调度能力。仅在预热完成后降低
# worker 优先级和亲和性,避免繁忙直播环境下启动时间被放大到超时。
process_limits = _apply_worker_process_limits(settings)
connection.send({ connection.send({
"type": "ready", "type": "ready",
"pid": os.getpid(), "pid": os.getpid(),
"load_ms": load_ms, "load_ms": load_ms,
"warmup_ms": warmup_ms, "warmup_ms": warmup_ms,
"graph_warmup_ms": graph_warmup_ms,
"voice_clone_warmup_ms": voice_clone_warmup_ms,
"max_new_tokens": MAX_NEW_TOKENS, "max_new_tokens": MAX_NEW_TOKENS,
"cpu_threads": cpu_threads, "cpu_threads": cpu_threads,
**process_limits, **process_limits,
@@ -278,7 +335,7 @@ class FasterQwenWorkerClient:
self.settings = dict(settings) self.settings = dict(settings)
self.logger = logger self.logger = logger
self.synthesis_timeout_seconds = max(1.0, float(synthesis_timeout_seconds)) self.synthesis_timeout_seconds = max(1.0, float(synthesis_timeout_seconds))
self.startup_timeout_seconds = max(10.0, float(startup_timeout_seconds)) self.startup_timeout_seconds = max(0.1, float(startup_timeout_seconds))
self._context = context or multiprocessing.get_context("spawn") self._context = context or multiprocessing.get_context("spawn")
self._process_target = process_target or faster_qwen_worker_main self._process_target = process_target or faster_qwen_worker_main
self._lock = threading.RLock() self._lock = threading.RLock()
@@ -330,21 +387,55 @@ class FasterQwenWorkerClient:
self._worker_pid = int(getattr(process, "pid", 0) or 0) self._worker_pid = int(getattr(process, "pid", 0) or 0)
self._log("info", "[FasterQwenTTS] worker 已启动, pid=%s,正在加载和预热", self.worker_pid) self._log("info", "[FasterQwenTTS] worker 已启动, pid=%s,正在加载和预热", self.worker_pid)
if not parent_connection.poll(self.startup_timeout_seconds): started = time.monotonic()
deadline = started + self.startup_timeout_seconds
last_stage = "process_spawned"
message: Any = None
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
self._next_start_after = time.monotonic() + STARTUP_FAILURE_BACKOFF_SECONDS self._next_start_after = time.monotonic() + STARTUP_FAILURE_BACKOFF_SECONDS
self._terminate_worker_locked("startup_timeout", graceful=False) self._terminate_worker_locked("startup_timeout", graceful=False)
raise FasterQwenWorkerError( raise FasterQwenWorkerError(
f"Faster-Qwen3-TTS worker startup exceeded {self.startup_timeout_seconds:.0f}s" f"Faster-Qwen3-TTS worker startup exceeded {self.startup_timeout_seconds:.0f}s "
f"(last_stage={last_stage})"
) )
if not parent_connection.poll(min(STARTUP_PROGRESS_POLL_SECONDS, remaining)):
if not process.is_alive():
exit_code = getattr(process, "exitcode", None)
self._next_start_after = time.monotonic() + STARTUP_FAILURE_BACKOFF_SECONDS
self._terminate_worker_locked("startup_process_exited", graceful=False)
raise FasterQwenWorkerError(
f"Faster-Qwen3-TTS worker exited during startup "
f"(exit_code={exit_code}, last_stage={last_stage})"
)
continue
try: try:
message = parent_connection.recv() message = parent_connection.recv()
except (EOFError, OSError) as exc: except (EOFError, OSError) as exc:
self._next_start_after = time.monotonic() + STARTUP_FAILURE_BACKOFF_SECONDS self._next_start_after = time.monotonic() + STARTUP_FAILURE_BACKOFF_SECONDS
self._terminate_worker_locked("startup_connection_closed", graceful=False) self._terminate_worker_locked("startup_connection_closed", graceful=False)
raise FasterQwenWorkerError("Faster-Qwen3-TTS worker exited during startup") from exc raise FasterQwenWorkerError(
f"Faster-Qwen3-TTS worker connection closed during startup "
f"(last_stage={last_stage})"
) from exc
if isinstance(message, dict) and message.get("type") == "progress":
last_stage = str(message.get("stage") or "unknown")
elapsed_ms = int((time.monotonic() - started) * 1000)
self._log(
"info",
"[FasterQwenTTS] worker 启动进度: %s, elapsed=%sms",
last_stage,
elapsed_ms,
)
continue
break
if not isinstance(message, dict) or message.get("type") != "ready": if not isinstance(message, dict) or message.get("type") != "ready":
if isinstance(message, dict): if isinstance(message, dict):
error = str(message.get("error") or message.get("error_type") or "unknown startup error") error = str(message.get("error") or message.get("error_type") or "unknown startup error")
worker_traceback = str(message.get("traceback") or "").strip()
if worker_traceback:
self._log("error", "[FasterQwenTTS] worker 启动异常:\n%s", worker_traceback)
else: else:
error = "invalid startup response" error = "invalid startup response"
self._next_start_after = time.monotonic() + STARTUP_FAILURE_BACKOFF_SECONDS self._next_start_after = time.monotonic() + STARTUP_FAILURE_BACKOFF_SECONDS
@@ -353,10 +444,13 @@ class FasterQwenWorkerClient:
self._next_start_after = 0.0 self._next_start_after = 0.0
self._log( self._log(
"info", "info",
"[FasterQwenTTS] worker 预热完成, pid=%s load=%sms warmup=%sms max_new_tokens=%s", "[FasterQwenTTS] worker 预热完成, pid=%s load=%sms warmup=%sms "
"graph=%sms voice_clone=%sms max_new_tokens=%s",
self.worker_pid, self.worker_pid,
message.get("load_ms"), message.get("load_ms"),
message.get("warmup_ms"), message.get("warmup_ms"),
message.get("graph_warmup_ms"),
message.get("voice_clone_warmup_ms"),
message.get("max_new_tokens"), message.get("max_new_tokens"),
) )
self._log( self._log(
+50 -2
View File
@@ -12,9 +12,16 @@
}, },
"bettergi": { "bettergi": {
"_说明": "BetterGI.exe 的完整路径。work_dir 留空则自动取 exe 所在目录。", "_说明": "BetterGI.exe 的完整路径。work_dir 留空则自动取 exe 所在目录。",
"exe_path": "C:\\Program Files\\BetterGI\\BetterGI.exe", "exe_path": "D:\\bgi-live\\BetterGI.exe",
"work_dir": "" "work_dir": ""
}, },
"daily": {
"one_dragon_template": "每日模板",
"managed_one_dragon_name": "直播系统自动每日",
"ley_line_craft_resin_before": true,
"commission_use_current_party": true,
"current_party_read_timeout_sec": 45
},
"global": { "global": {
"_说明": "default_cooldown: 冷却秒数防刷屏。admin_uids: 管理员UID绕过冷却。log_level: DEBUG/INFO/WARNING。restart_mode: gentle温和(执行中跳过)/aggressive激进(先杀BetterGI再重启,立即响应,适合直播)。", "_说明": "default_cooldown: 冷却秒数防刷屏。admin_uids: 管理员UID绕过冷却。log_level: DEBUG/INFO/WARNING。restart_mode: gentle温和(执行中跳过)/aggressive激进(先杀BetterGI再重启,立即响应,适合直播)。",
"default_cooldown": 30, "default_cooldown": 30,
@@ -52,7 +59,8 @@
"streaming": false, "streaming": false,
"cpu_threads": 3, "cpu_threads": 3,
"cpu_affinity_count": 6, "cpu_affinity_count": 6,
"process_priority": "below_normal" "process_priority": "below_normal",
"startup_timeout_seconds": 900
} }
}, },
"enable_danmu_reply": true, "enable_danmu_reply": true,
@@ -249,6 +257,46 @@
"active_operator" "active_operator"
] ]
}, },
"daily": {
"enabled": true,
"aliases": [
"自动每日",
"每日",
"日常",
"做每日",
"做日常"
],
"allowed_roles": [
"super_admin",
"active_operator"
]
},
"switch_party": {
"enabled": true,
"aliases": [
"切换队伍",
"更换队伍",
"换队伍",
"换队"
],
"allowed_roles": [
"super_admin",
"active_operator"
]
},
"edit_party": {
"enabled": true,
"aliases": [
"修改队员",
"更换队员",
"换队员",
"换角色"
],
"allowed_roles": [
"super_admin",
"active_operator"
]
},
"leave": { "leave": {
"enabled": true, "enabled": true,
"aliases": [ "aliases": [
+7
View File
@@ -0,0 +1,7 @@
{
"_说明": "键为 BetterGI 正式秘境名,值为可维护的俗称列表;修改后无需重启。",
"铭记之谷": [
"风本",
"少女套"
]
}
+194
View File
@@ -0,0 +1,194 @@
# 自动每日与队伍管理
本功能要求 BetterGI `0.63.0` 或更高版本,系统会通过以下命令行入口启动托管的一条龙配置:
```text
BetterGI.exe startOneDragon <配置名称>
```
## 弹幕指令
```text
自动每日
每日
日常
做每日
做日常
自动每日 秘境 <秘境正式名称或俗称>
自动每日 地脉 <经验|摩拉> <国家>
自动每日 委托
切换队伍 <队伍名称>
更换队伍 <队伍名称>
换队伍 <队伍名称>
换队 <队伍名称>
修改队员 <队员1> <队员2> <队员3> <队员4>
更换队员 <队员1> <队员2> <队员3> <队员4>
换队员 <队员1> <队员2> <队员3> <队员4>
换角色 <队员1> <队员2> <队员3> <队员4>
```
示例:
```text
自动每日
每日
日常 秘境 风本
自动每日 秘境 风本
自动每日 秘境 少女套
自动每日 地脉 经验 蒙德
自动每日 地脉 摩拉 枫丹
自动每日 委托
切换队伍 永冻队
换队 永冻队
修改队员 神里绫华 申鹤 枫原万叶 珊瑚宫心海
换角色 神里绫华 申鹤 枫原万叶 珊瑚宫心海
修改队员 神里绫华、申鹤、枫原万叶、珊瑚宫心海
```
队员支持空格、英文或中文逗号、顿号、斜杠分隔。连续输入四个角色名时,只有能够唯一拆分为四名已知角色才会执行。角色必须存在于 BGI 的 `AutoSwitchRoles` 角色数据中,且四人不能重复。新版脚本同时支持 `combat_avatar.json` 中维护的角色别名。
执行类指令可以省略指令后的空格,例如 `每日秘境 风本``换队永冻队``换角色芙宁娜纳西妲钟离雷电将军`。系统按最长别名优先解析,因此“换队员”不会被误识别为“换队”。
## 每日流程
每日流程固定为:
```text
领取邮件 > 合成树脂 > 可选其他任务 > 领取尘歌壶奖励 > 领取每日奖励
```
- 直接发送 `自动每日` 时不插入其他任务。
- 秘境模式一直执行到树脂耗尽或 BGI 报错。
- 地脉模式默认在执行前合成树脂,然后一直执行到树脂耗尽或 BGI 报错。
- 委托模式执行配置组“每日委托”。
- 委托启用“使用当前队伍”时,正式任务启动前会先读取游戏当前队伍预设名,并写入 AutoCommissionNova 的战斗与元素采集队伍配置;该准备步骤不进入正式计费任务。
- 其他任务报错时,由 BetterGI 一条龙继续执行后面的尘歌壶和每日奖励任务。
- 一条龙完成后复用现有配置组的计费、三分钟保留队列和任务结算规则。
`daily.ley_line_craft_resin_before` 设为 `false` 时,只有地脉模式会跳过“合成树脂”;其他每日模式仍会执行该步骤。
## 项目配置
`config/config.json`
```json
{
"daily": {
"one_dragon_template": "默认配置",
"managed_one_dragon_name": "直播系统自动每日",
"ley_line_craft_resin_before": true,
"commission_use_current_party": true,
"current_party_read_timeout_sec": 45
}
}
```
| 字段 | 说明 |
|------|------|
| `one_dragon_template` | BGI `User/OneDragon` 中作为来源的一条龙配置名,不含 `.json` |
| `managed_one_dragon_name` | 系统每次覆盖生成并通过命令行启动的配置名 |
| `ley_line_craft_resin_before` | 地脉模式执行前是否插入“合成树脂” |
| `commission_use_current_party` | 委托模式是否读取游戏当前队伍名并覆盖 AutoCommissionNova 的战斗与元素采集队伍 |
| `current_party_read_timeout_sec` | 读取当前队伍名称的超时时间,限制为 10 至 120 秒 |
托管配置每次执行都会从模板重新生成。模板中的合成地区、冒险家协会地区、战斗队伍、好感队和尘歌壶设置会保留,系统会强制把完成动作设为“无”,防止任务结束后退出游戏或关机。
秘境俗称维护在 `config/domain_aliases.json`
```json
{
"铭记之谷": ["风本", "少女套"]
}
```
键必须是当前 BGI `AutoDomain/settings.json` 中存在的正式秘境名。值是俗称列表。同一俗称不能指向多个秘境;文件在每次指令执行时读取,修改后无需重启本项目。
## BetterGI 配置
### 1. 一条龙模板
在 BetterGI 中创建一条龙配置,名称与 `daily.one_dragon_template` 相同,默认是“默认配置”。在模板中配置:
- 合成浓缩树脂使用的地区。
- 领取每日奖励使用的冒险家协会地区。
- 自动秘境和自动地脉需要的战斗队伍、好感队及其他树脂选项。
- 尘歌壶进入方式和奖励领取设置。
模板的任务勾选和顺序不会直接采用,系统会按固定流程重新生成托管配置。
### 2. 自动秘境
- 安装并配置 BGI 的 `AutoDomain` 脚本。
- 在 BGI 自动战斗配置中选择可用战斗策略,并确保策略文件存在于 `User/AutoFight`
- 系统会关闭一条龙周计划覆盖并写入解析后的正式秘境名。
- 系统会永久将 `User/config.json``autoDomainConfig.specifyResinUse` 修正为 `false`,以树脂耗尽为停止条件。
### 3. 自动地脉花
- 在 BGI 中配置自动地脉花使用的战斗策略和队伍。
- 系统会把地脉类型和国家写入星期一至星期日的全部字段。
- 系统会开启树脂耗尽模式,并关闭“运行次数取小值”。
- 系统会永久将 `autoLeyLineOutcropConfig.isGoToSynthesizer` 修正为 `false`。是否在地脉前合成树脂只由本项目的 `ley_line_craft_resin_before` 控制。
支持国家:蒙德、璃月、稻妻、须弥、枫丹、纳塔、挪德卡莱。
### 4. 每日委托配置组
在 BetterGI 的配置组页面创建“每日委托”,加入并启用实际完成每日委托所需的项目。系统不会自动创建此配置组,缺失或没有可执行项目时会拒绝启动自动每日,且不会停止当前任务。
使用 `AutoCommissionNova` 时,首次必须在 BetterGI 中手动运行并完成配置面板,至少保存用户配置和战斗策略。启用“使用当前队伍”后,战斗队伍与元素采集队伍都可以留空;未启用时,这两个队伍名称都必须保存。确认生成以下文件后,再关闭“启动时显示配置面板”:
```text
User/JsScript/AutoCommissionNova/Data/user-config.json
```
系统会在启动“自动每日 委托”前校验该文件。启用当前队伍后,项目会自动同步自维护脚本:
```text
integrations/LiveCurrentParty
```
系统自动生成并覆盖托管配置组“直播系统读取当前队伍”,通过请求 ID 对应的 `status.json` 获取游戏当前队伍预设名,然后只修改:
```text
AutoCommissionNova/Data/user-config.json
party.global.battleTeamName
party.global.elementTeamName
```
不会修改 `AutoCommissionNova` 的任何源码,因此第三方脚本更新不会覆盖本项目逻辑。两个字段使用同一次 OCR 得到的当前队伍预设名。
### 5. 切换队伍配置组
在 BetterGI 创建配置组“切换队伍”,加入并只启用一个文件夹名为 `AcceleratedEditionSwitchParty` 的 JavaScript 项目。系统会在执行前更新该项目的 `jsScriptSettingsObject.partyName`
### 6. 修改队员配置组
在 BetterGI 创建配置组“修改队员”,加入并只启用一个文件夹名为 `AutoSwitchRoles` 的 JavaScript 项目。系统会按顺序读取:
```text
User/JsScript/AutoSwitchRoles/settings.json
User/JsScript/AutoSwitchRoles/combat_avatar.json
```
旧版脚本会从 `settings.json` 的下拉选项读取角色;`AutoSwitchRoles 6.7.0+` 的位置参数是文本框,系统会改为读取 `combat_avatar.json` 中的正式名称和别名。校验后更新配置组中的 `position1``position4`
## 权限与计费
`daily``switch_party``edit_party` 默认权限与 `run` 相同:一级超管和已完成上号的二级队首。可以在后台“规则”页修改启停状态、别名和允许角色。
三个新功能都复用现有排队和计费规则。切换队伍与修改队员分别执行配置组“切换队伍”和“修改队员”,并按普通配置组的三分钟完成规则结算。
## 排障
- 提示找不到一条龙模板:检查 `User/OneDragon/<模板名>.json` 是否存在,且模板名与配置完全一致。
- 提示秘境未知:检查 BGI `AutoDomain/settings.json``config/domain_aliases.json` 的正式名称。
- 提示战斗策略不存在:在 BetterGI 中重新选择策略,并确认对应文件位于 `User/AutoFight`
- 提示配置组缺少 JavaScript 项目:检查配置组名称、项目的 `folderName` 和启用状态。
- 提示 AutoCommissionNova 未完成首次配置:在 BGI 中打开配置面板并保存用户配置与战斗策略;关闭“使用当前队伍”时还需保存战斗队伍和元素采集队伍。
- 提示读取当前队伍失败:确认当前场景允许打开配队界面;若 OCR 返回多个候选,查看 BetterGI 日志中的“当前队伍名称候选”记录。
- 提示角色未知或有歧义:检查 `AutoSwitchRoles/settings.json``combat_avatar.json`,使用其中唯一对应的名称或别名,并用空格明确分隔四人。
- BGI 启动失败:确认版本为 `0.63.0+``bettergi.exe_path``bettergi.work_dir` 正确,并查看 BetterGI 日志。
- 一条龙执行中子任务报错:查看 BetterGI 日志确认后续“领取尘歌壶奖励”和“领取每日奖励”是否继续;本项目只在整条一条龙完成标记出现后结算。
- 自动秘境异常后扫码上号卡在“退出秘境”:确认项目维护的 `integrations/扫码上号` 已同步。脚本会 OCR 识别“退出秘境”并点击右侧确认,OCR 按钮定位失败时使用 1920×1080 坐标 `(1164, 757)` 兜底。
+18 -1
View File
@@ -15,6 +15,9 @@ const COMMAND_ROLE_DEFAULTS = {
confirm_yes: ["super_admin", "pending_operator"], confirm_yes: ["super_admin", "pending_operator"],
confirm_no: ["super_admin", "pending_operator"], confirm_no: ["super_admin", "pending_operator"],
run: ["super_admin", "active_operator"], run: ["super_admin", "active_operator"],
daily: ["super_admin", "active_operator"],
switch_party: ["super_admin", "active_operator"],
edit_party: ["super_admin", "active_operator"],
leave: ["super_admin", "active_operator", "viewer"], leave: ["super_admin", "active_operator", "viewer"],
reset: ["super_admin"], reset: ["super_admin"],
points: ROLE_OPTIONS.map(([id]) => id), points: ROLE_OPTIONS.map(([id]) => id),
@@ -29,6 +32,9 @@ const CMD_LABELS = {
confirm_yes: "确认是", confirm_yes: "确认是",
confirm_no: "确认不是", confirm_no: "确认不是",
run: "执行配置组", run: "执行配置组",
daily: "自动每日",
switch_party: "切换队伍",
edit_party: "修改队员",
leave: "退出", leave: "退出",
reset: "重置", reset: "重置",
points: "积分", points: "积分",
@@ -36,7 +42,7 @@ const CMD_LABELS = {
help: "帮助" help: "帮助"
}; };
const CMD_HAS_ARG = { run: true }; const CMD_HAS_ARG = { run: true, daily: true, switch_party: true, edit_party: true };
const clone = (value) => JSON.parse(JSON.stringify(value ?? {})); const clone = (value) => JSON.parse(JSON.stringify(value ?? {}));
const splitList = (value) => String(value || "").replaceAll("", ",").split(",").map((item) => item.trim()).filter(Boolean); const splitList = (value) => String(value || "").replaceAll("", ",").split(",").map((item) => item.trim()).filter(Boolean);
@@ -82,6 +88,12 @@ const ensureDraftShape = (draft) => {
draft.bilibili.cookie_auto_refresh_enabled ??= true; draft.bilibili.cookie_auto_refresh_enabled ??= true;
draft.bilibili.cookie_check_interval_hours ??= 6; draft.bilibili.cookie_check_interval_hours ??= 6;
draft.bettergi ||= {}; draft.bettergi ||= {};
draft.daily ||= {};
draft.daily.one_dragon_template ||= "默认配置";
draft.daily.managed_one_dragon_name ||= "直播系统自动每日";
draft.daily.ley_line_craft_resin_before ??= true;
draft.daily.commission_use_current_party ??= true;
draft.daily.current_party_read_timeout_sec ??= 45;
draft.global ||= {}; draft.global ||= {};
draft.queue ||= {}; draft.queue ||= {};
draft.frontend ||= {}; draft.frontend ||= {};
@@ -1261,6 +1273,11 @@ createApp({
<h2>BetterGI</h2> <h2>BetterGI</h2>
<label>BetterGI.exe<input v-model="draft.bettergi.exe_path"></label> <label>BetterGI.exe<input v-model="draft.bettergi.exe_path"></label>
<label>工作目录<input v-model="draft.bettergi.work_dir"></label> <label>工作目录<input v-model="draft.bettergi.work_dir"></label>
<label>每日一条龙模板<input v-model="draft.daily.one_dragon_template"></label>
<label>托管一条龙名称<input v-model="draft.daily.managed_one_dragon_name"></label>
<label class="inline"><input v-model="draft.daily.ley_line_craft_resin_before" type="checkbox"> 地脉前合成树脂</label>
<label class="inline"><input v-model="draft.daily.commission_use_current_party" type="checkbox"> 委托战斗及采集均使用游戏当前队伍</label>
<label>读取当前队伍超时秒数<input v-model.number="draft.daily.current_party_read_timeout_sec" type="number" min="10" max="120"></label>
<label>超管 UID<input v-model="draft.global.admin_uidsText" @blur="draft.global.admin_uids = splitList(draft.global.admin_uidsText).map((item) => Number(item)).filter((item) => Number.isFinite(item))" placeholder="逗号分隔 UID"></label> <label>超管 UID<input v-model="draft.global.admin_uidsText" @blur="draft.global.admin_uids = splitList(draft.global.admin_uidsText).map((item) => Number(item)).filter((item) => Number.isFinite(item))" placeholder="逗号分隔 UID"></label>
</div> </div>
<div class="panel"> <div class="panel">
+154
View File
@@ -0,0 +1,154 @@
(async function () {
setGameMetrics(1920, 1080, 1);
const STATUS_FILE = "status.json";
const requestId = String(settings.requestId || "").trim();
let lastCandidates = [];
function writeStatus(state, data) {
const payload = Object.assign({
state: state,
request_id: requestId,
party_name: "",
candidates: lastCandidates,
message: "",
updated_at: new Date().toISOString()
}, data || {});
file.writeTextSync(STATUS_FILE, JSON.stringify(payload, null, 2));
}
function normalizeText(value) {
return String(value || "").replace(/\s+/g, "").trim();
}
function isPartyNameCandidate(text) {
if (!text || text.length > 20) {
return false;
}
const ignored = [
"队伍配置",
"快速编队",
"元素共鸣",
"调整队伍",
"当前队伍",
"部署",
"出战",
"详情"
];
if (ignored.some((label) => text === label || text.includes(label))) {
return false;
}
return !/^(esc|enter|space|l|f\d{1,2})$/i.test(text);
}
function isPartyPageOpen() {
const capture = captureGameRegion();
try {
const results = capture.findMulti(RecognitionObject.ocr(0, 0, 720, 180));
for (let index = 0; index < results.count; index++) {
const text = normalizeText(results[index].text);
if (text.includes("队伍配置")) {
return true;
}
}
return false;
} finally {
capture.dispose();
}
}
function recognizePartyNameCandidates() {
const capture = captureGameRegion();
try {
const results = capture.findMulti(RecognitionObject.ocr(0, 940, 520, 140));
const candidates = [];
for (let index = 0; index < results.count; index++) {
const result = results[index];
const text = normalizeText(result.text);
log.info(
"当前队伍名称候选位置:({x},{y},{w},{h}), 识别结果:{text}",
result.x,
result.y,
result.Width,
result.Height,
text
);
if (isPartyNameCandidate(text)) {
candidates.push({
text: text,
x: Number(result.x || 0),
y: Number(result.y || 0)
});
}
}
candidates.sort((left, right) => {
if (left.x !== right.x) {
return left.x - right.x;
}
return right.y - left.y;
});
return candidates.filter(
(candidate, index, values) =>
values.findIndex((value) => value.text === candidate.text) === index
);
} finally {
capture.dispose();
}
}
async function openPartyPageAndRead() {
for (let attempt = 0; attempt < 3; attempt++) {
keyPress("VK_L");
await sleep(2200);
for (let scan = 0; scan < 4; scan++) {
if (isPartyPageOpen()) {
const candidates = recognizePartyNameCandidates();
if (candidates.length > 0) {
return candidates;
}
}
await sleep(600);
}
keyPress("VK_ESCAPE");
await sleep(800);
await genshin.returnMainUi();
}
return [];
}
try {
if (!requestId) {
throw new Error("读取请求标识为空");
}
writeStatus("running");
await genshin.returnMainUi();
const candidates = await openPartyPageAndRead();
lastCandidates = candidates.map((candidate) => candidate.text);
if (candidates.length === 0) {
throw new Error("未识别到当前队伍名称,请确认游戏处于可打开配队界面的状态");
}
if (candidates.length > 1) {
throw new Error("识别到多个队伍名称候选:" + lastCandidates.join("、"));
}
const partyName = candidates[0].text;
await genshin.returnMainUi();
await sleep(500);
writeStatus("success", {
party_name: partyName,
candidates: lastCandidates
});
log.info("当前队伍名称读取成功: {name}", partyName);
} catch (error) {
try {
await genshin.returnMainUi();
} catch (_) {
}
const message = String(error && error.message ? error.message : error);
writeStatus("error", {
candidates: lastCandidates,
message: message
});
log.error("当前队伍名称读取失败: {message}", message);
}
})();
@@ -0,0 +1,17 @@
{
"manifest_version": 1,
"name": "直播系统读取当前队伍",
"version": "1.0.0",
"bgi_version": "0.63.0",
"description": "读取游戏当前队伍预设名称并通过状态文件返回给直播联动系统",
"authors": [
{
"name": "BGI直播项目"
}
],
"settings_ui": "settings.json",
"main": "main.js",
"saved_files": [
"status.json"
]
}
@@ -0,0 +1,8 @@
[
{
"name": "requestId",
"type": "input-text",
"label": "读取请求标识",
"default": ""
}
]
+77 -1
View File
@@ -1,6 +1,6 @@
(async function () { (async function () {
// ======================================== // ========================================
// 扫码上号 v3 // 扫码上号 v3.1
// 流程: 状态判定 → (已登录则先退出) → A0检测tap → A1选号 → A2等登录+点中心 → A3完成 // 流程: 状态判定 → (已登录则先退出) → A0检测tap → A1选号 → A2等登录+点中心 → A3完成
// 状态文件: status.txt (登录中 / 已登录) // 状态文件: status.txt (登录中 / 已登录)
// ======================================== // ========================================
@@ -20,6 +20,25 @@
// 脚本启动 → 写入"登录中" // 脚本启动 → 写入"登录中"
writeStatus("登录中"); writeStatus("登录中");
function settingNumber(name, fallback) {
const value = typeof settings !== "undefined" && settings ? Number(settings[name]) : NaN;
return Number.isFinite(value) ? value : fallback;
}
function settingBoolean(name, fallback) {
if (typeof settings === "undefined" || !settings || settings[name] === undefined) return fallback;
return settings[name] !== false && String(settings[name]).toLowerCase() !== "false";
}
const useExitDomainOcr = settingBoolean("useOcr", true);
let confirmExitX = settingNumber("confirmExitX", 1164);
let confirmExitY = settingNumber("confirmExitY", 757);
if (confirmExitX === 830 && confirmExitY === 600) {
confirmExitX = 1164;
confirmExitY = 757;
log.info("检测到旧版退出秘境坐标,已自动修正为 (1164, 757)");
}
// ---------- 加载图像资源 ---------- // ---------- 加载图像资源 ----------
const tapMat = file.readImageMatSync("assets/tap.png"); const tapMat = file.readImageMatSync("assets/tap.png");
const a0PhoneMat = file.readImageMatSync("assets/a0_phone.png"); const a0PhoneMat = file.readImageMatSync("assets/a0_phone.png");
@@ -115,6 +134,54 @@
return null; return null;
} }
function normalizeOcrText(text) {
return String(text || "").replace(/\s+/g, "");
}
function readOcrResults(x, y, w, h) {
const cap = captureGameRegion();
try {
const results = cap.findMulti(RecognitionObject.ocr(x, y, w, h));
return Array.from(results).map((result) => ({
text: normalizeOcrText(result.text),
x: Number(result.x || 0),
y: Number(result.y || 0),
w: Number(result.width || result.Width || 0),
h: Number(result.height || result.Height || 0)
}));
} catch (e) {
log.warn("退出秘境弹窗 OCR 失败: " + e);
return [];
} finally {
cap.dispose();
}
}
async function handleExitDomainDialog() {
const results = readOcrResults(480, 220, 960, 620);
const combinedText = results.map((result) => result.text).join("");
if (!combinedText.includes("退出秘境")) return false;
log.info("检测到退出秘境确认弹窗");
const confirmResult = results
.filter((result) => result.text.includes("确认") && result.x >= 900)
.sort((left, right) => right.x - left.x)[0];
if (useExitDomainOcr && confirmResult) {
const x = Math.round(confirmResult.x + confirmResult.w / 2);
const y = Math.round(confirmResult.y + confirmResult.h / 2);
log.info(`OCR 定位确认按钮,点击 (${x}, ${y})`);
click(x, y);
} else {
const reason = useExitDomainOcr ? "OCR 未定位到确认按钮" : "已关闭 OCR 按钮定位";
log.warn(`${reason},使用固定坐标 (${confirmExitX}, ${confirmExitY})`);
click(confirmExitX, confirmExitY);
}
log.info("已点击退出秘境确认,等待返回主界面...");
await sleep(6000);
return true;
}
// ======================================== // ========================================
// 前置处理:检测登录前的提示图标 // 前置处理:检测登录前的提示图标
// 检测到则点击 (1830, 985)1秒后点击 (1100, 675),再1秒后进入初始判定; // 检测到则点击 (1830, 985)1秒后点击 (1100, 675),再1秒后进入初始判定;
@@ -155,11 +222,20 @@
log.info("未同时检测到 tap 图标和手机图标 → 开始检测派蒙头像"); log.info("未同时检测到 tap 图标和手机图标 → 开始检测派蒙头像");
let paimonFound = findImageMatchRobust(paimonMat); let paimonFound = findImageMatchRobust(paimonMat);
if (!paimonFound && await handleExitDomainDialog()) {
paimonFound = findImageMatchRobust(paimonMat);
if (paimonFound) log.info("退出秘境后已检测到派蒙头像");
}
for (let i = 0; !paimonFound && i < 8; i++) { for (let i = 0; !paimonFound && i < 8; i++) {
log.info(`${i + 1}/8 次未检测到派蒙头像,按 ESC 后重试...`); log.info(`${i + 1}/8 次未检测到派蒙头像,按 ESC 后重试...`);
keyPress("Escape"); keyPress("Escape");
await sleep(1000); await sleep(1000);
if (await handleExitDomainDialog()) {
log.info("退出秘境弹窗已处理,重新检测主界面");
}
paimonFound = findImageMatchRobust(paimonMat); paimonFound = findImageMatchRobust(paimonMat);
if (paimonFound) log.info("已检测到派蒙头像,继续退出账号流程");
} }
if (!paimonFound) { if (!paimonFound) {
+1 -1
View File
@@ -1,7 +1,7 @@
{ {
"manifest_version": 1, "manifest_version": 1,
"name": "扫码上号", "name": "扫码上号",
"version": "1.0", "version": "1.1",
"bgi_version": "0.48.0", "bgi_version": "0.48.0",
"description": "直播用扫码上号脚本:判断在大世界则退出到开门页面→点击扫码登录→等待扫码→检测验证码", "description": "直播用扫码上号脚本:判断在大世界则退出到开门页面→点击扫码登录→等待扫码→检测验证码",
"authors": [ "authors": [
+5 -5
View File
@@ -44,7 +44,7 @@
{ {
"name": "useOcr", "name": "useOcr",
"type": "checkbox", "type": "checkbox",
"label": "使用OCR识别文本(否则用固定坐标)", "label": "退出秘境确认按钮使用 OCR 定位(关闭后使用固定坐标",
"default": true "default": true
}, },
{ {
@@ -53,14 +53,14 @@
{ {
"name": "confirmExitX", "name": "confirmExitX",
"type": "input-text", "type": "input-text",
"label": "确认退出按钮 X(固定坐标模式)", "label": "退出秘境确认按钮 X(1920×1080 固定坐标",
"default": "830" "default": "1164"
}, },
{ {
"name": "confirmExitY", "name": "confirmExitY",
"type": "input-text", "type": "input-text",
"label": "确认退出按钮 Y(固定坐标模式)", "label": "退出秘境确认按钮 Y(1920×1080 固定坐标",
"default": "600" "default": "757"
}, },
{ {
"name": "qrLoginX", "name": "qrLoginX",
+6 -1
View File
@@ -4,7 +4,12 @@ cd /d "%~dp0"
set PYTHON=%~dp0.venv-tts\Scripts\python.exe set PYTHON=%~dp0.venv-tts\Scripts\python.exe
if not exist "%PYTHON%" ( if not exist "%PYTHON%" (
echo Python virtual environment not found: %PYTHON% set PYTHON=%~dp0.venv\Scripts\python.exe
)
if not exist "%PYTHON%" (
echo Python virtual environment not found.
echo Checked .venv-tts and .venv under: %~dp0
echo Please run setup first or ask WorkBuddy to repair dependencies. echo Please run setup first or ask WorkBuddy to repair dependencies.
pause pause
exit /b 1 exit /b 1
+4 -1
View File
@@ -32,6 +32,8 @@ function Invoke-Checked([string]$File, [string[]]$Arguments) {
function Get-ProjectPython { function Get-ProjectPython {
$candidates = @( $candidates = @(
$env:LIVE_STREAMING_PYTHON, $env:LIVE_STREAMING_PYTHON,
(Join-Path $Root ".venv-tts\Scripts\python.exe"),
(Join-Path $Root ".venv\Scripts\python.exe"),
"E:\Programs\Anaconda3\envs\Live-streaming\python.exe", "E:\Programs\Anaconda3\envs\Live-streaming\python.exe",
"python" "python"
) | Where-Object { $_ } ) | Where-Object { $_ }
@@ -45,7 +47,7 @@ function Get-ProjectPython {
} catch { } catch {
} }
} }
throw "No usable Python found. Install Python 3.10+, set LIVE_STREAMING_PYTHON, or use the Live-streaming conda env." throw "No usable Python found. Create .venv, install Python 3.10+, or set LIVE_STREAMING_PYTHON."
} }
$Python = Get-ProjectPython $Python = Get-ProjectPython
@@ -197,6 +199,7 @@ New-Item -ItemType Directory -Force -Path (Join-Path $AppDist "logs") | Out-Null
Copy-Item -LiteralPath (Join-Path $Root "web") -Destination (Join-Path $AppDist "web") -Recurse -Force Copy-Item -LiteralPath (Join-Path $Root "web") -Destination (Join-Path $AppDist "web") -Recurse -Force
Copy-Item -LiteralPath (Join-Path $Root "config") -Destination (Join-Path $AppDist "config") -Recurse -Force Copy-Item -LiteralPath (Join-Path $Root "config") -Destination (Join-Path $AppDist "config") -Recurse -Force
Copy-Item -LiteralPath (Join-Path $Root "integrations") -Destination (Join-Path $AppDist "integrations") -Recurse -Force Copy-Item -LiteralPath (Join-Path $Root "integrations") -Destination (Join-Path $AppDist "integrations") -Recurse -Force
Copy-Item -LiteralPath (Join-Path $Root "docs") -Destination (Join-Path $AppDist "docs") -Recurse -Force
Copy-Item -LiteralPath (Join-Path $Root "vendor") -Destination (Join-Path $AppDist "vendor") -Recurse -Force Copy-Item -LiteralPath (Join-Path $Root "vendor") -Destination (Join-Path $AppDist "vendor") -Recurse -Force
Copy-Item -LiteralPath (Join-Path $Root "README.md") -Destination (Join-Path $AppDist "README.md") -Force Copy-Item -LiteralPath (Join-Path $Root "README.md") -Destination (Join-Path $AppDist "README.md") -Force
+102
View File
@@ -0,0 +1,102 @@
import json
import tempfile
import unittest
from pathlib import Path
from app.bettergi_current_party import (
CURRENT_PARTY_SCRIPT_NAME,
CurrentPartyReadError,
clear_current_party_status,
prepare_current_party_read,
read_current_party_status,
)
class BetterGICurrentPartyTests(unittest.TestCase):
def setUp(self):
self.temp_dir = tempfile.TemporaryDirectory()
self.root = Path(self.temp_dir.name)
self.group_dir = self.root / "User" / "ScriptGroup"
self._write_json(
self.group_dir / "切换队伍.json",
{
"index": 2,
"name": "切换队伍",
"config": {"marker": "preserved"},
"projects": [{"name": "旧项目"}],
},
)
self._write_json(
self.group_dir / "其他.json",
{"index": 5, "name": "其他", "config": {}, "projects": []},
)
def tearDown(self):
self.temp_dir.cleanup()
@staticmethod
def _write_json(path: Path, data):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
def test_prepare_reader_group_clones_config_and_uses_request_id(self):
prepared = prepare_current_party_read(self.root, "request-123")
update = prepared.updates[0]
self.assertEqual(update.data["index"], 6)
self.assertEqual(update.data["config"]["marker"], "preserved")
self.assertEqual(update.data["name"], "直播系统读取当前队伍")
project = update.data["projects"][0]
self.assertEqual(project["folderName"], CURRENT_PARTY_SCRIPT_NAME)
self.assertEqual(project["jsScriptSettingsObject"]["requestId"], "request-123")
self.assertEqual(
prepared.status_path,
self.root / "User" / "JsScript" / CURRENT_PARTY_SCRIPT_NAME / "status.json",
)
def test_status_reader_ignores_stale_and_running_results(self):
prepared = prepare_current_party_read(self.root, "request-123")
self._write_json(
prepared.status_path,
{"state": "success", "request_id": "old", "party_name": "旧队伍"},
)
self.assertIsNone(read_current_party_status(prepared.status_path, "request-123"))
self._write_json(
prepared.status_path,
{"state": "running", "request_id": "request-123"},
)
self.assertIsNone(read_current_party_status(prepared.status_path, "request-123"))
self._write_json(
prepared.status_path,
{
"state": "success",
"request_id": "request-123",
"party_name": "好感队",
"candidates": ["好感队"],
},
)
result = read_current_party_status(prepared.status_path, "request-123")
self.assertEqual(result.party_name, "好感队")
self.assertEqual(result.candidates, ("好感队",))
def test_status_reader_reports_script_error_and_can_be_cleared(self):
prepared = prepare_current_party_read(self.root, "request-123")
self._write_json(
prepared.status_path,
{
"state": "error",
"request_id": "request-123",
"message": "识别到多个候选",
},
)
with self.assertRaisesRegex(CurrentPartyReadError, "多个候选"):
read_current_party_status(prepared.status_path, "request-123")
clear_current_party_status(prepared.status_path)
self.assertFalse(prepared.status_path.exists())
if __name__ == "__main__":
unittest.main()
+503
View File
@@ -0,0 +1,503 @@
import json
import tempfile
import unittest
from pathlib import Path
from app.bettergi_daily import (
DAILY_MODE_COMMISSION,
DAILY_MODE_DOMAIN,
DAILY_MODE_LEY_LINE,
DAILY_MODE_NONE,
DailyAutomationError,
DomainAliasResolver,
apply_json_updates,
parse_daily_request,
prepare_commission_current_party_update,
prepare_daily_run,
prepare_edit_party_update,
prepare_switch_party_update,
resolve_party_members,
)
class BetterGIDailyTestCase(unittest.TestCase):
def setUp(self):
self.temp_dir = tempfile.TemporaryDirectory()
self.root = Path(self.temp_dir.name)
self.alias_path = self.root / "domain_aliases.json"
self._write_json(
self.alias_path,
{
"_说明": "test",
"铭记之谷": ["风本", "少女套"],
},
)
self._write_json(
self.root / "User" / "OneDragon" / "默认配置.json",
{
"Name": "默认配置",
"TaskEnabledList": {
"mail-template-id": True,
"serenitea-template-id": True,
"craft-template-id": True,
"domain-template-id": False,
"ley-line-template-id": False,
"daily-reward-template-id": True,
},
"TaskOrder": [
"mail-template-id",
"serenitea-template-id",
"craft-template-id",
"domain-template-id",
"ley-line-template-id",
"daily-reward-template-id",
],
"TaskDefinitions": {
"mail-template-id": "领取邮件",
"serenitea-template-id": "领取尘歌壶奖励",
"craft-template-id": "合成树脂",
"domain-template-id": "自动秘境",
"ley-line-template-id": "自动地脉花",
"daily-reward-template-id": "领取每日奖励",
},
"CraftingBenchCountry": "枫丹",
"AdventurersGuildCountry": "璃月",
"PartyName": "战斗队",
"DailyRewardPartyName": "好感队",
"SecretTreasureObjects": ["须臾树脂"],
"WeeklyDomainEnabled": True,
"DomainName": "旧秘境",
"CompletionAction": "关机",
},
)
self._write_json(
self.root / "User" / "config.json",
{
"marker": "preserved",
"autoFightConfig": {"strategyName": "策略A"},
"autoDomainConfig": {
"specifyResinUse": True,
"other": 1,
},
"autoLeyLineOutcropConfig": {
"isGoToSynthesizer": True,
"team": "战斗队",
"friendshipTeam": "",
"fightConfig": {"strategyName": "策略A"},
},
},
)
strategy = self.root / "User" / "AutoFight" / "策略A.txt"
strategy.parent.mkdir(parents=True, exist_ok=True)
strategy.write_text("战斗策略", encoding="utf-8")
self._write_json(
self.root / "User" / "JsScript" / "AutoDomain" / "settings.json",
[
{
"name": "domainName",
"type": "select",
"options": ["铭记之谷", "仲夏庭园"],
}
],
)
self._write_json(
self.root / "User" / "ScriptGroup" / "每日委托.json",
{"name": "每日委托", "projects": [{"name": "委托", "status": "Enabled"}]},
)
self._write_json(
self.root / "User" / "ScriptGroup" / "切换队伍.json",
{
"name": "切换队伍",
"projects": [
{
"folderName": "AcceleratedEditionSwitchParty",
"status": "Enabled",
"jsScriptSettingsObject": {"partyName": "旧队伍", "debug": True},
}
],
},
)
self._write_json(
self.root / "User" / "ScriptGroup" / "修改队员.json",
{
"name": "修改队员",
"projects": [
{
"folderName": "AutoSwitchRoles",
"status": "Enabled",
"jsScriptSettingsObject": {"position1": ""},
}
],
},
)
options = [
"水-单手剑-芙宁娜",
"冰-长枪-爱可菲",
"草-法器-纳西妲",
"雷-长枪-雷电将军",
"岩-长枪-钟离",
]
self._write_json(
self.root / "User" / "JsScript" / "AutoSwitchRoles" / "settings.json",
[
{"name": f"position{index}", "type": "select", "options": ["", *options]}
for index in range(1, 5)
],
)
def tearDown(self):
self.temp_dir.cleanup()
@staticmethod
def _write_json(path: Path, data):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
def _prepare(
self,
argument: str,
*,
craft_before: bool = True,
commission_use_current_party: bool = False,
):
return prepare_daily_run(
self.root,
self.alias_path,
argument,
template_name="默认配置",
managed_name="直播系统自动每日",
ley_line_craft_resin_before=craft_before,
commission_use_current_party=commission_use_current_party,
)
def test_parse_daily_modes(self):
self.assertEqual(parse_daily_request("").mode, DAILY_MODE_NONE)
domain = parse_daily_request("秘境 风本")
self.assertEqual((domain.mode, domain.domain_name), (DAILY_MODE_DOMAIN, "风本"))
ley_line = parse_daily_request("地脉 经验 蒙德")
self.assertEqual(
(ley_line.mode, ley_line.ley_line_type, ley_line.ley_line_country),
(DAILY_MODE_LEY_LINE, "启示之花", "蒙德"),
)
self.assertEqual(parse_daily_request("委托").mode, DAILY_MODE_COMMISSION)
def test_parse_rejects_missing_or_unknown_arguments(self):
with self.assertRaisesRegex(DailyAutomationError, "需要指定秘境"):
parse_daily_request("秘境")
with self.assertRaisesRegex(DailyAutomationError, "地脉模式格式"):
parse_daily_request("地脉 经验")
with self.assertRaisesRegex(DailyAutomationError, "不支持的地脉国家"):
parse_daily_request("地脉 摩拉 天空岛")
with self.assertRaisesRegex(DailyAutomationError, "仅支持"):
parse_daily_request("探索")
def test_domain_alias_resolves_canonical_and_common_names(self):
resolver = DomainAliasResolver(self.alias_path, self.root)
self.assertEqual(resolver.resolve("风本"), "铭记之谷")
self.assertEqual(resolver.resolve("少女套"), "铭记之谷")
self.assertEqual(resolver.resolve("仲夏庭园"), "仲夏庭园")
def test_domain_alias_collision_is_rejected(self):
self._write_json(
self.alias_path,
{
"铭记之谷": ["风本"],
"仲夏庭园": ["风 本"],
},
)
with self.assertRaisesRegex(DailyAutomationError, "同时指向"):
DomainAliasResolver(self.alias_path, self.root).resolve("风本")
def test_domain_run_generates_exact_flow_and_corrects_bgi_config(self):
prepared = self._prepare("秘境 风本")
self.assertEqual(prepared.request.domain_name, "铭记之谷")
self.assertEqual(prepared.task_name, "自动每日(秘境:铭记之谷)")
updates = {update.path.name: update for update in prepared.updates}
corrected = updates["config.json"].data
self.assertFalse(corrected["autoDomainConfig"]["specifyResinUse"])
self.assertEqual(corrected["autoDomainConfig"]["other"], 1)
self.assertEqual(corrected["marker"], "preserved")
managed = updates["直播系统自动每日.json"].data
names = [managed["TaskDefinitions"][task_id] for task_id in managed["TaskOrder"]]
self.assertEqual(
names,
["领取邮件", "合成树脂", "自动秘境", "领取尘歌壶奖励", "领取每日奖励"],
)
self.assertEqual(
managed["TaskOrder"],
[
"mail-template-id",
"craft-template-id",
"domain-template-id",
"serenitea-template-id",
"daily-reward-template-id",
],
)
self.assertFalse(managed["WeeklyDomainEnabled"])
self.assertEqual(managed["DomainName"], "铭记之谷")
self.assertEqual(managed["CompletionAction"], "")
self.assertEqual(managed["PartyName"], "战斗队")
def test_no_mode_skips_only_the_other_task(self):
prepared = self._prepare("")
managed = prepared.updates[-1].data
names = [managed["TaskDefinitions"][task_id] for task_id in managed["TaskOrder"]]
self.assertEqual(
names,
["领取邮件", "合成树脂", "领取尘歌壶奖励", "领取每日奖励"],
)
def test_ley_line_run_sets_all_days_and_respects_craft_toggle(self):
prepared = self._prepare("地脉 摩拉 枫丹", craft_before=False)
updates = {update.path.name: update for update in prepared.updates}
corrected = updates["config.json"].data
self.assertFalse(corrected["autoLeyLineOutcropConfig"]["isGoToSynthesizer"])
managed = updates["直播系统自动每日.json"].data
names = [managed["TaskDefinitions"][task_id] for task_id in managed["TaskOrder"]]
self.assertEqual(
names,
["领取邮件", "自动地脉花", "领取尘歌壶奖励", "领取每日奖励"],
)
self.assertTrue(managed["LeyLineResinExhaustionMode"])
self.assertFalse(managed["LeyLineOpenModeCountMin"])
for day in ("Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"):
self.assertTrue(managed[f"LeyLineRun{day}"])
self.assertEqual(managed[f"LeyLine{day}Type"], "藏金之花")
self.assertEqual(managed[f"LeyLine{day}Country"], "枫丹")
def test_commission_is_added_as_a_configuration_group_task(self):
prepared = self._prepare("委托")
managed = prepared.updates[-1].data
names = [managed["TaskDefinitions"][task_id] for task_id in managed["TaskOrder"]]
self.assertEqual(
names,
["领取邮件", "合成树脂", "每日委托", "领取尘歌壶奖励", "领取每日奖励"],
)
commission_id = next(
task_id
for task_id, name in managed["TaskDefinitions"].items()
if name == "每日委托"
)
self.assertNotIn(
commission_id,
{
"mail-template-id",
"craft-template-id",
"serenitea-template-id",
"daily-reward-template-id",
},
)
def test_commission_requires_nonempty_group(self):
(self.root / "User" / "ScriptGroup" / "每日委托.json").unlink()
with self.assertRaisesRegex(DailyAutomationError, "每日委托配置组"):
self._prepare("委托")
def test_auto_commission_nova_requires_complete_first_run_config(self):
self._write_json(
self.root / "User" / "ScriptGroup" / "每日委托.json",
{
"name": "每日委托",
"projects": [
{
"folderName": "AutoCommissionNova",
"status": "Enabled",
}
],
},
)
with self.assertRaisesRegex(DailyAutomationError, "缺少 Data/user-config.json"):
self._prepare("委托")
user_config_path = (
self.root
/ "User"
/ "JsScript"
/ "AutoCommissionNova"
/ "Data"
/ "user-config.json"
)
self._write_json(
user_config_path,
{
"party": {
"global": {
"battleTeamName": "战斗队",
"elementTeamName": "",
"battleStrategy": "策略A",
}
}
},
)
with self.assertRaisesRegex(DailyAutomationError, "元素采集队伍"):
self._prepare("委托")
self._write_json(
user_config_path,
{
"party": {
"global": {
"battleTeamName": "战斗队",
"elementTeamName": "采集队",
"battleStrategy": "策略A",
}
}
},
)
self.assertEqual(self._prepare("委托").request.mode, DAILY_MODE_COMMISSION)
def test_auto_commission_nova_can_inject_current_party_for_battle_and_collection(self):
self._write_json(
self.root / "User" / "ScriptGroup" / "每日委托.json",
{
"name": "每日委托",
"projects": [
{
"folderName": "AutoCommissionNova",
"status": "Enabled",
}
],
},
)
user_config_path = (
self.root
/ "User"
/ "JsScript"
/ "AutoCommissionNova"
/ "Data"
/ "user-config.json"
)
self._write_json(
user_config_path,
{
"marker": "preserved",
"party": {
"global": {
"battleTeamName": "",
"elementTeamName": "",
"battleStrategy": "策略A",
}
},
},
)
prepared = self._prepare("委托", commission_use_current_party=True)
self.assertTrue(prepared.requires_current_party)
update = prepare_commission_current_party_update(self.root, "当前队伍")
self.assertEqual(
update.data["party"]["global"]["battleTeamName"],
"当前队伍",
)
self.assertEqual(
update.data["party"]["global"]["elementTeamName"],
"当前队伍",
)
self.assertEqual(update.data["marker"], "preserved")
def test_switch_party_updates_only_target_setting(self):
update = prepare_switch_party_update(self.root, "深渊队")
project = update.data["projects"][0]
self.assertEqual(project["jsScriptSettingsObject"]["partyName"], "深渊队")
self.assertTrue(project["jsScriptSettingsObject"]["debug"])
def test_edit_party_accepts_separators_and_contiguous_names(self):
separated, display = resolve_party_members(
self.root,
"芙宁娜 爱可菲 纳西妲 雷电将军",
)
self.assertEqual(display, ["芙宁娜", "爱可菲", "纳西妲", "雷电将军"])
contiguous, contiguous_display = resolve_party_members(
self.root,
"芙宁娜爱可菲纳西妲雷电将军",
)
self.assertEqual(contiguous, separated)
self.assertEqual(contiguous_display, display)
update, names = prepare_edit_party_update(
self.root,
"芙宁娜、爱可菲、纳西妲、雷电将军",
)
settings = update.data["projects"][0]["jsScriptSettingsObject"]
self.assertEqual(names, ("芙宁娜", "爱可菲", "纳西妲", "雷电将军"))
self.assertEqual(settings["position4"], "雷-长枪-雷电将军")
def test_edit_party_supports_auto_switch_roles_67_character_data(self):
self._write_json(
self.root / "User" / "JsScript" / "AutoSwitchRoles" / "settings.json",
[
{"name": f"position{index}", "type": "input-text", "default": ""}
for index in range(1, 5)
],
)
self._write_json(
self.root / "User" / "JsScript" / "AutoSwitchRoles" / "combat_avatar.json",
[
{"name": "神里绫华", "alias": ["绫华"]},
{"name": "申鹤", "alias": []},
{"name": "枫原万叶", "alias": ["万叶"]},
{"name": "珊瑚宫心海", "alias": ["心海"]},
],
)
update, names = prepare_edit_party_update(
self.root,
"绫华 申鹤 万叶 心海",
)
settings = update.data["projects"][0]["jsScriptSettingsObject"]
self.assertEqual(names, ("神里绫华", "申鹤", "枫原万叶", "珊瑚宫心海"))
self.assertEqual(
[settings[f"position{index}"] for index in range(1, 5)],
["神里绫华", "申鹤", "枫原万叶", "珊瑚宫心海"],
)
resolved, display = resolve_party_members(
self.root,
"神里绫华申鹤枫原万叶珊瑚宫心海",
)
self.assertEqual(resolved, list(display))
def test_edit_party_rejects_ambiguous_combat_avatar_alias(self):
self._write_json(
self.root / "User" / "JsScript" / "AutoSwitchRoles" / "settings.json",
[
{"name": f"position{index}", "type": "input-text", "default": ""}
for index in range(1, 5)
],
)
self._write_json(
self.root / "User" / "JsScript" / "AutoSwitchRoles" / "combat_avatar.json",
[
{"name": "角色甲", "alias": ["同名"]},
{"name": "角色乙", "alias": ["同名"]},
{"name": "角色丙", "alias": []},
{"name": "角色丁", "alias": []},
{"name": "角色戊", "alias": []},
],
)
with self.assertRaisesRegex(DailyAutomationError, "未知或有歧义"):
resolve_party_members(self.root, "同名 角色丙 角色丁 角色戊")
def test_edit_party_requires_four_distinct_known_members(self):
with self.assertRaisesRegex(DailyAutomationError, "必须是4人"):
resolve_party_members(self.root, "芙宁娜 爱可菲 纳西妲")
with self.assertRaisesRegex(DailyAutomationError, "不能重复"):
resolve_party_members(self.root, "芙宁娜 芙宁娜 纳西妲 雷电将军")
with self.assertRaisesRegex(DailyAutomationError, "未知"):
resolve_party_members(self.root, "芙宁娜 爱可菲 纳西妲 不存在")
def test_apply_json_updates_writes_managed_files(self):
prepared = self._prepare("秘境 风本")
apply_json_updates(prepared.updates)
bgi_config = json.loads((self.root / "User" / "config.json").read_text(encoding="utf-8"))
managed = json.loads(
(self.root / "User" / "OneDragon" / "直播系统自动每日.json").read_text(encoding="utf-8")
)
self.assertFalse(bgi_config["autoDomainConfig"]["specifyResinUse"])
self.assertEqual(managed["DomainName"], "铭记之谷")
if __name__ == "__main__":
unittest.main()
+43 -1
View File
@@ -1,8 +1,10 @@
import asyncio
import logging
import unittest import unittest
from pathlib import Path from pathlib import Path
from tempfile import TemporaryDirectory from tempfile import TemporaryDirectory
from types import SimpleNamespace from types import SimpleNamespace
from unittest.mock import patch from unittest.mock import AsyncMock, patch
from app.danmu_queue import BgiLogMonitor, WebServer from app.danmu_queue import BgiLogMonitor, WebServer
@@ -69,5 +71,45 @@ class BgiLogDisplayFormatTests(unittest.TestCase):
) )
class BgiLogCompletionTests(unittest.IsolatedAsyncioTestCase):
async def test_configuration_group_marker_still_finishes_current_group(self):
monitor = BgiLogMonitor(".", logging.getLogger("test-bgi-group-log"))
callback = AsyncMock()
monitor.set_finish_callback(callback)
with patch.object(monitor, "_get_current_log_size", return_value=0):
monitor.set_current_group("薄荷", "group-run")
monitor._process_line('配置组 "薄荷" 执行结束')
await asyncio.sleep(0)
callback.assert_awaited_once_with("薄荷", "group-run")
self.assertIsNone(monitor._current_group)
async def test_one_dragon_marker_finishes_managed_daily(self):
monitor = BgiLogMonitor(".", logging.getLogger("test-bgi-one-dragon-log"))
callback = AsyncMock()
monitor.set_finish_callback(callback)
with patch.object(monitor, "_get_current_log_size", return_value=0):
monitor.set_current_one_dragon("自动每日(委托)", "daily-run")
monitor._process_line("一条龙和配置组任务结束")
await asyncio.sleep(0)
callback.assert_awaited_once_with("自动每日(委托)", "daily-run")
self.assertIsNone(monitor._current_group)
async def test_child_group_completion_does_not_finish_one_dragon_early(self):
monitor = BgiLogMonitor(".", logging.getLogger("test-bgi-child-group-log"))
callback = AsyncMock()
monitor.set_finish_callback(callback)
with patch.object(monitor, "_get_current_log_size", return_value=0):
monitor.set_current_one_dragon("自动每日", "daily-run")
monitor._process_line('配置组 "每日委托" 执行结束')
await asyncio.sleep(0)
callback.assert_not_awaited()
self.assertEqual(monitor._current_group, "自动每日")
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+439
View File
@@ -0,0 +1,439 @@
import asyncio
import json
import logging
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, call, patch
from app.bettergi_current_party import (
CurrentPartyReadError,
CurrentPartyReadResult,
PreparedCurrentPartyRead,
)
from app.bettergi_daily import (
DailyAutomationError,
DailyRequest,
JsonUpdate,
PreparedDailyRun,
)
from app.danmu_queue import (
COMMAND_ALLOWED_ROLE_DEFAULTS,
BetterGIRunner,
CommandHandler,
Config,
build_public_command_hints,
)
class DailyCommandIntegrationTests(unittest.IsolatedAsyncioTestCase):
def setUp(self):
self.temp_dir = tempfile.TemporaryDirectory()
self.root = Path(self.temp_dir.name)
self.config = SimpleNamespace(
data={
"commands": {
"run": {"enabled": True, "aliases": ["执行", ""]},
"daily": {
"enabled": True,
"aliases": ["自动每日", "每日", "日常", "做每日", "做日常"],
},
"switch_party": {
"enabled": True,
"aliases": ["切换队伍", "更换队伍", "换队伍", "换队"],
},
"edit_party": {
"enabled": True,
"aliases": ["修改队员", "更换队员", "换队员", "换角色"],
},
}
},
admin_uids=set(),
bettergi_work_dir=str(self.root),
daily_cfg={
"one_dragon_template": "默认配置",
"managed_one_dragon_name": "直播系统自动每日",
"ley_line_craft_resin_before": True,
"commission_use_current_party": True,
"current_party_read_timeout_sec": 45,
},
broadcast_cfg={"tts_categories": {"execution": True}},
music_monitor_cfg={"request_player": {}},
)
self.queue_mgr = MagicMock()
self.queue_mgr.state = {
"current_admin_uid": 123,
"login_status": "logged_in",
"current_group": None,
"default_running": False,
}
self.queue_mgr.is_admin.return_value = True
self.queue_mgr.start_group.return_value = {
"success": True,
"msg": "ok",
"run_id": "daily-run",
}
self.runner = MagicMock()
self.runner.kill_bgi = AsyncMock()
self.runner.start_groups = AsyncMock(return_value=True)
self.runner.start_one_dragon = AsyncMock(return_value=True)
self.runner.sync_js_script.return_value = True
self.log_monitor = MagicMock()
self.broadcaster = SimpleNamespace(broadcast=AsyncMock())
self.handler = CommandHandler(
self.config,
MagicMock(),
self.queue_mgr,
self.runner,
self.log_monitor,
MagicMock(),
logging.getLogger("test-daily-command"),
broadcaster=self.broadcaster,
)
def tearDown(self):
self.temp_dir.cleanup()
def _prepared_daily(self, *, requires_current_party: bool = False) -> PreparedDailyRun:
return PreparedDailyRun(
request=DailyRequest(mode="commission" if requires_current_party else "none"),
config_name="直播系统自动每日",
updates=(
JsonUpdate(
self.root / "User" / "OneDragon" / "直播系统自动每日.json",
{"Name": "直播系统自动每日"},
"test",
),
),
requires_current_party=requires_current_party,
)
async def test_daily_replaces_old_task_writes_config_and_starts_one_dragon(self):
self.queue_mgr.state["current_group"] = "旧任务"
prepared = self._prepared_daily()
with (
patch("app.danmu_queue.prepare_daily_run", return_value=prepared) as prepare,
patch("app.danmu_queue.apply_json_updates") as apply_updates,
patch("app.danmu_queue.uuid.uuid4", return_value=SimpleNamespace(hex="daily-run")),
):
await self.handler._cmd_daily(123, "测试用户", "")
prepare.assert_called_once()
self.queue_mgr.interrupt_group.assert_called_once_with(
"daily_replaced",
status="cancelled",
)
self.runner.kill_bgi.assert_awaited_once_with(reason="自动每日配置更新")
apply_updates.assert_called_once_with(prepared.updates)
self.queue_mgr.start_group.assert_called_once_with(
123,
"自动每日",
run_id="daily-run",
)
self.log_monitor.set_current_one_dragon.assert_called_once_with(
"自动每日",
"daily-run",
)
self.runner.start_one_dragon.assert_awaited_once_with("直播系统自动每日")
async def test_daily_validation_failure_does_not_stop_current_task(self):
self.queue_mgr.state["current_group"] = "旧任务"
with patch(
"app.danmu_queue.prepare_daily_run",
side_effect=DailyAutomationError("未找到一条龙模板"),
):
await self.handler._cmd_daily(123, "测试用户", "秘境 风本")
self.queue_mgr.interrupt_group.assert_not_called()
self.runner.kill_bgi.assert_not_awaited()
self.runner.start_one_dragon.assert_not_awaited()
async def test_daily_commission_reads_current_party_before_starting(self):
prepared = self._prepared_daily(requires_current_party=True)
reader_update = JsonUpdate(
self.root / "User" / "ScriptGroup" / "直播系统读取当前队伍.json",
{"projects": []},
"reader",
)
reader = PreparedCurrentPartyRead(
request_id="daily-run",
group_name="直播系统读取当前队伍",
status_path=self.root / "status.json",
updates=(reader_update,),
)
party_update = JsonUpdate(
self.root / "User" / "JsScript" / "AutoCommissionNova" / "Data" / "user-config.json",
{
"party": {
"global": {
"battleTeamName": "好感队",
"elementTeamName": "好感队",
}
}
},
"party",
)
capture = AsyncMock(return_value="好感队")
with (
patch("app.danmu_queue.prepare_daily_run", return_value=prepared) as prepare,
patch("app.danmu_queue.prepare_current_party_read", return_value=reader),
patch(
"app.danmu_queue.prepare_commission_current_party_update",
return_value=party_update,
) as prepare_party,
patch("app.danmu_queue.apply_json_updates") as apply_updates,
patch.object(self.handler, "_capture_current_party_name", capture),
patch("app.danmu_queue.uuid.uuid4", return_value=SimpleNamespace(hex="daily-run")),
):
await self.handler._cmd_daily(123, "测试用户", "委托")
self.assertTrue(prepare.call_args.kwargs["commission_use_current_party"])
capture.assert_awaited_once_with(reader, 45)
prepare_party.assert_called_once_with(str(self.root), "好感队")
apply_updates.assert_called_once_with((party_update, *prepared.updates))
self.runner.start_one_dragon.assert_awaited_once_with("直播系统自动每日")
async def test_daily_commission_reader_failure_does_not_start_billed_task(self):
prepared = self._prepared_daily(requires_current_party=True)
reader = PreparedCurrentPartyRead(
request_id="daily-run",
group_name="直播系统读取当前队伍",
status_path=self.root / "status.json",
updates=(),
)
with (
patch("app.danmu_queue.prepare_daily_run", return_value=prepared),
patch("app.danmu_queue.prepare_current_party_read", return_value=reader),
patch.object(
self.handler,
"_capture_current_party_name",
new=AsyncMock(side_effect=CurrentPartyReadError("未识别到队伍名称")),
),
patch("app.danmu_queue.apply_json_updates") as apply_updates,
patch("app.danmu_queue.uuid.uuid4", return_value=SimpleNamespace(hex="daily-run")),
):
await self.handler._cmd_daily(123, "测试用户", "委托")
apply_updates.assert_not_called()
self.queue_mgr.start_group.assert_not_called()
self.runner.start_one_dragon.assert_not_awaited()
async def test_current_party_capture_syncs_group_and_waits_for_matching_status(self):
reader_update = JsonUpdate(
self.root / "User" / "ScriptGroup" / "直播系统读取当前队伍.json",
{"projects": []},
"reader",
)
reader = PreparedCurrentPartyRead(
request_id="request-123",
group_name="直播系统读取当前队伍",
status_path=self.root / "status.json",
updates=(reader_update,),
)
with (
patch("app.danmu_queue.apply_json_updates") as apply_updates,
patch(
"app.danmu_queue.read_current_party_status",
return_value=CurrentPartyReadResult("好感队", ("好感队",)),
),
patch("app.danmu_queue.asyncio.sleep", new=AsyncMock()),
):
party_name = await self.handler._capture_current_party_name(reader, 45)
self.assertEqual(party_name, "好感队")
self.runner.sync_js_script.assert_called_once_with("LiveCurrentParty")
apply_updates.assert_called_once_with(reader.updates)
self.runner.start_groups.assert_awaited_once_with(["直播系统读取当前队伍"])
async def test_daily_requires_confirmed_login_before_preparing_files(self):
self.queue_mgr.state["login_status"] = "confirming"
with patch("app.danmu_queue.prepare_daily_run") as prepare:
await self.handler._cmd_daily(123, "测试用户", "")
prepare.assert_not_called()
self.runner.kill_bgi.assert_not_awaited()
async def test_daily_start_failure_clears_monitor_and_enters_reset_flow(self):
prepared = self._prepared_daily()
self.runner.start_one_dragon.return_value = False
reset = AsyncMock()
self.handler.system = SimpleNamespace(_reset_wait_and_retry_login=reset)
with (
patch("app.danmu_queue.prepare_daily_run", return_value=prepared),
patch("app.danmu_queue.apply_json_updates"),
patch("app.danmu_queue.uuid.uuid4", return_value=SimpleNamespace(hex="daily-run")),
):
await self.handler._cmd_daily(123, "测试用户", "")
self.log_monitor.set_current_one_dragon.assert_has_calls(
[call("自动每日", "daily-run"), call(None)]
)
reset.assert_awaited_once_with(
123,
"测试用户",
"自动每日启动失败,正在关闭原神并启动扫码上号",
)
async def test_switch_party_updates_group_then_reuses_group_execution(self):
update = JsonUpdate(
self.root / "User" / "ScriptGroup" / "切换队伍.json",
{"projects": []},
"test",
)
with (
patch("app.danmu_queue.prepare_switch_party_update", return_value=update),
patch("app.danmu_queue.apply_json_updates") as apply_updates,
patch("app.danmu_queue.uuid.uuid4", return_value=SimpleNamespace(hex="party-run")),
patch.object(
self.handler,
"_resolve_group_name",
return_value={
"success": True,
"name": "切换队伍",
"matched": False,
"suggestions": [],
},
),
patch.object(self.handler, "_group_uses_nahida_collect", return_value=False),
):
await self.handler._cmd_switch_party(123, "测试用户", "永冻队")
apply_updates.assert_called_once_with((update,))
self.queue_mgr.start_group.assert_called_once_with(
123,
"切换队伍",
run_id="party-run",
)
self.runner.start_groups.assert_awaited_once_with(["切换队伍"])
def test_execution_command_aliases_support_no_space_arguments(self):
self.assertEqual(self.handler._split_command_text("执行薄荷"), ("执行", "薄荷"))
self.assertEqual(
self.handler._split_command_text("自动每日秘境 风本"),
("自动每日", "秘境 风本"),
)
self.assertEqual(
self.handler._split_command_text("更换队伍永冻队"),
("更换队伍", "永冻队"),
)
self.assertEqual(
self.handler._split_command_text("修改队员芙宁娜纳西妲钟离雷电将军"),
("修改队员", "芙宁娜纳西妲钟离雷电将军"),
)
self.assertEqual(
self.handler._split_command_text("日常秘境 风本"),
("日常", "秘境 风本"),
)
self.assertEqual(
self.handler._split_command_text("换队永冻队"),
("换队", "永冻队"),
)
self.assertEqual(
self.handler._split_command_text("换角色芙宁娜纳西妲钟离雷电将军"),
("换角色", "芙宁娜纳西妲钟离雷电将军"),
)
def test_longest_new_alias_wins_for_contiguous_arguments(self):
self.assertEqual(
self.handler._split_command_text("换队员芙宁娜纳西妲钟离雷电将军"),
("换队员", "芙宁娜纳西妲钟离雷电将军"),
)
async def test_confirmed_login_announces_daily_and_party_usage(self):
self.queue_mgr.state["login_status"] = "confirming"
await self.handler._cmd_confirm_login(123, "测试用户", "")
self.assertEqual(self.broadcaster.broadcast.await_count, 2)
first, second = self.broadcaster.broadcast.await_args_list
self.assertIn("发送“每日”完成基础每日", first.args[0])
self.assertTrue(first.kwargs["tts"])
self.assertIn("每日 秘境 风本", second.args[0])
self.assertIn("换队 永冻队", second.args[0])
self.assertIn("换角色 四名角色", second.args[0])
self.assertIn("执行 组名", second.args[0])
self.assertFalse(second.kwargs["tts"])
self.assertTrue(second.kwargs["danmu"])
def test_public_command_hints_are_sanitized_and_use_shortest_alias(self):
config_data = {
"commands": self.config.data["commands"],
"bilibili": {"cookie": "secret"},
}
hints = build_public_command_hints(config_data)
self.assertEqual(set(hints), {"daily", "switch_party", "edit_party"})
self.assertEqual(hints["daily"]["preferred_alias"], "每日")
self.assertEqual(hints["switch_party"]["preferred_alias"], "换队")
self.assertEqual(hints["edit_party"]["preferred_alias"], "换角色")
self.assertNotIn("bilibili", hints)
def test_new_command_alias_defaults_are_complete(self):
config_path = self.root / "config.json"
config_path.write_text(json.dumps({}), encoding="utf-8")
config = Config(str(config_path))
self.assertEqual(
config.data["commands"]["daily"]["aliases"],
["自动每日", "每日", "日常", "做每日", "做日常"],
)
self.assertEqual(
config.data["commands"]["switch_party"]["aliases"],
["切换队伍", "更换队伍", "换队伍", "换队"],
)
self.assertEqual(
config.data["commands"]["edit_party"]["aliases"],
["修改队员", "更换队员", "换队员", "换角色"],
)
def test_disabled_command_hint_has_no_public_alias(self):
config_data = {
"commands": {
"daily": {"enabled": False, "aliases": ["每日"]},
}
}
hints = build_public_command_hints(config_data)
self.assertFalse(hints["daily"]["enabled"])
self.assertEqual(hints["daily"]["aliases"], [])
self.assertEqual(hints["daily"]["preferred_alias"], "")
def test_new_command_permissions_match_run(self):
for key in ("daily", "switch_party", "edit_party"):
self.assertEqual(
COMMAND_ALLOWED_ROLE_DEFAULTS[key],
COMMAND_ALLOWED_ROLE_DEFAULTS["run"],
)
class BetterGIRunnerOneDragonTests(unittest.IsolatedAsyncioTestCase):
async def test_start_one_dragon_uses_supported_cli_shape(self):
runner = BetterGIRunner(
r"C:\BetterGI\BetterGI.exe",
r"C:\BetterGI",
logging.getLogger("test-one-dragon-runner"),
)
create_process = AsyncMock(return_value=SimpleNamespace())
with (
patch("app.danmu_queue.asyncio.create_subprocess_exec", create_process),
patch("app.danmu_queue.asyncio.sleep", new=AsyncMock()),
):
result = await runner.start_one_dragon("直播系统自动每日")
self.assertTrue(result)
create_process.assert_awaited_once_with(
r"C:\BetterGI\BetterGI.exe",
"startOneDragon",
"直播系统自动每日",
cwd=r"C:\BetterGI",
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
if __name__ == "__main__":
unittest.main()
+86
View File
@@ -9,6 +9,7 @@ from app.faster_qwen_worker import (
MAX_NEW_TOKENS, MAX_NEW_TOKENS,
FasterQwenWorkerClient, FasterQwenWorkerClient,
FasterQwenWorkerTimeout, FasterQwenWorkerTimeout,
faster_qwen_worker_main,
_generation_kwargs, _generation_kwargs,
_configure_worker_environment, _configure_worker_environment,
) )
@@ -91,6 +92,14 @@ def _ready(pid):
} }
def _progress(stage):
return {
"type": "progress",
"stage": stage,
"pid": 4000,
}
def _probe_worker_main(connection, _settings): def _probe_worker_main(connection, _settings):
connection.send({ connection.send({
"type": "ready", "type": "ready",
@@ -153,6 +162,8 @@ class FasterQwenWorkerTests(unittest.TestCase):
def test_client_returns_worker_audio(self): def test_client_returns_worker_audio(self):
connection = _FakeConnection([ connection = _FakeConnection([
_progress("model_loading"),
_progress("warmup_started"),
_ready(4100), _ready(4100),
{ {
"type": "result", "type": "result",
@@ -181,6 +192,81 @@ class FasterQwenWorkerTests(unittest.TestCase):
self.assertEqual(len(context.processes), 1) self.assertEqual(len(context.processes), 1)
client.close() client.close()
def test_client_default_startup_timeout_allows_slow_model_load(self):
client = FasterQwenWorkerClient({})
self.assertEqual(client.startup_timeout_seconds, 900.0)
client.close()
def test_worker_applies_process_limits_after_model_warmup(self):
order = []
class FakeModel:
def warmup(self):
order.append("warmup")
class WorkerConnection:
def __init__(self):
self.sent = []
self.closed = False
def send(self, message):
self.sent.append(message)
def recv(self):
return {"command": "stop"}
def close(self):
self.closed = True
connection = WorkerConnection()
with (
patch("app.faster_qwen_worker._configure_worker_environment", return_value=3),
patch(
"app.faster_qwen_worker._load_runtime",
side_effect=lambda _settings, progress=None: (
progress("from_pretrained_started") if progress else None,
order.append("load"),
{"model": FakeModel()},
)[-1],
),
patch(
"app.faster_qwen_worker._apply_worker_process_limits",
side_effect=lambda _settings: order.append("limits") or {
"cpu_affinity_count": 6,
"process_priority": "below_normal",
},
),
patch(
"app.faster_qwen_worker._synthesize_wav",
side_effect=lambda _runtime, _text: order.append("voice_clone") or b"wav",
),
):
faster_qwen_worker_main(connection, {})
self.assertEqual(order, ["load", "warmup", "voice_clone", "limits"])
progress_stages = [
item.get("stage")
for item in connection.sent
if item.get("type") == "progress"
]
self.assertEqual(
progress_stages,
[
"process_started",
"environment_ready",
"model_loading",
"from_pretrained_started",
"model_loaded",
"warmup_started",
"graph_warmup_complete",
"voice_clone_warmup_started",
"voice_clone_warmup_complete",
"warmup_complete",
],
)
self.assertEqual(connection.sent[-1]["type"], "ready")
def test_timeout_terminates_worker_and_starts_a_prewarmed_replacement(self): def test_timeout_terminates_worker_and_starts_a_prewarmed_replacement(self):
first_connection = _FakeConnection([_ready(4200)]) first_connection = _FakeConnection([_ready(4200)])
replacement_connection = _FakeConnection([_ready(4300)]) replacement_connection = _FakeConnection([_ready(4300)])
+33
View File
@@ -0,0 +1,33 @@
import json
import unittest
from pathlib import Path
class LoginScriptDomainExitTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.script_dir = Path(__file__).resolve().parents[1] / "integrations" / "扫码上号"
cls.source = (cls.script_dir / "main.js").read_text(encoding="utf-8")
cls.settings = json.loads((cls.script_dir / "settings.json").read_text(encoding="utf-8"))
cls.manifest = json.loads((cls.script_dir / "manifest.json").read_text(encoding="utf-8"))
def test_domain_exit_requires_specific_ocr_text_before_click(self):
self.assertIn('combinedText.includes("退出秘境")', self.source)
self.assertIn('result.text.includes("确认") && result.x >= 900', self.source)
def test_legacy_and_default_confirm_coordinates_are_corrected(self):
defaults = {
item.get("name"): item.get("default")
for item in self.settings
if isinstance(item, dict) and item.get("name")
}
self.assertEqual(defaults["confirmExitX"], "1164")
self.assertEqual(defaults["confirmExitY"], "757")
self.assertIn("confirmExitX === 830 && confirmExitY === 600", self.source)
def test_private_login_script_version_is_updated(self):
self.assertEqual(self.manifest["version"], "1.1")
if __name__ == "__main__":
unittest.main()
+165 -1
View File
@@ -1,10 +1,12 @@
import asyncio import asyncio
import ctypes
import logging import logging
import unittest import unittest
from datetime import datetime, timedelta from datetime import datetime, timedelta
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
from app.danmu_queue import SystemScheduler from app.danmu_queue import CommandHandler, QueueSystem, SystemScheduler
class _FakeConfig: class _FakeConfig:
@@ -155,5 +157,167 @@ class StartupLiveCompensationTests(unittest.IsolatedAsyncioTestCase):
) )
class LivehimePushVerificationTests(unittest.IsolatedAsyncioTestCase):
def setUp(self):
self.logger = logging.getLogger("test-livehime-push-verification")
self.scheduler = SystemScheduler(
_FakeConfig({
"bilibili_push_window_keyword": "直播姬",
"bilibili_push_click_x_ratio": 0.787,
"bilibili_push_click_y_ratio": 0.927,
"bilibili_stop_push_click_x_ratio": 0.787,
"bilibili_stop_push_click_y_ratio": 0.927,
}),
self.logger,
)
async def test_push_skips_click_when_bilibili_is_already_live(self):
self.scheduler._get_bilibili_live_status = AsyncMock(return_value=1)
self.scheduler._click_bilibili_live = AsyncMock()
self.assertTrue(await self.scheduler.push_bilibili_live())
self.scheduler._click_bilibili_live.assert_not_awaited()
async def test_push_clicks_then_waits_for_confirmed_live_status(self):
self.scheduler._get_bilibili_live_status = AsyncMock(side_effect=[0, 0, 1])
self.scheduler._click_bilibili_live = AsyncMock(return_value=True)
with patch("app.danmu_queue.asyncio.sleep", new_callable=AsyncMock):
result = await self.scheduler.push_bilibili_live()
self.assertTrue(result)
self.scheduler._click_bilibili_live.assert_awaited_once()
async def test_stop_skips_click_when_bilibili_is_already_offline(self):
self.scheduler._get_bilibili_live_status = AsyncMock(return_value=0)
self.scheduler._click_bilibili_live = AsyncMock()
self.assertTrue(await self.scheduler.stop_bilibili_live())
self.scheduler._click_bilibili_live.assert_not_awaited()
async def test_live_status_timeout_returns_specific_error(self):
self.scheduler._get_bilibili_live_status = AsyncMock(return_value=0)
with patch("app.danmu_queue.asyncio.sleep", new_callable=AsyncMock):
result = await self.scheduler._wait_for_bilibili_live_status(
1,
timeout_seconds=0.01,
)
self.assertFalse(result)
self.assertIn("未确认开播", self.scheduler.last_live_action_error)
@patch("app.danmu_queue.os.name", "nt")
async def test_click_restores_and_focuses_before_reading_physical_window_rect(self):
hwnd = 12345
user32 = MagicMock()
call_order = []
user32.IsIconic.return_value = True
user32.GetForegroundWindow.return_value = hwnd
user32.GetSystemMetrics.side_effect = lambda index: {
76: 0,
77: 0,
78: 2560,
79: 1440,
}[index]
user32.SetCursorPos.return_value = True
user32.GetCursorPos.return_value = False
user32.SetThreadDpiAwarenessContext.return_value = None
def show_window(_hwnd, _command):
call_order.append("restore")
return True
def get_window_rect(_hwnd, pointer):
call_order.append("rect")
rect = ctypes.cast(pointer, ctypes.POINTER(ctypes.wintypes.RECT)).contents
rect.left = -11
rect.top = -11
rect.right = 2571
rect.bottom = 1379
return True
user32.ShowWindow.side_effect = show_window
user32.GetWindowRect.side_effect = get_window_rect
self.scheduler._find_window_by_keyword = MagicMock(return_value=hwnd)
self.scheduler._force_foreground_window = MagicMock(return_value=True)
with (
patch("app.danmu_queue.ctypes.windll.user32", user32),
patch("app.danmu_queue.asyncio.sleep", new_callable=AsyncMock),
):
result = await self.scheduler._click_bilibili_live(
x_ratio_key="bilibili_push_click_x_ratio",
y_ratio_key="bilibili_push_click_y_ratio",
action_name="开启推流按钮",
)
self.assertTrue(result)
self.assertLess(call_order.index("restore"), call_order.index("rect"))
user32.SetCursorPos.assert_called_once_with(2021, 1277)
self.assertEqual(user32.mouse_event.call_count, 2)
class ManualLiveOverrideTests(unittest.TestCase):
def setUp(self):
self.system = QueueSystem.__new__(QueueSystem)
self.system.logger = logging.getLogger("test-manual-live-override")
self.system._manual_live_override_until_monotonic = 0.0
self.system._manual_live_override_until_wall = 0.0
self.system._bilibili_live_status = None
self.system._bilibili_live_status_updated_monotonic = 0.0
self.system._bilibili_live_status_updated_wall = 0.0
def test_manual_live_override_allows_commands_outside_schedule(self):
handler = CommandHandler.__new__(CommandHandler)
handler.system = self.system
handler.config = SimpleNamespace(
system_cfg={"live_start_time": "08:30", "live_end_time": "23:30"}
)
with patch("app.danmu_queue.time.monotonic", return_value=100.0):
self.system.enable_manual_live_override(300)
self.assertTrue(handler._is_live_time())
snapshot = self.system.manual_live_override_snapshot()
self.assertTrue(snapshot["active"])
self.assertGreater(snapshot["until"], 0)
def test_clear_manual_live_override_restores_schedule_gate(self):
handler = CommandHandler.__new__(CommandHandler)
handler.system = self.system
handler.config = SimpleNamespace(
system_cfg={"live_start_time": "08:30", "live_end_time": "23:30"}
)
with (
patch("app.danmu_queue.time.monotonic", return_value=100.0),
patch("app.danmu_queue.datetime") as mocked_datetime,
):
mocked_datetime.now.return_value = datetime(2026, 8, 16, 3, 30)
mocked_datetime.strptime = datetime.strptime
self.system.enable_manual_live_override(300)
self.system.clear_manual_live_override()
self.assertFalse(handler._is_live_time())
def test_confirmed_bilibili_live_status_allows_commands_outside_schedule(self):
handler = CommandHandler.__new__(CommandHandler)
handler.system = self.system
handler.config = SimpleNamespace(
system_cfg={"live_start_time": "08:30", "live_end_time": "23:30"}
)
with (
patch("app.danmu_queue.time.monotonic", return_value=100.0),
patch("app.danmu_queue.datetime") as mocked_datetime,
):
mocked_datetime.now.return_value = datetime(2026, 8, 16, 3, 30)
mocked_datetime.strptime = datetime.strptime
self.system.set_bilibili_live_status(1)
self.assertTrue(handler._is_live_time())
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+44
View File
@@ -2,6 +2,7 @@ import asyncio
import logging import logging
import time import time
import unittest import unittest
from types import SimpleNamespace
from unittest.mock import patch from unittest.mock import patch
import numpy as np import numpy as np
@@ -144,6 +145,49 @@ class PipelineOverlapTests(unittest.IsolatedAsyncioTestCase):
await asyncio.gather(synth_worker, play_worker, return_exceptions=True) await asyncio.gather(synth_worker, play_worker, return_exceptions=True)
class BackgroundWarmupTests(unittest.IsolatedAsyncioTestCase):
async def test_start_returns_before_tts_warmup_finishes(self):
warmup_started = asyncio.Event()
release_warmup = asyncio.Event()
class FakeTTS:
enabled = True
async def warmup(self):
warmup_started.set()
await release_warmup.wait()
async def idle_worker():
await asyncio.Event().wait()
broadcaster = Broadcaster.__new__(Broadcaster)
broadcaster._tts_started = False
broadcaster._tts_warmup_done = asyncio.Event()
broadcaster._tts_warmup_task = None
broadcaster._tts_worker_tasks = set()
broadcaster._tts_queue_cfg = {"warmup_on_start": True}
broadcaster._tts_pending = SimpleNamespace(maxsize=8)
broadcaster._tts_playback_queue = SimpleNamespace(maxsize=2)
broadcaster._tts_synthesis_loop = idle_worker
broadcaster._tts_playback_loop = idle_worker
broadcaster.tts = FakeTTS()
broadcaster.logger = logging.getLogger("tts-background-warmup-test")
await asyncio.wait_for(broadcaster.start(), timeout=0.2)
await asyncio.wait_for(warmup_started.wait(), timeout=0.2)
self.assertFalse(broadcaster._tts_warmup_done.is_set())
self.assertIsNotNone(broadcaster._tts_warmup_task)
release_warmup.set()
await asyncio.wait_for(broadcaster._tts_warmup_done.wait(), timeout=0.2)
tasks = list(broadcaster._tts_worker_tasks)
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
class TTSExpiryPolicyTests(unittest.TestCase): class TTSExpiryPolicyTests(unittest.TestCase):
def test_default_expiry_windows_match_configured_policy(self): def test_default_expiry_windows_match_configured_policy(self):
broadcaster = Broadcaster.__new__(Broadcaster) broadcaster = Broadcaster.__new__(Broadcaster)
File diff suppressed because one or more lines are too long
+1 -1
View File
@@ -4,7 +4,7 @@
<meta charset="UTF-8" /> <meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" /> <meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>直播联动后台</title> <title>直播联动后台</title>
<script type="module" crossorigin src="/admin/assets/index-CIcFYlv1.js"></script> <script type="module" crossorigin src="/admin/assets/index-BdYZGJMQ.js"></script>
<link rel="stylesheet" crossorigin href="/admin/assets/index-BB8VhdaG.css"> <link rel="stylesheet" crossorigin href="/admin/assets/index-BB8VhdaG.css">
</head> </head>
<body> <body>
+124 -11
View File
@@ -47,9 +47,10 @@
width: 100vw; width: 100vw;
height: 100vh; height: 100vh;
display: flex; display: flex;
align-items: center; align-items: flex-start;
justify-content: center; justify-content: flex-start;
background: transparent; background: transparent;
overflow: hidden;
} }
body.preview #viewport { body.preview #viewport {
@@ -1315,6 +1316,34 @@
border: 1px solid rgba(255, 242, 201, 0.13); border: 1px solid rgba(255, 242, 201, 0.13);
} }
.command-quick-reference {
position: relative;
z-index: 1;
display: grid;
grid-template-columns: repeat(2, minmax(0, 1fr));
gap: 2px 12px;
margin: 0 12px 2px;
padding: 3px 8px;
border-top: 1px solid rgba(255, 242, 201, 0.15);
border-bottom: 1px solid rgba(255, 242, 201, 0.15);
color: rgba(244, 255, 247, 0.86);
font-size: 12px;
font-weight: 720;
line-height: 1.25;
}
.command-quick-reference span {
min-width: 0;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.command-quick-reference strong {
color: #fff0b5;
font-weight: 900;
}
.tutorial-stage { .tutorial-stage {
position: relative; position: relative;
z-index: 1; z-index: 1;
@@ -1322,19 +1351,20 @@
display: flex; display: flex;
align-items: center; align-items: center;
justify-content: center; justify-content: center;
padding: 0 14px 3px; min-height: 32px;
padding: 0 10px 1px;
overflow: hidden; overflow: hidden;
} }
.tutorial-text { .tutorial-text {
width: 100%; width: 100%;
color: var(--text); color: var(--text);
font-size: 26px; font-size: 20px;
font-weight: 900; font-weight: 900;
line-height: 1.35; line-height: 1.25;
text-align: center; text-align: center;
white-space: nowrap; white-space: nowrap;
letter-spacing: 0.7px; letter-spacing: 0;
text-shadow: 0 2px 5px rgba(0, 0, 0, 0.48), 0 0 14px rgba(255, 240, 181, 0.18); text-shadow: 0 2px 5px rgba(0, 0, 0, 0.48), 0 0 14px rgba(255, 240, 181, 0.18);
opacity: 1; opacity: 1;
transform: translateY(0); transform: translateY(0);
@@ -1342,12 +1372,12 @@
} }
.tutorial-text.long { .tutorial-text.long {
font-size: 23px; font-size: 18px;
letter-spacing: 0.35px; letter-spacing: 0;
} }
.tutorial-text.extra-long { .tutorial-text.extra-long {
font-size: 21px; font-size: 16px;
letter-spacing: 0; letter-spacing: 0;
} }
@@ -1852,6 +1882,7 @@
<!-- 右侧中部:教学轮播 --> <!-- 右侧中部:教学轮播 -->
<section class="panel tutorial-panel"> <section class="panel tutorial-panel">
<div class="panel-title tutorial-title">直播间使用说明</div> <div class="panel-title tutorial-title">直播间使用说明</div>
<div id="command-quick-reference" class="command-quick-reference"></div>
<div class="tutorial-stage"> <div class="tutorial-stage">
<div id="tutorial-text" class="tutorial-text">发送“签到”领取积分,发送“排队”加入队伍</div> <div id="tutorial-text" class="tutorial-text">发送“签到”领取积分,发送“排队”加入队伍</div>
</div> </div>
@@ -1898,7 +1929,7 @@
return m + ":" + String(sec % 60).padStart(2, "0"); return m + ":" + String(sec % 60).padStart(2, "0");
} }
const tutorialSlides = [ const baseTutorialSlides = [
"发送“签到”领取积分,发送“排队”加入队伍", "发送“签到”领取积分,发送“排队”加入队伍",
"成为队首后,请在90秒内发送“上号”", "成为队首后,请在90秒内发送“上号”",
"发送“上号”后,请在4分钟内完成扫码登录", "发送“上号”后,请在4分钟内完成扫码登录",
@@ -1907,8 +1938,85 @@
"发送“积分”查询积分,发送“队列”查看排队情况", "发送“积分”查询积分,发送“队列”查看排队情况",
"发送“退出”离开队伍,发送“点歌 歌名”点歌" "发送“退出”离开队伍,发送“点歌 歌名”点歌"
]; ];
const fallbackCommandHints = {
daily: { enabled: true, aliases: ["自动每日", "每日", "日常"], preferred_alias: "每日" },
switch_party: { enabled: true, aliases: ["切换队伍", "换队"], preferred_alias: "换队" },
edit_party: { enabled: true, aliases: ["修改队员", "换角色"], preferred_alias: "换角色" }
};
let commandHints = fallbackCommandHints;
let tutorialSlides = baseTutorialSlides.slice();
let tutorialIndex = 0; let tutorialIndex = 0;
function normalizeCommandHint(value, fallback) {
const source = value && typeof value === "object" ? value : fallback;
const aliases = Array.isArray(source.aliases)
? source.aliases.map((item) => String(item || "").trim()).filter(Boolean)
: fallback.aliases.slice();
const preferred = String(source.preferred_alias || "").trim()
|| aliases.reduce((best, item) => !best || item.length < best.length ? item : best, "")
|| fallback.preferred_alias;
return {
enabled: source.enabled !== false,
aliases,
preferred_alias: preferred
};
}
function renderCommandQuickReference() {
const el = document.getElementById("command-quick-reference");
if (!el) return;
const daily = commandHints.daily;
if (!daily || !daily.enabled) {
el.hidden = true;
el.innerHTML = "";
return;
}
const cmd = esc(daily.preferred_alias || "每日");
el.hidden = false;
el.innerHTML = [
`<span><strong>${cmd}</strong>:基础流程</span>`,
`<span><strong>${cmd} 秘境</strong> &lt;名称&gt;</span>`,
`<span><strong>${cmd} 地脉</strong> &lt;类型&gt; &lt;国家&gt;</span>`,
`<span><strong>${cmd} 委托</strong>:每日委托</span>`
].join("");
}
function rebuildTutorialSlides() {
const slides = baseTutorialSlides.slice();
const daily = commandHints.daily;
const switchParty = commandHints.switch_party;
const editParty = commandHints.edit_party;
if (daily && daily.enabled) {
const cmd = daily.preferred_alias || "每日";
slides.push(`发送“${cmd}”完成邮件、树脂、尘歌壶和每日奖励`);
slides.push(`发送“${cmd} 秘境 风本”刷到树脂耗尽`);
slides.push(`发送“${cmd} 地脉 经验 蒙德”刷经验地脉`);
slides.push(`发送“${cmd} 委托”执行每日委托配置`);
}
const partyParts = [];
if (switchParty && switchParty.enabled) {
partyParts.push(`${switchParty.preferred_alias || "换队"} 永冻队”切换队伍`);
}
if (editParty && editParty.enabled) {
partyParts.push(`${editParty.preferred_alias || "换角色"} 四名角色”修改队员`);
}
if (partyParts.length) slides.push(`发送${partyParts.join(",发送")}`);
tutorialSlides = slides;
tutorialIndex %= tutorialSlides.length;
renderCommandQuickReference();
showTutorialSlide(tutorialIndex, false);
}
function applyCommandHints(value) {
const source = value && typeof value === "object" ? value : {};
commandHints = {
daily: normalizeCommandHint(source.daily, fallbackCommandHints.daily),
switch_party: normalizeCommandHint(source.switch_party, fallbackCommandHints.switch_party),
edit_party: normalizeCommandHint(source.edit_party, fallbackCommandHints.edit_party)
};
rebuildTutorialSlides();
}
function renderTutorialDots() { function renderTutorialDots() {
const dots = document.getElementById("tutorial-dots"); const dots = document.getElementById("tutorial-dots");
if (!dots) return; if (!dots) return;
@@ -2344,8 +2452,10 @@
themeClasses.forEach((c) => document.body.classList.remove(c)); themeClasses.forEach((c) => document.body.classList.remove(c));
const theme = cfg.theme || "classic"; const theme = cfg.theme || "classic";
if (theme && theme !== "classic") document.body.classList.add("theme-" + theme); if (theme && theme !== "classic") document.body.classList.add("theme-" + theme);
applyCommandHints(cfg.command_hints);
} catch (e) { } catch (e) {
console.error("frontend-config", e); console.error("frontend-config", e);
applyCommandHints(fallbackCommandHints);
} }
} }
@@ -2647,9 +2757,12 @@
function scaleCanvas() { function scaleCanvas() {
const canvas = document.getElementById("canvas"); const canvas = document.getElementById("canvas");
const scale = Math.min(window.innerWidth / 1920, window.innerHeight / 1080); const scale = Math.min(window.innerWidth / 1920, window.innerHeight / 1080);
canvas.style.transform = "scale(" + scale + ")"; const offsetX = Math.max(0, (window.innerWidth - 1920 * scale) / 2);
const offsetY = Math.max(0, (window.innerHeight - 1080 * scale) / 2);
canvas.style.transform = `translate(${offsetX}px, ${offsetY}px) scale(${scale})`;
} }
applyCommandHints(fallbackCommandHints);
startTutorialCarousel(); startTutorialCarousel();
fetchFrontendConfig(); fetchFrontendConfig();
fetchState(); fetchState();