# 读取视频分析报告1.xlsx，调用coze工作流，获取需求列表 import pandas as pd import orjson import os from 视频理解任务 import GoogleVideoAnalyzer from prompt.prompt import HOOK_EXTRACTION_PROMPT_V2 import requests import json from datetime import datetime import re def get_first_row_and_call_coze(): import json import requests from datetime import datetime try: # 读取Excel文件 print(f"[{datetime.now()}] 开始读取Excel文件...") df = pd.read_excel('视频分析报告1.xlsx') # 获取第一行数据 first_row = df.iloc[0] # 获取第6、7、8列的数据（对应F、G、H列） summary = first_row.iloc[5] if pd.notna(first_row.iloc[5]) else "{}" timeline = first_row.iloc[7] if pd.notna(first_row.iloc[7]) else "{}" video_url = first_row.iloc[3] if pd.notna(first_row.iloc[3]) else "" # 准备数据 - 确保数据格式正确 data = { "summary": summary, "timeline": timeline } # 调用Coze工作流 url = "https://api.coze.cn/v1/workflow/run" headers = { "Content-Type": "application/json", "Authorization": "Bearer pat_ddPm5K5tCKXU2zH1NChGHoOKGOSECyzxmXPEcrtJg52UcCIndRAfiWRRxCH95pdB" } payload = { "workflow_id": "7507245138873450535", "parameters": data } print(f"[{datetime.now()}] 开始调用Coze工作流...") print(f"请求URL: {url}") print(f"请求头: {headers}") print(f"请求数据: {json.dumps(payload, ensure_ascii=False, indent=2)}") response = requests.post(url, json=payload, headers=headers, timeout=60) # 打印响应状态和头信息 print(f"\n响应状态码: {response.status_code}") print(f"响应头: {dict(response.headers)}") # 尝试获取响应内容 try: response_text = response.text print(f"响应内容: {response_text}") # 检查响应内容是否为空 if not response_text.strip(): print("警告: 响应内容为空") return None # 尝试解析JSON try: result = response.json() print(f"\n[{datetime.now()}] Coze工作流返回结果:") print(json.dumps(result, ensure_ascii=False, indent=2)) return result except json.JSONDecodeError as je: print(f"JSON解析错误: {str(je)}") print(f"原始响应内容: {response_text}") return None except Exception as e: print(f"读取响应内容时出错: {str(e)}") return None except pd.errors.EmptyDataError: print("错误: Excel文件为空") except requests.exceptions.RequestException as e: print(f"请求错误: {str(e)}") if hasattr(e, 'response') and e.response is not None: print(f"错误响应状态码: {e.response.status_code}") print(f"错误响应内容: {e.response.text}") except Exception as e: print(f"发生未知错误: {str(e)}") return None def call_coze_api(summary, timeline): url = "https://api.coze.cn/v1/workflow/run" headers = { "Content-Type": "application/json", "Authorization": "Bearer pat_ddPm5K5tCKXU2zH1NChGHoOKGOSECyzxmXPEcrtJg52UcCIndRAfiWRRxCH95pdB" } payload = { "workflow_id": "7511970477477904393", "parameters": { "summary": summary, "timeline": timeline } } response = requests.post(url, json=payload, headers=headers, timeout=600) response.raise_for_status() return response.json() def extract_fields_from_response(resp): """ 从响应中提取字段数据 Args: resp: API响应字典，可能包含多层嵌套的JSON字符串 Returns: tuple: (time_str, query_str, hook_str) 三个字符串，分别包含时间、问题和钩子话术 """ import json import re import logging logger = logging.getLogger(__name__) def extract_json_from_markdown(markdown_str): """从markdown代码块中提取JSON字符串""" # 匹配 ```json 和 ``` 之间的内容 match = re.search(r'```(?:json)?\s*([\s\S]*?)```', markdown_str) if match: return match.group(1).strip() return markdown_str.strip() def parse_nested_json(json_str): """递归解析嵌套的JSON字符串""" try: # 尝试直接解析 return json.loads(json_str) except json.JSONDecodeError: try: # 如果失败，尝试解析转义的JSON字符串 unescaped = json_str.encode().decode('unicode_escape') return json.loads(unescaped) except: # 如果还是失败，返回原始字符串 return json_str try: # 处理新的API响应格式 if isinstance(resp, dict) and 'data' in resp: # 如果data是字符串，尝试解析它 if isinstance(resp['data'], str): try: data = json.loads(resp['data']) if isinstance(data, dict) and 'output' in data: current_data = data['output'] else: current_data = data except json.JSONDecodeError: current_data = resp['data'] else: current_data = resp['data'] else: current_data = resp # 处理多层嵌套的JSON while isinstance(current_data, (str, dict)): if isinstance(current_data, dict): # 如果是字典，尝试获取data或output字段 if 'data' in current_data: current_data = current_data['data'] elif 'output' in current_data: current_data = current_data['output'] else: break elif isinstance(current_data, str): # 如果是字符串，尝试解析JSON try: parsed = parse_nested_json(current_data) if parsed == current_data: # 如果解析结果和输入相同，说明不是JSON break current_data = parsed except: break # 如果最终结果是字符串，尝试从markdown中提取 if isinstance(current_data, str): current_data = extract_json_from_markdown(current_data) current_data = parse_nested_json(current_data) logger.info(f"Final parsed data type: {type(current_data)}") logger.info(f"Final data preview: {str(current_data)[:200]}...") # 确保数据是列表 if not isinstance(current_data, list): logger.warning(f"Converting non-list data to list: {type(current_data)}") current_data = [current_data] # 提取并验证每个对象 time_list = [] query_list = [] hook_list = [] for item in current_data: if not isinstance(item, dict): logger.warning(f"Skipping invalid item: {item}") continue try: # 使用新的中文字段名获取值 time = str(item.get('钩子出现时间', '')).strip() query = str(item.get('钩子到AI大模型的问题', '')).strip() hook = str(item.get('钩子话术', '')).strip() if time or query or hook: # 只添加非空值 time_list.append(time) query_list.append(query) hook_list.append(hook) except Exception as e: logger.warning(f"Error extracting fields from item: {e}") continue # 将列表转换为字符串，用换行符连接 time_str = '\n'.join(time_list) if time_list else '' query_str = '\n'.join(query_list) if query_list else '' hook_str = '\n'.join(hook_list) if hook_list else '' logger.info(f"Extracted - Time: {time_str[:50]}..., Query: {query_str[:50]}..., Hook: {hook_str[:50]}...") return time_str, query_str, hook_str except json.JSONDecodeError as je: logger.error(f"JSON decode error: {str(je)}") logger.error(f"Error position: {je.pos}") logger.error(f"Character at error: {repr(str(current_data)[je.pos:je.pos+10])}") return '', '', '' except Exception as e: logger.error(f"Error processing response: {str(e)}") logger.error(f"Response data: {str(resp)[:200]}...") return '', '', '' def process_excel(input_excel, output_excel=None): """ 处理Excel文件，调用API并更新数据 """ try: df = pd.read_excel(input_excel) # 确保目标列存在且为字符串类型 for col in [8, 9, 10]: # 对应第9、10、11列 if col >= len(df.columns): df.insert(col, f'Column_{col}', '') df.iloc[:, col] = df.iloc[:, col].fillna('').astype(str) for idx, row in df.iterrows(): try: print(f"\n开始处理第{idx+1}行") summary = str(row.iloc[5]) if pd.notna(row.iloc[5]) else "{}" timeline = str(row.iloc[6]) if pd.notna(row.iloc[6]) else "{}" print(f"Summary: {summary[:100]}...") # 只打印前100个字符 print(f"Timeline: {timeline[:100]}...") resp = call_coze_api(summary, timeline) print(f"API Response: {json.dumps(resp, ensure_ascii=False)}") # 打印API响应 time, query, hook = extract_fields_from_response(resp) # 更新数据 - 使用单个值赋值而不是批量赋值 df.at[idx, df.columns[7]] = time df.at[idx, df.columns[8]] = hook df.at[idx, df.columns[9]] = query print(f"第{idx+1}行处理完成") print(f"Time: {time[:100]}...") print(f"Hook: {hook[:100]}...") print(f"Query: {query[:100]}...") except Exception as e: print(f"第{idx+1}行处理失败: {str(e)}") df.at[idx, df.columns[7]] = "error" df.at[idx, df.columns[8]] = "error" df.at[idx, df.columns[9]] = "error" # 每处理一行就保存一次，防止中断丢失 df.to_excel(input_excel, index=False) print(f"已保存到第{idx+1}行") print("全部处理完成，已保存。") except Exception as e: print(f"处理Excel文件时发生错误: {str(e)}") raise if __name__ == "__main__": process_excel("文件/视频分析报告_new.xlsx")