Update Live-streaming code (auto-daily features)
This commit is contained in:
@@ -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)
|
||||
Reference in New Issue
Block a user