File size: 10,729 Bytes
90884df | 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 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 | #!/usr/bin/env python3
from __future__ import annotations
import hashlib, json, os, shutil, stat, wave
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
ROOT=Path("/home/ubuntu/minimax-laion-corpus")
LABEL="interim_tranche_678_due_systemic_bot_auth"
FINAL=ROOT/"versions"/LABEL
STAGING=ROOT/"versions"/("."+LABEL+".staging")
RESULTS=ROOT/"manifests/ingest-results.jsonl"
CANDIDATES=ROOT/"manifests/candidates.jsonl"
KNOWN=ROOT/"manifests/known_user22_hashes.json"
STOPPED=ROOT/"manifests/tranche.stopped.json"
CANONICAL_ROOT=(ROOT/"canonical").resolve()
SEED="music3lab.laion.interim678.split.v1"
REVISION="6e7bf3758a77301e46a715af894fefd79bb1da53"
def canon(obj):
return json.dumps(obj,sort_keys=True,ensure_ascii=False,separators=(",",":")).encode()
def sha_bytes(data):
return hashlib.sha256(data).hexdigest()
def sha_file(path,chunk=8<<20):
h=hashlib.sha256()
with path.open("rb") as f:
while b:=f.read(chunk):h.update(b)
return h.hexdigest()
def pcm_info(path):
h=hashlib.sha256()
with wave.open(str(path),"rb") as w:
info={"frame_count":w.getnframes(),"sample_rate":w.getframerate(),"channels":w.getnchannels(),
"sample_width_bytes":w.getsampwidth(),"compression":w.getcomptype()}
if (info["sample_rate"],info["channels"],info["sample_width_bytes"],info["compression"])!=(44100,2,2,"NONE"):
raise RuntimeError(f"noncanonical WAV {path}: {info}")
while data:=w.readframes(65536):h.update(data)
info["pcm_sha256"]=h.hexdigest()
return info
def copy_verify(item):
row,record=item
src=Path(row["canonical_path"]).resolve()
if src.parent!=CANONICAL_ROOT:raise RuntimeError(f"source outside canonical root: {src}")
if not src.is_file():raise FileNotFoundError(src)
if sha_file(src)!=record["canonical_sha256"]:raise RuntimeError(f"source WAV hash mismatch: {src}")
srcinfo=pcm_info(src)
for key in ("frame_count","sample_rate","channels","pcm_sha256"):
if srcinfo[key]!=record[key]:raise RuntimeError(f"source {key} mismatch: {src}")
dst=STAGING/record["relative_path"]
dst.parent.mkdir(parents=True,exist_ok=True)
shutil.copyfile(src,dst)
if dst.stat().st_size!=record["canonical_bytes"]:raise RuntimeError(f"copy size mismatch: {dst}")
if sha_file(dst)!=record["canonical_sha256"]:raise RuntimeError(f"copy WAV hash mismatch: {dst}")
dstinfo=pcm_info(dst)
for key in ("frame_count","sample_rate","channels","pcm_sha256"):
if dstinfo[key]!=record[key]:raise RuntimeError(f"copy {key} mismatch: {dst}")
os.chmod(dst,0o444)
return record
def main():
if FINAL.exists():raise FileExistsError(FINAL)
if STAGING.exists():raise FileExistsError(STAGING)
rows=[json.loads(x) for x in RESULTS.read_text(encoding="utf-8").splitlines() if x.strip()]
accepted=sorted((x for x in rows if x.get("accepted")),key=lambda x:x["candidate_index"])
if len(accepted)!=678:raise RuntimeError(f"expected 678 accepted, got {len(accepted)}")
if len(rows)!=994:raise RuntimeError(f"expected 994 attempts, got {len(rows)}")
if [x["candidate_index"] for x in rows]!=list(range(994)):raise RuntimeError("attempt order is not frozen candidate order 0..993")
if len({x["song_id"] for x in accepted})!=678:raise RuntimeError("duplicate source ID")
wav_hashes=[x["canonical_sha256"] for x in accepted]
pcm_hashes=[x["audio_stats"]["pcm_sha256"] for x in accepted]
if len(set(wav_hashes))!=678 or len(set(pcm_hashes))!=678:raise RuntimeError("within-tranche exact duplicate")
known=json.loads(KNOWN.read_text(encoding="utf-8"))["files"]
if set(wav_hashes)&{x["canonical_sha256"] for x in known}:raise RuntimeError("user22 WAV overlap")
if set(pcm_hashes)&{x["audio_stats"]["pcm_sha256"] for x in known}:raise RuntimeError("user22 PCM overlap")
candidate_sha=sha_file(CANDIDATES);results_sha=sha_file(RESULTS);stopped_sha=sha_file(STOPPED)
if candidate_sha!="533d8623268b18f87de2774ecf45656520f37edd8d5bb30c69f9bb6401231a07":raise RuntimeError("candidate manifest changed")
stop=json.loads(STOPPED.read_text())
if stop.get("accepted_total")!=678 or stop.get("reason")!="systemic_rate_limit_or_auth_failure":raise RuntimeError("stop artifact mismatch")
records=[]
for row in accepted:
stats=row["audio_stats"];source_identity={
"candidate_index":row["candidate_index"],"candidate_rank_sha256":row["rank_sha256"],
"source_id":row["song_id"],"canonical_sha256":row["canonical_sha256"],
"pcm_sha256":stats["pcm_sha256"]}
split_rank=sha_bytes((SEED+"\0"+str(source_identity["candidate_index"])+"\0"+
source_identity["candidate_rank_sha256"]+"\0"+source_identity["source_id"]+"\0"+
source_identity["canonical_sha256"]+"\0"+source_identity["pcm_sha256"]).encode())
records.append({
"schema_version":"music3lab.laion.interim-record.v1",
"candidate_index":row["candidate_index"],"candidate_rank_sha256":row["rank_sha256"],
"source_dataset":"laion/LAION-DISCO-12M","source_dataset_revision":REVISION,
"source_id":row["song_id"],"source_url":row["url"],
"source_record_sha256":sha_bytes(canon(row)),
"canonical_sha256":row["canonical_sha256"],"pcm_sha256":stats["pcm_sha256"],
"canonical_bytes":row["canonical_bytes"],"frame_count":stats["frame_count"],
"sample_rate":stats["sample_rate"],"channels":stats["channels"],
"relative_path":"files/"+row["canonical_sha256"]+".wav","split_rank_sha256":split_rank,
})
by_split_rank=sorted(records,key=lambda x:(x["split_rank_sha256"],x["source_id"]))
for i,record in enumerate(by_split_rank):
record["split"]="train" if i<542 else ("validation" if i<610 else "heldout")
counts={x:sum(r["split"]==x for r in records) for x in ("train","validation","heldout")}
if counts!={"train":542,"validation":68,"heldout":68}:raise RuntimeError(counts)
for split in counts:
ids=[r["source_id"] for r in records if r["split"]==split]
if len(ids)!=len(set(ids)):raise RuntimeError(f"duplicate within {split}")
sets={s:{r["source_id"] for r in records if r["split"]==s} for s in counts}
if sets["train"]&sets["validation"] or sets["train"]&sets["heldout"] or sets["validation"]&sets["heldout"]:
raise RuntimeError("source leakage across splits")
STAGING.mkdir(mode=0o755)
(STAGING/"files").mkdir(mode=0o755)
row_by_id={x["song_id"]:x for x in accepted}
with ThreadPoolExecutor(max_workers=4) as ex:
copied=list(ex.map(copy_verify,((row_by_id[r["source_id"]],r) for r in records)))
if len(copied)!=678:raise RuntimeError("copy count mismatch")
file_paths=list((STAGING/"files").glob("*.wav"))
if len(file_paths)!=678:raise RuntimeError(f"copy file count {len(file_paths)}")
manifest_rows=sorted(records,key=lambda x:x["candidate_index"])
manifest_payload=b"".join(canon(x)+b"\n" for x in manifest_rows)
semantic_payload={"schema_version":"music3lab.laion.interim-semantic.v1","label":LABEL,
"complete_corpus":False,"incomplete_reason":"systemic_bot_auth","original_target":2000,
"source_dataset":"laion/LAION-DISCO-12M","source_dataset_revision":REVISION,
"candidate_manifest_sha256":candidate_sha,"stop_artifact_sha256":stopped_sha,
"split_seed":SEED,"split_rank":"sha256(seed NUL candidate_index NUL candidate_rank NUL source_id NUL canonical_sha256 NUL pcm_sha256)",
"split_assignment":"ascending split_rank: first 542 train, next 68 validation, final 68 heldout",
"records":[{k:r[k] for k in ("candidate_index","candidate_rank_sha256","source_id","canonical_sha256","pcm_sha256",
"canonical_bytes","frame_count","sample_rate","channels","relative_path","split_rank_sha256","split")} for r in manifest_rows]}
semantic_digest=sha_bytes(canon(semantic_payload))
files_payload="".join(f"{r['canonical_sha256']} {r['canonical_bytes']} {r['relative_path']}\n" for r in sorted(records,key=lambda x:x["relative_path"])).encode()
files_digest=sha_bytes(files_payload)
splits_obj={"schema_version":"music3lab.laion.interim-splits.v1","seed":SEED,
"algorithm":semantic_payload["split_rank"],"assignment":semantic_payload["split_assignment"],
"counts":counts,"splits":{s:[{"source_id":r["source_id"],"candidate_index":r["candidate_index"],
"relative_path":r["relative_path"]} for r in sorted((x for x in records if x["split"]==s),key=lambda x:x["split_rank_sha256"])] for s in counts}}
splits_payload=canon(splits_obj)+b"\n"
manifest_path=STAGING/"manifest.jsonl";splits_path=STAGING/"splits.json";summary_path=STAGING/"summary.json"
manifest_path.write_bytes(manifest_payload);splits_path.write_bytes(splits_payload)
summary={"schema_version":"music3lab.laion.interim-summary.v1","label":LABEL,"status":"INTERIM_NOT_COMPLETE",
"incomplete_reason":"systemic_bot_auth","original_target":2000,"accepted_count":678,"attempted_count":994,
"skip_unavailable_count":299,"systemic_failure_count":17,"next_candidate_index":994,
"source_dataset":"laion/LAION-DISCO-12M","source_dataset_revision":REVISION,
"candidate_manifest_sha256":candidate_sha,"ingest_results_sha256":results_sha,"stop_artifact_sha256":stopped_sha,
"manifest_sha256":sha_file(manifest_path),"splits_sha256":sha_file(splits_path),
"semantic_digest_sha256":semantic_digest,"files_aggregate_digest_sha256":files_digest,
"file_count":678,"canonical_bytes":sum(r["canonical_bytes"] for r in records),
"split_seed":SEED,"split_counts":counts,"within_wav_duplicates":0,"within_pcm_duplicates":0,
"user22_wav_overlap":0,"user22_pcm_overlap":0}
summary_path.write_bytes(canon(summary)+b"\n")
sums=[]
for r in sorted(records,key=lambda x:x["relative_path"]):sums.append(f"{r['canonical_sha256']} {r['relative_path']}\n")
for name in ("manifest.jsonl","splits.json","summary.json"):sums.append(f"{sha_file(STAGING/name)} {name}\n")
sums_path=STAGING/"SHA256SUMS";sums_path.write_text("".join(sums),encoding="utf-8")
for p in (manifest_path,splits_path,summary_path,sums_path):os.chmod(p,0o444)
os.chmod(STAGING/"files",0o555)
os.chmod(STAGING,0o555)
os.replace(STAGING,FINAL)
final_summary=json.loads((FINAL/"summary.json").read_text())
if final_summary!=summary:raise RuntimeError("published summary mismatch")
for r in manifest_rows:
p=FINAL/r["relative_path"]
if not p.is_file() or sha_file(p)!=r["canonical_sha256"]:raise RuntimeError(f"postpublish mismatch {p}")
if stat.S_IMODE(p.stat().st_mode)!=0o444:raise RuntimeError(f"mode mismatch {p}")
print(json.dumps({"status":"PASS","root":str(FINAL),"manifest":str(FINAL/"manifest.jsonl"),
"manifest_sha256":summary["manifest_sha256"],"summary_sha256":sha_file(FINAL/"summary.json"),
"splits_sha256":summary["splits_sha256"],"semantic_digest_sha256":semantic_digest,
"files_aggregate_digest_sha256":files_digest,"sha256sums_sha256":sha_file(FINAL/"SHA256SUMS"),
"accepted_count":678,"canonical_bytes":summary["canonical_bytes"],"split_counts":counts},sort_keys=True))
if __name__=="__main__":main()
|