VideoAgentTrek-ScreenFilter数据库设计：使用MySQL存储审核日志与模型元数据
今天咱们来聊聊一个挺实际的问题：当你部署好一个像VideoAgentTrek-ScreenFilter这样的视频内容审核服务后，那些海量的审核记录、视频信息、用户数据，到底该怎么存、怎么管？
你可能已经用Docker把服务跑起来了，界面也能正常访问，但很快就会发现，所有操作记录都只是暂时存在内存里，服务一重启，数据全没了。这显然不行。一个能投入实际使用的系统，必须有个可靠的后端数据库来持久化这些业务数据。
这篇文章，我就手把手带你走一遍这个数据库从设计到搭建的全过程。咱们不搞那些虚头巴脑的理论，直接聚焦在“怎么把事做成”上。我会用最直白的话，把表结构设计、关系梳理、以及用代码操作数据的核心步骤讲清楚。即使你之前对MySQL或ORM框架接触不多，跟着做下来，也能给自己的服务搭起一个稳固的“数据仓库”。
1. 先想清楚：我们要存什么？
在动手建表之前，咱们得先盘算一下，VideoAgentTrek-ScreenFilter这个服务在运行过程中，会产生哪些需要长期保存的数据。理清这个，后面的设计才不会跑偏。
审核记录：这是核心。每次有视频提交审核，不管是通过还是拒绝，这个动作本身的时间、谁操作的、用的哪个模型版本、结果是什么，这些信息都得记下来。以后查日志、做统计、回溯问题，全靠它。
视频元信息：视频本身的一些属性，比如文件名、存储路径、时长、分辨率、大小，还有审核后生成的截图或结果文件的路径。把这些信息和审核记录关联起来，查找和管理就方便多了。
用户信息：虽然可能是个轻量级的内部系统，但记录一下是哪个用户或哪个客户端发起的审核请求，对于权限管理和使用情况分析还是有帮助的。
模型版本：AI模型是会迭代更新的。记录下每次审核使用的是哪个版本的模型，对于后续分析不同版本模型的效果、定位特定版本的问题至关重要。
把这些实体和它们之间的关系在脑子里过一遍，我们就能画出个简单的逻辑图：一个用户可以发起多次审核，每次审核针对一个视频，并使用了某个特定版本的模型。
接下来，我们就基于这个关系，来设计具体的数据库表。
2. 动手设计：四张核心表的结构
我这里直接给出建议的表结构。你可以在MySQL客户端（如MySQL Workbench, Navicat）或者直接在命令行里执行这些SQL语句来创建它们。
首先，记得创建一个专用的数据库：
CREATE DATABASE IF NOT EXISTS video_agent_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE video_agent_db;2.1 用户表 (users)：记录操作者
这张表相对简单，主要目的是标识请求来源。
CREATE TABLE users ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY COMMENT '用户唯一ID', username VARCHAR(50) NOT NULL UNIQUE COMMENT '用户名，用于登录或标识', client_identifier VARCHAR(255) COMMENT '客户端标识，如IP或设备ID', is_active BOOLEAN DEFAULT TRUE COMMENT '账户是否有效', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '记录创建时间' ) ENGINE=InnoDB COMMENT='用户信息表';设计思路：id是自增主键，确保唯一。username可以是你后台管理系统的账号，也可以是外部系统传递的标识。client_identifier字段挺实用，比如记录调用API的机器IP，方便溯源。is_active是个软删除开关，比直接删记录更安全。
2.2 模型版本表 (model_versions)：跟踪AI模型
用来管理你部署的各个版本的审核模型。
CREATE TABLE model_versions ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY COMMENT '版本ID', version_name VARCHAR(100) NOT NULL UNIQUE COMMENT '版本名称，如 v1.2.0', model_path VARCHAR(500) COMMENT '模型文件在服务器上的存储路径', description TEXT COMMENT '版本描述，如更新内容', is_current BOOLEAN DEFAULT FALSE COMMENT '是否为当前使用版本', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '版本入库时间' ) ENGINE=InnoDB COMMENT='审核模型版本表';设计思路：version_name是业务上使用的版本号。model_path指向模型权重文件的位置。is_current这个标志位很重要，通常只有一个是TRUE，这样在代码里就能轻松知道该加载哪个模型，切换版本也只需更新这个标志。
2.3 视频元信息表 (video_metadata)：描述视频文件
保存视频本身的属性信息，它和审核记录是一对一的关系（一次审核处理一个视频）。
CREATE TABLE video_metadata ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY COMMENT '视频元数据ID', file_name VARCHAR(255) NOT NULL COMMENT '原始视频文件名', file_path VARCHAR(500) NOT NULL COMMENT '视频文件在服务器上的存储路径', file_size BIGINT UNSIGNED COMMENT '文件大小（字节）', duration FLOAT COMMENT '视频时长（秒）', resolution VARCHAR(50) COMMENT '分辨率，如 1920x1080', format VARCHAR(20) COMMENT '视频格式，如 mp4, avi', thumbnail_path VARCHAR(500) COMMENT '缩略图或关键帧截图路径', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '记录创建时间' ) ENGINE=InnoDB COMMENT='视频文件元信息表';设计思路：这里存储的都是静态信息。file_path是关键，它是你后续访问视频文件的依据。thumbnail_path可以存储审核过程中截取的典型帧，方便在管理后台快速预览。
2.4 审核记录表 (review_logs)：核心业务日志
这是最核心的表，记录了每一次审核动作的完整上下文和结果。
CREATE TABLE review_logs ( id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY COMMENT '审核记录唯一ID', video_id INT UNSIGNED NOT NULL COMMENT '关联的视频ID', user_id INT UNSIGNED COMMENT '关联的用户ID，可为空', model_version_id INT UNSIGNED NOT NULL COMMENT '关联的模型版本ID', status ENUM('pending', 'processing', 'passed', 'rejected', 'error') DEFAULT 'pending' COMMENT '审核状态', risk_score FLOAT COMMENT '风险评分，0-1之间', risk_category VARCHAR(100) COMMENT '识别出的风险类别', details JSON COMMENT '详细结果，以JSON格式存储，如具体帧信息、置信度', started_at TIMESTAMP NULL COMMENT '审核开始时间', finished_at TIMESTAMP NULL COMMENT '审核完成时间', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '记录创建时间', INDEX idx_video_id (video_id), INDEX idx_status_created (status, created_at), INDEX idx_user_id (user_id), FOREIGN KEY (video_id) REFERENCES video_metadata(id) ON DELETE RESTRICT, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE SET NULL, FOREIGN KEY (model_version_id) REFERENCES model_versions(id) ON DELETE RESTRICT ) ENGINE=InnoDB COMMENT='视频审核记录表';设计思路：
- 状态流：
status字段使用枚举类型，清晰定义审核的生命周期：待处理 -> 处理中 -> 通过/拒绝/错误。 - 结果存储：
risk_score和risk_category存储核心结论。details字段使用JSON类型，这是MySQL 5.7+的一个很棒的特性，可以灵活地存储结构化的详细数据，比如哪一帧有问题、对应的坐标和置信度，而不用频繁修改表结构。 - 性能优化：为
video_id,status等常用查询条件建立了索引，能大幅提升查询速度。 - 外键约束：通过外键确保了数据的一致性。例如，一条审核记录必须对应一个存在的视频。
ON DELETE RESTRICT防止误删关联数据，ON DELETE SET NULL表示用户被删除后，审核记录中的用户ID设为NULL，记录本身保留。
3. 让代码活起来：使用SQLAlchemy操作数据
表建好了，下一步就是让我们的Python服务能和数据库对话。这里我用Python生态里非常流行的ORM框架——SQLAlchemy来演示。ORM的好处是，你可以用操作Python对象的方式来操作数据库，不用写繁琐的SQL字符串。
3.1 定义数据模型（映射表结构）
首先，我们创建models.py文件，用SQLAlchemy的语法来定义上面那四张表。
from datetime import datetime from sqlalchemy import create_engine, Column, Integer, String, Float, Boolean, Text, Enum, JSON, BigInteger, TIMESTAMP, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship import enum # 创建基类 Base = declarative_base() # 定义审核状态枚举（与数据库ENUM对应） class ReviewStatus(enum.Enum): PENDING = 'pending' PROCESSING = 'processing' PASSED = 'passed' REJECTED = 'rejected' ERROR = 'error' # 1. 用户模型 class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True, autoincrement=True) username = Column(String(50), unique=True, nullable=False) client_identifier = Column(String(255)) is_active = Column(Boolean, default=True) created_at = Column(TIMESTAMP, default=datetime.utcnow) # 定义关系（一个用户有多条审核记录） reviews = relationship("ReviewLog", back_populates="user") # 2. 模型版本模型 class ModelVersion(Base): __tablename__ = 'model_versions' id = Column(Integer, primary_key=True, autoincrement=True) version_name = Column(String(100), unique=True, nullable=False) model_path = Column(String(500)) description = Column(Text) is_current = Column(Boolean, default=False) created_at = Column(TIMESTAMP, default=datetime.utcnow) # 定义关系（一个模型版本被用于多条审核记录） reviews = relationship("ReviewLog", back_populates="model_version") # 3. 视频元信息模型 class VideoMetadata(Base): __tablename__ = 'video_metadata' id = Column(Integer, primary_key=True, autoincrement=True) file_name = Column(String(255), nullable=False) file_path = Column(String(500), nullable=False) file_size = Column(BigInteger) # 使用BigInteger存储大文件大小 duration = Column(Float) resolution = Column(String(50)) format = Column(String(20)) thumbnail_path = Column(String(500)) created_at = Column(TIMESTAMP, default=datetime.utcnow) # 定义关系（一个视频对应一条审核记录，这里用一对一关系） review = relationship("ReviewLog", uselist=False, back_populates="video") # 4. 审核记录模型（核心） class ReviewLog(Base): __tablename__ = 'review_logs' id = Column(BigInteger, primary_key=True, autoincrement=True) video_id = Column(Integer, ForeignKey('video_metadata.id', ondelete='RESTRICT'), nullable=False) user_id = Column(Integer, ForeignKey('users.id', ondelete='SET NULL')) model_version_id = Column(Integer, ForeignKey('model_versions.id', ondelete='RESTRICT'), nullable=False) status = Column(Enum(ReviewStatus), default=ReviewStatus.PENDING) risk_score = Column(Float) risk_category = Column(String(100)) details = Column(JSON) # 存储灵活的JSON数据 started_at = Column(TIMESTAMP) finished_at = Column(TIMESTAMP) created_at = Column(TIMESTAMP, default=datetime.utcnow) # 定义关系（关联到其他表） video = relationship("VideoMetadata", back_populates="review") user = relationship("User", back_populates="reviews") model_version = relationship("ModelVersion", back_populates="reviews")3.2 连接数据库与会话管理
然后，我们创建一个database.py文件来处理数据库连接和会话。
from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, scoped_session from models import Base import os # 从环境变量或配置文件中读取数据库连接信息，更安全 DATABASE_URL = os.getenv( "DATABASE_URL", "mysql+pymysql://your_username:your_password@localhost:3306/video_agent_db?charset=utf8mb4" ) # 创建数据库引擎 engine = create_engine( DATABASE_URL, pool_pre_ping=True, # 每次连接前ping一下，防止连接失效 pool_recycle=3600, # 连接一小时后回收，避免数据库断开 echo=False # 设为True可以看到所有执行的SQL，调试时有用 ) # 创建会话工厂 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) # 创建一个线程安全的全局会话作用域（常用于Web框架） # from threading import local # ScopedSession = scoped_session(SessionLocal, scopefunc=local) def get_db(): """ 依赖注入函数，用于在请求中获取数据库会话。 确保每个请求使用独立的会话，并在结束后关闭。 """ db = SessionLocal() try: yield db finally: db.close() def init_db(): """初始化数据库，创建所有表（如果不存在）。""" Base.metadata.create_all(bind=engine) print("数据库表创建成功！")3.3 编写业务操作示例
最后，我们看看在业务代码里怎么用。假设在main.py或你的API处理函数中。
from datetime import datetime from sqlalchemy.orm import Session from models import User, VideoMetadata, ModelVersion, ReviewLog, ReviewStatus from database import get_db, init_db import json # 第一步：初始化数据库（通常只在首次部署时运行一次） # init_db() def create_review_record(db: Session, video_info: dict, user_id: int = None): """ 一个完整的业务函数示例：创建一次审核记录。 """ # 1. 插入或获取用户信息（这里简化处理，实际可能根据token获取） user = None if user_id: user = db.query(User).filter(User.id == user_id).first() # 2. 获取当前激活的模型版本 current_model = db.query(ModelVersion).filter(ModelVersion.is_current == True).first() if not current_model: raise ValueError("没有找到当前激活的模型版本！") # 3. 创建视频元数据记录 new_video = VideoMetadata( file_name=video_info['file_name'], file_path=video_info['file_path'], file_size=video_info.get('file_size'), duration=video_info.get('duration'), resolution=video_info.get('resolution'), format=video_info.get('format'), thumbnail_path=video_info.get('thumbnail_path') ) db.add(new_video) db.flush() # 立即执行插入，以便获取生成的ID，但不提交事务 # 4. 创建审核记录（初始状态为pending） new_review = ReviewLog( video_id=new_video.id, user_id=user.id if user else None, model_version_id=current_model.id, status=ReviewStatus.PENDING, started_at=datetime.utcnow(), details={} # 初始化为空JSON ) db.add(new_review) # 5. 提交事务，将所有更改保存到数据库 db.commit() # 刷新对象，获取数据库生成的ID等字段 db.refresh(new_review) print(f"审核记录创建成功，ID: {new_review.id}") return new_review.id def update_review_result(db: Session, review_id: int, result: dict): """ 更新审核结果。 """ review = db.query(ReviewLog).filter(ReviewLog.id == review_id).first() if not review: raise ValueError(f"未找到ID为 {review_id} 的审核记录") review.status = ReviewStatus.PASSED if result['risk_score'] < 0.5 else ReviewStatus.REJECTED review.risk_score = result['risk_score'] review.risk_category = result.get('category') review.details = result.get('details', {}) # 将Python字典存入JSON字段 review.finished_at = datetime.utcnow() db.commit() print(f"审核记录 {review_id} 已更新，状态: {review.status.value}") def query_recent_reviews(db: Session, limit: int = 10): """ 查询最近的审核记录（带关联信息）。 """ reviews = ( db.query(ReviewLog) .join(VideoMetadata) # 关联视频表 .join(ModelVersion) # 关联模型表 .outerjoin(User) # 左关联用户表（因为可能为空） .order_by(ReviewLog.created_at.desc()) .limit(limit) .all() ) result = [] for r in reviews: result.append({ 'review_id': r.id, 'video_name': r.video.file_name, 'status': r.status.value, 'risk_score': r.risk_score, 'model_version': r.model_version.version_name, 'operator': r.user.username if r.user else 'System', 'finished_at': r.finished_at.isoformat() if r.finished_at else None }) return result # 示例：在FastAPI等Web框架的路由中使用 # @app.post("/api/review") # async def start_review(video: UploadFile, current_user: UserInDB, db: Session = Depends(get_db)): # # 保存视频文件，获取元信息... # video_info = {...} # review_id = create_review_record(db, video_info, current_user.id) # # 异步触发审核任务... # return {"review_id": review_id, "message": "审核任务已提交"}4. 一些你可能遇到的坑和实用建议
跟着上面做，一个可用的数据库层就搭起来了。但在实际跑起来之前，还有几个小地方需要注意，能帮你避开一些常见的坑。
连接池和超时设置：如果你的服务并发量不小，一定要在创建数据库引擎时配置连接池参数（pool_size,max_overflow）。另外，网络不稳定时，设置合理的超时时间（connect_timeout）也很重要。
事务管理要小心：像上面create_review_record函数里，我们插入了VideoMetadata和ReviewLog两条记录。这应该在一个事务里完成，要么都成功，要么都失败。确保你的业务逻辑被正确的try...except包裹，并在异常时调用db.rollback()。
JSON字段的查询：review_logs.details是JSON类型，你可以直接用SQL进行条件查询，比如找details里某个置信度大于0.9的记录。在SQLAlchemy中，可以用ReviewLog.details['confidence'].as_float() > 0.9这样的方式。
定期归档旧数据：审核日志会增长得非常快。建议制定一个数据保留策略，比如将3个月前的记录转移到历史表，或者只保留摘要，原始细节归档到对象存储（如S3）。这能保证主表的查询性能。
别忘了备份：数据库里的数据是无价的。一定要设置定期的自动备份策略（例如使用mysqldump或云数据库的备份功能），并测试恢复流程。
5. 写在最后
好了，到这里，一个为VideoAgentTrek-ScreenFilter服务量身定做的数据库后端就设计并实现得差不多了。我们从分析要存什么数据开始，一步步设计了四张核心表，明确了它们之间的关系，然后用SQLAlchemy这个强大的工具把代码和数据库连接了起来。
整个过程其实并不复杂，核心就是想清楚业务实体和关系，然后用合适的工具把它们映射到数据库表和Python类上。你现在拥有的，不再是一个“失忆”的服务，而是一个能完整记录每一次审核轨迹、便于查询统计、支持业务分析的健壮系统了。
当然，这只是一个起点。随着业务复杂，你可能需要增加更多表（比如审核策略表、统计报表表），或者对查询做更深入的优化。但有了这个清晰的基础结构，后续的扩展都会顺理成章。接下来，你可以试着跑通上面的代码，然后把它集成到你的Web服务框架（比如Flask或FastAPI）里，让整个数据流真正运转起来。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
