forked from AlotOfBlahaj/Auto_Record_Matsuri
-
Notifications
You must be signed in to change notification settings - Fork 0
/
upload.py
90 lines (77 loc) · 3.23 KB
/
upload.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
import re
import subprocess
from urllib.parse import quote
from abc import ABCMeta, abstractmethod
from os import name
from minio import Minio
from botocore.exceptions import ClientError
from config import ddir, enable_db, s3_access_key, s3_secret_key, upload_by, s3_server
from tools import get_logger, ABSPATH, Database, bot
class Upload(metaclass=ABCMeta):
@abstractmethod
def upload_item(self, item_path, item_name):
pass
class S3Upload(Upload):
def __init__(self):
self.logger = get_logger('S3Upload')
self.minio = Minio(s3_server,
access_key=s3_access_key,
secret_key=s3_secret_key,
secure=True)
def upload_item(self, item_path, item_name):
self.minio.fput_object('matsuri', item_name, item_path)
class BDUpload(Upload):
def __init__(self):
self.logger = get_logger('BDUpload')
def upload_item(self, item_path: str, item_name: str) -> None:
if 'nt' in name:
command = [f"{ABSPATH}\\BaiduPCS-Go\\BaiduPCS-Go.exe", "upload", "--nofix"]
else:
command = [f"{ABSPATH}/BaiduPCS-Go/BaiduPCS-Go", "upload", "--nofix"]
command.append(item_path)
command.append("/")
subprocess.run(command)
def share_item(self, item_name: str) -> str:
if 'nt' in name:
command = [f'{ABSPATH}\\BaiduPCS-GO\\BaiduPCS-Go.exe', "share", "set"]
else:
command = [f"{ABSPATH}/BaiduPCS-Go/BaiduPCS-Go", "share", "set"]
command.append(item_name)
s2 = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
encoding='utf-8', universal_newlines=True)
share_info = s2.stdout
if 'https' in share_info:
share_info = share_info.replace('\n', '')
self.logger.info(f'{item_name}: Share successful {share_info}')
else:
self.logger.error('Share failed')
raise RuntimeError(f'{item_name} share failed')
reg = r'https://pan.baidu.com/s/([A-Za-z0-9_-]{23})'
linkre = re.compile(reg)
link = re.search(linkre, share_info)
try:
link = 'https://pan.baidu.com/s/' + link.group(1)
except AttributeError:
self.logger.exception('get share link error')
raise RuntimeError('get share link error')
return link
def upload_video(video_dict):
upload_way_dict = {'bd': BDUpload,
's3': S3Upload}
upload_way = upload_way_dict.get(upload_by)
uploader = upload_way()
uploader.upload_item(f"{ddir}/{video_dict['Title']}", video_dict['Title'])
if upload_by == 'bd':
share_url = uploader.share_item(video_dict['Title'])
if enable_db:
db = Database('Video')
db.insert(video_dict['Title'], share_url, video_dict['Date'])
elif upload_by == 's3':
if enable_db:
db = Database('Video')
db.insert(video_dict['Title'],
f"gets3/{quote(video_dict['Title'])}",
video_dict['Date'])
else:
raise RuntimeError(f'Upload {video_dict["Title"]} failed')
bot(f"[下载提示] {video_dict['Title']} 已上传, 请查看页面")