test_relationship_graph.py 8.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333
  1. #!/usr/bin/env python3
  2. # -*- coding: utf-8 -*-
  3. """
  4. 关系事件与关系图谱测试
  5. """
  6. import json
  7. import sys
  8. import pytest
  9. import data_modules.index_manager as index_manager_module
  10. from data_modules.config import DataModulesConfig
  11. from data_modules.index_manager import (
  12. EntityMeta,
  13. IndexManager,
  14. RelationshipEventMeta,
  15. RelationshipMeta,
  16. )
  17. @pytest.fixture
  18. def temp_project(tmp_path):
  19. cfg = DataModulesConfig.from_project_root(tmp_path)
  20. cfg.ensure_dirs()
  21. return cfg
  22. def test_relationship_events_timeline_and_subgraph(temp_project):
  23. manager = IndexManager(temp_project)
  24. manager.upsert_entity(
  25. EntityMeta(
  26. id="xiaoyan",
  27. type="角色",
  28. canonical_name="萧炎",
  29. tier="核心",
  30. current={},
  31. first_appearance=1,
  32. last_appearance=10,
  33. is_protagonist=True,
  34. )
  35. )
  36. manager.upsert_entity(
  37. EntityMeta(
  38. id="yaolao",
  39. type="角色",
  40. canonical_name="药老",
  41. tier="重要",
  42. current={},
  43. first_appearance=1,
  44. last_appearance=10,
  45. )
  46. )
  47. manager.upsert_entity(
  48. EntityMeta(
  49. id="lintian",
  50. type="角色",
  51. canonical_name="林天",
  52. tier="重要",
  53. current={},
  54. first_appearance=2,
  55. last_appearance=10,
  56. )
  57. )
  58. manager.upsert_relationship(
  59. RelationshipMeta(
  60. from_entity="xiaoyan",
  61. to_entity="yaolao",
  62. type="师徒",
  63. description="正式拜师",
  64. chapter=3,
  65. )
  66. )
  67. manager.upsert_relationship(
  68. RelationshipMeta(
  69. from_entity="yaolao",
  70. to_entity="lintian",
  71. type="敌对",
  72. description="理念冲突",
  73. chapter=5,
  74. )
  75. )
  76. event_id = manager.record_relationship_event(
  77. RelationshipEventMeta(
  78. from_entity="xiaoyan",
  79. to_entity="yaolao",
  80. type="师徒",
  81. chapter=3,
  82. action="create",
  83. polarity=1,
  84. strength=0.9,
  85. description="拜师",
  86. evidence="公开收徒",
  87. confidence=0.95,
  88. )
  89. )
  90. assert event_id > 0
  91. manager.record_relationship_event(
  92. RelationshipEventMeta(
  93. from_entity="yaolao",
  94. to_entity="lintian",
  95. type="敌对",
  96. chapter=5,
  97. action="create",
  98. polarity=-1,
  99. strength=0.8,
  100. description="结怨",
  101. evidence="比斗失手",
  102. confidence=0.8,
  103. )
  104. )
  105. events = manager.get_relationship_events("xiaoyan", direction="both", limit=20)
  106. assert events
  107. timeline = manager.get_relationship_timeline("xiaoyan", "yaolao", limit=20)
  108. assert timeline
  109. assert timeline[0]["type"] == "师徒"
  110. graph = manager.build_relationship_subgraph("xiaoyan", depth=2, chapter=10, top_edges=10)
  111. node_ids = {n["id"] for n in graph["nodes"]}
  112. assert "xiaoyan" in node_ids
  113. assert "yaolao" in node_ids
  114. assert "lintian" in node_ids
  115. assert graph["edges"]
  116. mermaid = manager.render_relationship_subgraph_mermaid(graph)
  117. assert "mermaid" in mermaid
  118. assert "师徒" in mermaid
  119. def test_relationship_subgraph_respects_chapter_slice(temp_project):
  120. manager = IndexManager(temp_project)
  121. manager.upsert_entity(
  122. EntityMeta(
  123. id="a",
  124. type="角色",
  125. canonical_name="甲",
  126. current={},
  127. first_appearance=1,
  128. last_appearance=3,
  129. is_protagonist=True,
  130. )
  131. )
  132. manager.upsert_entity(
  133. EntityMeta(
  134. id="b",
  135. type="角色",
  136. canonical_name="乙",
  137. current={},
  138. first_appearance=1,
  139. last_appearance=3,
  140. )
  141. )
  142. manager.record_relationship_event(
  143. RelationshipEventMeta(
  144. from_entity="a",
  145. to_entity="b",
  146. type="同盟",
  147. chapter=1,
  148. action="create",
  149. polarity=1,
  150. strength=0.6,
  151. )
  152. )
  153. manager.record_relationship_event(
  154. RelationshipEventMeta(
  155. from_entity="a",
  156. to_entity="b",
  157. type="同盟",
  158. chapter=2,
  159. action="remove",
  160. polarity=0,
  161. strength=0.0,
  162. )
  163. )
  164. graph_ch1 = manager.build_relationship_subgraph("a", depth=1, chapter=1, top_edges=10)
  165. graph_ch3 = manager.build_relationship_subgraph("a", depth=1, chapter=3, top_edges=10)
  166. assert len(graph_ch1["edges"]) == 1
  167. assert len(graph_ch3["edges"]) == 0
  168. def test_relationship_subgraph_fallbacks_to_snapshot_when_events_missing(temp_project):
  169. manager = IndexManager(temp_project)
  170. manager.upsert_entity(
  171. EntityMeta(
  172. id="a",
  173. type="角色",
  174. canonical_name="甲",
  175. current={},
  176. first_appearance=1,
  177. last_appearance=5,
  178. is_protagonist=True,
  179. )
  180. )
  181. manager.upsert_entity(
  182. EntityMeta(
  183. id="b",
  184. type="角色",
  185. canonical_name="乙",
  186. current={},
  187. first_appearance=1,
  188. last_appearance=5,
  189. )
  190. )
  191. # 只写 relationships 快照,不写 relationship_events
  192. manager.upsert_relationship(
  193. RelationshipMeta(
  194. from_entity="a",
  195. to_entity="b",
  196. type="同盟",
  197. description="旧版快照数据",
  198. chapter=3,
  199. )
  200. )
  201. graph = manager.build_relationship_subgraph("a", depth=1, chapter=3, top_edges=10)
  202. assert graph["edges"]
  203. assert graph["edges"][0]["action"] == "snapshot"
  204. assert graph["edges"][0]["type"] == "同盟"
  205. def test_relationship_graph_cli_commands(temp_project, monkeypatch, capsys):
  206. manager = IndexManager(temp_project)
  207. manager.upsert_entity(
  208. EntityMeta(
  209. id="hero",
  210. type="角色",
  211. canonical_name="主角",
  212. current={},
  213. first_appearance=1,
  214. last_appearance=1,
  215. is_protagonist=True,
  216. )
  217. )
  218. manager.upsert_entity(
  219. EntityMeta(
  220. id="mentor",
  221. type="角色",
  222. canonical_name="师父",
  223. current={},
  224. first_appearance=1,
  225. last_appearance=1,
  226. )
  227. )
  228. manager.record_relationship_event(
  229. RelationshipEventMeta(
  230. from_entity="hero",
  231. to_entity="mentor",
  232. type="师徒",
  233. chapter=1,
  234. action="create",
  235. polarity=1,
  236. strength=0.9,
  237. )
  238. )
  239. root = str(temp_project.project_root)
  240. def run_cli(args):
  241. monkeypatch.setattr(sys, "argv", ["index_manager"] + args)
  242. index_manager_module.main()
  243. output = capsys.readouterr().out.strip().splitlines()
  244. assert output
  245. return json.loads(output[-1])
  246. payload = run_cli(
  247. [
  248. "--project-root",
  249. root,
  250. "get-relationship-events",
  251. "--entity",
  252. "hero",
  253. "--direction",
  254. "both",
  255. "--limit",
  256. "10",
  257. ]
  258. )
  259. assert payload["status"] == "success"
  260. assert payload["data"]
  261. payload = run_cli(
  262. [
  263. "--project-root",
  264. root,
  265. "get-relationship-graph",
  266. "--center",
  267. "hero",
  268. "--depth",
  269. "1",
  270. "--chapter",
  271. "1",
  272. "--format",
  273. "mermaid",
  274. ]
  275. )
  276. assert payload["status"] == "success"
  277. assert "mermaid" in payload["data"]["mermaid"]
  278. payload = run_cli(
  279. [
  280. "--project-root",
  281. root,
  282. "get-relationship-timeline",
  283. "--a",
  284. "hero",
  285. "--b",
  286. "mentor",
  287. "--limit",
  288. "10",
  289. ]
  290. )
  291. assert payload["status"] == "success"
  292. assert payload["data"]
  293. payload = run_cli(
  294. [
  295. "--project-root",
  296. root,
  297. "record-relationship-event",
  298. "--data",
  299. json.dumps(
  300. {
  301. "from_entity": "hero",
  302. "type": "师徒",
  303. "chapter": 1,
  304. },
  305. ensure_ascii=False,
  306. ),
  307. ]
  308. )
  309. assert payload["status"] == "error"
  310. assert payload["error"]["code"] == "INVALID_RELATIONSHIP_EVENT"