init

.idea/.gitignore

@@ -0,0 +1,8 @@
+# 默认忽略的文件
+/shelf/
+/workspace.xml
+# 基于编辑器的 HTTP 客户端请求
+/httpRequests/
+# Datasource local storage ignored files
+/dataSources/
+/dataSources.local.xml

.idea/CardVideoSummary.iml

@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="PYTHON_MODULE" version="4">
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$" />
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>

.idea/inspectionProfiles/Project_Default.xml

@@ -0,0 +1,59 @@
+<component name="InspectionProjectProfileManager">
+  <profile version="1.0">
+    <option name="myName" value="Project Default" />
+    <inspection_tool class="DuplicatedCode" enabled="true" level="WEAK WARNING" enabled_by_default="true">
+      <Languages>
+        <language minSize="102" name="Python" />
+      </Languages>
+    </inspection_tool>
+    <inspection_tool class="Eslint" enabled="true" level="WARNING" enabled_by_default="true" />
+    <inspection_tool class="PyPackageRequirementsInspection" enabled="true" level="WARNING" enabled_by_default="true">
+      <option name="ignoredPackages">
+        <value>
+          <list size="31">
+            <item index="0" class="java.lang.String" itemvalue="webargs" />
+            <item index="1" class="java.lang.String" itemvalue="transformers" />
+            <item index="2" class="java.lang.String" itemvalue="timm" />
+            <item index="3" class="java.lang.String" itemvalue="fluent-logger" />
+            <item index="4" class="java.lang.String" itemvalue="towhee" />
+            <item index="5" class="java.lang.String" itemvalue="flask_restful" />
+            <item index="6" class="java.lang.String" itemvalue="opencv_python" />
+            <item index="7" class="java.lang.String" itemvalue="fastapi" />
+            <item index="8" class="java.lang.String" itemvalue="seaborn" />
+            <item index="9" class="java.lang.String" itemvalue="matplotlib" />
+            <item index="10" class="java.lang.String" itemvalue="minio" />
+            <item index="11" class="java.lang.String" itemvalue="ipython" />
+            <item index="12" class="java.lang.String" itemvalue="torch" />
+            <item index="13" class="java.lang.String" itemvalue="uvicorn" />
+            <item index="14" class="java.lang.String" itemvalue="python-multipart" />
+            <item index="15" class="java.lang.String" itemvalue="torchvision" />
+            <item index="16" class="java.lang.String" itemvalue="pymilvus" />
+            <item index="17" class="java.lang.String" itemvalue="psutil" />
+            <item index="18" class="java.lang.String" itemvalue="ultralytics" />
+            <item index="19" class="java.lang.String" itemvalue="picamera2" />
+            <item index="20" class="java.lang.String" itemvalue="posix_ipc" />
+            <item index="21" class="java.lang.String" itemvalue="websocket-client" />
+            <item index="22" class="java.lang.String" itemvalue="yolov10" />
+            <item index="23" class="java.lang.String" itemvalue="kornia" />
+            <item index="24" class="java.lang.String" itemvalue="prettytable" />
+            <item index="25" class="java.lang.String" itemvalue="huggingface_hub" />
+            <item index="26" class="java.lang.String" itemvalue="PIL" />
+            <item index="27" class="java.lang.String" itemvalue="sklearn" />
+            <item index="28" class="java.lang.String" itemvalue="faster_whisper" />
+            <item index="29" class="java.lang.String" itemvalue="pyserial" />
+            <item index="30" class="java.lang.String" itemvalue="requests" />
+          </list>
+        </value>
+      </option>
+    </inspection_tool>
+    <inspection_tool class="PyPep8NamingInspection" enabled="true" level="WEAK WARNING" enabled_by_default="true">
+      <option name="ignoredErrors">
+        <list>
+          <option value="N803" />
+          <option value="N802" />
+          <option value="N806" />
+        </list>
+      </option>
+    </inspection_tool>
+  </profile>
+</component>

.idea/inspectionProfiles/profiles_settings.xml

@@ -0,0 +1,6 @@
+<component name="InspectionProjectProfileManager">
+  <settings>
+    <option name="USE_PROJECT_PROFILE" value="false" />
+    <version value="1.0" />
+  </settings>
+</component>

.idea/misc.xml

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="Black">
+    <option name="sdkName" value="pytorch" />
+  </component>
+  <component name="ProjectRootManager" version="2" project-jdk-name="pytorch" project-jdk-type="Python SDK" />
+</project>

.idea/modules.xml

@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/CardVideoSummary.iml" filepath="$PROJECT_DIR$/.idea/CardVideoSummary.iml" />
+    </modules>
+  </component>
+</project>

.idea/vcs.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+  </component>
+</project>

Test/split_text_with_overlap.py

@@ -0,0 +1,160 @@
+import requests
+import json
+import re
+import os
+
+# ================= 配置参数区域 =================
+# 1. API 配置
+API_URL = "http://100.64.0.8/v1/workflows/run"
+API_KEY = "Bearer app-qR46FHcfLyKz2kb0tiiRfV50"
+USER_ID = "abc-123"
+
+# 2. 滑动窗口配置
+CHUNK_SIZE = 15000  # 每次发送给AI的字符长度
+OVERLAP_SIZE = 500  # 窗口重叠部分的长度 (防止截断关键句子，建议 200-500)
+
+
+# ===============================================
+
+def send(text_chunk):
+    """
+    发送单个文本片段给 AI 接口
+    """
+    headers = {
+        "Authorization": API_KEY,
+        "Content-Type": "application/json"
+    }
+
+    payload = {
+        "inputs": {
+            "question": text_chunk
+        },
+        "response_mode": "blocking",
+        "user": USER_ID
+    }
+
+    try:
+        response = requests.post(API_URL, headers=headers, json=payload)
+        response.raise_for_status()
+
+        # 获取 AI 返回的纯文本结果
+        result_text = response.json()['data']['outputs']['result']
+        return result_text
+
+    except requests.exceptions.RequestException as e:
+        print(f"❌ 请求发生错误: {e}")
+        return "[]"  # 发生错误返回空列表字符串，防止程序崩溃
+
+
+def split_text_with_overlap(text, chunk_size, overlap):
+    """
+    生成器：将长文本按滑动窗口分割
+    """
+    start = 0
+    text_len = len(text)
+
+    while start < text_len:
+        end = start + chunk_size
+        # 如果是最后一段，end 不能超过文本长度
+        if end > text_len:
+            end = text_len
+
+        yield text[start:end]
+
+        # 如果已经到达末尾，停止循环
+        if end == text_len:
+            break
+
+        # 下一次的起点 = 当前起点 + 步长 (步长 = 块大小 - 重叠量)
+        start += (chunk_size - overlap)
+
+
+def parse_ai_json(json_str):
+    """
+    清洗并解析 AI 返回的 JSON 字符串
+    """
+    if not json_str:
+        return []
+
+    # 1. 移除可能存在的 Markdown 标记 (```json ... ```)
+    cleaned_str = re.sub(r"```json\s*", "", json_str)
+    cleaned_str = re.sub(r"```\s*", "", cleaned_str)
+    cleaned_str = cleaned_str.strip()
+
+    # 2. 尝试解析
+    try:
+        return json.loads(cleaned_str)
+    except json.JSONDecodeError:
+        print(f"⚠️ 解析 JSON 失败，AI 返回原始内容: {cleaned_str[:100]}...")
+        return []
+
+
+def deduplicate_results(all_hits):
+    """
+    简单去重：如果时间戳和英文卡名完全一致，则视为重复
+    (处理滑动窗口重叠区域可能导致的重复识别)
+    """
+    seen = set()
+    unique_hits = []
+
+    for hit in all_hits:
+        # 创建一个唯一标识 Key
+        key = (hit.get('time'), hit.get('card_name_en'))
+
+        if key not in seen:
+            seen.add(key)
+            unique_hits.append(hit)
+
+    return unique_hits
+
+
+if __name__ == '__main__':
+    # 1. 读取文件
+    text_path = r"C:\Code\ML\Project\untitled10\Audio\temp\transcripts\vortexcards.txt"
+    try:
+        with open(text_path, "r", encoding="utf-8") as f:
+            full_text = f.read()
+    except FileNotFoundError:
+        print(f"❌ 找不到文件: {text_path}")
+        exit()
+
+    print(f"📄 原文总长度: {len(full_text)} 字符")
+
+    all_extracted_cards = []
+
+    # 2. 开始滑动窗口处理
+    chunks = list(split_text_with_overlap(full_text, CHUNK_SIZE, OVERLAP_SIZE))
+    total_chunks = len(chunks)
+
+    print(f"✂️ 将分割为 {total_chunks} 个片段进行处理 (Size: {CHUNK_SIZE}, Overlap: {OVERLAP_SIZE})...\n")
+
+    for i, chunk in enumerate(chunks):
+        print(f"⏳ 正在处理第 {i + 1}/{total_chunks} 个片段 (长度 {len(chunk)})...")
+
+        # 发送请求
+        ai_response_str = send(chunk)
+
+        # 解析结果
+        chunk_hits = parse_ai_json(ai_response_str)
+
+        if chunk_hits:
+            print(f"   ✅ 第 {i + 1} 段识别到 {len(chunk_hits)} 张卡片")
+            all_extracted_cards.extend(chunk_hits)
+        else:
+            print(f"   ⚪ 第 {i + 1} 段未发现目标")
+
+    # 3. 去重 (因为有重叠窗口，可能同一张卡在两段话里都被识别了)
+    final_results = deduplicate_results(all_extracted_cards)
+
+    # 4. 输出最终结果
+    print("\n" + "=" * 30)
+    print(f"🎉 处理完成! 共发现 {len(final_results)} 个高光时刻 (已去重)")
+    print("=" * 30)
+
+    # 打印结果 JSON
+    print(json.dumps(final_results, indent=2, ensure_ascii=False))
+
+    # 如果需要保存到文件
+    file_name = os.path.splitext(os.path.split(text_path)[-1])[0]
+    with open(f"{file_name}.json", "w", encoding="utf-8") as f:
+        json.dump(final_results, f, indent=2, ensure_ascii=False)

Test/test01.py

@@ -0,0 +1,3 @@
+
+if __name__ == '__main__':
+    print("1235456")

Test/获取视频帧.py

@@ -0,0 +1,36 @@
+import cv2
+
+
+def get_frame_at_time(video_path, time_ms, output_image_path):
+    """
+    获取视频指定毫秒数的帧
+    :param video_path: 视频文件路径
+    :param time_ms: 需要获取的时间点，单位为毫秒
+    :param output_image_path: 保存帧的路径
+    """
+    cap = cv2.VideoCapture(video_path)
+
+    # 设置视频的位置到指定的毫秒数 [5]
+    cap.set(cv2.CAP_PROP_POS_MSEC, time_ms)
+
+    # 读取帧 [2]
+    ret, frame = cap.read()
+
+    if ret:
+        # 保存图像
+        cv2.imwrite(output_image_path, frame)
+        print(f"Frame at {time_ms}ms saved to {output_image_path}")
+    else:
+        print("Failed to read frame")
+
+    cap.release()
+
+if __name__ == '__main__':
+    video_path = r"C:\Code\ML\Video\直播数据\video\vortexcards.mp4"
+    hours = 1
+    minutes = 53
+    seconds = 22
+
+    time_in_ms = (hours * 3600 + minutes * 60 + seconds) * 1000
+    # 示例：获取视频的第 5000 毫秒（5秒）的帧
+    get_frame_at_time(video_path, time_in_ms, "frame.jpg")

app/api/routes.py

@@ -0,0 +1,64 @@
+from fastapi import APIRouter, HTTPException, UploadFile, File
+from app.schemas.models import VideoFrameRequest, CardInfoOutput
+from app.services.llm_service import LLMService
+from app.services.video_service import VideoService
+from typing import List
+import logging
+
+# 获取 logger 实例（如果 routes 也要打印日志的话）
+logger = logging.getLogger("API")
+
+router = APIRouter()
+llm_service = LLMService()
+video_service = VideoService()
+
+
+# --- 接口 1: 文本总结 (改为文件上传) ---
+@router.post("/summarize", response_model=List[CardInfoOutput])
+async def summarize_text(file: UploadFile = File(...)):
+    """
+    上传 txt 文件，返回提取的卡片信息 JSON 列表
+    """
+    # 1. 基本校验：检查文件名或扩展名
+    if not file.filename.endswith(".txt"):
+        raise HTTPException(status_code=400, detail="Only .txt files are allowed")
+
+    try:
+        # 2. 读取文件内容
+        # file.read() 返回的是 bytes，需要解码
+        content_bytes = await file.read()
+
+        # 尝试解码 (优先 UTF-8，兼容 Windows 的 GBK)
+        try:
+            text = content_bytes.decode("utf-8")
+        except UnicodeDecodeError:
+            # 如果 utf-8 失败，尝试 gbk (常见于 Windows 记事本)
+            try:
+                text = content_bytes.decode("gbk")
+            except UnicodeDecodeError:
+                raise HTTPException(status_code=400, detail="File encoding not supported. Please use UTF-8.")
+
+        if not text.strip():
+            raise HTTPException(status_code=400, detail="File is empty")
+
+        logger.info(f"📂 接收到文件: {file.filename}, 大小: {len(content_bytes)} bytes")
+
+        # 3. 调用服务层处理
+        results = llm_service.process_text(text)
+        return results
+
+    except Exception as e:
+        logger.error(f"❌ 处理文件上传时出错: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+# --- 接口 2: 截取视频帧---
+@router.post("/capture-frames", response_model=List[CardInfoOutput])
+def capture_video_frames(request: VideoFrameRequest):
+    try:
+        result = video_service.capture_frames(request.video_path, request.cards)
+        return result
+    except FileNotFoundError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

app/core/config.py

@@ -0,0 +1,22 @@
+import os
+
+
+class Settings:
+    # 你的 API 配置
+    API_URL: str = "http://100.64.0.8/v1/workflows/run"
+    API_KEY: str = "Bearer app-qR46FHcfLyKz2kb0tiiRfV50"
+    USER_ID: str = "abc-123"
+
+    # 滑动窗口配置
+    CHUNK_SIZE: int = 10000
+    OVERLAP_SIZE: int = 500
+
+    # 图片保存目录 (项目根目录下的 static/frames)
+    STATIC_DIR: str = os.path.join(os.getcwd(), "static")
+    FRAMES_DIR: str = os.path.join(STATIC_DIR, "frames")
+
+
+settings = Settings()
+
+# 确保目录存在
+os.makedirs(settings.FRAMES_DIR, exist_ok=True)

app/core/logger.py

@@ -0,0 +1,22 @@
+import logging
+import sys
+
+
+# 创建一个 logger 实例
+def get_logger(name: str):
+    logger = logging.getLogger(name)
+    logger.setLevel(logging.INFO)
+
+    # 避免重复添加 handler
+    if not logger.handlers:
+        # 输出到控制台
+        handler = logging.StreamHandler(sys.stdout)
+        # 格式：[时间] [级别] [模块名] 消息
+        formatter = logging.Formatter(
+            '%(asctime)s - %(levelname)s - [%(name)s] - %(message)s',
+            datefmt='%Y-%m-%d %H:%M:%S'
+        )
+        handler.setFormatter(formatter)
+        logger.addHandler(handler)
+
+    return logger

app/main.py

@@ -0,0 +1,21 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+from app.api.routes import router
+from app.core.config import settings
+
+app = FastAPI(title="Card Extraction API", version="1.0")
+
+
+app.mount("/static", StaticFiles(directory=settings.STATIC_DIR), name="static")
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"]
+)
+
+
+# 注册路由
+app.include_router(router, prefix="/api")

app/schemas/models.py

@@ -0,0 +1,28 @@
+from pydantic import BaseModel
+from typing import List, Optional
+
+
+# --- 基础字段 (共用) ---
+class CardBase(BaseModel):
+    time: str
+    card_name_cn: Optional[str] = None
+    card_name_en: Optional[str] = None
+    series: Optional[str] = None
+    rarity_score: Optional[int] = None
+    rarity_desc: Optional[str] = None
+
+
+# --- 接口输入模型 ---
+class CardInfoInput(CardBase):
+    pass
+
+
+# --- 接口输出模型 ---
+class CardInfoOutput(CardBase):
+    frame_image_path: Optional[str] = None
+
+
+# --- 接口2 请求体 ---
+class VideoFrameRequest(BaseModel):
+    video_path: str
+    cards: List[CardInfoInput]

app/services/llm_service.py

@@ -0,0 +1,88 @@
+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

app/services/video_service.py

@@ -0,0 +1,68 @@
+import cv2
+import os
+import uuid
+from app.core.config import settings
+from app.core.logger import get_logger
+from app.schemas.models import CardInfoInput, CardInfoOutput
+
+logger = get_logger("VideoService")
+
+
+class VideoService:
+    def time_str_to_ms(self, time_str: str) -> int:
+        try:
+            parts = list(map(int, time_str.split(':')))
+            if len(parts) == 3:
+                h, m, s = parts
+                return (h * 3600 + m * 60 + s) * 1000
+            elif len(parts) == 2:
+                m, s = parts
+                return (m * 60 + s) * 1000
+            return 0
+        except ValueError:
+            return 0
+
+    def capture_frames(self, video_path: str, cards: list[CardInfoInput]) -> list[CardInfoOutput]:
+        if not os.path.exists(video_path):
+            logger.error(f"❌ 找不到视频文件: {video_path}")
+            raise FileNotFoundError(f"Video file not found: {video_path}")
+
+        logger.info(f"🎬 打开视频文件: {video_path}")
+        logger.info(f"📋 待处理卡片数量: {len(cards)}")
+
+        cap = cv2.VideoCapture(video_path)
+        output_list = []
+
+        success_count = 0
+
+        for idx, card_input in enumerate(cards):
+            # 将 Input 模型转为 Output 模型 (此时 path 为 None)
+            card_output = CardInfoOutput(**card_input.dict())
+
+            time_ms = self.time_str_to_ms(card_output.time)
+            logger.info(
+                f"📸 [{idx + 1}/{len(cards)}] 正在截取 {card_output.time} ({time_ms}ms) - {card_output.card_name_cn or '未知卡名'}")
+
+            # 设定位置
+            cap.set(cv2.CAP_PROP_POS_MSEC, time_ms)
+            ret, frame = cap.read()
+
+            if ret:
+                filename = f"{uuid.uuid4()}_{time_ms}.jpg"
+                save_path = os.path.join(settings.FRAMES_DIR, filename)
+
+                try:
+                    cv2.imwrite(save_path, frame)
+                    card_output.frame_image_path = save_path
+                    success_count += 1
+                    logger.info(f"   ✅ 保存成功: {filename}")
+                except Exception as e:
+                    logger.error(f"   ❌ 保存图片失败: {e}")
+            else:
+                logger.warning(f"   ⚠️ 无法读取视频帧 (可能时间戳超出视频长度)")
+
+            output_list.append(card_output)
+
+        cap.release()
+        logger.info(f"🏁 截取任务结束. 成功: {success_count}, 总数: {len(cards)}")
+        return output_list

run_CardVideoSummary.py

@@ -0,0 +1,8 @@
+import uvicorn
+import socket
+
+if __name__ == "__main__":
+    ip = socket.gethostbyname(socket.gethostname())
+    port = 7721
+    print(f"http://{ip}:{port}/docs")
+    uvicorn.run("app.main:app", host="0.0.0.0", port=port, reload=True)