4K视频内容分析：从FFmpeg预处理到Elasticsearch检索实战

在技术领域，我们常常需要处理海量的视频文件，无论是用户上传的内容、监控录像还是影视素材。4K超高清画质的普及，虽然带来了极致的视觉体验，但也对存储、传输和处理能力提出了巨大挑战。一个时长两小时的4K电影文件，体积可能高达数十GB，直接进行全文检索或内容分析几乎不可行。本文将以一部假设的4K影片《我会找到你》为例，探讨如何构建一套完整的视频内容分析与检索系统，实现从原始视频到可搜索、可管理结构化数据的转变。

这套系统尤其适合需要处理大量视频资料的开发者、运维工程师和架构师，例如在线教育平台的内容管理、安防监控的视频回溯、媒体资源的数字化归档等场景。通过本文，你将掌握如何利用FFmpeg进行视频预处理，如何使用OpenCV提取关键帧，如何借助OCR和语音识别技术转化音视频信息为文本，并最终通过Elasticsearch实现毫秒级的多维度检索。我们将从环境准备开始，逐步实现一个最小可用的视频分析管道，并深入讲解每个环节的配置要点和排错方法。

1. 理解视频内容分析的技术栈与核心挑战

处理4K视频内容分析，首先需要明确技术选型和面临的真实难题。单纯播放一个4K视频或许不难，但要从中精准提取信息并建立索引，就需要一套组合技术方案。

1.1 为什么4K视频分析需要特殊处理

4K分辨率（3840×2160像素）意味着单帧图像包含约830万像素，是1080p的四倍。这种高分辨率虽然保证了画面细节，但也带来了三个核心挑战：

• 计算资源消耗巨大：直接对4K视频流进行实时分析，对CPU和内存的压力是普通高清视频的4倍以上。在生产环境中，通常需要先进行分辨率下采样或关键帧提取。
• 存储与传输成本高：原始4K视频文件体积庞大，不适宜直接作为分析对象。需要先将其转换为更适合处理的中间格式，同时保留关键信息。
• 分析精度与效率的平衡：全帧分析精度最高但速度最慢，抽样分析速度快但可能遗漏关键画面。需要根据业务场景选择合适分析策略。

在实际项目中，我们通常采用"预处理-抽帧-特征提取-索引构建"的管道式架构，而不是直接处理原始视频流。

1.2 视频内容分析的核心技术组件

一个完整的视频分析系统通常包含以下核心组件：

这套技术栈的优势在于各层解耦，可以针对特定需求替换组件。比如对实时性要求高的场景，可以用更轻量的YOLO模型替代复杂的识别模型。

2. 环境准备与依赖配置

构建视频分析系统前，需要确保基础环境就绪。以下配置基于Ubuntu 20.04 LTS，其他Linux发行版需要调整包管理命令。

2.1 系统级依赖安装

FFmpeg和OpenCV需要大量的系统级依赖，正确的依赖安装是后续步骤的基础。

# 更新包管理器并安装基础编译工具 sudo apt update sudo apt install -y build-essential cmake pkg-config # 安装视频处理相关依赖 sudo apt install -y libsm6 libxext6 libxrender-dev libavcodec-dev libavformat-dev libswscale-dev libv4l-dev sudo apt install -y libxvidcore-dev libx264-dev libjpeg-dev libpng-dev libtiff-dev sudo apt install -y libgtk-3-dev libatlas-base-dev gfortran # 安装Python环境（推荐Python 3.8+） sudo apt install -y python3-dev python3-pip python3-venv

2.2 创建隔离的Python环境

视频分析项目依赖复杂，建议使用虚拟环境避免版本冲突。

# 创建项目目录并进入 mkdir video-analyzer && cd video-analyzer # 创建虚拟环境 python3 -m venv venv source venv/bin/activate # 升级pip pip install --upgrade pip

2.3 安装Python核心依赖

通过requirements.txt管理依赖版本，确保环境一致性。

# requirements.txt opencv-python==4.5.5.64 Pillow==9.0.1 numpy==1.21.6 ffmpeg-python==0.2.0 pytesseract==0.3.8 SpeechRecognition==3.8.1 elasticsearch==7.17.0 python-dotenv==0.19.2

安装命令：

pip install -r requirements.txt

2.4 验证关键组件安装

安装完成后，需要验证FFmpeg、Tesseract等关键组件的可用性。

# 检查FFmpeg版本 ffmpeg -version # 检查Tesseract安装 tesseract --version # Python环境验证脚本 python -c " import cv2 print(f'OpenCV版本: {cv2.__version__}') import pytesseract print('Tesseract导入成功') import speech_recognition as sr print('SpeechRecognition导入成功') "

如果任何一步出现错误，需要根据错误信息排查依赖安装问题。常见的坑包括权限不足、路径错误或依赖缺失。

3. 视频预处理与关键帧提取策略

原始4K视频不能直接用于分析，需要先进行预处理。我们将以《我会找到你》的假设片段为例，演示完整的处理流程。

3.1 视频元信息提取

分析前需要了解视频的基本信息，这些元数据有助于制定合适的处理策略。

import ffmpeg import json def get_video_metadata(video_path): """提取视频的元数据信息""" try: probe = ffmpeg.probe(video_path) video_info = next( stream for stream in probe['streams'] if stream['codec_type'] == 'video' ) metadata = { 'duration': float(video_info.get('duration', 0)), 'width': int(video_info.get('width', 0)), 'height': int(video_info.get('height', 0)), 'codec': video_info.get('codec_name', 'unknown'), 'frame_rate': eval(video_info.get('r_frame_rate', '0/1')), 'total_frames': int(video_info.get('nb_frames', 0)) } return metadata except Exception as e: print(f"元数据提取失败: {e}") return None # 使用示例 if __name__ == "__main__": metadata = get_video_metadata("/path/to/我会找到你.mp4") print(json.dumps(metadata, indent=2, ensure_ascii=False))

执行后会输出类似以下信息：

{ "duration": 7200.0, "width": 3840, "height": 2160, "codec": "h264", "frame_rate": 23.98, "total_frames": 172656 }

这些信息告诉我们：这是一个2小时（7200秒）的4K视频，采用H.264编码，帧率23.98fps，总帧数约17万帧。全帧分析显然不现实，需要制定抽帧策略。

3.2 智能关键帧提取算法

单纯按时间间隔抽帧会遗漏重要画面变化，我们需要更智能的关键帧检测。

import cv2 import numpy as np import os class KeyFrameExtractor: def __init__(self, min_interval=5, threshold=5000): """ 初始化关键帧提取器 :param min_interval: 最小抽帧间隔（秒） :param threshold: 帧差异阈值，用于检测场景变化 """ self.min_interval = min_interval self.threshold = threshold self.last_frame = None self.last_extract_time = 0 def extract_key_frames(self, video_path, output_dir): """提取关键帧到指定目录""" if not os.path.exists(output_dir): os.makedirs(output_dir) cap = cv2.VideoCapture(video_path) fps = cap.get(cv2.CAP_PROP_FPS) min_interval_frames = int(self.min_interval * fps) frame_count = 0 saved_count = 0 while True: ret, frame = cap.read() if not ret: break current_time = frame_count / fps # 转换为灰度图进行比较 gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) gray = cv2.resize(gray, (320, 180)) # 缩小尺寸加速计算 should_save = False # 第一帧总是保存 if self.last_frame is None: should_save = True else: # 计算帧间差异 diff = cv2.absdiff(gray, self.last_frame) non_zero_count = np.count_nonzero(diff) # 场景变化检测或达到最小时间间隔 if (non_zero_count > self.threshold or current_time - self.last_extract_time >= self.min_interval): should_save = True if should_save: filename = f"frame_{saved_count:06d}_{int(current_time)}s.jpg" output_path = os.path.join(output_dir, filename) cv2.imwrite(output_path, frame) self.last_extract_time = current_time saved_count += 1 print(f"已保存关键帧: {filename}") self.last_frame = gray frame_count += 1 # 每处理1000帧打印进度 if frame_count % 1000 == 0: print(f"处理进度: {frame_count}帧，已保存{saved_count}关键帧") cap.release() return saved_count # 使用示例 extractor = KeyFrameExtractor(min_interval=10, threshold=3000) frame_count = extractor.extract_key_frames( "/path/to/我会找到你.mp4", "./extracted_frames" ) print(f"总共提取了{frame_count}个关键帧")

这个算法结合了时间间隔和场景变化检测，确保在画面静止时不会保存过多相似帧，在激烈场景变化时又能及时捕获关键帧。

3.3 批量处理与进度监控

对于长视频，需要完善的进度监控和错误处理机制。

import time from concurrent.futures import ThreadPoolExecutor import threading class BatchVideoProcessor: def __init__(self, max_workers=2): self.max_workers = max_workers self.lock = threading.Lock() self.processed_count = 0 self.total_count = 0 def process_single_video(self, video_info): """处理单个视频文件""" video_path, output_dir = video_info try: start_time = time.time() # 创建视频专用的输出目录 video_name = os.path.splitext(os.path.basename(video_path))[0] video_output_dir = os.path.join(output_dir, video_name) extractor = KeyFrameExtractor() frame_count = extractor.extract_key_frames(video_path, video_output_dir) elapsed = time.time() - start_time with self.lock: self.processed_count += 1 print(f"[{self.processed_count}/{self.total_count}] " f"完成: {video_name}, 提取{frame_count}帧, 耗时{elapsed:.1f}秒") return True except Exception as e: print(f"处理失败 {video_path}: {e}") return False def process_batch(self, video_list, output_base_dir): """批量处理视频列表""" self.total_count = len(video_list) self.processed_count = 0 video_infos = [(video, output_base_dir) for video in video_list] with ThreadPoolExecutor(max_workers=self.max_workers) as executor: results = list(executor.map(self.process_single_video, video_infos)) success_count = sum(results) print(f"批量处理完成: 成功{success_count}/{self.total_count}") return success_count # 使用示例 processor = BatchVideoProcessor(max_workers=3) video_files = [ "/data/videos/我会找到你_part1.mp4", "/data/videos/我会找到你_part2.mp4", "/data/videos/我会找到你_part3.mp4" ] processor.process_batch(video_files, "./extracted_frames_batch")

4. 多模态内容提取技术

提取关键帧后，需要从视觉和听觉两个维度提取可搜索的内容信息。

4.1 基于OCR的字幕文本提取

视频中的字幕包含丰富的剧情信息，是重要的检索线索。

import pytesseract from PIL import Image import re class SubtitleExtractor: def __init__(self, lang='chi_sim+eng'): self.lang = lang # 字幕区域通常位于画面底部20%区域 self.subtitle_region = (0, 0.8, 1, 1) # (left, top, right, bottom) def preprocess_image(self, image_path): """图像预处理提升OCR识别率""" image = Image.open(image_path) width, height = image.size # 裁剪字幕区域 left = int(self.subtitle_region[0] * width) top = int(self.subtitle_region[1] * height) right = int(self.subtitle_region[2] * width) bottom = int(self.subtitle_region[3] * height) cropped = image.crop((left, top, right, bottom)) # 转换为灰度图 gray = cropped.convert('L') # 二值化处理 threshold = 150 binary = gray.point(lambda p: p > threshold and 255) return binary def extract_text(self, image_path): """从图像中提取文本""" try: processed_image = self.preprocess_image(image_path) # OCR识别 config = f'--psm 6 -l {self.lang}' text = pytesseract.image_to_string(processed_image, config=config) # 清理文本结果 cleaned_text = self.clean_text(text) return cleaned_text except Exception as e: print(f"OCR提取失败 {image_path}: {e}") return "" def clean_text(self, text): """清理OCR识别结果""" if not text: return "" # 移除多余空白字符 text = re.sub(r'\s+', ' ', text) # 移除特殊字符但保留中文和基本标点 text = re.sub(r'[^\w\s\u4e00-\u9fff，。！？：；""''()（）-]', '', text) return text.strip() def batch_extract(self, frame_dir, output_file): """批量处理帧目录中的图像""" results = [] frame_files = sorted([f for f in os.listdir(frame_dir) if f.endswith(('.jpg', '.png'))]) for frame_file in frame_files: frame_path = os.path.join(frame_dir, frame_file) # 从文件名提取时间戳 time_match = re.search(r'_(\d+)s\.', frame_file) timestamp = int(time_match.group(1)) if time_match else 0 text = self.extract_text(frame_path) if text and len(text) > 1: # 过滤掉过短的结果 result = { 'frame_file': frame_file, 'timestamp': timestamp, 'text': text, 'extraction_time': time.time() } results.append(result) print(f"帧{frame_file} 时间{timestamp}s: {text}") # 保存结果到JSON文件 import json with open(output_file, 'w', encoding='utf-8') as f: json.dump(results, f, ensure_ascii=False, indent=2) return results # 使用示例 extractor = SubtitleExtractor(lang='chi_sim') results = extractor.batch_extract('./extracted_frames', './subtitle_results.json')

4.2 语音识别与音频处理

对于没有字幕但对白重要的场景，需要从音频轨道提取信息。

import speech_recognition as sr import ffmpeg import tempfile import os class AudioProcessor: def __init__(self): self.recognizer = sr.Recognizer() def extract_audio_segment(self, video_path, start_time, duration=30): """从视频中提取音频片段""" try: # 创建临时文件保存音频 with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as temp_file: temp_path = temp_file.name # 使用FFmpeg提取音频 ( ffmpeg .input(video_path, ss=start_time, t=duration) .output(temp_path, ac=1, ar=16000) # 单声道，16kHz采样率 .overwrite_output() .run(quiet=True) ) return temp_path except Exception as e: print(f"音频提取失败: {e}") return None def transcribe_audio(self, audio_path, language='zh-CN'): """转录音频文件为文本""" try: with sr.AudioFile(audio_path) as source: audio_data = self.recognizer.record(source) text = self.recognizer.recognize_google(audio_data, language=language) return text except sr.UnknownValueError: print(f"无法识别音频内容: {audio_path}") return "" except sr.RequestError as e: print(f"语音识别服务错误: {e}") return "" finally: # 清理临时文件 if os.path.exists(audio_path): os.unlink(audio_path) def process_video_audio(self, video_path, interval=60): """按时间间隔处理视频音频""" import json # 获取视频时长 metadata = get_video_metadata(video_path) duration = metadata['duration'] results = [] for start_time in range(0, int(duration), interval): print(f"处理音频段: {start_time}-{start_time+interval}秒") audio_path = self.extract_audio_segment(video_path, start_time, interval) if audio_path: text = self.transcribe_audio(audio_path) if text: result = { 'start_time': start_time, 'end_time': start_time + interval, 'text': text, 'video_path': video_path } results.append(result) print(f"时间段{start_time}s: {text}") # 保存结果 output_file = f"./audio_results_{os.path.basename(video_path)}.json" with open(output_file, 'w', encoding='utf-8') as f: json.dump(results, f, ensure_ascii=False, indent=2) return results # 使用示例 audio_processor = AudioProcessor() audio_results = audio_processor.process_video_audio( "/path/to/我会找到你.mp4", interval=30 # 每30秒处理一次 )

4.3 视觉特征分析与场景识别

除了文本内容，视觉特征也是重要的检索维度。

class VisualFeatureExtractor: def __init__(self): # 使用OpenCV的预训练模型 self.face_cascade = cv2.CascadeClassifier( cv2.data.haarcascades + 'haarcascade_frontalface_default.xml' ) def extract_color_histogram(self, image_path, bins=8): """提取颜色直方图特征""" image = cv2.imread(image_path) histograms = [] # 分别计算每个通道的直方图 for channel in range(3): hist = cv2.calcHist([image], [channel], None, [bins], [0, 256]) histograms.extend(hist.flatten()) # 归一化 histograms = np.array(histograms) / np.sum(histograms) return histograms.tolist() def detect_faces(self, image_path): """检测人脸并返回数量和大致位置""" image = cv2.imread(image_path) gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) faces = self.face_cascade.detectMultiScale( gray, scaleFactor=1.1, minNeighbors=5, minSize=(30, 30) ) face_info = { 'count': len(faces), 'positions': [{'x': int(x), 'y': int(y), 'w': int(w), 'h': int(h)} for (x, y, w, h) in faces] } return face_info def analyze_scene_complexity(self, image_path): """分析场景复杂度（基于边缘检测）""" image = cv2.imread(image_path) gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) # 边缘检测 edges = cv2.Canny(gray, 50, 150) # 计算边缘密度作为复杂度指标 edge_density = np.sum(edges > 0) / (edges.shape[0] * edges.shape[1]) return { 'edge_density': float(edge_density), 'complexity_level': 'high' if edge_density > 0.1 else 'medium' if edge_density > 0.05 else 'low' } def extract_frame_features(self, frame_path): """提取帧的完整特征集""" features = { 'frame_file': os.path.basename(frame_path), 'color_histogram': self.extract_color_histogram(frame_path), 'face_info': self.detect_faces(frame_path), 'scene_complexity': self.analyze_scene_complexity(frame_path), 'timestamp': self._extract_timestamp(frame_path) } return features def _extract_timestamp(self, frame_path): """从文件名提取时间戳""" import re match = re.search(r'_(\d+)s\.', frame_path) return int(match.group(1)) if match else 0 # 使用示例 visual_extractor = VisualFeatureExtractor() frame_features = visual_extractor.extract_frame_features('./extracted_frames/frame_000001_120s.jpg') print(f"人脸数量: {frame_features['face_info']['count']}") print(f"场景复杂度: {frame_features['scene_complexity']['complexity_level']}")

5. 构建可搜索的视频内容索引

提取的多模态内容需要建立统一的搜索索引，Elasticsearch是理想的选择。

5.1 Elasticsearch索引设计与映射

合理的索引设计是高效检索的基础。

from elasticsearch import Elasticsearch import json class VideoSearchEngine: def __init__(self, hosts=['localhost:9200']): self.es = Elasticsearch(hosts) self.index_name = "video-content" def create_index(self): """创建视频内容索引""" if self.es.indices.exists(index=self.index_name): print("索引已存在，跳过创建") return mapping = { "mappings": { "properties": { "video_id": {"type": "keyword"}, "frame_file": {"type": "keyword"}, "timestamp": {"type": "integer"}, "subtitle_text": { "type": "text", "analyzer": "ik_max_word", # 中文分词 "search_analyzer": "ik_smart" }, "audio_text": { "type": "text", "analyzer": "ik_max_word", "search_analyzer": "ik_smart" }, "face_count": {"type": "integer"}, "scene_complexity": {"type": "keyword"}, "color_histogram": {"type": "float"}, # 实际使用nested类型更合适 "extraction_time": {"type": "date"}, "video_metadata": { "properties": { "duration": {"type": "integer"}, "resolution": {"type": "keyword"}, "codec": {"type": "keyword"} } } } }, "settings": { "number_of_shards": 3, "number_of_replicas": 1, "analysis": { "analyzer": { "ik_smart": { "type": "ik_smart" }, "ik_max_word": { "type": "ik_max_word" } } } } } self.es.indices.create(index=self.index_name, body=mapping) print("索引创建成功") def index_frame_data(self, frame_data): """索引单帧数据""" document = { "video_id": frame_data.get('video_id', 'unknown'), "frame_file": frame_data['frame_file'], "timestamp": frame_data['timestamp'], "subtitle_text": frame_data.get('subtitle_text', ''), "audio_text": frame_data.get('audio_text', ''), "face_count": frame_data.get('face_count', 0), "scene_complexity": frame_data.get('scene_complexity', 'unknown'), "color_histogram": frame_data.get('color_histogram', []), "extraction_time": frame_data.get('extraction_time'), "video_metadata": frame_data.get('video_metadata', {}) } # 使用帧文件名作为文档ID确保唯一性 doc_id = f"{frame_data['video_id']}_{frame_data['frame_file']}" response = self.es.index( index=self.index_name, id=doc_id, body=document ) return response['result'] == 'created' def bulk_index_frames(self, frames_data): """批量索引帧数据""" from elasticsearch.helpers import bulk actions = [] for frame_data in frames_data: doc_id = f"{frame_data['video_id']}_{frame_data['frame_file']}" action = { "_index": self.index_name, "_id": doc_id, "_source": { "video_id": frame_data.get('video_id', 'unknown'), "frame_file": frame_data['frame_file'], "timestamp": frame_data['timestamp'], "subtitle_text": frame_data.get('subtitle_text', ''), "audio_text": frame_data.get('audio_text', ''), "face_count": frame_data.get('face_count', 0), "scene_complexity": frame_data.get('scene_complexity', 'unknown'), "color_histogram": frame_data.get('color_histogram', []), "extraction_time": frame_data.get('extraction_time'), "video_metadata": frame_data.get('video_metadata', {}) } } actions.append(action) success, failed = bulk(self.es, actions, stats_only=True) print(f"批量索引完成: 成功{success}条，失败{failed}条") return success, failed # 初始化搜索引擎 search_engine = VideoSearchEngine() search_engine.create_index()

5.2 多维度搜索查询实现

支持文本、时间范围、视觉特征等多维度组合查询。

class VideoSearchQuery: def __init__(self, es_client, index_name): self.es = es_client self.index_name = index_name def search_by_text(self, query_text, fields=None, size=10): """文本搜索""" if fields is None: fields = ["subtitle_text", "audio_text"] search_body = { "query": { "multi_match": { "query": query_text, "fields": fields, "type": "best_fields" } }, "sort": [ {"_score": {"order": "desc"}}, {"timestamp": {"order": "asc"}} ], "size": size, "highlight": { "fields": { "subtitle_text": {}, "audio_text": {} } } } response = self.es.search(index=self.index_name, body=search_body) return self._format_search_results(response) def search_by_time_range(self, start_time, end_time, size=20): """时间范围搜索""" search_body = { "query": { "range": { "timestamp": { "gte": start_time, "lte": end_time } } }, "sort": [ {"timestamp": {"order": "asc"}} ], "size": size } response = self.es.search(index=self.index_name, body=search_body) return self._format_search_results(response) def search_by_visual_features(self, min_faces=0, complexity=None): """视觉特征搜索""" must_conditions = [] if min_faces > 0: must_conditions.append({ "range": { "face_count": { "gte": min_faces } } }) if complexity: must_conditions.append({ "term": { "scene_complexity": complexity } }) search_body = { "query": { "bool": { "must": must_conditions } }, "sort": [ {"timestamp": {"order": "asc"}} ], "size": 50 } response = self.es.search(index=self.index_name, body=search_body) return self._format_search_results(response) def advanced_search(self, text_query=None, time_range=None, min_faces=None, complexity=None, size=20): """高级组合搜索""" must_conditions = [] # 文本查询条件 if text_query: must_conditions.append({ "multi_match": { "query": text_query, "fields": ["subtitle_text", "audio_text"], "type": "best_fields" } }) # 时间范围条件 if time_range and 'start' in time_range and 'end' in time_range: must_conditions.append({ "range": { "timestamp": { "gte": time_range['start'], "lte": time_range['end'] } } }) # 视觉特征条件 if min_faces is not None and min_faces > 0: must_conditions.append({ "range": { "face_count": { "gte": min_faces } } }) if complexity: must_conditions.append({ "term": { "scene_complexity": complexity } }) search_body = { "query": { "bool": { "must": must_conditions } }, "sort": [ {"_score": {"order": "desc"}}, {"timestamp": {"order": "asc"}} ], "size": size, "highlight": { "fields": { "subtitle_text": {}, "audio_text": {} } } } response = self.es.search(index=self.index_name, body=search_body) return self._format_search_results(response) def _format_search_results(self, response): """格式化搜索结果""" results = [] for hit in response['hits']['hits']: source = hit['_source'] result = { 'frame_file': source['frame_file'], 'timestamp': source['timestamp'], 'score': hit['_score'], 'subtitle_text': source.get('subtitle_text', ''), 'audio_text': source.get('audio_text', ''), 'face_count': source.get('face_count', 0), 'scene_complexity': source.get('scene_complexity', 'unknown') } # 添加高亮显示 if 'highlight' in hit: result['highlight'] = hit['highlight'] results.append(result) return { 'total': response['hits']['total']['value'], 'results': results } # 使用示例 search_query = VideoSearchQuery(search_engine.es, search_engine.index_name) # 搜索包含"父亲"的片段 results = search_query.search_by_text("父亲") print(f"找到{results['total']}个相关片段") # 搜索30分钟到40分钟之间的内容 results = search_query.search_by_time_range(1800, 2400) # 高级搜索：包含"监狱"且有多个人物的复杂场景 results = search_query.advanced_search( text_query="监狱", min_faces=2, complexity="high" )

6. 系统集成与API服务

将分析能力封装为可调用的API服务，便于集成到其他系统。

6.1 Flask REST API实现

from flask import Flask, request, jsonify import logging from datetime import datetime app = Flask(__name__) # 初始化组件 search_engine = VideoSearchEngine() search_query = VideoSearchQuery(search_engine.es, search_engine.index_name) @app.route('/api/search', methods=['GET']) def search_video_content(): """视频内容搜索API""" try: # 获取查询参数 query_text = request.args.get('q', '') start_time = request.args.get('start_time', type=int) end_time = request.args.get('end_time', type=int) min_faces = request.args.get('min_faces', type=int) complexity = request.args.get('complexity') size = request.args.get('size', 20, type=int) time_range = None if start_time is not None and end_time is not None: time_range = {'start': start_time, 'end': end_time} # 执行搜索 results = search_query.advanced_search( text_query=query_text, time_range=time_range, min_faces=min_faces, complexity=complexity, size=size ) return jsonify({ 'success': True, 'query_time': datetime.now().isoformat(), 'results': results }) except Exception as e: logging.error(f"搜索失败: {e}") return jsonify({ 'success': False, 'error': str(e) }), 500 @app.route('/api/analyze', methods=['POST']) def analyze_video(): """视频分析API""" try: video_path = request