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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
| import os
import time
import boto3
import json5
import logging
import argparse
import traceback
class S3Operator():
"""s3操作类
"""
def __init__(self, access_key: str, secret_key: str, bucket_name: str, endpoint: str, region_name: str):
session = boto3.Session(aws_access_key_id=access_key, aws_secret_access_key=secret_key)
self.__resource = session.resource(service_name="s3", region_name=region_name, endpoint_url=endpoint)
self.__bucket_name = bucket_name
def uploadFile(self, src_path, dest_path) -> bool:
"""上传单个文件
Parameters:
src_path (str): 源文件, 必须是文件而不是目录
dest_path (str): 目标文件
Returns:
bool: 是否上传成功
注意: 可能会覆盖同名文件
"""
if os.path.isdir(src_path):
logging.warning(f"Upload file failed: {src_path} is a directory. {traceback.format_exc()}")
return False
try:
self.__resource.Bucket(self.__bucket_name).upload_file(src_path, dest_path)
logging.info(f"Upload file success: {src_path} -> {dest_path}")
except Exception as e:
logging.error(f"Upload file failed: {e}. {traceback.format_exc()}")
return False
return True
def uploadFolder(self, src_path, dest_path) -> bool:
"""上传整个目录
Parameters:
src_path (str): 源目录, 可以是文件也可以是目录
dest_path (str): 目标目录, 可以是文件也可以是目录
Returns:
bool: 是否上传成功
"""
if os.path.isdir(src_path):
files = self.getRelativeFilePaths(src_path)
logging.info(f"Files detected: {files}")
res = True
src_path = src_path if src_path.endswith(os.path.sep) else os.path.dirname(src_path)
for file in files:
logging.info(f"Uploading file: {os.path.join(src_path, file)} -> {os.path.join(dest_path, file)}")
res = self.uploadFile(os.path.join(src_path, file), os.path.join(dest_path, file))
return res
else:
logging.warning(f"{src_path} is a file. Try to upload as a file. {traceback.format_exc()}")
self.uploadFile(src_path, dest_path)
return False
def delFiles(self, files: list) -> bool:
"""批量删除文件
Parameters:
files (list): 文件列表, 元素为 {'Key': '<filename>'}
Returns:
bool: 是否删除成功
"""
try:
self.__resource.meta.client.delete_objects(Bucket=self.__bucket_name, Delete={"Objects": files})
return True
except Exception as e:
logging.error(f"Delete files failed: {e}. {traceback.format_exc()}")
return False
def delFolder(self, folder: str) -> bool:
"""删除目录
Parameters:
folder (str): 目录
Returns:
bool: 是否删除成功
"""
try:
files = self.walkfolder(folder)
paths = [{'Key': i['Key']} for i in files]
if paths == []:
logging.warning(f"Folder does not exist: {folder}")
return True
if self.delFiles(paths):
logging.info(f"Delete folder success: {folder}")
return True
else:
logging.error(f"Delete folder failed: {folder}")
return False
except Exception as e:
logging.error(f"Delete folder failed: {e}. {traceback.format_exc()}")
return False
def getRelativeFilePaths(self, base_dir) -> list:
"""返回目录下所有文件的相对路径
如果原目录以斜线结尾，则不保留原末级目录
如果原目录不以斜线结尾，则保留末级目录名字
Parameters:
base_dir (str): 原目录, 可以是绝对或者相对路径
Returns:
list: 相对路径列表
eg:
Files: /root/test/1.txt , /root/test/2.txt
IN: /root/test
OUT: ['test/1.txt', 'test/2.txt']
"""
relative_paths = []
for root, _, files in os.walk(base_dir):
for file in files:
file_path = os.path.join(root, file)
relative_path = os.path.relpath(file_path, base_dir)
if base_dir.endswith(os.path.sep):
relative_paths.append(relative_path)
else:
relative_paths.append(os.path.join(os.path.basename(base_dir), relative_path))
return relative_paths
def ls(self, path: str) -> list:
"""获取s3指定路径下的以path为前缀的单层文件/目录信息(包括目录本身), 类似ls path*命令
不能以 '/' 开头.
Parameters:
path (str): 指定路径
Returns:
list: 路径列表, 元素为<Path>
eg:
Files: test/test1/1.txt , /test/test2/2.txt
IN: test/test
OUT: ['test/test1', 'test/test2']
IN: test/test1/
OUT: [test/test1/, 'test/test1/1.txt']
"""
res = self.__resource.meta.client.list_objects_v2(Bucket=self.__bucket_name, Prefix=path, Delimiter='/')
res_content = []
if not res:
logging.error(f"S3 error: {path}")
return res_content
if 'CommonPrefixes' in res:
res_content += ([i['Prefix'] for i in res['CommonPrefixes']])
if 'Contents' in res:
res_content += ([i['Key'] for i in res['Contents']])
if res_content == []:
logging.warning(f"Folder does not exist: {path}")
return res_content
def walkfolder(self, folder) -> list:
"""获取s3指定目录下的以path为前缀的所有文件/目录及其子文件信息, 类似于tree path*命令
务必以/结尾以确保精准匹配
Parameters:
folder (str): 指定目录
Returns:
list: 字典列表, 元素为
{'Key': '<FilePath>',
'LastModified': datetime.datetime(, tzinfo=tzutc()),
'ETag': '"d41d8cd98f00b204e9800998ecf8427e"',
'Size': 0,
'StorageClass': 'Standard'}
"""
files = self.__resource.meta.client.list_objects_v2(Bucket=self.__bucket_name, Prefix=folder)
if 'Contents' in files:
return files['Contents']
else:
logging.warning(f"Folder does not exist: {folder}")
return []
def createSymbolFile(self, path: str, content: str) -> bool:
"""在S3创建一个标志文件
在s3上创建路径为path, 内容为content的文件
"""
try:
self.__resource.meta.client.put_object(Bucket=self.__bucket_name, Key=path, Body=content)
logging.info(f"Created symbol file: {path}. Content: {content}")
return True
except Exception as e:
logging.error(f"S3 error: {e}. {traceback.format_exc()}")
return False
|
