carry_efficiency_tool/utils/aliyun_oss.py

# -*- coding: utf-8 -*-
# @Time: 2023/12/26
import time
import uuid
from datetime import datetime
from typing import Dict, Any,  Optional
import oss2
import requests
OSS_ACCESS_KEY_ID = "LTAIP6x1l3DXfSxm"
OSS_ACCESS_KEY_SECRET = "KbTaM9ars4OX3PMS6Xm7rtxGr1FLon"
# OSS_BUCKET_ENDPOINT = "oss-cn-hangzhou-internal.aliyuncs.com"# 内网地址
OSS_BUCKET_ENDPOINT = "oss-cn-hangzhou.aliyuncs.com" # 外网地址
OSS_BUCKET_NAME = "art-crawler"
class Oss:

    @classmethod
    def channel_upload_oss(cls, src_url: str,
                              video_id: str,
                              referer: Optional[str] = None) -> Dict[str, Any]:
        headers = {
            'Accept': '*/*',
            'Accept-Language': 'zh-CN,zh;q=0.9',
            'Cache-Control': 'no-cache',
            'Pragma': 'no-cache',
            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) '
                          'Chrome/117.0.0.0 Safari/537.36',
        }
        if referer:
            headers.update({'Referer': referer})
        response = requests.request(url=src_url, method='GET', headers=headers, timeout=30)
        file_content = response.content
        content_type = response.headers.get('Content-Type', 'application/octet-stream')

        oss_object_key = f'carry/video/{video_id}'
        auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
        bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, OSS_BUCKET_NAME)
        response = bucket.put_object(oss_object_key, file_content, headers={'Content-Type': content_type})

        if 'Content-Length' in response.headers:
            return {
                'status': response.status,
                'oss_object_key': oss_object_key}
        raise AssertionError(f'OSS上传失败，请求ID: \n{response.headers["x-oss-request-id"]}')

    """
        视频发送到art-pubbucket
        """

    @classmethod
    def stitching_sync_upload_bytes_to_oss(cls, data: str, video_id: str) -> Dict[str, Any]:
        oss_object_key = f'carry/video/{video_id}'
        auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
        bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, "art-pubbucket")
        response = bucket.put_object(oss_object_key, data)

        if 'Content-Length' in response.headers:
            return {
                'status': response.status,
                'oss_object_key': oss_object_key,
                'save_oss_timestamp': int(datetime.now().timestamp() * 1000),
            }
        raise AssertionError(f'OSS上传失败，请求ID: \n{response.headers["x-oss-request-id"]}')

    """
    视频发送到art-pubbucket
    """
    @classmethod
    def stitching_sync_upload_oss(cls, src_url: str,
                        video_id: str) -> Dict[str, Any]:
        oss_object_key = f'carry/video/{video_id}'
        auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
        bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, "art-pubbucket")
        response = bucket.put_object_from_file(oss_object_key, src_url)

        if 'Content-Length' in response.headers:
            return {
                'status': response.status,
                'oss_object_key': oss_object_key,
                'save_oss_timestamp': int(datetime.now().timestamp() * 1000),
            }
        raise AssertionError(f'OSS上传失败，请求ID: \n{response.headers["x-oss-request-id"]}')

    """
    封面发送到art-pubbucket
    """
    @classmethod
    def stitching_fm_upload_oss(cls, src_url: str,
                                  video_id: str) -> Dict[str, Any]:
        oss_object_key = f'jq_oss/jpg/{video_id}.jpg'
        auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
        bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, "art-pubbucket")
        response = bucket.put_object_from_file(oss_object_key, src_url)

        if 'Content-Length' in response.headers:
            return {
                'status': response.status,
                'oss_object_key': oss_object_key,
                'save_oss_timestamp': int(datetime.now().timestamp() * 1000),
            }
        raise AssertionError(f'OSS上传失败，请求ID: \n{response.headers["x-oss-request-id"]}')

    """
    封面发送到art-pubbucket
    """
    @classmethod
    def mp3_upload_oss(cls, src_url: str,
                                video_id: str) -> Dict[str, Any]:
        oss_object_key = f'jq_audio/audio/{video_id}.mp3'
        auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
        bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, "art-crawler")
        response = bucket.put_object_from_file(oss_object_key, src_url)

        if 'Content-Length' in response.headers:
            return {
                'status': response.status,
                'oss_object_key': oss_object_key,
                'save_oss_timestamp': int(datetime.now().timestamp() * 1000),
            }
        raise AssertionError(f'OSS上传失败，请求ID: \n{response.headers["x-oss-request-id"]}')


    @classmethod
    def download_video_oss(cls, url, file_path, add_suffix: bool = True):
        video_path = file_path + 'video.mp4' if add_suffix else file_path
        oss_object_key = cls.channel_upload_oss(url, str(uuid.uuid4()))
        time.sleep(2)
        oss_object = oss_object_key.get("oss_object_key")
        if oss_object:
            auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
            bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, OSS_BUCKET_NAME)
            # 获取指定路径下的对象列表
            bucket.get_object_to_file(oss_object, video_path)
            time.sleep(5)
            return video_path
        else:
            return video_path

    @classmethod
    def download_sph_ls(cls, video_url, video_path_url, v_id):
        if "jpg" in video_url:
            video_path = video_path_url + str(v_id) + '.jpg'
        else:
            video_path = video_path_url + str(v_id) + '.mp4'
        auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
        bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, OSS_BUCKET_NAME)
        # 获取指定路径下的对象列表
        bucket.get_object_to_file(video_url, video_path)
        time.sleep(5)
        return video_path

    @classmethod
    def generate_url(cls, oss_object_key: str) -> str:
        auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
        bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, 'art-pubbucket')
        return bucket.sign_url(method='GET', key=oss_object_key, expires=3600, slash_safe=True)


if __name__ == '__main__':
    Oss.download_sph_ls('channel/video/sph/14374775553517295881.jpg','asa','1')