829 lines
30 KiB
Python
829 lines
30 KiB
Python
"""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)
|