245 lines
10 KiB
Python
245 lines
10 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import secrets
|
||
from datetime import date
|
||
from typing import Any
|
||
|
||
from backend.bootstrap.config import normalize_date
|
||
from backend.features.heaven.agent import (
|
||
HEAVEN_PROMPT_VERSIONS,
|
||
HeavenAgentError,
|
||
interpret_heaven,
|
||
)
|
||
from backend.features.heaven.engine import (
|
||
build_five_phase_field,
|
||
hexagram_from_lines,
|
||
)
|
||
from backend.features.heaven.knowledge import prepare_heaven_context
|
||
from backend.features.heaven.six_yao import build_six_yao_chart
|
||
from backend.features.market import MarketServiceMixin
|
||
|
||
|
||
class HeavenReadingMixin:
|
||
def heaven_personal(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||
trade_date = normalize_date(str(payload.get("trade_date") or date.today().isoformat()))
|
||
field = build_five_phase_field(
|
||
trade_date,
|
||
self.database.list_sector_phase_overrides(),
|
||
)
|
||
personal = self.account_personal_field(trade_date, field, public=True)
|
||
if not personal:
|
||
raise ValueError("请先在账号设置中保存个人命理资料。")
|
||
return personal
|
||
|
||
def heaven_hexagram(self, raw_lines: Any) -> dict[str, Any]:
|
||
if not isinstance(raw_lines, list):
|
||
raise ValueError("六爻起卦结果格式不正确。")
|
||
try:
|
||
lines = [int(value) for value in raw_lines]
|
||
except (TypeError, ValueError) as exc:
|
||
raise ValueError("六爻必须由六、七、八、九组成。") from exc
|
||
return hexagram_from_lines(lines)
|
||
|
||
def heaven_readings(
|
||
self, mode: str, context_date: str = "", limit: int = 100
|
||
) -> dict[str, Any]:
|
||
mode = str(mode or "").strip()
|
||
if mode not in {"trend", "fortune", "heart"}:
|
||
raise ValueError("解读记录类型不正确。")
|
||
normalized_date = normalize_date(context_date) if context_date else ""
|
||
return {
|
||
"mode": mode,
|
||
"items": self.database.list_heaven_readings(
|
||
self.current_user_id, mode, normalized_date, limit
|
||
),
|
||
}
|
||
|
||
@staticmethod
|
||
def _heaven_reading_identity(
|
||
mode: str, context_date: str, context: dict[str, Any]
|
||
) -> tuple[str, str]:
|
||
display_date = MarketServiceMixin._display_compact_date(context_date)
|
||
if mode == "trend":
|
||
stock = (context.get("selected_focus") or {}).get("stock") or {}
|
||
code = str(stock.get("code") or "").strip()
|
||
name = str(stock.get("name") or "").strip()
|
||
hexagram = context.get("hexagram") or {}
|
||
transformed = hexagram.get("transformed") or {}
|
||
subject = " ".join(item for item in (code, name) if item) or "观势"
|
||
detail = f"{display_date} · {hexagram.get('name') or '--'} → {transformed.get('name') or '--'}"
|
||
return subject, detail
|
||
if mode == "fortune":
|
||
field = context.get("five_phase_field") or {}
|
||
pillars = field.get("pillars") or {}
|
||
dominant = (field.get("balance") or [{}])[0]
|
||
subject = f"{display_date} 观气"
|
||
detail = (
|
||
f"{pillars.get('year') or '--'}年 · {pillars.get('month') or '--'}月 · "
|
||
f"{pillars.get('day') or '--'}日 · {dominant.get('element') or '--'}气偏显"
|
||
)
|
||
return subject, detail
|
||
hexagram = context.get("hexagram") or {}
|
||
transformed = hexagram.get("transformed") or {}
|
||
question = str(context.get("question") or "").strip()
|
||
question_detail = f" · {question[:48]}" if question else ""
|
||
return (
|
||
f"{display_date} 观心",
|
||
f"{hexagram.get('name') or '--'} → {transformed.get('name') or '--'}{question_detail}",
|
||
)
|
||
|
||
def heaven_interpret(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||
mode = str(payload.get("mode") or "").strip()
|
||
if mode not in {"trend", "fortune", "heart"}:
|
||
raise ValueError("问天解读模式不正确。")
|
||
trade_date = normalize_date(str(payload.get("trade_date") or date.today().isoformat()))
|
||
prompt_version = HEAVEN_PROMPT_VERSIONS[mode]
|
||
stale_fortune: dict[str, Any] | None = None
|
||
if mode == "fortune":
|
||
existing = self.database.latest_heaven_reading(
|
||
self.current_user_id, "fortune", trade_date
|
||
)
|
||
if self._legacy_truncated_heaven_reading(existing):
|
||
self.database.delete_heaven_reading(
|
||
self.current_user_id, int(existing["id"])
|
||
)
|
||
existing = None
|
||
if existing and self.database.heaven_reading_interpretation_version(
|
||
self.current_user_id, int(existing["id"])
|
||
) == prompt_version:
|
||
return {
|
||
"answer": existing["answer"],
|
||
"mode": mode,
|
||
"compiler": "stored",
|
||
"notice": "",
|
||
"reading": existing,
|
||
"reused": True,
|
||
}
|
||
stale_fortune = existing
|
||
if mode in {"trend", "fortune"}:
|
||
setup = self.heaven_setup(
|
||
trade_date,
|
||
str(payload.get("sector") or ""),
|
||
str(payload.get("stock_code") or ""),
|
||
payload.get("manual_data"),
|
||
)
|
||
if mode == "trend":
|
||
chart = setup["chart"]
|
||
if not chart.get("available"):
|
||
issues = "；".join((chart.get("quality") or {}).get("issues") or [])
|
||
raise ValueError(f"观势数据未通过六爻校验，暂不解势：{issues}")
|
||
hexagram_context = json.loads(json.dumps(chart["hexagram"], ensure_ascii=False))
|
||
for line in hexagram_context.get("lines", []):
|
||
line.pop("evidence", None)
|
||
line.pop("score", None)
|
||
line.pop("talent", None)
|
||
line.pop("layer", None)
|
||
line.pop("role", None)
|
||
if not line.get("moving"):
|
||
line.pop("text", None)
|
||
line.pop("image", None)
|
||
line.pop("line_name", None)
|
||
context = {
|
||
"data_trade_date": setup["trade_date"],
|
||
"selected_focus": {
|
||
"sector": chart.get("sector") or "",
|
||
"stock": chart.get("stock") or {},
|
||
},
|
||
"hexagram": hexagram_context,
|
||
"movement": chart.get("movement") or {},
|
||
}
|
||
else:
|
||
personal_profile = self.account_personal_field(
|
||
setup["calendar_date"],
|
||
setup["field"],
|
||
public=False,
|
||
)
|
||
context = {
|
||
"calendar_date": setup["calendar_date"],
|
||
"five_phase_field": setup["field"],
|
||
"personal_profile": personal_profile,
|
||
}
|
||
context_date = setup["calendar_date"]
|
||
if mode == "trend":
|
||
context_date = setup["trade_date"]
|
||
else:
|
||
question = str(payload.get("question") or "").strip()
|
||
if len(question) > 300:
|
||
raise ValueError("观心问题不能超过300个字符。")
|
||
question_preset = str(payload.get("question_preset") or "unthemed").strip()
|
||
if question_preset not in {"trade", "mind", "unthemed", "custom"}:
|
||
question_preset = "custom"
|
||
if not question:
|
||
question = "不设具体问题，只观此刻一念。"
|
||
question_preset = "unthemed"
|
||
raw_lines = payload.get("lines")
|
||
hexagram = self.heaven_hexagram(raw_lines)
|
||
context = {
|
||
"question": question,
|
||
"question_preset": question_preset,
|
||
"hexagram": hexagram,
|
||
"six_yao": build_six_yao_chart(
|
||
[int(value) for value in raw_lines],
|
||
str(payload.get("cast_at") or ""),
|
||
),
|
||
"ritual": {
|
||
"breathing": "用户已完成1秒准备与五轮吸3秒、顿2秒、呼4秒的静心呼吸。",
|
||
"casting": "用户以三枚铜钱自初爻至上爻投掷六次。",
|
||
"reflection": "用户已在看见卦象后察看第一念。",
|
||
},
|
||
}
|
||
context_date = trade_date
|
||
agent_context = prepare_heaven_context(mode, context)
|
||
agent_context["interpretation_version"] = prompt_version
|
||
result, compiler = self._call_heaven_agent(mode, agent_context)
|
||
subject, subject_detail = self._heaven_reading_identity(
|
||
mode, context_date, context
|
||
)
|
||
dedupe_key = (
|
||
f"fortune:{context_date}"
|
||
if mode == "fortune"
|
||
else f"{mode}:{context_date}:{secrets.token_urlsafe(12)}"
|
||
)
|
||
if stale_fortune:
|
||
self.database.delete_heaven_reading(
|
||
self.current_user_id, int(stale_fortune["id"])
|
||
)
|
||
reading = self.database.save_heaven_reading(
|
||
self.current_user_id,
|
||
mode,
|
||
context_date,
|
||
subject,
|
||
subject_detail,
|
||
str(result.get("answer") or ""),
|
||
agent_context,
|
||
dedupe_key,
|
||
)
|
||
return {
|
||
**result,
|
||
"mode": mode,
|
||
"compiler": compiler,
|
||
"notice": "当前智能服务繁忙，已自动切换备用服务。" if compiler == "fallback" else "",
|
||
"reading": reading,
|
||
"reused": False,
|
||
}
|
||
|
||
@staticmethod
|
||
def _legacy_truncated_heaven_reading(reading: dict[str, Any] | None) -> bool:
|
||
return bool(reading and str(reading.get("answer") or "").rstrip().endswith("……"))
|
||
|
||
def _call_heaven_agent(self, mode: str, context: dict[str, Any]) -> tuple[dict[str, Any], str]:
|
||
prompt_version = HEAVEN_PROMPT_VERSIONS[mode]
|
||
result = self.llm_gateway.call(
|
||
f"heaven_{mode}",
|
||
prompt_version,
|
||
lambda profile: interpret_heaven(
|
||
mode,
|
||
context,
|
||
profile.api_key,
|
||
profile.base_url,
|
||
profile.model,
|
||
),
|
||
(HeavenAgentError,),
|
||
)
|
||
return result.value, result.role
|