|
|
@@ -12,10 +12,12 @@ FEISHU_APP_ID = os.getenv("FEISHU_APP_ID", "cli_a90fe317987a9cc9")
|
|
|
FEISHU_APP_SECRET = os.getenv("FEISHU_APP_SECRET", "nn2dWuXTiRA2N6xodbm4g0qz1AfM2ayi")
|
|
|
|
|
|
CONTACTS_FILE = os.path.join(os.path.dirname(__file__), "contacts.json")
|
|
|
+CHAT_HISTORY_DIR = os.path.join(os.path.dirname(__file__), "chat_history")
|
|
|
+UNREAD_SUMMARY_FILE = os.path.join(CHAT_HISTORY_DIR, "unread_summary.json")
|
|
|
|
|
|
# ==================== 一、文件内使用的功能函数 ====================
|
|
|
|
|
|
-def _load_contacts() -> List[Dict[str, Any]]:
|
|
|
+def load_contacts() -> List[Dict[str, Any]]:
|
|
|
"""读取 contacts.json 中的所有联系人"""
|
|
|
if not os.path.exists(CONTACTS_FILE):
|
|
|
return []
|
|
|
@@ -25,7 +27,7 @@ def _load_contacts() -> List[Dict[str, Any]]:
|
|
|
except Exception:
|
|
|
return []
|
|
|
|
|
|
-def _save_contacts(contacts: List[Dict[str, Any]]):
|
|
|
+def save_contacts(contacts: List[Dict[str, Any]]):
|
|
|
"""保存联系人信息到 contacts.json"""
|
|
|
try:
|
|
|
with open(CONTACTS_FILE, 'w', encoding='utf-8') as f:
|
|
|
@@ -38,7 +40,7 @@ def list_contacts_info() -> List[Dict[str, str]]:
|
|
|
1. 列出所有联系人信息
|
|
|
读取 contacts.json 中的每一个联系人的 name、description，以字典列表返回
|
|
|
"""
|
|
|
- contacts = _load_contacts()
|
|
|
+ contacts = load_contacts()
|
|
|
return [{"name": c.get("name", ""), "description": c.get("description", "")} for c in contacts]
|
|
|
|
|
|
def get_contact_full_info(name: str) -> Optional[Dict[str, Any]]:
|
|
|
@@ -46,18 +48,25 @@ def get_contact_full_info(name: str) -> Optional[Dict[str, Any]]:
|
|
|
2. 根据联系人名称获取联系人完整字典信息
|
|
|
从 contacts.json 中读取每一个联系人做名称匹配，返回数据中的所有字段为一个字典对象
|
|
|
"""
|
|
|
- contacts = _load_contacts()
|
|
|
+ contacts = load_contacts()
|
|
|
for c in contacts:
|
|
|
if c.get("name") == name:
|
|
|
return c
|
|
|
return None
|
|
|
|
|
|
+def get_contact_by_id(id_value: str) -> Optional[Dict[str, Any]]:
|
|
|
+ """根据 chat_id 或 open_id 获取联系人信息"""
|
|
|
+ contacts = load_contacts()
|
|
|
+ for c in contacts:
|
|
|
+ if c.get("chat_id") == id_value or c.get("open_id") == id_value:
|
|
|
+ return c
|
|
|
+ return None
|
|
|
+
|
|
|
def update_contact_chat_id(name: str, chat_id: str):
|
|
|
"""
|
|
|
3. 更新某一个联系人的 chat_id
|
|
|
- 根据第二个函数找出联系人信息，如果信息中的 chat_id 为空，那么就将传进来的 chat_id 更新进去
|
|
|
"""
|
|
|
- contacts = _load_contacts()
|
|
|
+ contacts = load_contacts()
|
|
|
updated = False
|
|
|
for c in contacts:
|
|
|
if c.get("name") == name:
|
|
|
@@ -66,9 +75,59 @@ def update_contact_chat_id(name: str, chat_id: str):
|
|
|
updated = True
|
|
|
break
|
|
|
if updated:
|
|
|
- _save_contacts(contacts)
|
|
|
+ save_contacts(contacts)
|
|
|
+
|
|
|
+# ==================== 二、聊天记录文件管理 ====================
|
|
|
+
|
|
|
+def _ensure_chat_history_dir():
|
|
|
+ if not os.path.exists(CHAT_HISTORY_DIR):
|
|
|
+ os.makedirs(CHAT_HISTORY_DIR)
|
|
|
|
|
|
-# ==================== 二、@tool 工具 ====================
|
|
|
+def get_chat_file_path(contact_name: str) -> str:
|
|
|
+ _ensure_chat_history_dir()
|
|
|
+ return os.path.join(CHAT_HISTORY_DIR, f"{contact_name}.json")
|
|
|
+
|
|
|
+def load_chat_history(contact_name: str) -> List[Dict[str, Any]]:
|
|
|
+ path = get_chat_file_path(contact_name)
|
|
|
+ if os.path.exists(path):
|
|
|
+ try:
|
|
|
+ with open(path, 'r', encoding='utf-8') as f:
|
|
|
+ return json.load(f)
|
|
|
+ except Exception:
|
|
|
+ return []
|
|
|
+ return []
|
|
|
+
|
|
|
+def save_chat_history(contact_name: str, history: List[Dict[str, Any]]):
|
|
|
+ path = get_chat_file_path(contact_name)
|
|
|
+ try:
|
|
|
+ with open(path, 'w', encoding='utf-8') as f:
|
|
|
+ json.dump(history, f, ensure_ascii=False, indent=2)
|
|
|
+ except Exception as e:
|
|
|
+ print(f"保存聊天记录失败: {e}")
|
|
|
+
|
|
|
+def update_unread_count(contact_name: str, increment: int = 1, reset: bool = False):
|
|
|
+ """更新未读消息摘要"""
|
|
|
+ _ensure_chat_history_dir()
|
|
|
+ summary = {}
|
|
|
+ if os.path.exists(UNREAD_SUMMARY_FILE):
|
|
|
+ try:
|
|
|
+ with open(UNREAD_SUMMARY_FILE, 'r', encoding='utf-8') as f:
|
|
|
+ summary = json.load(f)
|
|
|
+ except Exception:
|
|
|
+ summary = {}
|
|
|
+
|
|
|
+ if reset:
|
|
|
+ summary[contact_name] = 0
|
|
|
+ else:
|
|
|
+ summary[contact_name] = summary.get(contact_name, 0) + increment
|
|
|
+
|
|
|
+ try:
|
|
|
+ with open(UNREAD_SUMMARY_FILE, 'w', encoding='utf-8') as f:
|
|
|
+ json.dump(summary, f, ensure_ascii=False, indent=2)
|
|
|
+ except Exception as e:
|
|
|
+ print(f"更新未读摘要失败: {e}")
|
|
|
+
|
|
|
+# ==================== 三、@tool 工具 ====================
|
|
|
|
|
|
@tool(
|
|
|
display={
|
|
|
@@ -170,6 +229,21 @@ async def feishu_send_message_to_contact(
|
|
|
if last_res:
|
|
|
# 更新 chat_id
|
|
|
update_contact_chat_id(contact_name, last_res.chat_id)
|
|
|
+
|
|
|
+ # [待开启] 发送即记录：为了维护完整的聊天记录，将机器人发出的消息也保存到本地文件
|
|
|
+ # try:
|
|
|
+ # history = load_chat_history(contact_name)
|
|
|
+ # history.append({
|
|
|
+ # "role": "assistant",
|
|
|
+ # "message_id": last_res.message_id,
|
|
|
+ # "content": content if isinstance(content, list) else [{"type": "text", "text": content}]
|
|
|
+ # })
|
|
|
+ # save_chat_history(contact_name, history)
|
|
|
+ # # 机器人回复了，将该联系人的未读计数重置为 0
|
|
|
+ # update_unread_count(contact_name, reset=True)
|
|
|
+ # except Exception as e:
|
|
|
+ # print(f"记录发送的消息失败: {e}")
|
|
|
+
|
|
|
return ToolResult(
|
|
|
title=f"消息已成功发送至 {contact_name}",
|
|
|
output=f"发送成功。消息 ID: {last_res.message_id}",
|
tanjingyu