cross_doctor/tool/qcos_func.py

# -*- coding=utf-8
from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
from qcloud_cos.cos_exception import CosServiceError
import sys
import os
import logging

# 正常情况日志级别使用 INFO，需要定位时可以修改为 DEBUG，此时 SDK 会打印和服务端的通信信息
# logging.basicConfig(level=logging.INFO, stream=sys.stdout)

g_cos_root = 'https://xinglu-1324629296.cos.ap-beijing.myqcloud.com'
g_cos_bucket = 'xinglu-1324629296'

def get_client():
    # 1. 设置用户属性, 包括 secret_id, secret_key, region等。Appid 已在 CosConfig 中移除，请在参数 Bucket 中带上 Appid。Bucket 由 BucketName-Appid 组成
    # secret_id = os.environ['COS_SECRET_ID']     # 用户的 SecretId，建议使用子账号密钥，授权遵循最小权限指引，降低使用风险。子账号密钥获取可参见 https://cloud.tencent.com/document/product/598/37140
    # secret_key = os.environ['COS_SECRET_KEY']   # 用户的 SecretKey，建议使用子账号密钥，授权遵循最小权限指引，降低使用风险。子账号密钥获取可参见 https://cloud.tencent.com/document/product/598/37140
    """
    SecretId:AKIDFLFvOfjbFi171mLnEXp370rHZozymN2c
    SecretKey:a6RYUDl6EAtQmurb8To4AFRCoDOM6g3D
    """
    secret_id = 'AKIDFLFvOfjbFi171mLnEXp370rHZozymN2c'
    secret_key = 'a6RYUDl6EAtQmurb8To4AFRCoDOM6g3D'
    region = 'ap-beijing'  # 替换为用户的 region，已创建桶归属的 region 可以在控制台查看，https://console.cloud.tencent.com/cos5/bucket
    # COS 支持的所有 region 列表参见https://cloud.tencent.com/document/product/436/6224
    token = None  # 如果使用永久密钥不需要填入 token，如果使用临时密钥需要填入，临时密钥生成和使用指引参见 https://cloud.tencent.com/document/product/436/14048
    scheme = 'https'  # 指定使用 http/https 协议来访问 COS，默认为 https，可不填

    config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token, Scheme=scheme)
    client = CosS3Client(config)
    return client


def upload_file_to_cos(local_filename: str, cos_path: str):
    client = get_client()
    # 本地路径 简单上传
    response = client.put_object_from_local_file(
        Bucket=g_cos_bucket,
        LocalFilePath=local_filename,
        Key=cos_path
    )
    print(response['ETag'])
    return response


def get_file_from_cos(cos_path: str, local_filename: str):
    """
    从cos上下载文件
    :param cos_path:
    :param local_filename:
    :return:
    """
    client = get_client()
    response = client.get_object(
        Bucket=g_cos_bucket,
        Key=cos_path
    )
    response['Body'].get_stream_to_file(local_filename)



def upload_order_images_to_cos(idx_to_localfile: dict, orderid: str):
    """
    将一组本地图片，上传到对应的cos的订单目录下
    :param idx_to_localfile:
    :param orderid:
    :return: idx_to_cosfile
    """
    xlsyn_cos_host = 'https://xinglu-1324629296.cos.ap-beijing.myqcloud.com'
    cos_order_path = '/user/wave_survey/%s' % orderid
    idx_to_cosfile = {}
    for idx, local_filepath in idx_to_localfile.items():
        image_filename = local_filepath.split('/')[-1]
        cos_path = '%s/%s' % (cos_order_path, image_filename)
        upload_file_to_cos(local_filepath, cos_path)
        cos_file = '%s/%s' % (xlsyn_cos_host, cos_path)
        idx_to_cosfile[idx] = cos_file
    return idx_to_cosfile


class CosFolderManager:
    def __init__(self, client, bucket):
        self.client = client
        self.bucket = bucket

    def _normalize_folder(self, folder_path):
        """确保文件夹路径以 / 结尾"""
        if not folder_path:
            return ''
        # 去除开头的 /
        if folder_path.startswith('/'):
            folder_path = folder_path[1:]
        # 确保结尾有 /
        if not folder_path.endswith('/'):
            folder_path += '/'
        return folder_path

    def folder_exists(self, folder_path):
        """
        判断文件夹是否存在
        :param folder_path: 文件夹路径，如 'data/logs/' 或 'data/logs'
        :return: True/False
        """
        key = self._normalize_folder(folder_path)
        if not key:
            return True  # 根目录默认存在

        try:
            self.client.head_object(Bucket=self.bucket, Key=key)
            return True
        except CosServiceError as e:
            if e.get_status_code() == 404:
                return False
            raise  # 其他错误向上抛出

    def create_folder(self, folder_path):
        """
        创建文件夹（上传一个以 / 结尾的空对象）
        :param folder_path: 文件夹路径
        :return: None
        """
        key = self._normalize_folder(folder_path)
        if not key:
            return  # 根目录无需创建

        self.client.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=b''  # 空内容
        )

    def ensure_folder(self, folder_path):
        """
        确保文件夹存在，不存在则创建
        :param folder_path: 文件夹路径
        :return: True(已存在) / False(刚创建)
        """
        if self.folder_exists(folder_path):
            return True
        else:
            self.create_folder(folder_path)
            return False


if __name__ == '__main__':
    idx_to_localfile = {0:'D:/slgwork/slgcode/wave_survey/1739430848000.jpg',
                           2:'D:/slgwork/slgcode/wave_survey/1739430851000.jpg'}
    idx_to_cosfile = upload_order_images_to_cos(idx_to_localfile, 'slg_orderid')
    print(idx_to_cosfile)