knowledge-agent/offline/feishu_evaluation.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import json
import requests
from typing import List, Dict, Any, Tuple
from dotenv import load_dotenv
import sys
import time
import logging
import threading
from concurrent.futures import ThreadPoolExecutor

# 添加项目根目录到系统路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# 导入项目中的openrouter模块
from llm.openrouter import OpenRouterProcessor, OpenRouterModel

# 配置日志记录
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(os.path.join(os.path.dirname(os.path.abspath(__file__)), 'feishu_evaluation.log')),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# 加载环境变量
load_dotenv()

# 检查环境变量
if not os.getenv('FEISHU_APP_ID') or not os.getenv('FEISHU_APP_SECRET') or not os.getenv('OPENROUTER_API_TOKEN'):
    logger.error("缺少必要的环境变量，请检查.env文件")
    sys.exit(1)

# 飞书API相关配置
APP_ID = os.getenv('FEISHU_APP_ID')
APP_SECRET = os.getenv('FEISHU_APP_SECRET')
SHEET_TOKEN = "ESmNsJ3FWhcfbmtvwlAcvcWMngf"
SHEET_NAME = "Zsag5j"

# 最大并发数
MAX_CONCURRENCY = 5

class FeishuAPI:
    """飞书API处理类"""
    
    def __init__(self):
        self.app_id = APP_ID
        self.app_secret = APP_SECRET
        self.base_url = "https://open.feishu.cn/open-apis"
        self.access_token = None
        self.token_expires = 0
        self.lock = threading.Lock()  # 添加线程锁保证线程安全
    
    def get_access_token(self):
        """获取飞书访问令牌"""
        # 使用线程锁确保线程安全
        with self.lock:
            # 如果令牌有效且未过期，直接返回
            if self.access_token and time.time() < self.token_expires:
                return self.access_token
                
            url = f"{self.base_url}/auth/v3/tenant_access_token/internal"
            payload = {
                "app_id": self.app_id,
                "app_secret": self.app_secret
            }
            
            try:
                response = requests.post(url, json=payload)
                result = response.json()
                if result.get("code") == 0:
                    self.access_token = result.get("tenant_access_token")
                    # 设置过期时间（提前5分钟过期）
                    self.token_expires = time.time() + result.get("expire") - 300
                    logger.info("成功获取飞书访问令牌")
                    return self.access_token
                else:
                    error_msg = f"获取飞书访问令牌失败: {result}"
                    logger.error(error_msg)
                    raise Exception(error_msg)
            except Exception as e:
                logger.error(f"获取飞书访问令牌时发生异常: {e}")
                raise
    
    def get_sheet_data(self):
        """获取飞书表格数据"""
        try:
            token = self.get_access_token()
            url = f"{self.base_url}/sheets/v2/spreadsheets/{SHEET_TOKEN}/values/{SHEET_NAME}"
            headers = {"Authorization": f"Bearer {token}"}
            
            logger.info(f"正在获取表格数据: {SHEET_TOKEN}/{SHEET_NAME}")
            response = requests.get(url, headers=headers)
            result = response.json()
            if result.get("code") == 0:
                data = result.get("data", {}).get("valueRange", {}).get("values", [])
                logger.info(f"成功获取表格数据，共 {len(data)} 行")
                return data
            else:
                error_msg = f"获取飞书表格数据失败: {result}"
                logger.error(error_msg)
                raise Exception(error_msg)
        except Exception as e:
            logger.error(f"获取表格数据时发生异常: {e}")
            raise
    
    def update_sheet_cell(self, row_index, col_index, value, max_retries=1, retry_delay=2):
        """更新飞书表格单元格，带重试机制"""
        # 计算单元格范围（例如：'H2'）
        col_letter = chr(ord('A') + col_index)
        range_str = f"{SHEET_NAME}!{col_letter}{row_index + 1}:{col_letter}{row_index + 1}"
        
        for retry in range(max_retries):
            try:
                # 每次重试都重新获取token，以防token过期
                token = self.get_access_token()
                url = f"{self.base_url}/sheets/v2/spreadsheets/{SHEET_TOKEN}/values"
                
                payload = {
                    "valueRange": {
                        "range": range_str,
                        "values": [[value]]
                    }
                }
                
                headers = {"Authorization": f"Bearer {token}"}
                
                logger.debug(f"正在更新单元格: {range_str} (尝试 {retry+1}/{max_retries})")
                response = requests.put(url, headers=headers, json=payload)
                result = response.json()
                
                if result.get("code") == 0:
                    logger.debug(f"成功更新单元格: {range_str}")
                    return True
                elif result.get("code") == 91403 and retry < max_retries - 1:  # Forbidden错误，可能是权限问题
                    logger.warning(f"更新单元格 {range_str} 返回Forbidden错误，将在 {retry_delay} 秒后重试")
                    time.sleep(retry_delay)
                    # 下次重试延迟时间加倍
                    retry_delay *= 2
                else:
                    logger.warning(f"更新单元格 {range_str} 失败: {result}")
                    return False
                    
            except Exception as e:
                if retry < max_retries - 1:
                    logger.warning(f"更新单元格 {range_str} 时发生异常: {e}，将在 {retry_delay} 秒后重试")
                    time.sleep(retry_delay)
                    # 下次重试延迟时间加倍
                    retry_delay *= 2
                else:
                    logger.error(f"更新单元格 {range_str} 时发生异常: {e}，已达到最大重试次数")
                    return False
        
        return False  # 所有重试都失败

class EvaluationProcessor:
    """评估处理类"""
    
    def __init__(self):
        # 加载评估和打分的prompt
        self.consistency_prompt = self._load_prompt("p1.md")
        self.scoring_prompt = self._load_prompt("p2.md")
        # 初始化OpenRouter处理器
        self.llm_processor = OpenRouterProcessor(OpenRouterModel.GEMINI_25_FLASH)
        
    def _load_prompt(self, filename):
        """加载prompt文件"""
        current_dir = os.path.dirname(os.path.abspath(__file__))
        file_path = os.path.join(current_dir, filename)
        
        with open(file_path, 'r', encoding='utf-8') as file:
            return file.read()
    
    def evaluate_consistency(self, keyword, structured_result):
        """评估一致性"""
        try:
            input_data = {
                "query": keyword,
                "query结果文本": structured_result
            }
            
            logger.info(f"正在进行一致性评估: {keyword[:30]}...")
            # 调用LLM进行一致性评估
            result = self.llm_processor.process(input_data, self.consistency_prompt)
            
            try:
                # 尝试解析JSON结果
                # 处理可能的不完整JSON字符串
                result = result.strip()
                if result.startswith('```json') and '```' in result:
                    # 提取JSON部分
                    json_str = result.split('```json', 1)[1].split('```', 1)[0].strip()
                    json_result = json.loads(json_str)
                else:
                    json_result = json.loads(result)
                    
                consistency = json_result.get("consistency", "")
                reason = json_result.get("reason", [])
                reason_str = "\n".join(reason) if isinstance(reason, list) else str(reason)
                logger.info(f"一致性评估结果: {consistency}")
                return consistency, reason_str
            except json.JSONDecodeError as e:
                # 如果结果不是有效的JSON，尝试修复并重新解析
                logger.warning(f"一致性评估结果解析失败: {result[:200]}... 错误: {e}")
                try:
                    # 尝试从文本中提取JSON部分
                    if '{' in result and '}' in result:
                        json_part = result[result.find('{'):result.rfind('}')+1]
                        json_result = json.loads(json_part)
                        consistency = json_result.get("consistency", "")
                        reason = json_result.get("reason", [])
                        reason_str = "\n".join(reason) if isinstance(reason, list) else str(reason)
                        logger.info(f"修复后解析成功，一致性评估结果: {consistency}")
                        return consistency, reason_str
                except:
                    pass
                return "解析错误", result[:500]  # 限制返回长度
        except Exception as e:
            logger.error(f"一致性评估过程中发生异常: {e}")
            return "评估异常", str(e)
    
    def evaluate_score(self, keyword, structured_result):
        """评估打分"""
        try:
            input_data = {
                "query词": keyword,
                "文本知识": structured_result
            }
            
            logger.info(f"正在进行打分评估: {keyword[:30]}...")
            # 调用LLM进行打分评估
            result = self.llm_processor.process(input_data, self.scoring_prompt)
            
            try:
                # 尝试解析JSON结果
                # 处理可能的不完整JSON字符串
                result = result.strip()
                if result.startswith('```json') and '```' in result:
                    # 提取JSON部分
                    json_str = result.split('```json', 1)[1].split('```', 1)[0].strip()
                    json_result = json.loads(json_str)
                else:
                    json_result = json.loads(result)
                    
                score = json_result.get("分数", "")
                reason = json_result.get("原因", {})
                # 将原因字典转换为字符串
                reason_str = "\n".join([f"{k}: {v}" for k, v in reason.items()])
                logger.info(f"打分评估结果: {score}")
                return score, reason_str
            except json.JSONDecodeError as e:
                # 如果结果不是有效的JSON，尝试修复并重新解析
                logger.warning(f"打分评估结果解析失败: {result[:200]}... 错误: {e}")
                try:
                    # 尝试从文本中提取JSON部分
                    if '{' in result and '}' in result:
                        json_part = result[result.find('{'):result.rfind('}')+1]
                        json_result = json.loads(json_part)
                        score = json_result.get("分数", "")
                        reason = json_result.get("原因", {})
                        reason_str = "\n".join([f"{k}: {v}" for k, v in reason.items()])
                        logger.info(f"修复后解析成功，打分评估结果: {score}")
                        return score, reason_str
                except:
                    pass
                return "解析错误", result[:500]  # 限制返回长度
        except Exception as e:
            logger.error(f"打分评估过程中发生异常: {e}")
            return "评估异常", str(e)

def process_row(row_index, row_data, feishu_api, evaluator):
    """处理单行数据"""
    try:
        # 获取关键词（C列）和结构化结果（G列）
        keyword = row_data[2] if len(row_data) > 2 else ""
        structured_result = row_data[6] if len(row_data) > 6 else ""
        
        if not keyword or not structured_result:
            logger.warning(f"行 {row_index+1} 数据不完整，跳过处理")
            return
        
        logger.info(f"开始处理行 {row_index+1}: {keyword[:30]}...")
        
        # 检查H列和J列是否已有数据，如果有则跳过处理
        h_value = row_data[7] if len(row_data) > 7 else ""
        j_value = row_data[9] if len(row_data) > 9 else ""
        
        # 评估一致性（如果H列为空）
        if not h_value:
            consistency, consistency_reason = evaluator.evaluate_consistency(keyword, structured_result)
            
            # 更新一致性结果（H列和I列）
            feishu_api.update_sheet_cell(row_index, 7, consistency)  # H列
            feishu_api.update_sheet_cell(row_index, 8, consistency_reason)  # I列
        else:
            logger.info(f"行 {row_index+1} 的一致性评估结果已存在，跳过处理")
        
        # 评估打分（如果J列为空）
        if not j_value:
            score, score_reason = evaluator.evaluate_score(keyword, structured_result)
            
            # 更新打分结果（J列和K列）
            feishu_api.update_sheet_cell(row_index, 9, score)  # J列
            feishu_api.update_sheet_cell(row_index, 10, score_reason)  # K列
        else:
            logger.info(f"行 {row_index+1} 的打分评估结果已存在，跳过处理")
        
        logger.info(f"行 {row_index+1} 处理完成")
        
    except Exception as e:
        logger.error(f"处理行 {row_index+1} 时出错: {e}", exc_info=True)

def main():
    """主函数"""
    try:
        logger.info("开始执行飞书表格评估程序")
        
        # 初始化飞书API和评估处理器
        feishu_api = FeishuAPI()
        evaluator = EvaluationProcessor()
        
        # 获取表格数据
        sheet_data = feishu_api.get_sheet_data()
        
        if not sheet_data or len(sheet_data) <= 1:  # 考虑表头
            logger.warning("表格数据为空或只有表头")
            return
        
        logger.info(f"共获取到 {len(sheet_data)-1} 行数据（不含表头）")
        
        # 创建线程池
        with ThreadPoolExecutor(max_workers=MAX_CONCURRENCY) as executor:
            # 跳过表头，处理数据行
            futures = []
            for i, row in enumerate(sheet_data[1:], 1):  # 从索引1开始（跳过表头）
                future = executor.submit(process_row, i, row, feishu_api, evaluator)
                futures.append(future)
            
            # 等待所有任务完成
            for future in futures:
                try:
                    future.result()  # 获取任务结果，如果有异常会在这里抛出
                except Exception as e:
                    logger.error(f"任务执行失败: {e}", exc_info=True)
        
        logger.info("所有数据处理完成")
    except Exception as e:
        logger.error(f"程序执行过程中发生异常: {e}", exc_info=True)
        sys.exit(1)

if __name__ == "__main__":
    # 运行主函数
    main()