Chen Yi
e0073e90bd
主角线：Guspu Frillyknots / Ral Fastenhatchets / Alath Blottedmine 由线索挖掘器自动选定，按因果链切集，经 5 维度自评闸门（≥12 分）后入库。
288 lines
10 KiB
Python
288 lines
10 KiB
Python
"""人物线索挖掘：从史料里找出「关系密切、有戏」的 3–6 人小圈子。
|
||
|
||
设计依据全部来自对 Mon Sagus 真实数据（403853 条事件）的实测：
|
||
|
||
1. **不能用裸互动次数排序。** 实测排名第一的簇 49 次互动里有 35 次是
|
||
``hf relationship denied``——反复请求建立关系、反复被拒的循环，是统计噪声。
|
||
因此整类剔除（访谈已定）。
|
||
2. **同一对同一类型反复出现也不是故事。** 修掉上一条之后，候选又变成
|
||
``对战×27`` 这种「同一对人反复互殴」的机械循环。因此在计数时对
|
||
(人物对, 事件类型) 做封顶，避免刷量取胜——这也正是访谈定的
|
||
「转折/冲突多样性为主、互动次数为次」。
|
||
3. **主线必须是"人"。** 种族分布实测：ELF/GOBLIN/DWARF/HUMAN/KOBOLD 五族加
|
||
``*_MAN`` 人形族占全部历史人物的 96.5%，其余 400 多种是夜行怪、野兽、
|
||
泰坦、实验体。所以这里用白名单，而不是越列越长的黑名单。
|
||
4. 强边阈值 ≥5 次反复互动；簇规模 3–6 人；跨度 ≥30 年。
|
||
实测：阈值提到 8 会一条候选都不剩，5 是正确档位。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from collections import Counter, defaultdict
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
|
||
from dfannals.legends import Event, World
|
||
|
||
# 事件类型 → (权重, 中文标签, 是否算转折点)
|
||
SIGNALS: dict[str, tuple[int, str, bool]] = {
|
||
"hf simple battle event": (1, "对战", False),
|
||
"add hf hf link": (1, "结缘", False),
|
||
"hfs formed reputation relationship": (1, "结缘", False),
|
||
"competition": (1, "竞争", False),
|
||
"hf wounded": (2, "搏杀", True),
|
||
"hf confronted": (2, "对峙", True),
|
||
"hf interrogated": (2, "审讯", True),
|
||
"failed intrigue corruption": (2, "阴谋", True),
|
||
"remove hf hf link": (3, "反目", True),
|
||
"hf abducted": (3, "绑架", True),
|
||
"hf convicted": (3, "定罪", True),
|
||
"entity persecuted": (3, "迫害", True),
|
||
"hf enslaved": (3, "奴役", True),
|
||
"hf ransomed": (3, "贖金", True),
|
||
"entity overthrown": (3, "推翻", True),
|
||
"failed frame attempt": (3, "构陷", True),
|
||
}
|
||
|
||
# 明确剔除：重复性请求，实测会刷满排序（访谈已定）
|
||
EXCLUDED_SIGNALS = ("hf relationship denied",)
|
||
|
||
# 文明种族白名单：实测覆盖 96.5% 的历史人物
|
||
CIVILIZED_RACES = frozenset({"DWARF", "ELF", "HUMAN", "GOBLIN", "KOBOLD"})
|
||
|
||
DEFAULT_MIN_INTERACTIONS = 5 # 强边阈值：≥5 次反复互动
|
||
DEFAULT_SIZE_RANGE = (3, 6)
|
||
DEFAULT_MIN_SPAN = 30 # 跨度 ≥30 年才撑得起连载
|
||
PAIR_TYPE_CAP = 4 # 同一对、同一类型最多计 4 次
|
||
MIN_SIGNAL_KINDS = 2 # 至少两种不同信号，否则只是单一类型的重复
|
||
INTERACTION_CAP = 40 # 互动总量封顶，防止刷量取胜
|
||
|
||
|
||
@dataclass
|
||
class Edge:
|
||
a: int
|
||
b: int
|
||
raw: int = 0
|
||
types: Counter = field(default_factory=Counter)
|
||
first_year: int = 0
|
||
last_year: int = 0
|
||
|
||
@property
|
||
def effective(self) -> int:
|
||
"""封顶后的有效互动次数。"""
|
||
return sum(min(n, PAIR_TYPE_CAP) for n in self.types.values())
|
||
|
||
@property
|
||
def turns(self) -> int:
|
||
"""封顶后的转折点次数。"""
|
||
return sum(min(n, PAIR_TYPE_CAP) for t, n in self.types.items() if SIGNALS[t][2])
|
||
|
||
|
||
@dataclass
|
||
class Thread:
|
||
members: list[int]
|
||
interactions: int
|
||
turns: int
|
||
span: int
|
||
first_year: int
|
||
last_year: int
|
||
types: Counter
|
||
races: Counter
|
||
non_person_races: list[str]
|
||
events: list[Event]
|
||
|
||
@property
|
||
def kinds(self) -> int:
|
||
return len(self.types)
|
||
|
||
@property
|
||
def score(self) -> int:
|
||
"""转折为主、多样性次之、互动次数封顶计入（访谈定的排序原则）。"""
|
||
return self.turns * 10 + self.kinds * 6 + min(self.interactions, INTERACTION_CAP)
|
||
|
||
@property
|
||
def label(self) -> str:
|
||
return f"{self.first_year}–{self.last_year}（{self.span} 年）"
|
||
|
||
def signal_line(self) -> str:
|
||
return "、".join(f"{SIGNALS[t][1]}×{n}" for t, n in self.types.most_common() if t in SIGNALS)
|
||
|
||
|
||
def is_person(race: str) -> bool:
|
||
"""是否属于"可作为主角的人"：文明种族或人形族（*_MAN）。"""
|
||
r = (race or "").upper()
|
||
return r in CIVILIZED_RACES or r.endswith("_MAN")
|
||
|
||
|
||
def build_edges(world: World) -> dict[tuple[int, int], Edge]:
|
||
"""按有叙事含义的信号建立人物之间的边。"""
|
||
edges: dict[tuple[int, int], Edge] = {}
|
||
for e in world.events:
|
||
if e.type not in SIGNALS:
|
||
continue
|
||
people = e.figure_ids()
|
||
if len(people) < 2:
|
||
continue
|
||
for i in range(len(people)):
|
||
for j in range(i + 1, len(people)):
|
||
key = (min(people[i], people[j]), max(people[i], people[j]))
|
||
edge = edges.get(key)
|
||
if edge is None:
|
||
edge = Edge(a=key[0], b=key[1], first_year=e.year, last_year=e.year)
|
||
edges[key] = edge
|
||
edge.raw += 1
|
||
edge.types[e.type] += 1
|
||
edge.first_year = min(edge.first_year, e.year)
|
||
edge.last_year = max(edge.last_year, e.year)
|
||
return edges
|
||
|
||
|
||
def find_threads(
|
||
world: World,
|
||
min_interactions: int = DEFAULT_MIN_INTERACTIONS,
|
||
size_range: tuple[int, int] = DEFAULT_SIZE_RANGE,
|
||
min_span: int = DEFAULT_MIN_SPAN,
|
||
require_people: bool = True,
|
||
min_kinds: int = MIN_SIGNAL_KINDS,
|
||
limit: int = 0,
|
||
) -> list[Thread]:
|
||
"""返回按剧情张力排序的候选线索。"""
|
||
edges = build_edges(world)
|
||
strong = {k: v for k, v in edges.items() if v.effective >= min_interactions}
|
||
|
||
parent: dict[int, int] = {}
|
||
|
||
def find(x: int) -> int:
|
||
parent.setdefault(x, x)
|
||
while parent[x] != x:
|
||
parent[x] = parent[parent[x]]
|
||
x = parent[x]
|
||
return x
|
||
|
||
for (a, b) in strong:
|
||
ra, rb = find(a), find(b)
|
||
if ra != rb:
|
||
parent[ra] = rb
|
||
|
||
members_of: dict[int, set[int]] = defaultdict(set)
|
||
for x in parent:
|
||
members_of[find(x)].add(x)
|
||
|
||
events_by_pair: dict[tuple[int, int], list[Event]] = defaultdict(list)
|
||
for e in world.events:
|
||
if e.type not in SIGNALS:
|
||
continue
|
||
people = e.figure_ids()
|
||
if len(people) < 2:
|
||
continue
|
||
for i in range(len(people)):
|
||
for j in range(i + 1, len(people)):
|
||
events_by_pair[(min(people[i], people[j]), max(people[i], people[j]))].append(e)
|
||
|
||
lo_size, hi_size = size_range
|
||
threads: list[Thread] = []
|
||
for members in members_of.values():
|
||
if not (lo_size <= len(members) <= hi_size):
|
||
continue
|
||
|
||
races = Counter(world.figures[m].race for m in members if m in world.figures)
|
||
if require_people and not all(is_person(r) for r in races):
|
||
continue
|
||
|
||
inner = {k: v for k, v in strong.items() if k[0] in members and k[1] in members}
|
||
if not inner:
|
||
continue
|
||
|
||
types: Counter = Counter()
|
||
for v in inner.values():
|
||
types.update({t: min(n, PAIR_TYPE_CAP) for t, n in v.types.items()})
|
||
if len(types) < min_kinds:
|
||
continue
|
||
|
||
first = min(v.first_year for v in inner.values())
|
||
last = max(v.last_year for v in inner.values())
|
||
if last - first < min_span:
|
||
continue
|
||
|
||
evs: list[Event] = []
|
||
seen_ids: set[int] = set()
|
||
for k in inner:
|
||
for e in events_by_pair.get(k, []):
|
||
if e.id not in seen_ids:
|
||
seen_ids.add(e.id)
|
||
evs.append(e)
|
||
evs.sort(key=lambda e: (e.year, e.seconds72, e.id))
|
||
|
||
threads.append(
|
||
Thread(
|
||
members=sorted(members),
|
||
interactions=sum(v.effective for v in inner.values()),
|
||
turns=sum(v.turns for v in inner.values()),
|
||
span=last - first,
|
||
first_year=first,
|
||
last_year=last,
|
||
types=types,
|
||
races=races,
|
||
non_person_races=sorted(r for r in races if not is_person(r)),
|
||
events=evs,
|
||
)
|
||
)
|
||
|
||
threads.sort(key=lambda t: (-t.score, -t.span))
|
||
return threads[:limit] if limit else threads
|
||
|
||
|
||
def render_report(world: World, threads: list[Thread], per_thread: int = 3) -> str:
|
||
"""给人看的候选明细。"""
|
||
lines = [
|
||
f"候选线索 {len(threads)} 条（转折为主、多样性次之；同一对同类事件已封顶）",
|
||
"筛选：强边 ≥5 次互动、3–6 人、跨度 ≥30 年、含巨兽的簇已剔除",
|
||
"",
|
||
]
|
||
for i, t in enumerate(threads, 1):
|
||
races = "、".join(f"{r}×{n}" for r, n in t.races.most_common())
|
||
monsters = "、".join(t.non_person_races) if t.non_person_races else "无"
|
||
lines.append(
|
||
f"{i:2d}. 得分 {t.score:4d} | {len(t.members)} 人 | 有效互动 {t.interactions:3d} | "
|
||
f"转折 {t.turns:2d} | 信号 {t.kinds} 种 | 跨度 {t.label} | 巨兽：{monsters}"
|
||
)
|
||
lines.append(f" 种族：{races}")
|
||
lines.append(f" 信号：{t.signal_line()}")
|
||
lines.append(f" 成员：{'、'.join(world.figure_name(m) for m in t.members)}")
|
||
for e in t.events[:per_thread]:
|
||
lines.append(f" · {e.year}年 {e.type} " + " · ".join(world.render_refs(e)))
|
||
lines.append("")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def to_json(world: World, threads: list[Thread]) -> dict:
|
||
return {
|
||
"world": world.name,
|
||
"candidates": [
|
||
{
|
||
"rank": i,
|
||
"score": t.score,
|
||
"members": [
|
||
{"id": m, "name": world.figure_name(m),
|
||
"race": world.figures[m].race if m in world.figures else ""}
|
||
for m in t.members
|
||
],
|
||
"interactions": t.interactions,
|
||
"turns": t.turns,
|
||
"kinds": t.kinds,
|
||
"span": t.span,
|
||
"first_year": t.first_year,
|
||
"last_year": t.last_year,
|
||
"signals": {SIGNALS[k][1]: v for k, v in t.types.items() if k in SIGNALS},
|
||
"event_ids": [e.id for e in t.events],
|
||
}
|
||
for i, t in enumerate(threads, 1)
|
||
],
|
||
}
|
||
|
||
|
||
def save_json(world: World, threads: list[Thread], path: Path) -> Path:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(json.dumps(to_json(world, threads), ensure_ascii=False, indent=2), encoding="utf-8")
|
||
return path
|