处理视频抽帧并转换成json

#每一步用检查包裹

#统计与检查，先检查后统计，中间业务逻辑

模块一：视频处理

cap = cv2.VideoCapture(str(video_path))  # 创建视频捕获对象，用于读取视频帧# 检查视频能否正常打开
if not cap.isOpened():print(f"无法打开视频文件: {video_path}")continue# 逐帧处理视频
while True:ret, frame = cap.read()  # 读取下一帧if not ret:  # 视频结束或读取失败break......  # 其他帧处理逻辑cap.release()  # 释放视频资源

模块二，上传oss：

bucket.put_object_from_file(oss_key, str(file_path))with ThreadPoolExecutor(max_workers=10) as executor:# 创建并维护任务顺序futures = []for path in frame_paths:future = executor.submit(upload_to_oss, path, bucket, OSS_TARGET_DIR)futures.append((get_frame_index(path), future))# 按帧索引排序结果futures.sort(key=lambda x: x[0])# 处理上传结果for frame_index, future in futures:try:url = future.result()if url:urls.append(url)else:upload_errors += 1except Exception as e:print(f"上传失败: 帧 {frame_index}, 错误信息: {str(e)}")upload_errors += 1if not urls:print(f"上传错误: 视频 {video_name} 所有帧均上传失败")

文件操作

import os
# 路径操作
os.path.join('dir', 'file.txt')  # 拼接路径
os.path.exists('/path/to/file')  # 检查路径是否存在
os.path.isfile('/path/to/file')  # 判断是否为文件
os.path.isdir('/path/to/dir')    # 判断是否为目录
os.path.getsize('/path/to/file') # 获取文件大小# 文件操作
os.rename('old.txt', 'new.txt')  # 文件重命名
os.remove('file.txt')            # 删除文件
os.listdir('/path/to/dir')       # 获取目录内容from pathlib import Path
# 创建Path对象
p = Path('/path/to/file.txt')# 路径操作
p.name      # 文件名 'file.txt'
p.stem      # 无后缀文件名 'file'
p.suffix    # 文件扩展名 '.txt'
p.parent    # 父目录 Path('/path/to')
p.exists()  # 检查路径是否存在
p.is_file() # 判断是否为文件# 文件操作
p.rename('new.txt')      # 重命名文件
p.unlink()              # 删除文件
p.write_text('content')  # 写入文本内容
p.read_text()           # 读取文本内容import shutil
shutil.copy('src.txt', 'dst.txt')  # 复制文件
shutil.copy2('src.txt', 'dst.txt') # 复制文件并保留元数据
shutil.move('src.txt', 'dst.txt')  # 移动文件
shutil.rmtree('/path/to/dir')      # 递归删除目录
shutil.make_archive('backup', 'zip', '/dir/to/compress') # 创建压缩包import glob
# 文件查找
txt_files = glob.glob('*.txt')       # 查找当前目录所有txt文件
all_files = glob.glob('**/*.py', recursive=True)  # 递归查找所有py文件import os
import tempfile
from pathlib import Pathdef safe_write(file_path, content):"""原子性写入文件，避免写入过程中断导致文件损坏"""file_path = Path(file_path)# 创建临时文件with tempfile.NamedTemporaryFile(mode='w',dir=file_path.parent,delete=False,suffix='.tmp') as tf:# 写入内容tf.write(content)temp_name = tf.name# 原子性替换原文件os.replace(temp_name, file_path)

Package：

cv2：用来处理视频

shutil：复制移动文件

question：

中文路径问题(不同包处理编码不一样，跨系统会出现显示问题)

solve： 全过程显式指定UTF-8。 之前想的用哈希创造处理中间文件名，处理完之后再把文件名换回去这个方法只能在小数量的情况下临时用用，哈希会冲突，表太大了内存不够。一旦丢失很麻烦。

tips：

一个好的函数参数不宜过多(少于3-4个)，名字要清晰，功能要单一。

1. 名字：它的名字是否清楚地表明了它只做的那一件事？

2. 长度：它是否足够短？（理想情况下一屏内显示）

3. 职责：它是否只有一个修改代码状态的理由？（单一职责）

4. 参数：它的参数数量是否很少（<4）？是否使用了关键字参数和默认值来简化调用？

5. 副作用：它是否有不可预知的副作用？是否避免了修改外部变量和输入参数？

6. 返回：它是否有明确、单一的返回值？

7. 文档：它是否有清晰的文档字符串和类型提示？

慎用.get,因为会有默认值，在数据量大的时候会掩盖问题。最好还是精确匹配，不然会有意想不到的错误。

处理成json

converted_data = []
# 遍历结构体中的所有URL列表
for urls in find_urls_in_structure(original_data):if urls:# 仅当存在URL时添加条目new_entry = {"info": {"url": urls  # 使用标准字段名"url"}}converted_data.append(new_entry)

y)