40 lines
1.4 KiB
Python
40 lines
1.4 KiB
Python
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from config import USER_DB_PATH
|
|
|
|
|
|
def load_user_profile(user_name: str, db_path: str | Path = USER_DB_PATH) -> str:
|
|
"""在 Python 层直接读档案，注入到消息上下文，模型无需主动调用 get_user_profile。"""
|
|
try:
|
|
with sqlite3.connect(db_path) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
user = conn.execute(
|
|
"SELECT * FROM users WHERE name = ?", (user_name,)
|
|
).fetchone()
|
|
if not user:
|
|
return f"用户 {user_name} 尚无历史记录，这是第一次见面。"
|
|
prefs = conn.execute(
|
|
"SELECT category, content FROM preferences WHERE user_name = ?",
|
|
(user_name,)
|
|
).fetchall()
|
|
conn.execute(
|
|
"UPDATE users SET last_seen = datetime('now') WHERE name = ?",
|
|
(user_name,)
|
|
)
|
|
return json.dumps(
|
|
{
|
|
"基本信息": {
|
|
"姓名": user["name"],
|
|
"年龄": user["age"],
|
|
"上次见面": user["last_seen"],
|
|
},
|
|
"偏好习惯": {p["category"]: p["content"] for p in prefs},
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
except Exception as e:
|
|
return f"档案读取失败（{e}），当作第一次见面。"
|
|
|