Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 0 additions & 2 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -222,8 +222,6 @@ __marimo__/
# Backup files
*.bak

# GUI 状态持久化(用户本地偏好)
gui_state.json
config/config.yml

# 调度运行参数(运行时由 schedule.example.yml 拷贝生成,含邮件授权码,不进版本库;模板见 config/schedule.example.yml)
Expand Down
2 changes: 1 addition & 1 deletion CONVENTIONS.md
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ except subprocess.TimeoutExpired as e:

## 9. GUI 持久化边界

`gui_state.json` 只存 `dungeon` / `sequence`;`enabled` 纯内存态,重启恢复全开,不持久化。
无 UI 状态文件:日常副本/序列的真源是子脚本 config(编辑期实时落盘);`set_dungeon` 为 no-op 的脚本(绝区零/崩铁,上游自身已支持)不提供选择,chip 直接呈现 `dungeon_list.yml` 声明项。`enabled` 纯内存态,重启恢复全开,不持久化。

## 10. 不随意修改 `.bak` / 备份文件

Expand Down
50 changes: 26 additions & 24 deletions src/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,10 @@
import warnings

from src.config.set_config import supports_weekly
from src.config.subscript import get_script_name
from src.service.chain_service import ChainService
from src.service.script_service import ScriptService
from src.service.app_service import AppService
from src.utils_config import get_script
from src.utils_shutdown import shutdown_sys
from src.utils_sub_config import get_script_name

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -217,8 +217,8 @@ def _run_selftest(out_path: str | None) -> int:
"""
result: dict = {"status": "fail", "checks": {}}
try:
service = ChainService()
data = service.load_config()
app_service = AppService()
data = app_service.load_config()
result["checks"]["service_ready"] = True
result["checks"]["script_count"] = len(data["script_list"])
result["checks"]["config_loaded"] = True
Expand All @@ -237,9 +237,9 @@ def _run_check_config(out_path: str | None) -> int:
返回退出码 0=全部合法 / 1=存在不合法项。
"""
try:
service = ChainService()
all_config_data = service.load_config()
invalid = service.collect_invalid_scripts(all_config_data["script_list"])
app_service = AppService()
all_config_data = app_service.load_config()
invalid = app_service.collect_invalid_scripts(all_config_data["script_list"])
result = {
"status": "ok" if not invalid else "invalid",
"script_count": len(all_config_data["script_list"]),
Expand All @@ -254,8 +254,8 @@ def _run_check_config(out_path: str | None) -> int:

def _run_list_scripts(out_path: str | None) -> int:
"""CLI: 列出所有脚本唯一标识(exe 用进程名,脚本文件用 display_name)。"""
service = ChainService()
all_config_data = service.load_config()
app_service = AppService()
all_config_data = app_service.load_config()
names = [get_script_name(s) for s in all_config_data["script_list"]]
result = {"script_count": len(names), "scripts": names}
_emit_json("list_scripts", result, out_path)
Expand All @@ -267,7 +267,7 @@ def _run_get_script(script_name: str, out_path: str | None) -> int:

返回退出码 0=找到 / 1=不存在。
"""
script = ScriptService().get_script(script_name)
script = get_script(script_name)
if script is None:
_emit_json(
"get_script",
Expand All @@ -281,8 +281,8 @@ def _run_get_script(script_name: str, out_path: str | None) -> int:

def _run_dump_config(out_path: str | None) -> int:
"""CLI: 导出完整 config.yml(JSON)。"""
service = ChainService()
all_config_data = service.load_config()
app_service = AppService()
all_config_data = app_service.load_config()
_emit_json("dump_config", all_config_data, out_path)
return 0

Expand All @@ -296,7 +296,7 @@ def _run_check_weekly(out_path: str | None) -> int:

返回退出码 0=一致 / 1=存在不一致。
"""
result = ScriptService().check_weekly()
result = AppService().check_weekly()
_emit_json("check_weekly", result, out_path)
assert "status" in result, "[cli] check_weekly 结果缺少 status"
return 0 if result["status"] == "ok" else 1
Expand Down Expand Up @@ -354,8 +354,8 @@ def _resolve_enable_keys(

def _run_generate_chain(args) -> int:
"""CLI: 生成脚本链配置。返回退出码 0=成功 / 1=失败。"""
service = ChainService()
all_config_data = service.load_config()
app_service = AppService()
all_config_data = app_service.load_config()

known = {get_script_name(s) for s in all_config_data["script_list"]}
enabled_keys, err = _resolve_enable_keys(args.enable, known)
Expand Down Expand Up @@ -407,12 +407,14 @@ def _run_generate_chain(args) -> int:
)
return 1
# 仅做持久化(周几跑是长期配置),不实时写子脚本 config
service.set_weekly_start(script_name, start_day)
app_service.set_weekly_start(script_name, start_day)

out = args.out
if out:
out = os.path.abspath(out)
out = service.generate_chain(all_config_data, enabled_keys, args.name, out_path=out)
out = app_service.generate_chain(
all_config_data, enabled_keys, args.name, out_path=out
)
_emit_cli("generate_chain", f"已生成脚本链配置: {out}")
return 0

Expand All @@ -425,10 +427,10 @@ def _run_run_chain(args) -> int:
return 1

extra_args = []
service = ChainService()
command, cwd, _env = service.build_chain_command(chain_path, extra_args)
app_service = AppService()
command, cwd, _env = app_service.build_chain_command(chain_path, extra_args)
_emit_cli("run_chain", f"运行: {cwd} {' '.join(command)}")
code = service.run_chain_command(
code = app_service.run_chain_command(
chain_path, block=not args.no_block, extra_args=extra_args
)
if args.no_block:
Expand All @@ -449,14 +451,14 @@ def _run_scheduled(args) -> int:
``CREATE_NEW_CONSOLE`` 起),故等待阻塞无害;关闭该控制台即取消。链在点火时
才生成(按当天星期)。
"""
service = ChainService()
all_config_data = service.load_config()
app_service = AppService()
all_config_data = app_service.load_config()
known = {get_script_name(s) for s in all_config_data["script_list"]}
enabled_keys, err = _resolve_enable_keys(args.enable, known)
if err:
_emit_cli("schedule_run", err)
return 1
service.schedule_run(
app_service.schedule_run(
enabled_keys,
args.schedule_run,
chain_name=args.name or "today",
Expand Down
8 changes: 4 additions & 4 deletions src/config/bgi.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,16 +2,16 @@
import os
import shutil

from src.config.subscript import (
get_process_name,
resolve_script_path,
)
from src.utils import (
get_our_bgi_user_dir,
require_config_yml_path,
safe_path_join,
)
from src.utils_logger import setup_logging
from src.utils_sub_config import (
get_process_name,
resolve_script_path,
)
from src.utils_yaml import load_yaml

logger = logging.getLogger(__name__)
Expand Down
73 changes: 71 additions & 2 deletions src/config/dungeon_config.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,13 @@
import os
from typing import Any

from src.utils import get_root_dir, safe_path_join
from src.utils_yaml import load_yaml_optional
from src.config.set_config import get_dungeon_lists
from src.utils import (
get_root_dir,
get_weekly_list_yml_path_under_root,
safe_path_join,
)
from src.utils_yaml import load_yaml, load_yaml_optional

DungeonOptions = list[str]
SequenceOptionsMap = dict[str, list[tuple[str, Any]]]
Expand Down Expand Up @@ -98,3 +104,66 @@ def get_display_name(
if val == actual_value:
return display_name
return str(actual_value)


def _load_weekly_map() -> dict:
"""读取 weekly_list.yml(周常声明配置,进 git,必存在)。

结构:{script_name: [{"name", "dungeons"?}, ...]}。周常起始日(周几起)另存于
weekly_start.yml,不在本文件。
"""
weekly_list_path = get_weekly_list_yml_path_under_root()
assert os.path.exists(weekly_list_path), (
f"[dungeon_config] 周常声明配置缺失: {weekly_list_path}"
)
data = load_yaml(weekly_list_path)
# 空文件或内容非 dict 都是声明配置损坏,直接暴露而非静默当成「无声明」。
assert isinstance(data, dict), (
f"[dungeon_config] 周常声明配置应为 dict(空文件或格式错误): {weekly_list_path}"
)
return data


def get_weekly_map(script_name: str) -> list:
"""返回某脚本支持的周常声明清单(weekly_list.yml)。

每项:{"name", "dungeons"?}。dungeons 存在且有内容即表示该周常需选副本。
声明项若带 ``dungeons_source`` 标记,副本清单取自游戏脚本自身配置(运行期读取,
get_dungeon_lists),读不到时降级为 dungeons=[]。文件缺失或该脚本无声明时返回空列表。
"""
defs_map = _load_weekly_map()
if script_name not in defs_map:
return []
defs = list(defs_map[script_name])
for d in defs:
source = d.get("dungeons_source")
if source:
# 副本清单来自外部(如 M7A 的 instance_names.json),运行期读取,不再手动维护;
# 读不到则降级为无可选副本(has_dungeon=False)。
names = get_dungeon_lists(script_name, d["name"], source)
d["dungeons"] = names if names is not None else []
return defs


def get_dungeon_map() -> dict:
"""返回日常副本/序列配置映射(dungeon_list.yml)。

声明项若带 ``dungeons_source`` 标记,其二级序列取自游戏脚本自身配置(运行期读取,
get_dungeon_lists),读不到时降级为 sequences=[]。文件缺失时返回空 dict。
"""
data = load_dungeon_map()
for script_name, cfg in data.items():
if not isinstance(cfg, dict):
continue
for d in cfg.get("dungeons", []):
if not isinstance(d, dict):
continue
source = d.get("dungeons_source")
if source:
# 二级序列来自外部(如 ok-ef 的 world_map.json),运行期读取,不手动维护;
# 读不到则降级为无可选序列(show_seq=False)。
names = get_dungeon_lists(script_name, d["name"], source)
d["sequences"] = (
[{"display": n, "value": n} for n in names] if names else []
)
return data
6 changes: 3 additions & 3 deletions src/config/set_config.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

> 设计定位:`set_config` 是适配器,把异构 config 适配成统一调用;不是外观模式,外观整合职责归 `src/service/` 的 ChainService。

> script_name 为全链路内部唯一标识,由 `get_script_name(script)` 获取,与进程名 `get_process_name` 区分。exe 脚本的 script_name 即进程名 basename 去后缀,如 `ok-ww`;python/bat 脚本文件的 script_name 即 display_name。注册表、`dungeon_list.yml`、`gui_state.json`、`weekly_timeouts.yml` 的 key 全用 script_name,display_name 仅用于展示。config.yml 加载经 `check_script_name_uniqueness` 断言唯一。
> script_name 为全链路内部唯一标识,由 `get_script_name(script)` 获取,与进程名 `get_process_name` 区分。exe 脚本的 script_name 即进程名 basename 去后缀,如 `ok-ww`;python/bat 脚本文件的 script_name 即 display_name。注册表、`dungeon_list.yml`、`weekly_timeouts.yml` 的 key 全用 script_name,display_name 仅用于展示。config.yml 加载经 `check_script_name_uniqueness` 断言唯一。

## 架构

Expand Down Expand Up @@ -84,9 +84,9 @@
| 鸣潮 | 否 | `Which to Farm` | 是 | 经 `super()._update_task(config, dungeon_name, None)` 复用副本写入,再按 `_sequence_map` 写序列;模拟领域需映射值,凝素/无音区直接用 sequence |
| 原神 | 否 | `DomainName` | 否 | — |
| 终末地 | 否 | `体力本` | 否 | — |
| 崩铁 | 否 | — | 否 | 日常无需适配(set_dungeon 为 no-op),反读回退 gui_state |
| 崩铁 | 否 | — | 否 | 日常无需适配(set_dungeon 为 no-op,上游自身已支持),chip 呈现声明项 |
| 异环 | 否,覆盖做互斥切换 | `任务类型`(声明于 `_mode_specs`) | 是 | 完全自定义(不调 super):副本→模式经 `_dungeon_to_mode` 反查 `_mode_specs` 声明式映射(含 `task_field`+`seq_fields`);`DailyRoutineTask.json` 切换 `daily_anomaly`↔`daily_anomaly_hunter` 互斥启用,复用基类 `_load`/`_save`,仅路径 `_routine_config_rel_path` 不同 |
| 绝区零 | 是,空实现仅 print | — | — | 无需适配副本选择 |
| 绝区零 | 是,空实现仅 print | — | — | 无需适配副本选择(上游自身已支持) |
| 粥 | 是,完全自定义 | — | 是 | 操作 `TaskQueue` 禁用全部→启用剿灭+选定+土,不写二级序列 |

标准流程:不覆盖 set_dungeon,靠 `_task_key` 适配;需二级序列支持则覆盖 `_update_task`(在 `super()._update_task(config, dungeon_name, None)` 后补序列);需完全自定义如粥或无需适配如绝区零才覆盖 set_dungeon。
Expand Down
57 changes: 22 additions & 35 deletions src/config/set_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,10 @@
import os
from typing import Any

from src.config.subscript import (
get_config_path as _get_config_path_impl,
from src.utils_sub_config import (
get_sub_config_path as _get_config_path_impl,
)
from src.config.subscript import (
from src.utils_sub_config import (
load_config,
load_game_config,
load_template,
Expand Down Expand Up @@ -338,7 +338,7 @@ def _read_dungeon(self) -> tuple[str | None, str | int | None]:
(无 ``_task_key`` / 非 ``_task_key`` + ``_task_map``),可完全自行实现而不调 super。

仅「脚本未安装」与「用户未选择」的副本部分返回 None;config 损坏或字段值未知
属异常,直接 assert 暴露,不静默回退(否则会被 gui_state 兜底掩盖)。
属异常,直接 assert 暴露,不静默回退(否则会被日常副本的声明项回退掩盖)。

Returns:
``(副本中文名, 序列值)``;无 _task_key(无适应)/ 脚本未安装 / 未选择时
Expand Down Expand Up @@ -1030,7 +1030,7 @@ class NTEConfig(ScriptConfig):
"seq_fields": _anomaly_seq_key_map, # 副本 → 序号字段
},
"daily_anomaly_hunter": {
"task_field": None,
"task_field": None, # 追猎目标无任务类型通道
"seq_fields": {"追猎目标": "追猎目标"}, # 副本名即 boss 名
},
}
Expand Down Expand Up @@ -1177,16 +1177,10 @@ def set_dungeon(self, dungeon_name: str, sequence: str | int | None = None) -> N
def _read_dungeon(self) -> tuple[str | None, str | int | None]:
"""反读当前日常副本与二级序号(与 set_dungeon / _update_task 对称)。

当前玩法由 DailyRoutineTask.json 的 Routine Items 启用状态判定,经 ``_mode_specs``
查表解析当前模式(异象界域 / 追猎目标)。追猎目标无任务类型通道,set_dungeon
不写 daily_anomaly.任务类型(陈旧值),故必须优先用启用状态,不能直接读 任务类型。

两种模式数据落点不同(NTE 既有结构):追猎目标 boss 存于 routine 文件
DailyRoutineTask.json 的 daily_anomaly_hunter 段;异象界域副本名+序号存于
config 文件 DailyRoutineTaskConfigs.json 的 daily_anomaly 段。故按模式 id 分流读取。

NTE 无标准存储结构(不依赖 _task_key + _task_map 反转),完全自行实现。
脚本未安装(routine 缺失)返回 (None, None);routine/config 损坏属异常,assert 暴露。
当前玩法由 DailyRoutineTask.json 的 Routine Items 启用状态判定(非 任务类型 字段),
经 ``_mode_specs`` 查表解析模式;两种模式的副本/序列数据均落 config 文件
DailyRoutineTaskConfigs.json,routine 文件仅用于判定启用模式。脚本未安装或 routine
缺失返回 (None, None);routine/config 损坏属异常,assert 暴露。

Returns:
(副本中文名, 序号值);无启用玩法/未安装/未选择返回 (None, None)。
Expand All @@ -1213,33 +1207,26 @@ def _read_dungeon(self) -> tuple[str | None, str | int | None]:
)
if mode_id is None:
return None, None # 无启用玩法 → 未选择副本
if mode_id == "daily_anomaly_hunter":
# 追猎目标:boss 名存于 routine 文件 daily_anomaly_hunter 段。
# 该段/字段可能尚未落盘(用户在 NTE 自身 UI 启用追猎但未选 boss,
# 不经本工具 set_dungeon 写入):段或字段缺失按「已识别模式、未选 boss」
# 处理为 None,而非断言——与读路径「容忍未配置」一致;结构性损坏(段
# 类型非 dict)已在下方 isinstance 断言覆盖。
section = routine.get("daily_anomaly_hunter", {})
assert isinstance(section, dict), (
f"[set_config][{self.display_name}] daily_anomaly_hunter 段必须是 dict"
)
boss = section.get("追猎目标")
return "追猎目标", boss if boss not in (None, "") else None
# 异象界域:副本名+序号存于 config 文件 daily_anomaly 段。
# 两种模式的副本/序列数据都落在 config 文件(与写入侧 _daily_section_dict
# 对称),故在此一次性加载;routine 文件只用于判定启用的模式。
config = self._load(allow_missing=True)
if config is None:
return None, None # 脚本未安装/未配置
assert isinstance(config, dict), (
f"[set_config][{self.display_name}] DailyTask config.yaml 必须是 dict"
f"[set_config][{self.display_name}] DailyRoutineTaskConfigs.json 必须是 dict"
)
# daily_anomaly 段缺失按「未选副本」处理(读路径容忍未配置;段类型非 dict
# 的结构性损坏由下方 isinstance 断言覆盖,字段级缺失不视为损坏)。
section = config.get("daily_anomaly", {})
# 段名 = 模式 id;段缺失按「未配置」处理(容忍未落盘),段类型非 dict 属损坏、assert。
section = config.get(mode_id, {})
assert isinstance(section, dict), (
f"[set_config][{self.display_name}] daily_anomaly 段必须是 dict"
f"[set_config][{self.display_name}] {mode_id} 段必须是 dict"
)
dungeon = section.get("任务类型") # 段缺失时为 None(未选副本)
if dungeon in (None, ""): # 值为空串同样视为未选具体副本
if mode_id == "daily_anomaly_hunter":
# 追猎目标:副本名即 boss 字段名(段缺失/空串按未选 boss,容忍未落盘)。
boss = section.get("追猎目标")
return "追猎目标", boss if boss else None
# 异象界域:副本名在 任务类型 字段,序号经 _anomaly_seq_key_map 反查。
dungeon = section.get("任务类型")
if dungeon in (None, ""): # 段缺失/字段为空串均视为未选具体副本
return None, None
key = self._anomaly_seq_key_map.get(dungeon)
sequence = section.get(key) if key else None
Expand Down
Loading
Loading