123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134 |
- # -*- 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_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
- if __name__ == '__main__':
- Oss.download_sph_ls('channel/video/sph/14374775553517295881.jpg','asa','1')
|