# Import FastAPI Libs from fastapi import FastAPI from pydantic import BaseModel # Import Qcloud COS Service Libs from qcloud_cos import CosConfig from qcloud_cos import CosS3Client # Import Other Common Libs import sys import os import logging from datetime import datetime import mimetypes import json # ------------------------------------------------ # Define constants # ------------------------------------------------ c_ret_code_success = 0 c_ret_code_error = 1 region = 'ap-shanghai' bucket = 'clark-apex-saz-1320304559' user_name = 'oss-program' credential_file = '/home/oracle/python/apex/credential/credential.json' # ------------------------------------------------ # Get credentials # ------------------------------------------------ def get_credentials(file: str, bucket_name: str, user_name: str) -> tuple: try: # Open and read the JSON file with open(file, "r") as credential_file: data = json.load(credential_file) # Access data from the JSON secret_id = data[bucket_name][user_name]['secret_id'] secret_key = data[bucket_name][user_name]['secret_key'] return secret_id, secret_key except Exception as e: print(e) return None, None # ------------------------------------------------ # Function of upload single file to Qcloud COS # ------------------------------------------------ def upload_single_file(dir: str, file_name: str, rename_prefix: str, client: CosS3Client, bucket: str, index: int) -> tuple: file_key = "" file_etag = "" ret_code = c_ret_code_success err_msg = "" try: # Check if file is empty if file_name == None or file_name == "": file_key = "" else: # Check file existance file_location = os.path.join(dir, file_name) mime_type, _ = mimetypes.guess_type(file_location) if os.path.exists(file_location): file_key = rename_prefix + "_" + str(index) + "_" + file_name # Call COS service to upload with open(file_location, 'rb') as fp: response = client.put_object( Bucket=bucket, # Bucket 由 BucketName-APPID 组成 Body=fp, Key=file_key, StorageClass='STANDARD', ContentType=mime_type ) file_etag = response['ETag'].strip('"') # delete file after upload os.remove(file_location) else: file_key = "" return ret_code, err_msg, file_key, file_etag except Exception as e: ret_code = c_ret_code_error err_msg = str(e) return ret_code, err_msg, file_key, file_etag # ------------------------------------------------ # Define File Upload Service Request and Response # ------------------------------------------------ class File_Upload_Req(BaseModel): dir: str file1_name: str file2_name: str file3_name: str file4_name: str file5_name: str class File_Upload_Res(BaseModel): result: int result_message: str file1_key: str file1_etag: str file2_key: str file2_etag: str file3_key: str file3_etag: str file4_key: str file4_etag: str file5_key: str file5_etag: str app = FastAPI() # ------------------------------------------------ # Define File Upload Service Process Detail # ------------------------------------------------ @app.post("/upload/") async def upload(file_uplaod_req: File_Upload_Req): secret_id, secret_key = get_credentials(credential_file, bucket, user_name) #secret_id = # 用户的 SecretId，建议使用子账号密钥，授权遵循最小权限指引，降低使用风险。子账号密钥获取可参见 https://cloud.tencent.com/document/product/598/37140 #secret_key = # 用户的 SecretKey，建议使用子账号密钥，授权遵循最小权限指引，降低使用风险。子账号密钥获取可参见 https://cloud.tencent.com/document/product/598/37140 # --------------------------------------------------- # 10. Set Logging Level # --------------------------------------------------- # 正常情况日志级别使用 INFO，需要定位时可以修改为 DEBUG，此时 SDK 会打印和服务端的通信信息 logging.basicConfig(level=logging.INFO, stream=sys.stdout) # --------------------------------------------------- # 20. Set Credential # --------------------------------------------------- # 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 # secret_id = 'AKIDQjrAHBsKctxoSzNYXSHtITVKH0Yv64tb' # 用户的 SecretId，建议使用子账号密钥，授权遵循最小权限指引，降低使用风险。子账号密钥获取可参见 https://cloud.tencent.com/document/product/598/37140 # secret_key = 'd2NZ1KXcfrriPwRnxvhKpFlHiy0w6CD2' # 用户的 SecretKey，建议使用子账号密钥，授权遵循最小权限指引，降低使用风险。子账号密钥获取可参见 https://cloud.tencent.com/document/product/598/37140 # region = 'ap-shanghai' # 替换为用户的 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，可不填 # --------------------------------------------------- # 30. Config Client # --------------------------------------------------- config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token, Scheme=scheme) client = CosS3Client(config) # --------------------------------------------------- # 40. Compose File Keys # --------------------------------------------------- # Get current date yyyy/mm/dd current_time = datetime.now() formatted_date = current_time.strftime("%Y/%m/%d/%H%M%S") # --------------------------------------------------- # 50. Upload Files # --------------------------------------------------- result = c_ret_code_success error_list = [] # File 1 ret_code, err_msg, file1_key, file1_etag = upload_single_file(file_uplaod_req.dir, file_uplaod_req.file1_name, formatted_date, client, bucket, 1) if ret_code != c_ret_code_success: result = c_ret_code_error error_list.append("File 1 upload failed: " + err_msg) # File 2 ret_code, err_msg, file2_key, file2_etag = upload_single_file(file_uplaod_req.dir, file_uplaod_req.file2_name, formatted_date, client, bucket, 2) if ret_code != c_ret_code_success: result = c_ret_code_error error_list.append("File 2 upload failed: " + err_msg) # File 3 ret_code, err_msg, file3_key, file3_etag = upload_single_file(file_uplaod_req.dir, file_uplaod_req.file3_name, formatted_date, client, bucket, 3) if ret_code != c_ret_code_success: result = c_ret_code_error error_list.append("File 3 upload failed: " + err_msg) # File 4 ret_code, err_msg, file4_key, file4_etag = upload_single_file(file_uplaod_req.dir, file_uplaod_req.file4_name, formatted_date, client, bucket, 4) if ret_code != c_ret_code_success: result = c_ret_code_error error_list.append("File 4 upload failed: " + err_msg) # File 5 ret_code, err_msg, file5_key, file5_etag = upload_single_file(file_uplaod_req.dir, file_uplaod_req.file5_name, formatted_date, client, bucket, 5) if ret_code != c_ret_code_success: result = c_ret_code_error error_list.append("File 5 upload failed: " + err_msg) file_upload_res = File_Upload_Res( result = result, result_message = ', '.join(error_list), file1_key = file1_key, file1_etag = file1_etag, file2_key = file2_key, file2_etag = file2_etag, file3_key = file3_key, file3_etag = file3_etag, file4_key = file4_key, file4_etag = file4_etag, file5_key = file5_key, file5_etag = file5_etag ) return file_upload_res # ------------------------------------------------ # Define File Download URL Service Request and Response # ------------------------------------------------ class File_Download_Req(BaseModel): key: str class File_Download_Res(BaseModel): result: int result_message: str download_url: str # ------------------------------------------------ # Define File Download URL Service Process Detail # ------------------------------------------------ @app.post("/get_download_url/") async def get_download_url(file_download_req: File_Download_Req): # Define constants secret_id, secret_key = get_credentials(credential_file, bucket, user_name) # --------------------------------------------------- # 10. Set Logging Level # --------------------------------------------------- # 正常情况日志级别使用 INFO，需要定位时可以修改为 DEBUG，此时 SDK 会打印和服务端的通信信息 logging.basicConfig(level=logging.INFO, stream=sys.stdout) # --------------------------------------------------- # 20. Set Credential # --------------------------------------------------- # 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 # secret_id = 'AKIDQjrAHBsKctxoSzNYXSHtITVKH0Yv64tb' # 用户的 SecretId，建议使用子账号密钥，授权遵循最小权限指引，降低使用风险。子账号密钥获取可参见 https://cloud.tencent.com/document/product/598/37140 # secret_key = 'd2NZ1KXcfrriPwRnxvhKpFlHiy0w6CD2' # 用户的 SecretKey，建议使用子账号密钥，授权遵循最小权限指引，降低使用风险。子账号密钥获取可参见 https://cloud.tencent.com/document/product/598/37140 # region = 'ap-shanghai' # 替换为用户的 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，可不填 # --------------------------------------------------- # 30. Config Client # --------------------------------------------------- config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token, Scheme=scheme) client = CosS3Client(config) try: # 生成下载 URL，未限制请求头部和请求参数 url = client.get_presigned_url( Method='GET', Bucket=bucket, Key=file_download_req.key, Expired=120 # 120秒后过期，过期时间请根据自身场景定义 ) ret_code = c_ret_code_success err_msg = "" except Exception as e: ret_code = c_ret_code_error err_msg = str(e) file_download_res = File_Download_Res( result = ret_code, result_message = err_msg, download_url = url ) return file_download_res