RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
75 lines
2.8 KiB
Python
75 lines
2.8 KiB
Python
"""Persona rendering for structured asynchronous notification events."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
|
|
from notifications import Event
|
|
|
|
|
|
API_UNAVAILABLE_NOTICE = "【聊天模式 API 暂时不可用】"
|
|
|
|
|
|
def _with_api_notice(text: str, mode: str) -> str:
|
|
if mode == "chat" and API_UNAVAILABLE_NOTICE not in text:
|
|
return text.rstrip() + "\n\n" + API_UNAVAILABLE_NOTICE
|
|
return text
|
|
|
|
|
|
def _fallback(event: Event) -> str:
|
|
p = event.payload
|
|
if event.kind == "mail":
|
|
mid = p.get("mid", "")
|
|
sender = p.get("sender") or p.get("from") or "一位发件人"
|
|
subject = p.get("subject") or "(无主题)"
|
|
when = p.get("time") or "刚才"
|
|
url = p.get("url") or "https://mail.rrrosmontis.icu"
|
|
return (
|
|
f"{when}有一封新邮件到了。它来自{sender}，标题是「{subject}」，"
|
|
f"编号是 {mid}。邮件在这里：{url}"
|
|
)
|
|
if event.kind.startswith("ginkgo"):
|
|
return "白果云课堂刚才有新的进展。\n" + event.legacy_text
|
|
return event.legacy_text
|
|
|
|
|
|
def _critical_values(event: Event) -> list[str]:
|
|
p = event.payload
|
|
values = []
|
|
for key in ("mid", "url", "link", "code"):
|
|
value = p.get(key)
|
|
if value not in (None, ""):
|
|
values.append(str(value))
|
|
values.extend(re.findall(r"https?://[^\s]+", event.legacy_text))
|
|
return values
|
|
|
|
|
|
async def render_event(event: Event, mode: str, target: int) -> str:
|
|
"""Render facts without allowing the model to alter IDs or links."""
|
|
fallback = _fallback(event)
|
|
try:
|
|
from chat import engine
|
|
facts = json.dumps(event.payload, ensure_ascii=False)
|
|
prompt = (
|
|
engine.user_persona_prompt(event.uid)
|
|
+ "\n\n你现在要把一条系统通知自然地说给用户听。保持白花的个性，"
|
|
"像日常聊天一样表达，不要机械地逐条罗列，不要暴露程序术语或异常堆栈。"
|
|
"通知事实必须准确，编号、时间、链接、课程和签到码不得编造或改写。"
|
|
"不要以‘需要我……吗’之类的套路问句结尾。只输出要发送的正文，不要JSON。"
|
|
)
|
|
import skill_runtime
|
|
scene = "reminder" if event.kind == "reminder" else "notification"
|
|
with skill_runtime.skill_session(scene, output_mode="text"):
|
|
raw = await engine.call_deepseek(prompt, facts, 0)
|
|
if not raw:
|
|
return _with_api_notice(fallback, mode)
|
|
text = raw.strip()
|
|
if text.startswith("```"):
|
|
return fallback
|
|
if any(value not in text for value in _critical_values(event)):
|
|
return fallback
|
|
return text[:3500]
|
|
except Exception as exc:
|
|
print(f"[notification-render] {type(exc).__name__}: {exc}")
|
|
return fallback
|