时光相册-视频生成

Ansjer/urls.py

@@ -25,7 +25,7 @@ from Controller import FeedBack, EquipmentOTA, EquipmentInfo, AdminManage, AppIn
     RegionController, VPGController, LanguageController, TestController, DeviceConfirmRegion, S3GetStsController, \
     DetectControllerV2, PcInfo, PctestController, DeviceDebug, PaymentCycle, \
     DeviceLogController, CouponController, AiController, ShadowController, AppAccountManagement, InitController, \
-    WeatherControl, SmartReplyController, InAppPurchaseController, DeviceCommonController
+    WeatherControl, SmartReplyController, InAppPurchaseController, DeviceCommonController, TimeAlbumController
 from Controller.Cron import CronTaskController
 from Controller.CustomCustomer import CustomCustomerController
 from Controller.MessagePush import EquipmentMessagePush
@@ -284,6 +284,7 @@ urlpatterns = [
     re_path('deviceCommon/(?P<operation>.*)$', DeviceCommonController.DeviceCommonView.as_view()),
     re_path('customCustomer/(?P<operation>.*)$', CustomCustomerController.CustomCustomerView.as_view()),
     re_path('adDepartment/(?P<operation>.*)$', AdDepartmentController.AdDepartmentView.as_view()),
+    re_path('timeAlbum/(?P<operation>.*)$', TimeAlbumController.TimeAlbum.as_view()),
 
 
 

CeleryTask/TimeAlbumTask.py

@@ -0,0 +1,63 @@
+import logging
+import os
+import subprocess
+from celery import shared_task
+from Ansjer.config import LOGGER
+import django
+
+# 设置 Django 的环境变量
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", 'Ansjer.local_config.local_settings')  # 确保替换为实际的 settings 模块路径
+django.setup()
+
+@shared_task
+def generate_video(image_files, output_path):
+    video_files = []
+    music_path = "static/temp/music.mp3"
+    # 定义特效列表
+    transitions = [
+        "BowTieHorizontal.glsl",
+        "burn.glsl",
+        "cube.glsl",
+        "pinwheel.glsl",
+        "windowslice.glsl",
+        "Radial.glsl",
+        "rotateTransition.glsl",
+        "wind.glsl",
+        "squeeze.glsl"
+    ]
+
+    # 生成相邻图片的视频
+    for i in range(len(image_files) - 1):
+        output_video = os.path.join(output_path, f'{i + 1}_{i + 2}.mp4')
+        video_files.append(output_video)
+
+        cmd = [
+            'ffmpeg', '-loop', '1', '-i', image_files[i],
+            '-loop', '1', '-i', image_files[i + 1],
+            '-filter_complex',
+            f'gltransition=duration=1:offset=0.7:source=gl-transitions/{transitions[i]}',
+            '-t', '2', output_video
+        ]
+
+        subprocess.run(cmd, check=True)
+
+    # 生成拼接列表文件
+    with open(os.path.join(output_path, 'filelist.txt'), 'w') as f:
+        for video in video_files:
+            f.write(f"file '{video}'\n")
+
+    # 拼接视频
+    final_output = os.path.join(output_path, 'final_output.mp4')
+    subprocess.run(['ffmpeg', '-f', 'concat', '-safe', '0', '-i', 'filelist.txt', '-c', 'copy', final_output],
+                   check=True)
+
+    # 添加背景音乐
+    final_output_with_music = os.path.join(output_path, 'final_output_with_music.mp4')
+    subprocess.run(['ffmpeg', '-i', final_output, '-i', music_path, '-c:v', 'copy', '-c:a', 'aac', '-b:a', '192k',
+                    '-shortest', final_output_with_music], check=True)
+
+    # 清理临时文件
+    for video in video_files:
+        os.remove(video)
+    os.remove('filelist.txt')
+

Controller/TimeAlbumController.py

@@ -0,0 +1,42 @@
+from django.views import View
+
+from CeleryTask.TimeAlbumTask import generate_video
+from Object.ResponseObject import ResponseObject
+
+
+class TimeAlbum(View):
+    def get(self, request, *args, **kwargs):
+        request.encoding = 'utf-8'
+        operation = kwargs.get('operation')
+        request_dict = request.GET
+        return self.validation(request, request_dict, operation)
+
+    def post(self, request, *args, **kwargs):
+        request.encoding = 'utf-8'
+        operation = kwargs.get('operation')
+        request_dict = request.POST
+        return self.validation(request, request_dict, operation)
+
+    def validation(self, request, request_dict, operation):
+        language = request_dict.get('language', 'en')
+        response = ResponseObject(language)
+        if operation == 'createTimeAlbum':
+            return self.create_time_album(response)
+        else:
+            return response.json(414)
+
+    @staticmethod
+    def create_time_album(response):
+        # 从AWS桶获取图片链接
+
+        # 下载图片保存地址
+
+        # 收集图片文件路径
+        image_files = ['static/temp/image1', 'static/temp/image2', 'static/temp/image3', 'static/temp/image4',
+                       'static/temp/image5']
+
+        output_path = 'static/temp/'
+
+        # 调用 Celery 任务
+        generate_video.delay(image_files, output_path)
+        return response.json(0)