mongo.py 6.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236
  1. from pymongo import MongoClient
  2. from bson import ObjectId
  3. from datetime import datetime
  4. from dotenv import load_dotenv
  5. import os
  6. load_dotenv()
  7. MONGO_URI = os.getenv("ARK_LOGS_MONGO_URI")
  8. client = MongoClient(MONGO_URI, serverSelectionTimeoutMS=5000)
  9. # 数据库
  10. db = client["arklogs"]
  11. # 豆包大模型的对话日志
  12. chat_logs = db["chat_logs"]
  13. # 聊天历史记录
  14. chat_history_col = db["chat_history"]
  15. # 兴趣圈集合
  16. circle_prompts = db["circle_prompt"]
  17. # 历史人物集合
  18. historical_figures = db["historical_figures"]
  19. def _ensure_index():
  20. try:
  21. chat_logs.create_index([("user_id", 1), ("asked_at", -1)])
  22. chat_history_col.create_index([("user_id", 1), ("session_id", 1), ("timestamp", -1)])
  23. except Exception:
  24. pass
  25. def save_chat_log(
  26. user_id: str,
  27. question: str,
  28. stream_mode: bool,
  29. raw_response: str = None,
  30. status: str = "success",
  31. error: str = None,
  32. ):
  33. """
  34. 保存聊天原始响应日志到 MongoDB
  35. Args:
  36. user_id: 提问人
  37. question: 提问的问题
  38. stream_mode: 回答方式(流式或非流式)
  39. raw_response: API 原始响应的 repr 字符串
  40. status: 响应状态 success | error
  41. error: 异常时的错误信息
  42. """
  43. try:
  44. _ensure_index()
  45. chat_logs.insert_one({
  46. "user_id": user_id,
  47. "question": question,
  48. "stream_mode": stream_mode,
  49. "raw_response": raw_response,
  50. "status": status,
  51. "error": error,
  52. "asked_at": datetime.now(),
  53. })
  54. except Exception as e:
  55. print(f"MongoDB 日志写入失败: {e}")
  56. def save_chat_history(
  57. user_id: str,
  58. session_id: str,
  59. role: str,
  60. content: str,
  61. timestamp: datetime,
  62. response_id: str = None,
  63. thinking: str = None,
  64. searching: str = None,
  65. attachments: list = None,
  66. input_tokens: int = None,
  67. output_tokens: int = None,
  68. total_tokens: int = None,
  69. app_name:str=None,
  70. ):
  71. try:
  72. _ensure_index()
  73. doc = {
  74. "user_id": user_id,
  75. "session_id": session_id,
  76. "role": role,
  77. "content": content,
  78. "thinking": thinking,
  79. "searching": searching,
  80. "response_id": response_id,
  81. "input_tokens": input_tokens,
  82. "output_tokens": output_tokens,
  83. "total_tokens": total_tokens,
  84. "timestamp": timestamp,
  85. "app_name": app_name,
  86. }
  87. if attachments:
  88. doc["attachments"] = attachments
  89. chat_history_col.insert_one(doc)
  90. except Exception as e:
  91. print(f"MongoDB 聊天历史写入失败: {e}")
  92. def get_chat_history(user_id: str, session_id: str) -> list:
  93. try:
  94. docs = chat_history_col.find(
  95. {"user_id": user_id, "session_id": session_id},
  96. {"_id": 0}
  97. ).sort("timestamp", 1)
  98. return list(docs)
  99. except Exception as e:
  100. print(f"MongoDB 聊天历史读取失败: {e}")
  101. return []
  102. def get_last_response_id(user_id: str, session_id: str) -> str | None:
  103. try:
  104. doc = chat_history_col.find_one(
  105. {"user_id": user_id, "session_id": session_id, "role": "assistant", "response_id": {"$ne": None}},
  106. {"response_id": 1, "_id": 0},
  107. sort=[("timestamp", -1)]
  108. )
  109. return doc["response_id"] if doc else None
  110. except Exception as e:
  111. print(f"MongoDB 查询 response_id 失败: {e}")
  112. return None
  113. def delete_chat_history(user_id: str, session_id: str) -> int:
  114. try:
  115. result = chat_history_col.delete_many({"user_id": user_id, "session_id": session_id})
  116. return result.deleted_count
  117. except Exception as e:
  118. print(f"MongoDB 聊天历史删除失败: {e}")
  119. return 0
  120. def get_sessions(user_id: str) -> list:
  121. try:
  122. pipeline = [
  123. {"$match": {"user_id": user_id, "role": "user"}},
  124. {"$sort": {"timestamp": 1}},
  125. {"$group": {
  126. "_id": "$session_id",
  127. "createdAt": {"$first": "$timestamp"},
  128. "preview": {"$first": "$content"},
  129. }},
  130. {"$sort": {"createdAt": -1}},
  131. {"$project": {
  132. "_id": 0,
  133. "sessionId": "$_id",
  134. "createdAt": 1,
  135. "preview": {"$substrCP": ["$preview", 0, 20]},
  136. }},
  137. ]
  138. return list(chat_history_col.aggregate(pipeline))
  139. except Exception as e:
  140. print(f"MongoDB 会话列表查询失败: {e}")
  141. return []
  142. _DEFAULT_PROMPT_CONFIG = {
  143. "name": "兴趣圈",
  144. "role": "活跃用户",
  145. "style": "自然亲切,有活人感",
  146. "keywords": [],
  147. "forbidden": [],
  148. }
  149. def get_circle_prompt(app_name: str) -> dict:
  150. try:
  151. doc = circle_prompts.find_one({"appName": app_name})
  152. return doc if doc else _DEFAULT_PROMPT_CONFIG
  153. except Exception:
  154. return _DEFAULT_PROMPT_CONFIG
  155. def upsert_circle_prompt(data: dict) -> None:
  156. circle_prompts.update_one(
  157. {"appName": data["appName"]},
  158. {"$set": data},
  159. upsert=True,
  160. )
  161. # ===================== 历史人物 =====================
  162. def get_all_figures() -> list:
  163. try:
  164. docs = historical_figures.find({})
  165. return [{"_id": str(doc["_id"]), **{k: v for k, v in doc.items() if k != "_id"}} for doc in docs]
  166. except Exception as e:
  167. print(f"MongoDB 历史人物列表查询失败: {e}")
  168. return []
  169. def get_figure_by_id(figure_id: str) -> dict | None:
  170. try:
  171. doc = historical_figures.find_one({"_id": ObjectId(figure_id)})
  172. if doc:
  173. doc["_id"] = str(doc["_id"])
  174. return doc
  175. except Exception as e:
  176. print(f"MongoDB 历史人物查询失败: {e}")
  177. return None
  178. def insert_figure(data: dict) -> str:
  179. try:
  180. result = historical_figures.insert_one(data)
  181. return str(result.inserted_id)
  182. except Exception as e:
  183. print(f"MongoDB 历史人物新增失败: {e}")
  184. return None
  185. def update_figure(figure_id: str, data: dict) -> int:
  186. try:
  187. result = historical_figures.update_one(
  188. {"_id": ObjectId(figure_id)},
  189. {"$set": data},
  190. )
  191. return result.matched_count
  192. except Exception as e:
  193. print(f"MongoDB 历史人物修改失败: {e}")
  194. return 0
  195. def delete_figure(figure_id: str) -> int:
  196. try:
  197. result = historical_figures.delete_one({"_id": ObjectId(figure_id)})
  198. return result.deleted_count
  199. except Exception as e:
  200. print(f"MongoDB 历史人物删除失败: {e}")
  201. return 0