#!/usr/bin/env python # -*- coding: utf-8 -*- """ 内容分析模块 提供自动关键词提取、文本摘要和标签生成功能 """ import jieba import jieba.analyse from collections import Counter from typing import List, Tuple, Dict from config import settings from model_manager import model_router from exceptions import GenerationError from cache_manager import cache_manager import hashlib class ContentAnalyzer: def __init__(self): # 初始化jieba jieba.initialize() # 使用模型路由器,不需要单独初始化客户端 self.client = None def extract_keywords(self, text: str, top_k: int = 3) -> List[str]: """ 提取文本关键词 Args: text: 输入文本 top_k: 返回关键词数量 Returns: 关键词列表 """ # 使用jieba的TF-IDF算法提取关键词 keywords = jieba.analyse.extract_tags(text, topK=top_k, withWeight=False) return keywords def generate_summary(self, text: str, sentences: int = 3) -> str: """ 生成文本摘要 Args: text: 输入文本 sentences: 摘要句子数 Returns: 文本摘要 """ # 生成缓存键 cache_key = f"summary:{hashlib.md5(f'{text}:{sentences}'.encode()).hexdigest()}" # 尝试从缓存获取结果 cached_result = cache_manager.get(cache_key) if cached_result is not None: return cached_result # 如果没有配置模型提供商,使用简单的句子抽取方法 if not model_router.providers: result = self._simple_summary(text, sentences) # 缓存结果 cache_manager.put(cache_key, result) return result try: # 使用OpenAI API生成摘要 prompt = f"""请为以下文本生成一个{sentences}句话的摘要: {text} 要求: 1. 保留最重要的信息 2. 语言简洁明了 3. 不超过{sentences}句话 """ summary = model_router.generate_text( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": "你是一个专业的文本摘要生成器。"}, {"role": "user", "content": prompt} ], max_tokens=200, temperature=0.3 ) result = summary.strip() if summary else "" # 缓存结果 cache_manager.put(cache_key, result) return result except Exception as e: # 如果API调用失败,回退到简单方法 print(f"OpenAI API调用失败: {e}") result = self._simple_summary(text, sentences) # 缓存结果 cache_manager.put(cache_key, result) return result def _simple_summary(self, text: str, sentences: int = 3) -> str: """ 简单的文本摘要方法(基于句子长度和位置) Args: text: 输入文本 sentences: 摘要句子数 Returns: 文本摘要 """ # 简单的句子分割 import re sentence_endings = r'[。!?.!?]' sentences_list = re.split(sentence_endings, text) sentences_list = [s.strip() for s in sentences_list if s.strip()] if len(sentences_list) <= sentences: return text # 选择前几句作为摘要 summary_sentences = sentences_list[:sentences] return '。'.join(summary_sentences) + '。' def generate_tags(self, text: str, top_k: int = 3) -> List[str]: """ 生成文本标签 Args: text: 输入文本 top_k: 返回标签数量 Returns: 标签列表 """ # 提取关键词作为标签 keywords = self.extract_keywords(text, top_k) # 可以添加更多标签生成逻辑 # 例如基于词性标注、实体识别等 return keywords def analyze_content(self, text: str) -> Dict: """ 综合分析文本内容 Args: text: 输入文本 Returns: 包含关键词、摘要和标签的字典 """ keywords = self.extract_keywords(text, 3) summary = self.generate_summary(text, 3) tags = self.generate_tags(text, 3) return { "keywords": keywords, "summary": summary, "tags": tags } # 创建全局实例 analyzer = ContentAnalyzer()