video_processing.py 1.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758
  1. import re
  2. import time
  3. import requests
  4. import json
  5. from common.aliyun_log import AliyunLogger
  6. from common.redis import get_video_data, install_video_data, in_video_data
  7. class VideoProcessing:
  8. def get_ai_data(self, video_path):
  9. url = "http://8.219.186.16:8080/process_video/"
  10. payload = json.dumps( {
  11. "video_path": video_path
  12. } )
  13. headers = {
  14. 'Content-Type': 'application/json'
  15. }
  16. try:
  17. response = requests.request( "POST", url, headers=headers, data=payload )
  18. response = response.json()
  19. result = response['result']
  20. mark = response['mark']
  21. cleaned_string = result.replace( "```json", '' ).replace( "```", '' ).strip()
  22. return cleaned_string, str(mark)
  23. except Exception as e:
  24. print(f"视频请求异常:{e}")
  25. return None
  26. def get_video(self, redis_task):
  27. video_data = get_video_data(redis_task)
  28. if not video_data:
  29. print("没有获取到视频内容")
  30. time.sleep(120)
  31. return
  32. # 解码为字符串
  33. data_str = video_data.decode( 'utf-8' )
  34. # 解析为 JSON 对象
  35. data_json = json.loads( data_str )
  36. video_id = data_json['video_id']
  37. title = data_json['title']
  38. video_path = data_json['video_path']
  39. type = data_json['type']
  40. print(video_path)
  41. data, mark = self.get_ai_data(video_path)
  42. AliyunLogger.logging(str(video_id), title, video_path, mark, type, data)
  43. print("写入日志成功")
  44. if __name__ == '__main__':
  45. video_processor = VideoProcessing()
  46. video_processor.get_video()