coze_hook.py 4.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133
  1. import requests
  2. import json
  3. class CozeHook(object):
  4. def __init__(self):
  5. self.url = "https://api.coze.cn/v1/workflow/run"
  6. self.headers = {
  7. "Content-Type": "application/json",
  8. "Authorization": "Bearer pat_ddPm5K5tCKXU2zH1NChGHoOKGOSECyzxmXPEcrtJg52UcCIndRAfiWRRxCH95pdB"
  9. }
  10. self.hook_id = "7507245138873450535"
  11. def call_coze_api(self, summary, timeline):
  12. url = self.url
  13. headers = self.headers
  14. payload = {
  15. "workflow_id": self.hook_id,
  16. "parameters": {
  17. "summary": summary,
  18. "timeline": timeline
  19. }
  20. }
  21. response = requests.post(url, json=payload, headers=headers, timeout=600)
  22. response.raise_for_status()
  23. return response.json()
  24. def extract_fields_from_response(self, resp):
  25. import re
  26. # Define patterns at the function level
  27. JSON_PATTERNS = [
  28. r"```json\\n(.*?)```", # 转义的换行
  29. r"```json\n(.*?)```", # 普通换行
  30. r"```(.*?)```", # 无语言标记
  31. r"\{.*\}" # 直接JSON对象
  32. ]
  33. def try_unescape_json_string(s):
  34. # 递归反序列化所有层级的转义JSON字符串
  35. for _ in range(3): # 最多尝试3层
  36. if isinstance(s, str):
  37. try:
  38. s2 = json.loads(s)
  39. # 如果反序列化后类型有变化,继续递归
  40. if type(s2) != str:
  41. s = s2
  42. else:
  43. break
  44. except Exception as e:
  45. print(f"JSON反序列化失败: {str(e)}")
  46. break
  47. else:
  48. break
  49. return s
  50. def extract_json_from_string(s):
  51. """Helper function to extract and parse JSON from a string"""
  52. if not isinstance(s, str):
  53. return s
  54. # First try direct JSON parsing
  55. try:
  56. return json.loads(s)
  57. except json.JSONDecodeError:
  58. pass
  59. # Then try each pattern
  60. for pattern in JSON_PATTERNS:
  61. json_str = re.search(pattern, s, re.DOTALL)
  62. if json_str:
  63. try:
  64. content = json_str.group(1)
  65. return json.loads(content)
  66. except Exception as e:
  67. print(f"使用模式 {pattern} 解析失败: {str(e)}")
  68. continue
  69. return s
  70. try:
  71. data = resp.get("data")
  72. if not data:
  73. print("响应中没有data字段")
  74. return ("", "", "")
  75. # First parse the outer JSON structure
  76. try:
  77. data = json.loads(data)
  78. except json.JSONDecodeError as e:
  79. print(f"解析外层data失败: {str(e)}")
  80. return ("", "", "")
  81. # Then handle the output field
  82. output = data.get("output")
  83. if not output:
  84. print("data中没有output字段")
  85. return ("", "", "")
  86. print(f"\n原始output字段: {output}")
  87. output = extract_json_from_string(output)
  88. if isinstance(output, str):
  89. print("output解析后仍为字符串")
  90. return ("", "", "")
  91. if isinstance(output, dict):
  92. # 按优先级检查不同的字段名
  93. if "需求列表" in output:
  94. demand_list = output["需求列表"]
  95. elif "questions" in output:
  96. demand_list = output["questions"]
  97. elif "interactive_questions" in output:
  98. demand_list = output["interactive_questions"]
  99. else:
  100. print("output中没有找到需求列表、questions或interactive_questions字段")
  101. return []
  102. else:
  103. demand_list = output
  104. if not demand_list or not isinstance(demand_list, list):
  105. print(f"需求列表无效: {demand_list}")
  106. return demand_list
  107. except Exception as e:
  108. print(f"解析返回数据出错: {str(e)}")
  109. print(f"原始响应: {json.dumps(resp, ensure_ascii=False, indent=2)}")
  110. return []
  111. def run(self, summary, timeline):
  112. resp = self.call_coze_api(summary, timeline)
  113. list = self.extract_fields_from_response(resp)
  114. return list