进入m3u8文件对应目录，执行以下命令 ，注意修改对应文件名

ffmpeg -allowed_extensions ALL -i *.m3u8 -c copy XXX.mp4

情况1

1. linux 使用宿主机的网络地址 使用172.17.0.1（Docker默认桥接网络的网关地址）
2. Windows或Mac，使用实际的IP地址或特定的地址（如host.docker.internal）

情况2

如果docker版本比较低，可以使用

docker run --add-host=host.docker.internal:host-gateway

对于 docker-compose.yml

services:
  your-service:
    extra_hosts:
      - "host.docker.internal:host-gateway"

如果无法识别 host-gateway，改为宿主机host

--add-host=host.docker.internal:<宿主机IP> ...

如果访问宿主机无法识别 host.docker.internal
报错
got error dial tcp: lookup host.docker.internal on 127.0.0.11:53: no such host
宿主机手动添加host

127.0.0.1 host.docker.internal

终极大招

修改宿主机防火墙规则，增加ip来源

sudo firewall-cmd --permanent --zone=public --add-rich-rule='rule family="ipv4" source address="192.168.1.0/24" port port="3399" protocol="tcp" accept'

sudo firewall-cmd --reload

model

class Video(models.Model):
    view_count = models.IntegerField(verbose_name='浏览量', default=0)
    like_count = models.IntegerField(verbose_name='点赞量', default=0)
    collect_count = models.IntegerField(verbose_name='收藏量', default=0)
    comment_count = models.IntegerField(verbose_name='评论量', default=0)
    # 忽略其他字段

view

from django.db.models import Sum, Count

data = Video.objects.aggregate(
        count=Count('id'),
        view_count=Sum('view_count'),
        like_count=Sum('like_count'),
        collect_count=Sum('collect_count'),
        comment_count=Sum('comment_count')
    )

安装 crypto-js

pnpm install crypto-js

代码

import * as CryptoJS from 'crypto-js';


export async function decryptText(text: string): Promise<string> {
    // 16进制
    const keyHex = "cb69976b953380e5ff637b0bf9af35fdf1a56815c81296b3e5314e506d209a05";
    const ivHex = 'a312cc7df85e4d26dc65fb4f3fe82919'
    // console.log(`text: ${text}`)
    // Convert hex strings to Uint8Array
    const keyArray = new Uint8Array(
        keyHex.match(/.{1,2}/g).map(byte => parseInt(byte, 16))
    );
    const ivArray = new Uint8Array(
        ivHex.match(/.{1,2}/g).map(byte => parseInt(byte, 16))
    );

    const encryptedData = hexStringToByteArray(text)

    // AES decryption using crypto-js
    const decryptedData = CryptoJS.AES.decrypt(
        {
            ciphertext: CryptoJS.enc.Hex.parse(
                Array.from(encryptedData).map(byte => ('0' + (byte & 0xFF).toString(16)).slice(-2)).join('')
            )
        },
        CryptoJS.enc.Hex.parse(
            Array.from(keyArray).map(byte => ('0' + byte.toString(16)).slice(-2)).join('')
        ),
        {
            iv: CryptoJS.enc.Hex.parse(
                Array.from(ivArray).map(byte => ('0' + byte.toString(16)).slice(-2)).join('')
            ),
            mode: CryptoJS.mode.CBC,
            padding: CryptoJS.pad.Pkcs7,
        }
    );

    return decryptedData.toString(CryptoJS.enc.Utf8)
}

定义模型

class TaskTracking(models.Model):
    task_id = models.CharField(max_length=100, null=False, blank=False, verbose_name='任务id', unique=True)
    task_type = models.CharField(max_length=100, null=False, blank=False, verbose_name='任务类型')
    task_pk = models.IntegerField(verbose_name='任务主键', null=False, blank=False, default=0)
    status = models.CharField(max_length=100, null=False, blank=False, verbose_name='任务状态')
    result = models.CharField(max_length=100, null=True, blank=True, verbose_name='任务结果', )
    created_at = models.DateTimeField(verbose_name='创建时间', editable=False, null=False, auto_now_add=True)
    updated_at = models.DateTimeField(verbose_name='更新时间', editable=False, null=False, auto_now=True)

    class Meta:
        verbose_name = "任务追踪"
        verbose_name_plural = verbose_name

迁移

python manage.py makemigrations && python manage.py migrate

定义装饰器

from functools import wraps
from celery.app.task import Task
from study.celery import app


def task_tracker(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 获取方法名作为 task_type
        task_type = func.__name__

        # 获取任务ID
        task_id = None
        if isinstance(args[0], Task):  # 检查第一个参数是否是 Celery 任务实例
            task_id = args[0].request.id

        # 获取 x 和 y 的值
        task_pk = args[1] if len(args) > 1 else kwargs.get('pk')

        # 任务开始，记录到数据库（包括 x 和 y 的值）
        task = TaskTracking.objects.create(task_id=task_id, task_type=task_type, task_pk=task_pk, status='started')

        try:
            result = func(*args, **kwargs)
            # 任务成功，更新数据库状态
            task.status = 'completed'
            task.result = result
            task.save()
            return result
        except Exception as e:
            # 任务失败，更新数据库状态
            task.status = 'failed'
            task.result = str(e)
            task.save()
            raise e

    return wrapper

use

@app.task(bind=True)
@task_tracker
def dosomething(self, pk):
    pass