File size: 3,466 Bytes
6d63e5b | 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 | import json
import os
import time
from dataclasses import asdict, dataclass
from typing import Optional, Set
from .common import PathOrStr
@dataclass
class Meta:
"""
Any resource that is downloaded to - or extracted in - the cache directory will
have a meta JSON file written next to it, which corresponds to an instance
of this class.
In older versions of AllenNLP, this meta document just had two fields: 'url' and
'etag'. The 'url' field is now the more general 'resource' field, but these old
meta files are still compatible when a `Meta` is instantiated with the `.from_path()`
class method.
"""
resource: str
"""
URL or normalized path to the resource.
"""
cached_path: str
"""
Path to the corresponding cached version of the resource.
"""
creation_time: float
"""
The unix timestamp of when the corresponding resource was cached or extracted.
"""
size: int = 0
"""
The size of the corresponding resource, in bytes.
"""
etag: Optional[str] = None
"""
Optional ETag associated with the current cached version of the resource.
"""
extraction_dir: bool = False
"""
Does this meta corresponded to an extraction directory?
"""
@classmethod
def new(
cls,
resource: PathOrStr,
cached_path: PathOrStr,
*,
etag: Optional[str] = None,
extraction_dir: bool = False,
) -> "Meta":
return cls( # type: ignore
resource=str(resource),
cached_path=str(cached_path),
creation_time=time.time(),
size=cls.get_resource_size(cached_path),
etag=etag,
extraction_dir=extraction_dir,
)
def to_file(self) -> None:
with open(self.cached_path + ".json", "w") as meta_file:
json.dump(asdict(self), meta_file)
@classmethod
def from_path(cls, path: PathOrStr) -> "Meta":
path = str(path)
with open(path) as meta_file:
data = json.load(meta_file)
# For backwards compat:
if "resource" not in data:
data["resource"] = data.pop("url")
if "creation_time" not in data:
data["creation_time"] = os.path.getmtime(path[:-5])
if "extraction_dir" not in data and path.endswith("-extracted.json"):
data["extraction_dir"] = True
if "cached_path" not in data:
data["cached_path"] = path[:-5]
if "size" not in data:
data["size"] = cls.get_resource_size(data["cached_path"])
return cls(**data) # type: ignore
@staticmethod
def get_resource_size(path: PathOrStr) -> int:
"""
Get the size of a file or directory.
"""
if os.path.isfile(path):
return os.path.getsize(path)
inodes: Set[int] = set()
total_size = 0
for dirpath, _, filenames in os.walk(str(path)):
for f in filenames:
fp = os.path.join(dirpath, f)
# skip if it is symbolic link or the same as a file we've already accounted
# for (this could happen with hard links).
inode = os.stat(fp).st_ino
if not os.path.islink(fp) and inode not in inodes:
inodes.add(inode)
total_size += os.path.getsize(fp)
return total_size
|