google_ai_studio/video_processing/video_processing.py

import re

import requests
import json

from common.aliyun_log import AliyunLogger
from common.redis import get_video_data, install_video_data


class VideoProcessing:
    def get_ai_data(self, video_path):
        url = "http://8.219.186.16:8080/process_video/"

        payload = json.dumps( {
            "video_path": video_path
        } )
        headers = {
            'Content-Type': 'application/json'
        }

        response = requests.request( "POST", url, headers=headers, data=payload )
        try:
            response = response.json()
            print(response)
            result = response['result']
            print(result)
            cleaned_string = result.split( 'json\n', 1 )[-1].strip()
            # 清除控制字符
            cleaned_string = re.sub( r'[\x00-\x1F]+', '', cleaned_string )

            # 修复引号等格式问题（确保字符串用双引号包围）
            cleaned_string = cleaned_string.replace( "\n", "" ).replace( "\\", "" ).strip()
            json_data = json.loads( cleaned_string )
            print(json_data)
            return json_data
        except Exception as e:
            print(f"视频请求异常:{e}")
            return None


    def get_video(self):
        video_data = get_video_data()
        if not video_data:
            print("没有获取到视频内容")
            return
        # 解码为字符串
        data_str = video_data.decode( 'utf-8' )

        # 解析为 JSON 对象
        data_json = json.loads( data_str )
        video_id = data_json['video_id']
        title = data_json['title']
        video_path = data_json['video_path']
        data = self.get_ai_data(video_path)
        AliyunLogger.logging(str(video_id), title, video_path, data)
        print("写入日志成功")




if __name__ == '__main__':
    video_processor = VideoProcessing()
    video_processor.get_video()