feat(pipeline): 会话上下文/任务结果/打勾落盘/中文名生成

pipeline_bot/pipeline/__init__.py

@@ -0,0 +1 @@
+"""pipeline — 新手九任务串行流水线(共用一次登录会话)。"""

pipeline_bot/pipeline/base.py

@@ -0,0 +1,12 @@
+"""任务基协议: 每任务文件暴露 run_task(ctx) -> TaskResult。"""
+from pipeline.ctx import PipelineContext, TaskResult
+
+__all__ = ["TASK_TITLE", "PipelineContext", "TaskResult", "run_task"]
+
+TASK_TITLE: str = "(未命名任务)"
+
+
+def run_task(ctx: PipelineContext) -> TaskResult:  # pragma: no cover - 接口文档
+    """任务文件必须实现: 执行任务并返回 TaskResult。"""
+    _ = ctx
+    raise NotImplementedError("任务文件必须实现 run_task(ctx)")

pipeline_bot/pipeline/ctx.py

@@ -0,0 +1,107 @@
+"""PipelineContext — 一次登录会话贯穿九任务; TaskResult 与打勾落盘。"""
+from __future__ import annotations
+
+import logging
+import time
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Final
+
+import requests
+
+from pbot.bot.runner import pump as _pump
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from pbot.transport.gateway import GatewayConnection
+    from pbot.types import ZoneId
+    from pbot.world.mirror import WorldMirror
+    from pbot.world.update import WorldUpdater
+
+_FMT: Final[str] = "%Y-%m-%d %H:%M:%S"
+
+
+@dataclass
+class TaskResult:
+    """单任务结果。"""
+
+    ok: bool
+    detail: str = ""
+
+    @staticmethod
+    def done(detail: str = "") -> TaskResult:
+        """成功结果。"""
+        return TaskResult(ok=True, detail=detail)
+
+    @staticmethod
+    def fail(detail: str) -> TaskResult:
+        """失败结果。"""
+        return TaskResult(ok=False, detail=detail)
+
+
+class ChecklistWriter:
+    """打勾文件实时写: 每任务一行, 完成即勾, 失败即叉, 支持更新。"""
+
+    def __init__(self, path: Path, zone_label: str, role: str) -> None:
+        """首写 header, 建立任务清单骨架。"""
+        self._path: Path = path
+        self._role: str = role
+        self._marks: dict[str, tuple[bool, str]] = {}
+        self._prefix: list[str] = [
+            f"# 传奇流水线进度 — {zone_label} — {role}\n",
+            "\n",
+            f"- 启动: {time.strftime(_FMT)}\n",
+            "\n",
+        ]
+        path.parent.mkdir(parents=True, exist_ok=True)
+        self._rewrite()
+
+    def _rewrite(self) -> None:
+        lines = [*self._prefix, "## 任务清单\n\n"]
+        for t, (o, d) in self._marks.items():
+            mark = "☑" if o else "☒"
+            lines.append(f"- {time.strftime(_FMT)} {mark} **{t}**"
+                         + (f" — {d}" if d else "") + "\n")
+        tmp = self._path.with_suffix(".tmp")
+        _ = tmp.write_text("".join(lines), encoding="utf-8")
+        _ = tmp.replace(self._path)
+
+    def note(self, text: str) -> None:
+        """追加一条备注行(重写时保留)。"""
+        line = f"> {time.strftime(_FMT)} {text}\n"
+        self._prefix.append(line)
+        self._rewrite()
+
+    def mark(self, title: str, ok: bool, detail: str = "") -> None:
+        """勾/叉一个任务并实时重写文件, 同名任务更新而非追加。"""
+        self._marks[title] = (ok, detail)
+        self._rewrite()
+        mark = "✓" if ok else "✗"
+        print(  # noqa: T201 — 流水线控制台实时进度为硬需求
+            f"[{mark}] {title}" + (f" — {detail}" if detail else ""), flush=True)
+
+
+@dataclass
+class PipelineContext:
+    """共用会话上下文: 登录一次, 九任务复用 conn/world/sid。"""
+
+    user: str
+    password: str
+    zone: ZoneId
+    zone_label: str
+    role_name: str
+    woma_minutes: float
+    conn: GatewayConnection
+    world: WorldMirror
+    updater: WorldUpdater
+    sid: int
+    stop: list[bool]
+    checklist: ChecklistWriter
+    http: requests.Session = field(default_factory=requests.Session)
+    jwt: str = ""
+    uid: int = 0
+    log: logging.Logger = field(default_factory=lambda: logging.getLogger("pipeline"))
+
+    def pump(self, window: float = 0.5) -> None:
+        """收帧窗口(转发 runner.pump)。"""
+        _ = _pump(self.conn, self.world, self.updater, window)

pipeline_bot/pipeline/naming.py

@@ -0,0 +1,39 @@
+"""随机中文角色名 + 文件名清洗。"""
+import random
+import re
+
+# 行 1-4: 单姓池(每串取一字); 行 5: 复姓(整串使用)
+_SURNAMES: tuple[str, ...] = (
+    "赵钱孙李周吴郑王冯陈褚卫蒋沈韩杨朱秦许何吕施张孔曹严华金魏陶姜",
+    "谢邹柏水窦章云苏潘葛范彭郎鲁韦昌马苗凤花方俞任袁柳鲍史唐",
+    "萧叶阎余潘杜夏宋钟汪田任姜范方石姚谭廖邹郝孔崔秦邱侯邵龚",
+    "林卓唐温贾雷贺汤罗毕安于时傅皮卞齐康伍余元卜顾孟平黄穆萧尹",
+    "欧阳司马上官诸葛夏侯皇甫尉迟公羊澹台公冶宗政濮阳淳于单于",
+)
+_GIVEN_CHARS: str = (
+    "风云雷电霜雪星月山河江海林泉石松竹梅兰菊荷峰岳城铁剑刀"
+    "狂傲凌飞跃震天啸苍穹宇辰晓旭尧舜禹轩辕擎苍痕无极道玄清真"
+    "长庚启明北斗天涯孤鸿惊鸿流云追风逐月听雨眠霜寒影夜白轻语"
+    "凌波青丝白首丹心碧血铁骨侠客行止言笑怒悲欢离合恩怨情仇录"
+    "一乙二十丁厂七卜人入八九几儿了力乃刀又三于干亏士工土才寸"
+)
+_GIVEN: tuple[str, ...] = tuple(c for c in _GIVEN_CHARS if "\u4e00" <= c <= "\u9fff")
+
+
+def random_cn_name() -> str:
+    """纯中文随机名: 姓(1-2字) + 名(1-2字), 全字 CJK 检查兜底。"""
+    surname = random.choice(_SURNAMES)  # noqa: S311
+    if _SURNAMES.index(surname) == 4 and random.random() < 0.1:  # noqa: S311
+        i = random.randrange(0, len(surname), 2)  # noqa: S311
+        surname = surname[i:i + 2]
+    else:
+        surname = random.choice(surname)  # noqa: S311
+    given_len = random.choice((1, 2))  # noqa: S311
+    given = "".join(random.choice(_GIVEN) for _ in range(given_len))  # noqa: S311
+    name = "".join(c for c in surname + given if "\u4e00" <= c <= "\u9fff")
+    return name[:4]
+
+
+def safe_filename(s: str) -> str:
+    """路径非法字符 → 下划线。"""
+    return re.sub(r'[\\/:*?\"<>|\s]+', "_", s).strip("_") or "unnamed"

pipeline_bot/pyproject.toml

@@ -102,4 +102,5 @@ indent-style = "space"
 [tool.pytest.ini_options]
 minversion = "8.0"
 testpaths = ["tests"]
+pythonpath = ["."]
 addopts = ["-ra", "--strict-config", "--strict-markers"]

pipeline_bot/tests/test_pipeline_units.py

@@ -28,3 +28,43 @@ def test_sc52_level_zero_ignored() -> None:
     u = WorldUpdater()
     u.apply(w, parse_frame(_sc52_frame(0)))
     assert w.me.level == 0  # 0 不产生误报
+
+
+# ---- naming ----
+from pathlib import Path  # noqa: E402
+
+from pipeline.ctx import ChecklistWriter  # noqa: E402
+from pipeline.naming import random_cn_name, safe_filename  # noqa: E402
+
+
+def test_random_cn_name_all_chinese() -> None:
+    for _ in range(50):
+        n = random_cn_name()
+        assert 2 <= len(n) <= 4
+        assert all("\u4e00" <= c <= "\u9fff" for c in n)
+
+
+def test_safe_filename() -> None:
+    assert safe_filename("玖扬复古176.38区/新角色") == "玖扬复古176.38区_新角色"
+    assert safe_filename('a\\b:*?"<>|') == "a_b"
+    assert safe_filename("   ") == "unnamed"
+
+
+# ---- checklist ----
+def test_checklist_writer_marks(tmp_path: Path) -> None:
+    w = ChecklistWriter(tmp_path / "进度.md", "玖扬复古176.38区", "凌风")
+    w.mark("T1 喝药保护", True, "state=on")
+    w.mark("T2 比奇杀怪", False, "超时")
+    txt = (tmp_path / "进度.md").read_text(encoding="utf-8")
+    assert "☑" in txt and "☒" in txt and "凌风" in txt
+    # header 与 note 行在 mark 重写后仍保留
+    w.note("登录成功")
+    w.mark("T3 新增项", True)
+    txt = (tmp_path / "进度.md").read_text(encoding="utf-8")
+    assert "# 传奇流水线进度" in txt
+    assert "> " in txt and "登录成功" in txt
+    # 二次 mark 同名任务 = 更新而非重复追加
+    w.mark("T2 比奇杀怪", True, "20/20")
+    txt = (tmp_path / "进度.md").read_text(encoding="utf-8")
+    assert txt.count("T2 比奇杀怪") == 1
+    assert txt.count("凌风") >= 1