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
from oss2 import Auth, Bucket, exceptions
from loguru import logger
from utils.config import OssConfig
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_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):
        video_path = file_path + 'video.mp4'
        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 upload_to_aliyun(cls,local_path):
        """
        上传文件到阿里云 OSS

        参数:
        local_path (str): 本地文件路径
        oss_path (str): OSS 目标路径

        返回:
        bool: 上传成功返回 True，失败返回 False
        """
        try:
            oss_path = generate_oss_path()
            logger.info(f"开始上传到阿里云 OSS: {oss_path}")


            # 初始化认证和存储空间
            auth = Auth(OssConfig["OSS_ACCESS_KEY_ID"], OssConfig["OSS_ACCESS_KEY_SECRET"])
            bucket = Bucket(auth, OssConfig["ENDPOINT"], OssConfig["BUCKETNAME"])

            # 上传文件
            result = bucket.put_object_from_file(oss_path, local_path)

            # 检查上传结果
            if result.status == 200:
                logger.info(f"上传成功，OSS 路径: {oss_path}")
                return OssConfig["BUCKETNAME_HOST"]+oss_path
            else:
                logger.error(f"上传失败，状态码: {result.status}")
                return False

        except exceptions.OssError as e:
            logger.error(f"阿里云 OSS 错误: {e}")
            return False
        except Exception as e:
            logger.error(f"上传异常: {e}")
            return

def generate_oss_path():
    """生成唯一OSS路径（日期+UUID）"""
    # 生成日期目录（格式：YYYYMMDD）
    date_dir = datetime.now().strftime("%Y%m%d")
    # 生成随机文件名（使用 UUID）
    random_filename = f"{uuid.uuid4().hex}.mp3"
    # 构建完整的 OSS 路径
    oss_base_dir = OssConfig["OSS_BASE_DIR"]
    oss_path = f"{oss_base_dir}/{date_dir}/{random_filename}"
    return f"{oss_path}"


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