from dataclasses import dataclass from datetime import datetime from typing import Optional from sqlalchemy import Column, String, TIMESTAMP, func from database.database import Base, get_session from utils import utils @dataclass class VideoScript(Base): __tablename__ = 't_video_script' id: str = Column(String, primary_key=True, comment='唯一标识') title: str = Column(String, nullable=False, comment='标题') description: Optional[str] = Column(String, nullable=True, comment='描述') keywords: Optional[str] = Column(String, nullable=True, comment='话题关键词') url: str = Column(String, nullable=False, comment='话题链接') script: str = Column(String, nullable=True, comment='视频脚本') content: str = Column(String, nullable=True, comment='话题内容') create_time: datetime = Column(TIMESTAMP(timezone=True), server_default=func.now(), nullable=False, comment='创建时间') def __repr__(self): return f"" def create_video_script(video_script: VideoScript): if video_script.id is None: video_script.id = utils.get_md5(video_script.url) with get_session() as db: db.add(video_script) db.commit() db.refresh(video_script) return video_script def video_script_not_exists(url_list: list): """ url_list如果在数据库中已经存在,则去除掉 :param url_list: :return: """ with get_session() as db: video_scripts = db.query(VideoScript).filter(VideoScript.url.in_(url_list)).all() for video_script in video_scripts: url_list.remove(video_script.url) return url_list