qzl/eryao
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 接入起卦后端流程并完善积分扣减链路

Browse Source
This commit is contained in:
qzl
2026-04-03 19:04:46 +08:00
parent a136e42290
commit d87b2e1e3a
56 changed files with 3310 additions and 809 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/core/agentscope/events/agui_codec.py
+26 -24
View File
@@ -17,36 +17,38 @@ from schemas.agent.ui_hints import UiHintsPayload
if TYPE_CHECKING:
pass
_INTERNAL_TO_AGUI: dict[str, EventType] = {
"run.started": EventType.RUN_STARTED,
"run.finished": EventType.RUN_FINISHED,
"run.error": EventType.RUN_ERROR,
"step.start": EventType.STEP_STARTED,
"step.finish": EventType.STEP_FINISHED,
"text.end": EventType.TEXT_MESSAGE_END,
"tool.start": EventType.TOOL_CALL_START,
"tool.args": EventType.TOOL_CALL_ARGS,
"tool.end": EventType.TOOL_CALL_END,
"tool.result": EventType.TOOL_CALL_RESULT,
"state.snapshot": EventType.STATE_SNAPSHOT,
"messages.snapshot": EventType.MESSAGES_SNAPSHOT,
_INTERNAL_TO_AGUI: dict[str, str] = {
"run.started": EventType.RUN_STARTED.value,
"run.finished": EventType.RUN_FINISHED.value,
"run.error": EventType.RUN_ERROR.value,
"step.start": EventType.STEP_STARTED.value,
"step.finish": EventType.STEP_FINISHED.value,
"text.end": EventType.TEXT_MESSAGE_END.value,
"tool.start": EventType.TOOL_CALL_START.value,
"tool.args": EventType.TOOL_CALL_ARGS.value,
"tool.end": EventType.TOOL_CALL_END.value,
"tool.result": EventType.TOOL_CALL_RESULT.value,
"state.snapshot": EventType.STATE_SNAPSHOT.value,
"messages.snapshot": EventType.MESSAGES_SNAPSHOT.value,
}
_ALLOWED_WIRE_TYPES: set[str] = {event_type.value for event_type in EventType}
_ALLOWED_WIRE_TYPES.add("DIVINATION_DERIVED")
def _convert_to_agui_type(internal_type: str) -> EventType:
def _convert_to_agui_type(internal_type: str) -> str:
mapped = _INTERNAL_TO_AGUI.get(internal_type)
if mapped is not None:
return mapped
return EventType(internal_type.upper().replace(".", "_"))
candidate = internal_type.upper().replace(".", "_")
if candidate in _ALLOWED_WIRE_TYPES:
return candidate
raise ValueError(f"unsupported ag-ui event type: {internal_type}")
def _is_agui_event(event: dict[str, Any]) -> bool:
event_type = event.get("type", "")
try:
EventType(event_type)
return True
except ValueError:
return False
event_type = str(event.get("type", "")).strip().upper()
return event_type in _ALLOWED_WIRE_TYPES
def _sanitize_agui_event(event: dict[str, Any]) -> dict[str, Any]:
@@ -153,7 +155,7 @@ def to_agui_wire_event(event: dict[str, Any] | BaseEvent) -> dict[str, Any]:
if internal_type == "text.end" and isinstance(data, dict):
text_end_payload: dict[str, Any] = {
"type": _convert_to_agui_type(internal_type).value,
"type": _convert_to_agui_type(internal_type),
}
if isinstance(thread_id, str) and thread_id:
text_end_payload["threadId"] = thread_id
@@ -174,7 +176,7 @@ def to_agui_wire_event(event: dict[str, Any] | BaseEvent) -> dict[str, Any]:
if internal_type == "tool.result" and isinstance(data, dict):
tool_result_payload: dict[str, Any] = {
"type": _convert_to_agui_type(internal_type).value,
"type": _convert_to_agui_type(internal_type),
}
if isinstance(thread_id, str) and thread_id:
tool_result_payload["threadId"] = thread_id
@@ -203,7 +205,7 @@ def to_agui_wire_event(event: dict[str, Any] | BaseEvent) -> dict[str, Any]:
wire_type = _convert_to_agui_type(internal_type)
payload: dict[str, Any] = {
"type": wire_type.value,
"type": wire_type,
}
if isinstance(thread_id, str) and thread_id:
payload["threadId"] = thread_id
backend/src/core/agentscope/prompts/user_prompt.py
+18
View File
@@ -0,0 +1,18 @@
from __future__ import annotations
from schemas.domain.divination import DerivedDivinationData
def build_divination_user_prompt(*, derived: DerivedDivinationData) -> str:
structured_json = derived.model_dump_json(
by_alias=True,
exclude_none=True,
ensure_ascii=False,
)
return (
f"用户问题：{derived.question}\n"
f"问题类型：{derived.question_type}\n"
"以下是后端推导后的六爻结构化数据（JSON）：\n"
f"{structured_json}\n"
"请仅基于以上六爻数据做专业解读。"
)
backend/src/core/agentscope/runtime/runner.py
+62 -27
View File
@@ -13,11 +13,14 @@ from agentscope.message import Msg
from agentscope.tool import Toolkit
from agentscope.model import OpenAIChatModel
from core.agentscope.prompts.system_prompt import build_system_prompt
from core.agentscope.prompts.user_prompt import build_divination_user_prompt
from core.agentscope.schemas.agui_input import extract_latest_user_payload
from core.divination import derive_divination
from core.agentscope.runtime.json_react_agent import JsonReActAgent
from core.agentscope.runtime.model_tracking import TrackingChatModel
from core.agentscope.runtime.stage_emitter import PipelineStageEmitter
from core.agentscope.utils import patch_agentscope_json_repair_compat
from core.agentscope.utils.json_finalize import finalize_json_response
from core.config.settings import config
from core.db.session import AsyncSessionLocal
from models.llm import Llm
@@ -26,9 +29,11 @@ from models.system_agents import SystemAgents
from schemas.agent.forwarded_props import (
ClientTimeContext,
RuntimeMode,
parse_forwarded_props_divination_payload,
parse_forwarded_props_client_time,
parse_forwarded_props_runtime_mode,
)
from schemas.domain.divination import DerivedDivinationData
from schemas.agent.runtime_models import (
WorkerAgentOutputLite,
resolve_worker_output_model,
@@ -97,6 +102,21 @@ class AgentScopeRunner:
worker_toolkit = self._build_toolkit()
if cancel_checker is not None and await cancel_checker():
raise asyncio.CancelledError("run canceled by user")
derived_divination = self._resolve_derived_divination(
run_input=run_input
)
await self._emit_step_event(
pipeline=pipeline,
run_input=run_input,
step_name="divination",
event_type="DIVINATION_DERIVED",
runtime_mode=runtime_mode,
extra_event={
"divination": derived_divination.model_dump(
mode="json", by_alias=True, exclude_none=True
)
},
)
worker_output = await self._execute_worker_step(
pipeline=pipeline,
run_input=run_input,
@@ -106,6 +126,7 @@ class AgentScopeRunner:
stage_config=worker_config,
runtime_client_time=runtime_client_time,
runtime_mode=runtime_mode,
derived_divination=derived_divination,
)
return {
"worker": worker_output.model_dump(mode="json", exclude_none=True),
@@ -187,6 +208,7 @@ class AgentScopeRunner:
stage_config: SystemAgentRuntimeConfig,
runtime_client_time: ClientTimeContext | None,
runtime_mode: RuntimeMode,
derived_divination: DerivedDivinationData,
) -> WorkerAgentOutputLite:
worker_output_model = resolve_worker_output_model()
await self._emit_step_event(
@@ -201,6 +223,7 @@ class AgentScopeRunner:
input_messages=self._build_worker_input_messages(
context_messages=context_messages,
run_input=run_input,
derived_divination=derived_divination,
),
toolkit=toolkit,
run_input=run_input,
@@ -234,6 +257,7 @@ class AgentScopeRunner:
runtime_mode: RuntimeMode,
) -> StageExecutionResult:
tracking_model = self._build_model(stage_config=stage_config)
formatter = OpenAIChatFormatter()
emitter = PipelineStageEmitter(
pipeline=pipeline,
session_id=run_input.thread_id,
@@ -243,32 +267,24 @@ class AgentScopeRunner:
emit_text_events=True,
emit_tool_events=False,
)
agent = self._build_agent(
agent_name=stage_config.agent_type.value,
system_prompt=build_system_prompt(
agent_type=stage_config.agent_type,
llm_config=stage_config.llm_config,
user_context=user_context,
now_utc=datetime.now(timezone.utc),
runtime_client_time=runtime_client_time,
extra_context=stage_config.extra_context,
tools=None,
),
toolkit=toolkit,
model=tracking_model,
emitter=emitter,
system_prompt = build_system_prompt(
agent_type=stage_config.agent_type,
llm_config=stage_config.llm_config,
user_context=user_context,
now_utc=datetime.now(timezone.utc),
runtime_client_time=runtime_client_time,
extra_context=stage_config.extra_context,
tools=None,
)
async with self._active_agent_lock:
self._active_agent = agent
try:
response_msg = await agent.reply_json(
input_messages, output_model=worker_output_model
)
finally:
async with self._active_agent_lock:
if self._active_agent is agent:
self._active_agent = None
worker_payload = worker_output_model.model_validate(response_msg.metadata or {})
_, worker_payload_raw = await finalize_json_response(
model=tracking_model,
formatter=formatter,
base_messages=[Msg("system", system_prompt, "system"), *input_messages],
output_model=worker_output_model,
retries=2,
)
worker_payload = worker_output_model.model_validate(worker_payload_raw)
response_metadata = self._llm_pricing_service.build_usage_metadata(
model=stage_config.model_code,
usage_summary=tracking_model.usage_summary(),
@@ -278,7 +294,12 @@ class AgentScopeRunner:
response_metadata=response_metadata,
)
return StageExecutionResult(
message=response_msg,
message=Msg(
name=stage_config.agent_type.value,
role="assistant",
content=worker_payload.answer,
metadata=worker_payload.model_dump(mode="json", exclude_none=True),
),
payload=worker_payload.model_dump(mode="json", exclude_none=True),
response_metadata=response_metadata,
)
@@ -288,13 +309,16 @@ class AgentScopeRunner:
*,
context_messages: list[Msg],
run_input: RunAgentInput,
derived_divination: DerivedDivinationData,
) -> list[Msg]:
if context_messages:
last = context_messages[-1]
if last.role == "user":
return context_messages
user_text, user_blocks = extract_latest_user_payload(run_input)
_, _ = extract_latest_user_payload(run_input)
user_text = build_divination_user_prompt(derived=derived_divination)
user_blocks = [{"type": "text", "text": user_text}]
if (
user_blocks
and isinstance(user_blocks[0], dict)
@@ -307,6 +331,17 @@ class AgentScopeRunner:
user_msg = Msg(name="user", role="user", content=content)
return [*context_messages, user_msg]
@staticmethod
def _resolve_derived_divination(
*, run_input: RunAgentInput
) -> DerivedDivinationData:
payload = parse_forwarded_props_divination_payload(
getattr(run_input, "forwarded_props", None)
)
if payload is None:
raise ValueError("forwardedProps.divinationPayload is required")
return derive_divination(payload)
def _build_model(
self, *, stage_config: SystemAgentRuntimeConfig
) -> TrackingChatModel:
backend/src/core/divination/__init__.py
+3
View File
@@ -0,0 +1,3 @@
from core.divination.derivation import derive_divination
__all__ = ["derive_divination"]
backend/src/core/divination/derivation.py
+281
View File
@@ -0,0 +1,281 @@
from __future__ import annotations
from datetime import datetime
from lunar_python import Solar
from core.divination.gua_catalog_loader import GuaCatalogItem, load_gua_catalog
from schemas.domain.divination import (
DerivedDivinationData,
DivinationPayload,
FushenDetail,
GanzhiDetail,
SpecialMark,
YaoDetail,
YaoType,
)
_DI_ZHI_ORDER = ("子", "丑", "寅", "卯", "辰", "巳", "午", "未", "申", "酉", "戌", "亥")
_TIAN_GAN_ORDER = ("甲", "乙", "丙", "丁", "戊", "己", "庚", "辛", "壬", "癸")
def _yao_to_bit(yao: YaoType) -> str:
if yao in (YaoType.SHAO_YANG, YaoType.LAO_YANG):
return "1"
return "0"
def _yao_to_changed_bit(yao: YaoType) -> str:
if yao == YaoType.SHAO_YANG:
return "1"
if yao == YaoType.SHAO_YIN:
return "0"
if yao == YaoType.LAO_YANG:
return "0"
return "1"
def _yao_is_yang(yao: YaoType) -> bool:
return yao in (YaoType.SHAO_YANG, YaoType.LAO_YANG)
def _yao_is_changing(yao: YaoType) -> bool:
return yao in (YaoType.LAO_YANG, YaoType.LAO_YIN)
def _resolve_liu_shou(day_gan: str) -> tuple[str, ...]:
start = {
"甲": 0,
"乙": 0,
"丙": 1,
"丁": 1,
"戊": 2,
"己": 3,
"庚": 4,
"辛": 4,
"壬": 5,
"癸": 5,
}.get(day_gan, 0)
base = ("龙", "雀", "勾", "蛇", "虎", "玄")
return tuple(base[(start + i) % 6] for i in range(6))
def _di_zhi_wu_xing(di_zhi: str) -> str:
if di_zhi in ("寅", "卯"):
return "木"
if di_zhi in ("巳", "午"):
return "火"
if di_zhi in ("申", "酉"):
return "金"
if di_zhi in ("亥", "子"):
return "水"
return "土"
def _chong_di_zhi(di_zhi: str) -> str:
return {
"子": "午",
"午": "子",
"丑": "未",
"未": "丑",
"寅": "申",
"申": "寅",
"卯": "酉",
"酉": "卯",
"辰": "戌",
"戌": "辰",
"巳": "亥",
"亥": "巳",
}.get(di_zhi, "")
def _wu_xing_status(month_di_zhi: str, wu_xing: str) -> str:
table = {
"寅": {"木": "旺", "火": "相", "水": "休", "金": "囚", "土": "死"},
"卯": {"木": "旺", "火": "相", "水": "休", "金": "囚", "土": "死"},
"巳": {"火": "旺", "土": "相", "木": "休", "水": "囚", "金": "死"},
"午": {"火": "旺", "土": "相", "木": "休", "水": "囚", "金": "死"},
"申": {"金": "旺", "水": "相", "土": "休", "火": "囚", "木": "死"},
"酉": {"金": "旺", "水": "相", "土": "休", "火": "囚", "木": "死"},
"亥": {"水": "旺", "木": "相", "金": "休", "土": "囚", "火": "死"},
"子": {"水": "旺", "木": "相", "金": "休", "土": "囚", "火": "死"},
"辰": {"土": "旺", "金": "相", "火": "休", "木": "囚", "水": "死"},
"未": {"土": "旺", "金": "相", "火": "休", "木": "囚", "水": "死"},
"戌": {"土": "旺", "金": "相", "火": "休", "木": "囚", "水": "死"},
"丑": {"土": "旺", "金": "相", "火": "休", "木": "囚", "水": "死"},
}
return table.get(month_di_zhi, {}).get(wu_xing, "")
def _get_kong_wang(gan_zhi: str) -> str:
tian_gan = gan_zhi[0]
di_zhi = gan_zhi[1]
tian_gan_value = _TIAN_GAN_ORDER.index(tian_gan) + 1
di_zhi_value = _DI_ZHI_ORDER.index(di_zhi) + 1
start = di_zhi_value - tian_gan_value - 1
if start < 0:
start += 12
start %= 12
if start == 0:
start = 12
first = _DI_ZHI_ORDER[start - 1]
second = _DI_ZHI_ORDER[start % 12]
return f"{first}{second}"
def _get_all_wu_xing_status(month_gan_zhi: str) -> dict[str, str]:
month_di_zhi = month_gan_zhi[1]
return {
"木": _wu_xing_status(month_di_zhi, "木"),
"火": _wu_xing_status(month_di_zhi, "火"),
"土": _wu_xing_status(month_di_zhi, "土"),
"金": _wu_xing_status(month_di_zhi, "金"),
"水": _wu_xing_status(month_di_zhi, "水"),
}
def _resolve_special_mark(
*, index: int, world_position: int, response_position: int
) -> SpecialMark:
position = index + 1
if position == world_position:
return SpecialMark.SHI
if position == response_position:
return SpecialMark.YING
return SpecialMark.NONE
def _build_target_relation_names(
base_item: GuaCatalogItem, target_item: GuaCatalogItem
) -> tuple[str, ...]:
gua_gong_wu_xing = {
"乾": "金",
"坤": "土",
"震": "木",
"巽": "木",
"坎": "水",
"离": "火",
"艮": "土",
"兑": "金",
}.get(base_item.upper_name, "金")
def calculate(yao_wu_xing: str) -> str:
if gua_gong_wu_xing == yao_wu_xing:
return "兄弟"
table = {
"木": {"火": "子孙", "土": "妻财", "金": "官鬼", "水": "父母"},
"火": {"土": "子孙", "金": "妻财", "水": "官鬼", "木": "父母"},
"土": {"金": "子孙", "水": "妻财", "木": "官鬼", "火": "父母"},
"金": {"水": "子孙", "木": "妻财", "火": "官鬼", "土": "父母"},
"水": {"木": "子孙", "火": "妻财", "土": "官鬼", "金": "父母"},
}
return table.get(gua_gong_wu_xing, {}).get(yao_wu_xing, "兄弟")
return tuple(calculate(element) for element in target_item.yao_elements)
def derive_divination(payload: DivinationPayload) -> DerivedDivinationData:
catalog = load_gua_catalog()
binary_code = "".join(_yao_to_bit(yao) for yao in payload.yao_lines)
changed_binary_code = "".join(_yao_to_changed_bit(yao) for yao in payload.yao_lines)
base_item = catalog[binary_code]
target_item = catalog[changed_binary_code]
has_changing_yao = any(_yao_is_changing(yao) for yao in payload.yao_lines)
dt = datetime.fromisoformat(payload.divination_time_iso.replace("Z", "+00:00"))
solar = Solar.fromYmdHms(dt.year, dt.month, dt.day, dt.hour, dt.minute, dt.second)
lunar = solar.getLunar()
year_gan_zhi = lunar.getYearInGanZhi()
month_gan_zhi = lunar.getMonthInGanZhi()
day_gan_zhi = lunar.getDayInGanZhi()
time_gan_zhi = lunar.getTimeInGanZhi()
liu_shou = _resolve_liu_shou(day_gan_zhi[0])
target_relations = _build_target_relation_names(base_item, target_item)
yao_info_list = [
YaoDetail(
position=index + 1,
spiritName=liu_shou[index],
relationName=base_item.yao_relations[index],
tiganName=base_item.yao_tigan[index],
elementName=base_item.yao_elements[index],
isYang=_yao_is_yang(payload.yao_lines[index]),
isChanging=_yao_is_changing(payload.yao_lines[index]),
specialMark=_resolve_special_mark(
index=index,
world_position=base_item.world_position,
response_position=base_item.response_position,
),
)
for index in range(6)
]
target_yao_info_list = [
YaoDetail(
position=index + 1,
spiritName=liu_shou[index],
relationName=target_relations[index],
tiganName=target_item.yao_tigan[index],
elementName=target_item.yao_elements[index],
isYang=_yao_is_yang(payload.yao_lines[index])
if not _yao_is_changing(payload.yao_lines[index])
else not _yao_is_yang(payload.yao_lines[index]),
isChanging=False,
specialMark=_resolve_special_mark(
index=index,
world_position=target_item.world_position,
response_position=target_item.response_position,
),
)
for index in range(6)
]
month_di_zhi = month_gan_zhi[1]
month_chong = _chong_di_zhi(month_di_zhi)
day_di_zhi = day_gan_zhi[1]
day_chong = _chong_di_zhi(day_di_zhi)
fushen_info_list = [
FushenDetail(
position=base_item.fushen_positions[idx] + 1,
relationName=base_item.fushen_relations[idx],
tiganName=base_item.fushen_tigan[idx],
elementName=base_item.fushen_elements[idx],
)
for idx in range(len(base_item.fushen_positions))
]
return DerivedDivinationData(
question=payload.question,
questionType=payload.question_type,
divinationMethod=payload.divination_method,
divinationTime=dt.strftime("%Y年%m月%d日 %H:%M"),
binaryCode=binary_code,
changedBinaryCode=changed_binary_code,
guaName=base_item.name,
upperName=base_item.upper_name,
lowerName=base_item.lower_name,
targetGuaName=target_item.name,
worldPosition=base_item.world_position,
responsePosition=base_item.response_position,
hasChangingYao=has_changing_yao,
ganzhi=GanzhiDetail(
yearGanZhi=year_gan_zhi,
monthGanZhi=month_gan_zhi,
dayGanZhi=day_gan_zhi,
timeGanZhi=time_gan_zhi,
yearKongWang=_get_kong_wang(year_gan_zhi),
monthKongWang=_get_kong_wang(month_gan_zhi),
dayKongWang=_get_kong_wang(day_gan_zhi),
timeKongWang=_get_kong_wang(time_gan_zhi),
yueJian=f"{month_di_zhi}{_di_zhi_wu_xing(month_di_zhi)}",
riChen=f"{day_di_zhi}{_di_zhi_wu_xing(day_di_zhi)}",
yuePo=f"{month_chong}{_di_zhi_wu_xing(month_chong)}",
riChong=f"{day_chong}{_di_zhi_wu_xing(day_chong)}",
),
wuXingStatuses=_get_all_wu_xing_status(month_gan_zhi),
yaoInfoList=yao_info_list,
targetYaoInfoList=target_yao_info_list if has_changing_yao else [],
fushenPositions=[item + 1 for item in base_item.fushen_positions],
fushenInfoList=fushen_info_list,
)
backend/src/core/divination/gua_catalog_loader.py
+139
View File
@@ -0,0 +1,139 @@
from __future__ import annotations
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
import re
@dataclass(frozen=True)
class GuaCatalogItem:
name: str
binary: str
upper_name: str
lower_name: str
yao_relations: tuple[str, ...]
yao_tigan: tuple[str, ...]
yao_elements: tuple[str, ...]
world_position: int
response_position: int
fushen_positions: tuple[int, ...]
fushen_relations: tuple[str, ...]
fushen_tigan: tuple[str, ...]
fushen_elements: tuple[str, ...]
_ENTRY_HEAD_RE = re.compile(r'put\("([01]{6})",\s*GuaInfo\(', re.MULTILINE)
_STRING_FIELD_RE = re.compile(r'\b%s\s*=\s*"([^"]*)"')
_INT_FIELD_RE = re.compile(r"\b%s\s*=\s*(\d+)")
_LIST_STRING_FIELD_RE = re.compile(r"\b%s\s*=\s*listOf\((.*?)\)", re.DOTALL)
_LIST_INT_FIELD_RE = re.compile(r"\b%s\s*=\s*listOf\((.*?)\)", re.DOTALL)
def _extract_gua_body(source: str, start_idx: int) -> tuple[str, int]:
depth = 1
idx = start_idx
while idx < len(source):
ch = source[idx]
if ch == "(":
depth += 1
elif ch == ")":
depth -= 1
if depth == 0:
return source[start_idx:idx], idx
idx += 1
raise ValueError("invalid Guaxiang.kt structure: unmatched parenthesis")
def _parse_string_field(body: str, field_name: str) -> str:
match = _STRING_FIELD_RE.pattern % re.escape(field_name)
found = re.search(match, body)
if found is None:
raise ValueError(f"missing field: {field_name}")
return found.group(1)
def _parse_int_field(body: str, field_name: str) -> int:
match = _INT_FIELD_RE.pattern % re.escape(field_name)
found = re.search(match, body)
if found is None:
raise ValueError(f"missing field: {field_name}")
return int(found.group(1))
def _parse_list_of_strings(
body: str, field_name: str, *, optional: bool = False
) -> tuple[str, ...]:
if f"{field_name} = emptyList()" in body:
return ()
match = _LIST_STRING_FIELD_RE.pattern % re.escape(field_name)
found = re.search(match, body)
if found is None:
if optional:
return ()
raise ValueError(f"missing list field: {field_name}")
inner = found.group(1)
values = re.findall(r'"([^"]+)"', inner)
return tuple(values)
def _parse_list_of_ints(
body: str, field_name: str, *, optional: bool = False
) -> tuple[int, ...]:
if f"{field_name} = emptyList()" in body:
return ()
match = _LIST_INT_FIELD_RE.pattern % re.escape(field_name)
found = re.search(match, body)
if found is None:
if optional:
return ()
raise ValueError(f"missing list field: {field_name}")
inner = found.group(1)
values = [int(item.strip()) for item in inner.split(",") if item.strip()]
return tuple(values)
def _resolve_guaxiang_file() -> Path:
current = Path(__file__).resolve()
root = current.parents[4]
target = (
root / "old/app/src/main/java/com/example/eryaoapp/screens/result/Guaxiang.kt"
)
if not target.exists():
raise FileNotFoundError(f"Guaxiang.kt not found: {target}")
return target
@lru_cache(maxsize=1)
def load_gua_catalog() -> dict[str, GuaCatalogItem]:
source = _resolve_guaxiang_file().read_text(encoding="utf-8")
result: dict[str, GuaCatalogItem] = {}
for head in _ENTRY_HEAD_RE.finditer(source):
binary = head.group(1)
body, _ = _extract_gua_body(source, head.end())
item = GuaCatalogItem(
name=_parse_string_field(body, "name"),
binary=_parse_string_field(body, "binary"),
upper_name=_parse_string_field(body, "upperName"),
lower_name=_parse_string_field(body, "lowerName"),
yao_relations=_parse_list_of_strings(body, "yaoRelations"),
yao_tigan=_parse_list_of_strings(body, "yaoTiGan"),
yao_elements=_parse_list_of_strings(body, "yaoElements"),
world_position=_parse_int_field(body, "worldPosition"),
response_position=_parse_int_field(body, "responsePosition"),
fushen_positions=_parse_list_of_ints(
body, "fushenPositions", optional=True
),
fushen_relations=_parse_list_of_strings(
body, "fushenRelations", optional=True
),
fushen_tigan=_parse_list_of_strings(body, "fushenTiGan", optional=True),
fushen_elements=_parse_list_of_strings(
body, "fushenElements", optional=True
),
)
result[binary] = item
if len(result) != 64:
raise ValueError(f"invalid gua catalog size: {len(result)}")
return result