my-video-workflow/controller/api/bilirecorder_blueprint.py

import os.path
import threading
from datetime import datetime, timedelta
from glob import glob
from flask import Blueprint, jsonify, request, current_app
from typing import Optional

from config import BILILIVE_RECORDER_DIRECTORY, VIDEO_TITLE, XIGUALIVE_RECORDER_DIRECTORY
from model import db
from model.DanmakuClip import DanmakuClip
from model.VideoClip import VideoClip
from model.Workflow import Workflow
from worker.danmaku import do_workflow

blueprint = Blueprint("api_bilirecorder", __name__, url_prefix="/api/bilirecorder")

bili_record_workflow_item: Optional[Workflow] = None


def auto_submit_task():
    global bili_record_workflow_item
    if not bili_record_workflow_item.editing:
        if len(bili_record_workflow_item.video_clips) > 0 and len(bili_record_workflow_item.danmaku_clips) > 0:
            threading.Thread(target=do_workflow, args=(
                bili_record_workflow_item.video_clips[0].full_path,
                bili_record_workflow_item.danmaku_clips[0].full_path,
                *[clip.full_path for clip in bili_record_workflow_item.danmaku_clips[1:]]
            )).start()
            clear_item()


def clear_item():
    global bili_record_workflow_item
    bili_record_workflow_item = None


def commit_item():
    global bili_record_workflow_item
    if bili_record_workflow_item is None:
        return
    bili_record_workflow_item.calculate_start_time()
    db.session.commit()


def safe_create_item():
    global bili_record_workflow_item
    if bili_record_workflow_item is None:
        bili_record_workflow_item = Workflow.query.filter(
            Workflow.update_time > (datetime.now() - timedelta(hours=8)),
            Workflow.automatic == 1
        ).first()
        if bili_record_workflow_item is None:
            bili_record_workflow_item = Workflow()
    else:
        if bili_record_workflow_item is not None and bili_record_workflow_item.id is not None:
            bili_record_workflow_item.editing = False
            commit_item()
            bili_record_workflow_item = Workflow()
    if bili_record_workflow_item is None:
        bili_record_workflow_item = Workflow()
    bili_record_workflow_item.name = VIDEO_TITLE.format(datetime.now().strftime("%Y%m%d"))
    bili_record_workflow_item.automatic = True
    bili_record_workflow_item.editing = True
    db.session.commit()
    return bili_record_workflow_item


def safe_get_item() -> Workflow:
    global bili_record_workflow_item
    if bili_record_workflow_item is None:
        return safe_create_item()
    return bili_record_workflow_item


def collect_danmaku_files(workflow: Optional[Workflow]):
    if workflow is None:
        return
    clip: VideoClip
    for clip in workflow.video_clips:
        full_path = clip.full_path
        pre_file_name = os.path.splitext(full_path)[0]
        # 理论上也只有一个结果
        for danmaku_file in glob("{}*xml".format(pre_file_name)):
            relpath = os.path.relpath(danmaku_file, BILILIVE_RECORDER_DIRECTORY)
            danmaku = DanmakuClip.query.filter(
                DanmakuClip.file == relpath,
                DanmakuClip.base_path == BILILIVE_RECORDER_DIRECTORY
            ).first()
            if danmaku is None:
                danmaku = DanmakuClip()
                danmaku.file = relpath
                danmaku.base_path = BILILIVE_RECORDER_DIRECTORY
                danmaku.offset = 0
                danmaku.workflow = workflow
                db.session.add(danmaku)
                workflow.danmaku_clips.append(danmaku)
        for danmaku_file in glob(os.path.join(XIGUALIVE_RECORDER_DIRECTORY, "*.xml")):
            relpath = os.path.relpath(danmaku_file, XIGUALIVE_RECORDER_DIRECTORY)
            danmaku = DanmakuClip.query.filter(
                DanmakuClip.file == relpath,
                DanmakuClip.base_path == XIGUALIVE_RECORDER_DIRECTORY
            ).first()
            if danmaku is None:
                danmaku = DanmakuClip()
                danmaku.file = relpath
                danmaku.base_path = XIGUALIVE_RECORDER_DIRECTORY
                danmaku.offset = 0
                danmaku.workflow = workflow
                db.session.add(danmaku)
                workflow.danmaku_clips.append(danmaku)
    commit_item()


@blueprint.post("/")
def bilirecorder_event():
    payload = request.json
    current_app.logger.debug(payload)
    if 'EventType' not in payload:
        response = jsonify({
            'error': "异常",
            'payload': payload,
        })
        response.status_code = 403
        return response

    if payload['EventType'] == "SessionStarted":
        # 录制开始
        safe_create_item()
    elif payload['EventType'] == "SessionEnded":
        # 录制结束
        item = safe_get_item()
        item.editing = False
        commit_item()
        auto_submit_task()
        return jsonify(item.to_dict())
    elif payload['EventType'] == "FileClosed":
        # 文件关闭
        item = safe_get_item()
        event_data = payload.get("EventData", {})
        video_file = event_data.get("RelativePath", None)
        # 判断是否重复
        already_add = False
        for clip in item.video_clips:
            if video_file == clip.file and clip.base_path == BILILIVE_RECORDER_DIRECTORY:
                already_add = True
                break
        if not already_add:
            video_clip = VideoClip()
            video_clip.file = video_file
            video_clip.base_path = BILILIVE_RECORDER_DIRECTORY
            video_clip.duration = event_data.get("Duration", 0)
            item.video_clips.append(video_clip)
            commit_item()
        collect_danmaku_files(item)
        auto_submit_task()
        return jsonify(item.to_dict())
    commit_item()
    item = safe_get_item()
    return jsonify(item.to_dict())


@blueprint.get("/")
def query_current_status():
    item = safe_get_item()
    return jsonify(item.to_dict())