"""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)