in alibabacloud_oss_v2/checkpoint.py [0:0]
def _is_valid(self) -> bool:
try:
ucp_info = {}
with open(self.cp_filepath, 'rb') as f:
ucp_info = json.loads(f.read())
if not isinstance(ucp_info, Dict):
return False
js = json.dumps(ucp_info.get("Data", {})).encode()
h = hashlib.md5()
h.update(js)
md5sum = h.hexdigest()
if (CHECKPOINT_MAGIC != ucp_info.get("Magic") or
md5sum != ucp_info.get("MD5")):
return False
cpid = self.cp_info["Data"]
ucpid = ucp_info["Data"]
#compare
if (cpid["ObjectInfo"] != ucpid["ObjectInfo"] or
cpid["FileMeta"] != ucpid["FileMeta"] or
cpid["FilePath"] != ucpid["FilePath"] or
cpid["PartSize"] != ucpid["PartSize"]):
return False
#upload info
uploadid = ucpid["UploadInfo"]["UploadId"]
if not isinstance(uploadid, str) or uploadid == '':
return False
self.upload_id = uploadid
return True
except Exception:
#print(f"err = {err}")
pass
return False