def _is_valid()

in alibabacloud_oss_v2/checkpoint.py [0:0]
56 lines of code
21 McCabe index (conditional complexity)

    def _is_valid(self) -> bool:
        try:
            dcp_info = {}
            with open(self.cp_filepath, 'rb') as f:
                dcp_info = json.loads(f.read())
                if not isinstance(dcp_info, Dict):
                    return False

                js = json.dumps(dcp_info.get("Data", {})).encode()
                h = hashlib.md5()
                h.update(js)
                md5sum = h.hexdigest()

                if (CHECKPOINT_MAGIC != dcp_info.get("Magic") or
                    md5sum != dcp_info.get("MD5")):
                    return False

            cpid = self.cp_info["Data"]
            dcpid = dcp_info["Data"]

            #compare
            if (cpid["ObjectInfo"] != dcpid["ObjectInfo"] or
                cpid["ObjectMeta"] != dcpid["ObjectMeta"] or
                cpid["FilePath"] != dcpid["FilePath"] or
                cpid["PartSize"] != dcpid["PartSize"]):
                return False

	        #download info
            offset = dcpid["DownloadInfo"].get('Offset', 0)
            crc64 = dcpid["DownloadInfo"].get('CRC64', 0)
            if (not isinstance(offset, int) or
                not isinstance(crc64, int)):
                return False

            if offset == 0 and crc64 != 0:
                return False

            roffset = 0
            if len(cpid["ObjectInfo"]["Range"]) > 0:
                range_header = parse_http_range(cpid["ObjectInfo"]["Range"])
                if offset < range_header[0]:
                    return False
                roffset = range_header[0]

            remains = (offset - roffset) % dcpid["PartSize"]
            if remains != 0:
                return False

            #valid data
            if self.verify_data and crc64 != 0:
                try:
                    with open(dcpid["FilePath"], 'rb') as f:
                        chash = Crc64(0)
                        limitn = offset - roffset
                        r = LimitReader(f, limitn)
                        chunk = 32 * 1024
                        for _ in range(0, limitn, chunk):
                            chash.write(r.read(chunk))
                        if chash.sum64() != crc64:
                            return False
                except Exception:
                    return False

            self.doffset = offset
            self.dcrc64 = crc64

            return True
        except Exception:
            #print(f"err = {err}")
            pass

        return False