dwarf-fortress-annals/dfannals/cli.py

# pyright: reportMissingImports=false, reportAttributeAccessIssue=false
# 说明：LSP 的 Python 环境看不到本项目包（未安装进它的 site-packages），
# 会把「包内互相 import」误报为缺失模块。运行时导入已由执行验证通过。

"""命令行入口。

    python3 -m dfannals.cli status                 查看当前进度
    python3 -m dfannals.cli index                  解析 exports → 年表 + 人物索引
    python3 -m dfannals.cli plan                   显示切章方案（不调用模型）
    python3 -m dfannals.cli next                   生成下一章 → 校验 → 推送
    python3 -m dfannals.cli run                    一条命令跑完：索引 + 下一章 + 推送
"""
from __future__ import annotations

import argparse
import sys
from pathlib import Path

from dfannals import chronicle, factcheck, legends, publish, timeline
from dfannals import slice as chapter_slice
from dfannals.config import EXPORT_DIR, GITEA_WEB, chapter_dir, ensure_dirs, volume_dir


def find_exports(export_dir: Path) -> list[Path]:
    """找出 legends 导出文件。原生 legends.xml 与 DFHack 的 legends_plus.xml 都要。"""
    if not export_dir.is_dir():
        return []
    files = sorted(export_dir.glob("*.xml"))
    base = [f for f in files if not f.name.endswith("-legends_plus.xml") and "legends_plus" not in f.name]
    plus = [f for f in files if "legends_plus" in f.name]
    return base + plus


def load_world(export_dir: Path) -> legends.World:
    paths = find_exports(export_dir)
    if not paths:
        raise SystemExit(
            f"在 {export_dir} 里没找到 legends XML。\n"
            "请先在游戏里导出（DF 的 Export XML + DFHack 的 exportlegends），"
            "再把文件复制进这个目录。"
        )
    return legends.load(paths)


def _current_volume(state: chronicle.State, world: legends.World) -> int:
    """世界换了就开新卷。"""
    if state.world != world.name:
        if state.world:
            state.volume += 1
            state.done = []
        state.world = world.name
    return state.volume


def cmd_status(args: argparse.Namespace) -> int:
    state = chronicle.State.load()
    if not state.world:
        print("尚未开始。先运行 index。")
        return 0
    vdir = volume_dir(state.volume)
    print(f"世界：{state.world}")
    print(f"卷号：第 {state.volume} 卷")
    print(f"已完成章节：{len(state.done)}")
    for key in state.done:
        print(f"  · {key}")
    print(f"成品目录：{vdir}")
    return 0


def cmd_index(args: argparse.Namespace) -> int:
    ensure_dirs()
    world = load_world(EXPORT_DIR)
    state = chronicle.State.load()
    volume = _current_volume(state, world)
    chapters = chapter_slice.plan(world)

    vdir = volume_dir(volume)
    timeline.write_outputs(world, vdir)
    state.save()

    print(legends.describe(world))
    print()
    print(chapter_slice.overview(world, chapters))
    print()
    print(f"年表与人物索引已写入 {vdir}")

    if not args.no_push:
        publish.ensure_repo()
        head = publish.commit_and_push(
            f"第 {volume} 卷索引：{world.name}（{len(chapters)} 章规划）",
            [vdir / "timeline.md", vdir / "figures.md"],
        )
        print(f"已推送索引：{head or '(无改动)'} → {GITEA_WEB}")
    return 0


def cmd_plan(args: argparse.Namespace) -> int:
    world = load_world(EXPORT_DIR)
    chapters = chapter_slice.plan(world)
    print(chapter_slice.overview(world, chapters))
    return 0


def cmd_next(args: argparse.Namespace) -> int:
    ensure_dirs()
    world = load_world(EXPORT_DIR)
    state = chronicle.State.load()
    volume = _current_volume(state, world)
    chapters = chapter_slice.plan(world)
    cdir = chapter_dir(volume)

    todo = [c for c in chapters if c.key not in state.done]
    if not todo:
        print("本卷史料已全部写完。需要新世界才能开新卷。")
        return 0

    chapter = todo[0]
    print(f"正在写第 {chapter.index} 章：{chapter.span}（{len(chapter.events)} 条事件）")
    tail = chronicle.previous_tail(cdir, chapters, chapter.key)
    body, suspicions, reply = chronicle.generate(world, chapter, tail=tail)
    length = chronicle.cjk_length(body)
    print(f"生成完成：{length} 字" + (f" | 总 token {reply.total_tokens}" if reply else ""))

    print(factcheck.report(suspicions))

    if args.dry_run:
        print("\n--- 试运行，不写文件 ---\n")
        print(body[:1200])
        return 0

    path = chronicle.write_chapter(cdir, chapter, body)
    state.mark_done(chapter.key)
    state.save()
    print(f"已写入 {path}")

    publish.ensure_repo()
    head = publish.commit_and_push(
        f"第 {volume} 卷第 {chapter.index} 章：{chapter.span}"
        + (f"（专名校验可疑 {len(suspicions)} 处）" if suspicions else "")
    )
    print(f"已推送：{head or '(无改动)'} → {GITEA_WEB}")
    return 0


def cmd_run(args: argparse.Namespace) -> int:
    rc = cmd_index(args)
    if rc:
        return rc
    return cmd_next(args)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(prog="dfannals", description="矮人要塞编年史生成管道")
    sub = parser.add_subparsers(dest="cmd", required=True)

    p = sub.add_parser("status", help="查看当前进度")
    p.set_defaults(func=cmd_status)

    p = sub.add_parser("index", help="解析 exports → 年表 + 人物索引")
    p.add_argument("--no-push", action="store_true", help="只写本地，不推送")
    p.set_defaults(func=cmd_index)

    p = sub.add_parser("plan", help="显示切章方案")
    p.set_defaults(func=cmd_plan)

    p = sub.add_parser("next", help="生成下一章")
    p.add_argument("--dry-run", action="store_true", help="只生成不落盘、不推送")
    p.set_defaults(func=cmd_next)

    p = sub.add_parser("run", help="一条命令：索引 + 下一章 + 推送")
    p.add_argument("--no-push", action="store_true")
    p.add_argument("--dry-run", action="store_true")
    p.set_defaults(func=cmd_run)

    args = parser.parse_args(argv)
    ensure_dirs()
    return args.func(args)


if __name__ == "__main__":
    sys.exit(main())