video-comprehension/获取trigger任务.py

# 读取视频分析报告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": "7507245138873450535",
        "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):
    import re
    
    # Define patterns at the function level
    JSON_PATTERNS = [
        r"```json\\n(.*?)```",  # 转义的换行
        r"```json\n(.*?)```",   # 普通换行
        r"```(.*?)```",         # 无语言标记
        r"\{.*\}"               # 直接JSON对象
    ]
    
    def try_unescape_json_string(s):
        # 递归反序列化所有层级的转义JSON字符串
        for _ in range(3):  # 最多尝试3层
            if isinstance(s, str):
                try:
                    s2 = json.loads(s)
                    # 如果反序列化后类型有变化，继续递归
                    if type(s2) != str:
                        s = s2
                    else:
                        break
                except Exception as e:
                    print(f"JSON反序列化失败: {str(e)}")
                    break
            else:
                break
        return s

    def extract_json_from_string(s):
        """Helper function to extract and parse JSON from a string"""
        if not isinstance(s, str):
            return s
            
        # First try direct JSON parsing
        try:
            return json.loads(s)
        except json.JSONDecodeError:
            pass
            
        # Then try each pattern
        for pattern in JSON_PATTERNS:
            json_str = re.search(pattern, s, re.DOTALL)
            if json_str:
                try:
                    content = json_str.group(1)
                    return json.loads(content)
                except Exception as e:
                    print(f"使用模式 {pattern} 解析失败: {str(e)}")
                    continue
        return s

    try:
        data = resp.get("data")
        if not data:
            print("响应中没有data字段")
            return ("", "", "")
            
        
        # First parse the outer JSON structure
        try:
            data = json.loads(data)
        except json.JSONDecodeError as e:
            print(f"解析外层data失败: {str(e)}")
            return ("", "", "")
            
        # Then handle the output field
        output = data.get("output")
        if not output:
            print("data中没有output字段")
            return ("", "", "")
            
        print(f"\n原始output字段: {output}")
        output = extract_json_from_string(output)
        
        if isinstance(output, str):
            print("output解析后仍为字符串")
            return ("", "", "")
            

        if isinstance(output, dict):
            # 按优先级检查不同的字段名
            if "需求列表" in output:
                demand_list = output["需求列表"]
            elif "questions" in output:
                demand_list = output["questions"]
            elif "interactive_questions" in output:
                demand_list = output["interactive_questions"]
            else:
                print("output中没有找到需求列表、questions或interactive_questions字段")
                return ("", "", "")
        else:
            demand_list = output

        if not demand_list or not isinstance(demand_list, list):
            print(f"需求列表无效: {demand_list}")
            return ("", "", "")

        times = []
        queries = []
        hooks = []
        
        for item in demand_list:
            if not isinstance(item, dict):
                print(f"跳过非字典项: {item}")
                continue
                
            time = item.get("需求钩子出现时间", "")
            query = item.get("需求详细query", "")
            hook = item.get("需求钩子话术", "")
            
            if not all([time, query, hook]):
                print(f"跳过不完整项: {item}")
                continue
                
            # Handle time format
            if time == "end":
                time = "视频结束"
            elif "-" in time:
                time = time.split("-")[0]  # Take the start time
                
            times.append(time)
            queries.append(query)
            hooks.append(hook)

        if not times:
            print("没有提取到有效的需求项")
            return ("", "", "")

        return ("\n".join(times), "\n".join(queries), "\n".join(hooks))

    except Exception as e:
        print(f"解析返回数据出错: {str(e)}")
        print(f"原始响应: {json.dumps(resp, ensure_ascii=False, indent=2)}")
    return ("", "", "")

def process_excel(input_excel, output_excel=None):
    df = pd.read_excel(input_excel)
    for idx, row in df.iterrows():
        summary = row.iloc[5] if pd.notna(row.iloc[5]) else "{}"
        timeline = row.iloc[7] if pd.notna(row.iloc[7]) else "{}"
        try:
            print(f"开始处理第{idx+1}行")
            resp = call_coze_api(summary, timeline)
            time, query, hook = extract_fields_from_response(resp)
            df.iat[idx, 9] = time    # 第8列
            df.iat[idx, 10] = query   # 第9列
            df.iat[idx, 11] = hook    # 第10列
            print(f"第{idx+1}行处理完成")
            print(hook)
        except Exception as e:
            print(f"第{idx+1}行处理失败: {e}")
            df.iat[idx, 9] = "error"
            df.iat[idx, 10] = "error"
            df.iat[idx, 11] = "error"
        # 每处理一行就保存一次，防止中断丢失
        df.to_excel(input_excel, index=False)
        print(f"已保存到第{idx+1}行")
    df.to_excel(input_excel, index=False)
    print("全部处理完成，已保存。")

# 读取视频分析报告1.xlsx，找出第10列为空的行，重新调用coze工作流分析
def process_empty_rows(input_excel, output_excel=None):
    df = pd.read_excel(input_excel)
    for idx, row in df.iterrows():
        if pd.notna(row.iloc[10]):
            continue
        summary = row.iloc[5] if pd.notna(row.iloc[5]) else "{}"
        timeline = row.iloc[7] if pd.notna(row.iloc[7]) else "{}"
        try:
            print(f"开始处理第{idx+1}行")
            resp = call_coze_api(summary, timeline)
            time, query, hook = extract_fields_from_response(resp)
            df.iat[idx, 9] = time    # 第8列
            df.iat[idx, 10] = query   # 第9列
            df.iat[idx, 11] = hook    # 第10列
            print(f"第{idx+1}行处理完成")
            print(hook)
        except Exception as e:
            print(f"第{idx+1}行处理失败: {e}")
            df.iat[idx, 9] = "error"
            df.iat[idx, 10] = "error"
            df.iat[idx, 11] = "error"
        # 每处理一行就保存一次，防止中断丢失
        df.to_excel(input_excel, index=False)
        print(f"已保存到第{idx+1}行")
    df.to_excel(input_excel, index=False)
    print("全部处理完成，已保存。")

if __name__ == "__main__":
    process_excel("文件/视频分析报告_new.xlsx")