in sapp/sharded_files.py [0:0]
def __init__(self, filepattern) -> None:
# pyre-fixme[4]: Attribute must be annotated.
self.directory, root = os.path.split(filepattern)
m = re.match(r"([^@]+)@([^.@]+)(\.[^.@]*)?$", root)
if not m:
raise ValueError("Not a sharded file: {}".format(filepattern))
# pyre-fixme[58]: `>=` is not supported for operand types `Optional[int]`
# and `int`.
self.extension: str = m.group(3) if m.lastindex >= 3 else ""
# pyre-fixme[58]: `>=` is not supported for operand types `Optional[int]`
# and `int`.
self.stem: str = m.group(1) if m.lastindex >= 1 else ""
# pyre-fixme[58]: `>=` is not supported for operand types `Optional[int]`
# and `int`.
shards = m.group(2) if m.lastindex >= 2 else ""
if shards == "*":
self.shard_index: int = -1
self.shard_total: int = -1
else:
try:
self.shard_total = int(shards)
self.shard_index = -1
except ValueError:
# Look for ?????-of-????? pattern.
m = re.match(r"(\d{5})-of-(\d{5})$", shards)
if not m:
raise ValueError("Invalid shard specification: {}".format(shards))
self.shard_index = int(m.group(1))
self.shard_total = int(m.group(2))
if self.directory == "":
self.directory = "."
# Sanity check shard total.
if self.shard_total == 0:
raise ValueError("Invalid shard total 0")