# coding:utf-8 import sys import io import os import time import re import json import io import oauthlib import requests import datetime from requests_oauthlib import OAuth2Session web_dir = os.getcwd() + "/web" if os.path.exists(web_dir): sys.path.append(web_dir) os.chdir(web_dir) import core.mw as mw DEBUG = False def setDebug(d=False): DEBUG = d class UnauthorizedError(Exception): pass class ObjectNotFoundError(Exception): pass class msodclient: plugin_dir = '' server_dir = '' credential_file = 'credentials.json' user_conf = "user.conf" token_file = 'token.pickle' def __init__(self, plugin_dir, server_dir): self.plugin_dir = plugin_dir self.server_dir = server_dir self.load() def setDebug(self, d=False): DEBUG = d def load(self): credential_path = os.path.join(self.plugin_dir, self.credential_file) credential = json.loads(mw.readFile(credential_path)) # print(credential) self.credential = credential["onedrive-international"] self.authorize_url = '{0}{1}'.format( self.credential['authority'], self.credential['authorize_endpoint']) self.token_url = '{0}{1}'.format( self.credential['authority'], self.credential['token_endpoint']) self.token_path = os.path.join(self.server_dir, self.token_file) self.root_uri = self.credential["api_uri"] + "/me/drive/root" self.backup_path = 'backup' def store_token(self, token): """存储token""" enstr = mw.enDoubleCrypt('msodc', json.dumps(token)) mw.writeFile(self.token_path, enstr) return True def get_store_token(self): rdata = mw.readFile(self.token_path) destr = mw.deDoubleCrypt('msodc', rdata) return json.loads(destr) def clear_token(self): """清除token记录""" try: if os.path.isfile(self.token_path): os.remove(self.token_path) except: if DEBUG: print("清除token失败。") def refresh_token(self, origin_token): """刷新token""" os.environ['OAUTHLIB_RELAX_TOKEN_SCOPE'] = '1' os.environ['OAUTHLIB_IGNORE_SCOPE_CHANGE'] = '1' refresh_token = origin_token["refresh_token"] aad_auth = OAuth2Session( self.credential["client_id"], scope=self.credential["scopes"], redirect_uri=self.credential["redirect_uri"]) new_token = aad_auth.refresh_token( self.token_url, refresh_token=refresh_token, client_id=self.credential["client_id"], client_secret=self.credential["client_secret"]) return new_token def get_token_from_authorized_url(self, authorized_url, expected_state=None): """通过授权编码获取访问token""" # 忽略token scope与已请求的scope不一致 os.environ['OAUTHLIB_RELAX_TOKEN_SCOPE'] = '1' os.environ['OAUTHLIB_IGNORE_SCOPE_CHANGE'] = '1' aad_auth = OAuth2Session(self.credential["client_id"], state=expected_state, scope=self.credential['scopes'], redirect_uri=self.credential['redirect_uri']) token = aad_auth.fetch_token( self.token_url, client_secret=self.credential["client_secret"], authorization_response=authorized_url) return token def get_token(self): token = self.get_store_token() now = time.time() expire_time = token["expires_at"] - 300 if now >= expire_time: new_token = self.refresh_token(token) self.store_token(new_token) return new_token return token def get_sign_in_url(self): """生成签名地址""" # Initialize the OAuth client aad_auth = OAuth2Session(self.credential["client_id"], scope=self.credential['scopes'], redirect_uri=self.credential['redirect_uri']) sign_in_url, state = aad_auth.authorization_url(self.authorize_url, prompt='login') return sign_in_url, state def get_authorized_header(self): token_obj = self.get_token() token = token_obj["access_token"] header = { "Authorization": "Bearer " + token, 'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) ' 'AppleWebKit/537.36 (KHTML, like Gecko) ' 'Chrome/67.0.3396.99 Safari/537.36' } return header def get_user_from_ms(self): """查询用户信息""" try: headers = self.get_authorized_header() user_api_base = self.credential["api_uri"] + "/me" # select_user_info_uri = self.build_uri(base=user_api_base) response = requests.get(user_api_base, headers=headers) if DEBUG: print("Debug get user:") print(response.status_code) print(response.text) if response.status_code == 200: response_data = response.json() user_principal_name = response_data["userPrincipalName"] return user_principal_name except oauthlib.oauth2.rfc6749.errors.InvalidGrantError: self.clear_auth() if DEBUG: print("用户授权已过期。") return None def clear_auth(self): self.clear_token() self.clear_user() def clear_user(self): try: # 清空user path = os.path.join(self.server_dir, self.user_conf) if os.path.isfile(path): os.remove(path) except: if DEBUG: print("清除user失败。") def store_user(self): """更新并存储用户信息""" user = self.get_user_from_ms() if user: path = os.path.join(self.server_dir, self.user_conf) mw.writeFile(path, user) else: raise RuntimeError("无法获取用户信息。") # --------------------- 文件操作功能 ---------------------- # 取目录路径 def get_path(self, path): sep = ":" if path == '/': path = '' if path[-1:] == '/': path = path[:-1] if path[:1] != "/" and path[:1] != sep: path = "/" + path if path == '/': path = '' # if path[:1] != sep: # path = sep + path try: from urllib.parse import quote except: from urllib import quote # path = quote(path) return path.replace('//', '/') def build_uri(self, path="", operate=None, base=None): """构建请求URL API请求URI格式参考: https://graph.microsoft.com/v1.0/me/drive/root:/bt_backup/:content --------------------------------------------- ---------- -------- base path operate 各部分之间用“：”连接。 :param path 子资源路径 :param operate 对文件进行的操作，比如content,children :return 请求url """ if base is None: base = self.root_uri path = self.get_path(path) sep = ":" if operate: if operate[:1] != "/": operate = "/" + operate if path: uri = base + sep + path if operate: uri += sep + operate else: uri = base if operate: uri += operate return uri def get_list(self, path="/"): """获取存储空间中的所有文件对象""" list_uri = self.build_uri(path, operate="/children") if DEBUG: print("List uri:") print(list_uri) data = [] response = requests.get(list_uri, headers=self.get_authorized_header()) status_code = response.status_code if status_code == 200: if DEBUG: print("DEBUG:") print(response.json()) response_data = response.json() drive_items = response_data["value"] for item in drive_items: tmp = {} tmp['name'] = item["name"] tmp['size'] = item["size"] if "folder" in item: # print("{} is folder:".format(item["name"])) # print(item["folder"]) tmp["type"] = None tmp['download'] = "" if "file" in item: tmp["type"] = "File" tmp['download'] = item["@microsoft.graph.downloadUrl"] # print("{} is file:".format(item["name"])) # print(item["file"]) formats = ["%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%dT%H:%M:%SZ"] t = None for time_format in formats: try: t = datetime.datetime.strptime( item["lastModifiedDateTime"], time_format) break except: continue t += datetime.timedelta(hours=8) ts = int( (time.mktime(t.timetuple()) + t.microsecond / 1000000.0)) tmp['time'] = ts data.append(tmp) mlist = {'path': path, 'list': data} return mlist def get_object(self, object_name): """查询对象信息""" try: get_uri = self.build_uri(path=object_name) if DEBUG: print("Get uri:") print(get_uri) response = requests.get(get_uri, headers=self.get_authorized_header()) if response.status_code in [200]: response_data = response.json() if DEBUG: print("Object info:") print(response_data) return response_data if response.status_code == 404: if DEBUG: print("对象不存在。") if DEBUG: print("Get Object debug:") print(response.status_code) print(response.text) except Exception as e: if DEBUG: print("Get object has excepiton:") print(e) return None def is_folder(self, obj): if "folder" in obj: return True return False def delete_object_by_os(self, object_name): """删除对象 :param object_name: :return: True 删除成功 其他 删除失败 """ obj = self.get_object(object_name) if obj is None: if DEBUG: print("对象不存在，删除操作未执行。") return True if self.is_folder(obj): child_count = obj["folder"]["childCount"] if child_count > 0: if DEBUG: print("文件夹不是空文件夹无法删除。") return False headers = self.get_authorized_header() delete_uri = self.build_uri(object_name) response = requests.delete(delete_uri, headers=headers) if response.status_code == 204: if DEBUG: print("对象: {} 已被删除。".format(object_name)) return True return False def delete_object(self, object_name, retries=2): """删除对象 :param object_name: :param retries: 重试次数，默认2次 :return: True 删除成功 其他 删除失败 """ try: return self.delete_object_by_os(object_name) except Exception as e: print("删除文件异常：") print(e) # 重试 if retries > 0: print("重新尝试删除文件{}...".format(object_name)) return self.delete_object( object_name, retries=retries - 1) return False def build_object_name(self, data_type, file_name): """根据数据类型构建对象存储名称 :param data_type: :param file_name: :return: """ import re prefix_dict = { "site": "web", "database": "db", "path": "path", } if not prefix_dict.get(data_type): print("data_type 类型错误!!!") exit(1) file_regx = prefix_dict.get(data_type) + "_(.+)_20\d+_\d+(?:\.|_)" sub_search = re.search(file_regx, file_name) sub_path_name = "" if sub_search: sub_path_name = sub_search.groups()[0] sub_path_name += '/' # 构建OS存储路径 object_name = self.backup_path + '/' + \ data_type + '/' + \ sub_path_name + \ file_name if object_name[:1] == "/": object_name = object_name[1:] return object_name def delete_file(self, file_name, data_type=None): """删除文件 根据传入的文件名称和文件数据类型构建对象名称，再删除 :param file_name: :param data_type: 数据类型 site/database/path :return: True 删除成功 其他 删除失败 """ object_name = self.build_object_name(data_type, file_name) return self.delete_object(object_name) def create_dir_by_step(self, parent_folder, sub_folder): create_uri = self.build_uri(path=parent_folder, operate="/children") if DEBUG: print("Create dir uri:") print(create_uri) post_data = { "name": sub_folder, "folder": {"@odata.type": "microsoft.graph.folder"}, "@microsoft.graph.conflictBehavior": "fail" } headers = self.get_authorized_header() headers.update({"Content-type": "application/json"}) response = requests.post(create_uri, headers=headers, json=post_data) if response.status_code in [201, 409]: if DEBUG: if response.status_code == 409: print("目录：{} 已经存在。".format(sub_folder)) return True else: if DEBUG: print("目录：{} 创建失败：".format(sub_folder)) print(response.status_code) print(response.text) return False def create_dir(self, dir_name): """创建远程目录 # API 请求结构 # POST /me/drive/root/children # or # POST /me/drive/root:/bt_backup/:/children # Content - Type: application / json # { # "name": "New Folder", # "folder": {}, # "@microsoft.graph.conflictBehavior": "rename" # } # Response: status code == 201 新创建/ 409 已存在 # @microsoft.graph.conflictBehavior: fail/rename/replace :param dir_name: 目录名称 :param parent_id: 父目录ID :return: True/False """ dir_name = self.get_path(dir_name.strip()) onedrive_business_reserved = r"[\*<>?:|#%]" if re.search(onedrive_business_reserved, dir_name) \ or dir_name[-1] == "." or dir_name[:1] == "~": if DEBUG: print("文件夹名称包含非法字符。") return False parent_folder = self.get_path(os.path.split(dir_name)[0]) sub_folder = os.path.split(dir_name)[1] # print("create_dir:", dir_name) obj = self.get_object(dir_name) # 判断对象是否存在 if obj is None: if not self.create_dir_by_step(parent_folder, sub_folder): # 兼容OneDrive 商业版文件夹创建 folder_array = dir_name.split("/") parent_folder = self.get_path(folder_array[0]) for i in range(1, len(folder_array)): sub_folder = folder_array[i] if DEBUG: print("Parent folder: {}".format(parent_folder)) print("Sub folder: {}".format(sub_folder)) if self.create_dir_by_step(parent_folder, sub_folder): parent_folder += "/" + folder_array[i] else: return False return True else: if self.is_folder(obj): if DEBUG: print("文件夹已存在。") return True def resumable_upload(self, local_file_name, object_name=None, progress_callback=None, progress_file_name=None, multipart_threshold=1024 * 1024 * 2, part_size=1024 * 1024 * 5, store_dir="/tmp", auto_cancel=True, retries=5, ): """断点续传 :param local_file_name: 本地文件名称 :param object_name: 指定OS中存储的对象名称 :param part_size: 指定分片上传的每个分片的大小。必须是320*1024的整数倍。 :param multipart_threshold: 文件长度大于该值时，则用分片上传。 :param progress_callback: 进度回调函数，默认是把进度信息输出到标准输出。 :param progress_file_name: 进度信息保存文件，进度格式参见[report_progress] :param store_dir: 上传分片存储目录, 默认/tmp。 :param auto_cancel: 当备份失败是否自动取消上传记录 :param retries: 上传重试次数 :return: True上传成功/False or None上传失败 """ try: file_size_separation_value = 4 * 1024 * 1024 if part_size % 320 != 0: if DEBUG: print("Part size 必须是320的整数倍。") return False if object_name is None: temp_file_name = os.path.split(local_file_name)[1] object_name = os.path.join(self.backup_path, temp_file_name) # if progress_file_name: # os.environ[PROGRESS_FILE_NAME] = progress_file_name # progress_callback = report_progress print("|-正在上传到 {}...".format(object_name)) dir_name = os.path.split(object_name)[0] if not self.create_dir(dir_name): if DEBUG: print("目录创建失败！") return False local_file_size = os.path.getsize(local_file_name) # if local_file_size < file_size_separation_value: if False: # 小文件上传 upload_uri = self.build_uri(path=object_name, operate="/content") if DEBUG: print("Upload uri:") print(upload_uri) headers = self.get_authorized_header() # headers.update({ # "Content-Type": "application/octet-stream" # }) # files = {"file": (object_name, open(local_file_name, "rb"))} file_data = open(local_file_name, "rb") response = requests.put(upload_uri, headers=headers, data=file_data) if DEBUG: print("status code:") print(response.status_code) # print(response.text) if response.status_code in [201, 200]: if DEBUG: print("文件上传成功！") return True else: # 大文件上传 # 1. 创建上传session create_session_uri = self.build_uri( path=object_name, operate="createUploadSession") headers = self.get_authorized_header() response = requests.post(create_session_uri, headers=headers) if response.status_code == 200: response_data = response.json() upload_url = response_data["uploadUrl"] expiration_date_time = response_data["expirationDateTime"] if DEBUG: print("上传session已建立。") print("Upload url: {}".format(upload_url)) print("Expiration datetime: {}".format( expiration_date_time)) # 2. 分片上传文件 requests.adapters.DEFAULT_RETRIES = 1 session = requests.session() session.keep_alive = False # 开始分片上传 import math parts = int(math.ceil(local_file_size / part_size)) for i in range(parts): if DEBUG: if i == parts - 1: num = "最后" else: num = "第{}".format(i + 1) print("正在上传{}部分...".format(num)) upload_range_start = i * part_size upload_range_end = min(upload_range_start + part_size, local_file_size) content_length = upload_range_end - upload_range_start headers = { 'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) ' 'AppleWebKit/537.36 (KHTML, like Gecko) ' 'Chrome/67.0.3396.99 Safari/537.36' } # 开发记录 # Content-Range和标准的http请求头中的Range作用有所不同 # Content-Range是OneDrive自定义的分片上传标识，格式也不一样 headers.update({ "Content-Length": repr(content_length), "Content-Range": "bytes {}-{}/{}".format( upload_range_start, upload_range_end - 1, local_file_size), "Content-Type": "application/octet-stream" }) if DEBUG: print("Headers:") print(headers) '''# TODO 优化read的读取占用内存''' f = io.open(local_file_name, "rb") f.seek(upload_range_start) upload_data = f.read(content_length) sub_response = session.put(upload_url, headers=headers, data=upload_data) expected_status_code = [200, 201, 202] if sub_response.status_code in expected_status_code: if DEBUG: print("Response status code: {}, " "bytes {}-{} 已上传成功。".format( sub_response.status_code, upload_range_start, upload_range_end - 1) ) print(sub_response.text) if sub_response.status_code in [200, 201]: if DEBUG: print("文件 {} 上传成功。".format(object_name)) return True else: print(sub_response.status_code) print(sub_response.text) _error_msg = "Bytes {}-{} 分片上传失败。".format( upload_range_start, upload_range_end ) if self.error_msg: self.error_msg += r"\n" self.error_msg += _error_msg raise RuntimeError(_error_msg) time.sleep(0.5) else: raise RuntimeError("session创建失败。") except UnauthorizedError as e: _error_msg = str(e) if self.error_msg: self.error_msg += r"\n" self.error_msg += _error_msg print(_error_msg) return False except Exception as e: print("文件上传出现错误：") print(e) if self.error_msg: self.error_msg += r"\n" self.error_msg += "文件{}上传出现错误：{}".format(object_name, str(e)) try: if upload_url: if DEBUG: print("正在清理上传session.") session.delete(upload_url) except: pass finally: try: f.close() except: pass try: session.close() except: pass # 重试断点续传 if retries > 0: print("重试上传文件....") return self.resumable_upload( local_file_name, object_name=object_name, store_dir=store_dir, part_size=part_size, multipart_threshold=multipart_threshold, progress_callback=progress_callback, progress_file_name=progress_file_name, retries=retries - 1, ) else: if self.error_msg: self.error_msg += r"\n" self.error_msg += "文件{}上传失败。".format(object_name) return False def upload_abs_file(self, file_name, remote_dir, *args, **kwargs): """按照数据类型上传文件 :param file_name: 上传文件名称 :param data_type: 数据类型 site/database/path :return: True/False """ try: import re # 根据数据类型提取子分类名称 # 比如data_type=database，子分类名称是数据库的名称。 # 提取方式是从file_name中利用正则规则去提取。 self.error_msg = "" file_name = os.path.abspath(file_name) temp_name = os.path.split(file_name)[1] object_name = 'backup/' + temp_name print(file_name) print(object_name) return self.resumable_upload(file_name, object_name=object_name, *args, **kwargs) except Exception as e: if self.error_msg: self.error_msg += r"\n" self.error_msg += "文件上传出现错误：{}".format(str(e)) return False def upload_file(self, file_name, data_type, *args, **kwargs): """按照数据类型上传文件 :param file_name: 上传文件名称 :param data_type: 数据类型 site/database/path :return: True/False """ try: import re # 根据数据类型提取子分类名称 # 比如data_type=database，子分类名称是数据库的名称。 # 提取方式是从file_name中利用正则规则去提取。 self.error_msg = "" if not file_name or not data_type: _error_msg = "文件参数错误。" print(_error_msg) self.error_msg = _error_msg return False file_name = os.path.abspath(file_name) temp_name = os.path.split(file_name)[1] object_name = self.build_object_name(data_type, temp_name) # dir_name = os.path.dirname(object_name) # self.create_dir(dir_name) if DEBUG: print(file_name) print(object_name) print(dir_name) return self.resumable_upload(file_name, object_name=object_name, *args, **kwargs) except Exception as e: if self.error_msg: self.error_msg += r"\n" self.error_msg += "文件上传出现错误：{}".format(str(e)) return False