CardVideoSummary/app/services/llm_service.py

import requests
import json
import re
from app.core.config import settings
from app.core.logger import get_logger
from app.schemas.models import CardInfoOutput

logger = get_logger("LLMService")


class LLMService:
    def split_text_with_overlap(self, text, chunk_size, overlap):
        start = 0
        text_len = len(text)
        while start < text_len:
            end = min(start + chunk_size, text_len)
            yield text[start:end]
            if end == text_len:
                break
            start += (chunk_size - overlap)

    def parse_ai_json(self, json_str):
        if not json_str: return []
        cleaned_str = re.sub(r"```json\s*", "", json_str)
        cleaned_str = re.sub(r"```\s*", "", cleaned_str).strip()
        try:
            return json.loads(cleaned_str)
        except json.JSONDecodeError:
            logger.warning(f"⚠️ 解析 JSON 失败，AI 返回原始内容: \n{cleaned_str}\n")
            return []

    def send_to_ai(self, text_chunk):
        headers = {
            "Authorization": settings.API_KEY,
            "Content-Type": "application/json"
        }
        payload = {
            "inputs": {"question": text_chunk},
            "response_mode": "blocking",
            "user": settings.USER_ID
        }
        try:
            response = requests.post(settings.API_URL, headers=headers, json=payload)
            response.raise_for_status()
            return response.json().get('data', {}).get('outputs', {}).get('result', "[]")
        except Exception as e:
            logger.error(f"❌ 请求发生错误: {e}")
            return "[]"

    def process_text(self, full_text: str) -> list[CardInfoOutput]:
        logger.info(f"📄 开始处理文本，原文总长度: {len(full_text)} 字符")

        chunks = list(self.split_text_with_overlap(full_text, settings.CHUNK_SIZE, settings.OVERLAP_SIZE))
        total_chunks = len(chunks)
        logger.info(
            f"✂️ 将分割为 {total_chunks} 个片段进行处理 (Size: {settings.CHUNK_SIZE}, Overlap: {settings.OVERLAP_SIZE})")

        all_hits = []

        for i, chunk in enumerate(chunks):
            logger.info(f"⏳ [进度 {i + 1}/{total_chunks}] 正在发送请求，片段长度 {len(chunk)}...")

            ai_resp = self.send_to_ai(chunk)
            chunk_hits = self.parse_ai_json(ai_resp)

            if chunk_hits:
                logger.info(f"   ✅ 第 {i + 1} 段识别到 {len(chunk_hits)} 张卡片")
                all_hits.extend(chunk_hits)
            else:
                logger.info(f"   ⚪ 第 {i + 1} 段未发现目标")

        # 去重
        seen = set()
        unique_results = []
        for hit in all_hits:
            try:
                # 转换为输出模型
                card = CardInfoOutput(**hit)
                # key: 时间 + 英文名
                key = (card.time, card.card_name_en)
                if key not in seen:
                    seen.add(key)
                    unique_results.append(card)
            except Exception as e:
                logger.error(f"⚠️ 数据校验失败: {e}")

        logger.info(f"🎉 处理完成! 原始识别 {len(all_hits)} 条，去重后 {len(unique_results)} 条")
        return unique_results