32 lines
1.4 KiB
Python
32 lines
1.4 KiB
Python
|
"""Reacquire the locked licensed corpus. Updated bytes are never accepted."""
|
||
|
|
import argparse
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
import urllib.request
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
parser = argparse.ArgumentParser()
|
||
|
|
parser.add_argument("--root", type=Path, default=Path("fixtures/media"))
|
||
|
|
args = parser.parse_args()
|
||
|
|
root = args.root.resolve()
|
||
|
|
fixture_path = root / "fixture.json"
|
||
|
|
lock = json.loads((root / "labels.lock.json").read_text())
|
||
|
|
assert hashlib.sha256(fixture_path.read_bytes()).hexdigest() == lock["fixture_sha256"]
|
||
|
|
for asset in json.loads(fixture_path.read_text(encoding="utf-8"))["assets"]:
|
||
|
|
destination = (root / asset["path"]).resolve()
|
||
|
|
assert destination.is_relative_to(root)
|
||
|
|
if destination.is_file() and hashlib.sha256(destination.read_bytes()).hexdigest() == asset["sha256"]:
|
||
|
|
continue
|
||
|
|
request = urllib.request.Request(asset["source_url"], headers={"User-Agent": "SearchiMediaEvaluation/1.0"})
|
||
|
|
with urllib.request.urlopen(request, timeout=60) as response:
|
||
|
|
data = response.read(asset["bytes"] + 1)
|
||
|
|
assert len(data) == asset["bytes"] and hashlib.sha256(data).hexdigest() == asset["sha256"], f"Publisher changed {asset['id']}; keep frozen corpus, investigate a new version"
|
||
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
destination.write_bytes(data)
|
||
|
|
print("Verified", asset["id"])
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|