dwarf-fortress-annals/dfannals/episodes.py

"""因果链切集：把一条人物线索切成"一集一个完整回合"。

参数来自访谈决策：

- 相邻事件间隔 >2 年即断开（比推荐值更紧，节奏更快）
- 短链只合并，不设线索总量下限
- 链路过长按字数预算拆上下集
- 每集正文预算落在 800–1500 字

字数预算用「事件条数 × 每条展开字数」估算。系数 110 是按此前实测校准的：
12 条史料事件生成的中文正文约 1100–1400 字。
"""
from __future__ import annotations

import math
from dataclasses import dataclass, field

from dfannals.legends import Event, World
from dfannals.threads import Thread

# 相邻事件间隔超过它即断开
GAP_YEARS = 2
# 每条史料在正文里大致展开的字数。实测区间约 85–115（同样 12 条事件，
# 两次生成分别得到 1026 字与 1388 字），取中值偏保守，避免把集拆得过碎。
CHARS_PER_EVENT = 95
MIN_CHARS = 800               # 每集预算下限
MAX_CHARS = 1500              # 每集预算上限


@dataclass
class Episode:
    index: int
    start_year: int
    end_year: int
    events: list[Event] = field(default_factory=list)
    part: int = 1
    parts: int = 1

    @property
    def key(self) -> str:
        base = f"{self.index:03d}-{self.start_year}-{self.end_year}"
        return base if self.parts == 1 else f"{base}-p{self.part}"

    @property
    def span(self) -> str:
        if self.start_year == self.end_year:
            return f"{self.start_year} 年"
        return f"{self.start_year}–{self.end_year} 年"

    @property
    def estimated_chars(self) -> int:
        return len(self.events) * CHARS_PER_EVENT

    def in_budget(self) -> bool:
        return MIN_CHARS <= self.estimated_chars <= MAX_CHARS or self.parts > 1


def _min_events() -> int:
    return max(1, math.ceil(MIN_CHARS / CHARS_PER_EVENT))


def _max_events() -> int:
    return max(1, MAX_CHARS // CHARS_PER_EVENT)


def _split_chains(events: list[Event], gap_years: int) -> list[list[Event]]:
    """按时间间隔把事件串切成因果链。"""
    chains: list[list[Event]] = []
    current: list[Event] = []
    prev_year: int | None = None
    for e in events:
        if current and prev_year is not None and e.year - prev_year > gap_years:
            chains.append(current)
            current = []
        current.append(e)
        prev_year = e.year
    if current:
        chains.append(current)
    return chains


def _merge_short(chains: list[list[Event]], min_events: int) -> list[list[Event]]:
    """短链只合并：攒够预算就收一集，尾部不足的并入上一集。"""
    merged: list[list[Event]] = []
    buffer: list[Event] = []
    for chain in chains:
        buffer.extend(chain)
        if len(buffer) >= min_events:
            merged.append(buffer)
            buffer = []
    if buffer:
        if merged:
            merged[-1].extend(buffer)
        else:
            merged = [buffer]
    return merged


def _split_long(chain: list[Event]) -> list[list[Event]]:
    """过长链路按预算拆成若干段（对应正文的上/下集）。

    注意不能均分：14 条事件均分成 7+7 会得到两个 770 字的碎片，
    两端都跌破 800 字下限。所以先取最小段数，再确保每段不低于下限。
    """
    n = len(chain)
    max_events = _max_events()
    min_events = _min_events()
    if n <= max_events:
        return [chain]
    parts = math.ceil(n / max_events)
    while parts > 1 and math.ceil(n / parts) < min_events:
        parts -= 1
    size = math.ceil(n / parts)
    return [chain[i:i + size] for i in range(0, n, size)]


def plan_episodes(
    thread: Thread,
    gap_years: int = GAP_YEARS,
    min_chars: int = MIN_CHARS,
    max_chars: int = MAX_CHARS,
) -> list[Episode]:
    """给定主角组，切出剧集序列。"""
    if not thread.events:
        return []

    chains = _split_chains(thread.events, gap_years)
    merged = _merge_short(chains, _min_events())

    episodes: list[Episode] = []
    for chain in merged:
        segments = _split_long(chain)
        for i, segment in enumerate(segments, 1):
            episodes.append(
                Episode(
                    index=len(episodes) + 1,
                    start_year=segment[0].year,
                    end_year=segment[-1].year,
                    events=segment,
                    part=i,
                    parts=len(segments),
                )
            )
    return episodes


def cast_of(world: World, thread: Thread, episode: Episode) -> tuple[list[int], list[int]]:
    """返回 (本集出场的主角, 本集出现的对手/外人)。"""
    counter: dict[int, int] = {}
    for e in episode.events:
        for fid in e.figure_ids():
            counter[fid] = counter.get(fid, 0) + 1
    members = set(thread.members)
    heros = sorted((f for f in counter if f in members), key=lambda f: -counter[f])
    others = sorted((f for f in counter if f not in members), key=lambda f: -counter[f])
    return heros, others


def render_plan(world: World, thread: Thread, episodes: list[Episode], top_others: int = 3) -> str:
    """剧集清单：集号、年份范围、事件条数、主角与对手。"""
    lines = [
        f"线索主角：{'、'.join(world.figure_name(m) for m in thread.members)}",
        f"线索跨度：{thread.label} | 事件 {len(thread.events)} 条 | 规划 {len(episodes)} 集",
        "",
    ]
    for ep in episodes:
        heros, others = cast_of(world, thread, ep)
        hero_names = "、".join(world.figure_name(h) for h in heros[:4]) or "（本集无主角出场）"
        other_names = "、".join(world.figure_name(o) for o in others[:top_others])
        budget = f"{ep.estimated_chars} 字估算"
        mark = "" if MIN_CHARS <= ep.estimated_chars <= MAX_CHARS else " ⚠ 超出预算"
        lines.append(
            f"  第 {ep.index:2d} 集（{ep.span}）| {len(ep.events):2d} 条事件 | {budget}{mark}"
            + (f" | 上/下：{ep.part}/{ep.parts}" if ep.parts > 1 else "")
        )
        lines.append(f"        主角：{hero_names}")
        if other_names:
            lines.append(f"        对手/外人：{other_names}")
    return "\n".join(lines)


def budget_report(episodes: list[Episode]) -> str:
    if not episodes:
        return "无剧集"
    inside = sum(1 for e in episodes if MIN_CHARS <= e.estimated_chars <= MAX_CHARS)
    sizes = [len(e.events) for e in episodes]
    return (
        f"共 {len(episodes)} 集；事件条数 {min(sizes)}–{max(sizes)}；"
        f"估算字数 {min(e.estimated_chars for e in episodes)}–{max(e.estimated_chars for e in episodes)}；"
        f"落在 {MIN_CHARS}–{MAX_CHARS} 预算内的 {inside}/{len(episodes)} 集"
        "（估算按每条约 95 字，真实字数在生成时会再校正）"
    )