| |
| """Collect text-only .gd references from explicitly allowed GitHub repositories. |
| |
| No archive is downloaded, and no image, audio, font, scene, or binary is fetched. |
| Remote descriptions/code are data only; this script never executes them. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| from concurrent.futures import ThreadPoolExecutor |
| from datetime import datetime, timezone |
| import json |
| from pathlib import Path, PurePosixPath |
| import re |
| import sys |
| import time |
| import urllib.error |
| import urllib.parse |
| import urllib.request |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) |
| from vibegame.data import Deduper, FORBIDDEN_LICENSE, has_conflicting_header, identify_license, sha256, write_jsonl |
|
|
| DEFAULT_REPOS = ["godotengine/godot-demo-projects@master", "nathanhoad/godot_dialogue_manager@main"] |
| NOTICE_NAME = re.compile(r"license|licence|copying|copyright|notice", re.I) |
| EXCLUDED_PARTS = {"third_party", "thirdparty", "vendor", "node_modules", ".git", ".godot"} |
|
|
|
|
| def fetch(url: str, retries: int = 3) -> bytes: |
| request = urllib.request.Request(url, headers={"User-Agent": "VibeGame-8B-license-audited-corpus/0.1"}) |
| for attempt in range(retries): |
| try: |
| with urllib.request.urlopen(request, timeout=40) as response: |
| return response.read(2_000_001) |
| except urllib.error.HTTPError as exc: |
| if exc.code not in {429, 500, 502, 503, 504} or attempt + 1 == retries: |
| raise |
| except (TimeoutError, urllib.error.URLError): |
| if attempt + 1 == retries: |
| raise |
| time.sleep(2 ** attempt) |
| raise RuntimeError("request retries exhausted") |
|
|
|
|
| def api(path: str) -> dict: |
| return json.loads(fetch("https://api.github.com/" + path)) |
|
|
|
|
| def raw(repo: str, commit: str, path: str) -> str: |
| return fetch(f"https://raw.githubusercontent.com/{repo}/{commit}/{urllib.parse.quote(path)}").decode("utf-8") |
|
|
|
|
| def collect_repo(spec: str, output: Path, max_files: int, deduper: Deduper) -> tuple[list, dict]: |
| repo, separator, ref = spec.partition("@") |
| if not re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repo): |
| raise ValueError("repository must be OWNER/REPO[@REF]") |
| metadata = api(f"repos/{repo}") |
| ref = ref if separator else metadata["default_branch"] |
| commit_info = api(f"repos/{repo}/commits/{urllib.parse.quote(ref, safe='')}") |
| commit = commit_info["sha"] |
| tree = api(f"repos/{repo}/git/trees/{commit}?recursive=1") |
| if tree.get("truncated"): |
| raise ValueError(f"{repo}: refusing an incomplete repository tree") |
| |
| blobs = [entry for entry in tree["tree"] if entry["type"] == "blob" and entry.get("mode") == "100644"] |
| notices = [entry for entry in blobs if NOTICE_NAME.search(PurePosixPath(entry["path"]).name)] |
| root_notices = [entry for entry in notices if "/" not in entry["path"]] |
| license_notices = [entry for entry in root_notices if re.match(r"licen[sc]e(?:\.|$)", entry["path"], re.I)] |
| if not license_notices: |
| raise ValueError(f"{repo}: no unambiguous root LICENSE file") |
| texts = {} |
| for entry in root_notices: |
| texts[entry["path"]] = raw(repo, commit, entry["path"]) |
| licenses = {identify_license(texts[entry["path"]]) for entry in license_notices} |
| if len(licenses) != 1 or None in licenses: |
| raise ValueError(f"{repo}: mixed, unknown, or disallowed root license; manual review required") |
| license_id = licenses.pop() |
| |
| for entry in root_notices: |
| if entry not in license_notices and has_conflicting_header("\n".join("# " + line for line in texts[entry["path"]].splitlines())): |
| raise ValueError(f"{repo}: conflicting root notice {entry['path']}") |
| nested_notice_dirs = {str(PurePosixPath(entry["path"]).parent) for entry in notices if "/" in entry["path"]} |
| |
| |
| rejected = [] |
| candidates = [] |
| for entry in sorted(blobs, key=lambda value: value["path"]): |
| path = entry["path"] |
| if not path.endswith(".gd"): |
| continue |
| if entry.get("size", 0) > 80_000: |
| rejected.append({"path": path, "reason": "over_80kb"}); continue |
| if set(PurePosixPath(path).parts) & EXCLUDED_PARTS: |
| rejected.append({"path": path, "reason": "vendor_directory"}); continue |
| if any(path.startswith(directory + "/") for directory in nested_notice_dirs): |
| rejected.append({"path": path, "reason": "nested_notice_boundary"}); continue |
| candidates.append(entry) |
| |
| if max_files > 0 and len(candidates) > max_files: |
| indices = {round(i * (len(candidates) - 1) / (max_files - 1)) for i in range(max_files)} if max_files > 1 else {0} |
| candidates = [entry for i, entry in enumerate(candidates) if i in indices] |
| |
| |
| |
| documentation = [entry for entry in blobs |
| if re.match(r"(?:readme|credits)(?:\.|$)", PurePosixPath(entry["path"]).name, re.I) |
| and any(str(PurePosixPath(entry["path"]).parent) == "." |
| or candidate["path"].startswith(str(PurePosixPath(entry["path"]).parent) + "/") |
| for candidate in candidates)] |
| documentation_exclusions = [] |
| with ThreadPoolExecutor(max_workers=6) as pool: |
| docs = list(pool.map(lambda entry: (entry["path"], raw(repo, commit, entry["path"])), documentation)) |
| for doc_path, doc_text in docs: |
| |
| if FORBIDDEN_LICENSE.search(doc_text): |
| documentation_exclusions.append(str(PurePosixPath(doc_path).parent)) |
| filtered = [] |
| for entry in candidates: |
| if any(directory == "." or entry["path"].startswith(directory + "/") for directory in documentation_exclusions): |
| rejected.append({"path": entry["path"], "reason": "README_or_CREDITS_license_boundary"}) |
| else: |
| filtered.append(entry) |
| candidates = filtered |
| slug = repo.replace("/", "__") |
| license_paths = [] |
| for path, content in texts.items(): |
| target = output / "licenses" / slug / commit / path |
| target.parent.mkdir(parents=True, exist_ok=True) |
| target.write_text(content, encoding="utf-8") |
| license_paths.append(str(target.relative_to(output))) |
| records = [] |
| def download(entry): |
| try: |
| return entry, raw(repo, commit, entry["path"]), None |
| except Exception as exc: |
| return entry, None, type(exc).__name__ + ": " + str(exc) |
| with ThreadPoolExecutor(max_workers=6) as pool: |
| for entry, code, error in pool.map(download, candidates): |
| path = entry["path"] |
| reason = None |
| if error: |
| reason = "download_failed: " + error |
| elif not code.strip() or "\x00" in code: |
| reason = "empty_or_nontext" |
| elif has_conflicting_header(code, expected_license=license_id): |
| reason = "conflicting_file_header" |
| elif deduper.match(code): |
| reason = "exact_or_near_duplicate" |
| if reason: |
| rejected.append({"path": path, "reason": reason}); continue |
| identifier = f"{repo}@{commit}:{path}" |
| deduper.add(identifier, code) |
| records.append({"id": identifier, "repository": repo, "commit": commit, |
| "path": path, "source_url": f"https://github.com/{repo}/blob/{commit}/{path}", |
| "license": license_id, "license_files": license_paths, |
| "sha256": sha256(code), "git_blob_sha": entry["sha"], |
| "language": "GDScript", "code": code, |
| "use": "reference_only_not_supervised_training"}) |
| manifest = {"repository": repo, "requested_ref": ref, "commit": commit, |
| "commit_date": commit_info["commit"]["committer"]["date"], |
| "repository_license_api_hint": (metadata.get("license") or {}).get("spdx_id"), |
| "verified_license": license_id, "license_files": license_paths, |
| "gd_files_in_tree": sum(entry["path"].endswith(".gd") for entry in blobs), |
| "candidate_downloads": len(candidates), "accepted": len(records), "rejected": rejected, |
| "readme_credits_checked": [path for path, _ in docs], |
| "documentation_excluded_directories": documentation_exclusions, |
| "filter": "only regular .gd text; entire nested-notice subtrees excluded; root license signatures + file comment checks; no assets"} |
| return records, manifest |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--repo", action="append", help="OWNER/REPO[@REF]; repeatable; commit SHA recommended") |
| parser.add_argument("--output", type=Path, default=Path("data")) |
| parser.add_argument("--max-files-per-repo", type=int, default=180, help="0 means all eligible files") |
| args = parser.parse_args() |
| args.output.mkdir(parents=True, exist_ok=True) |
| deduper = Deduper(threshold=0.85) |
| all_records, manifests = [], [] |
| for spec in args.repo or DEFAULT_REPOS: |
| records, manifest = collect_repo(spec, args.output, args.max_files_per_repo, deduper) |
| all_records.extend(records); manifests.append(manifest) |
| print(f"{manifest['repository']}: {len(records)} accepted .gd files at {manifest['commit']}", flush=True) |
| write_jsonl(args.output / "reference_gd.jsonl", all_records) |
| report = {"schema_version": 1, "collected_at_utc": datetime.now(timezone.utc).isoformat(), |
| "record_count": len(all_records), "repositories": manifests, |
| "stack_v2": {"gdscript_available": True, "included": False, |
| "reason": "gated terms, original-license obligations, and Software Heritage bulk-download agreement require separate resolution", |
| "source": "https://huggingface.co/datasets/bigcode/the-stack-v2"}, |
| "limitations": ["Conservative automated filtering is not a legal guarantee.", |
| "Code-only reference corpus is not instruction-tuning data.", |
| "A root MIT license does not relicense assets; no assets were downloaded."]} |
| (args.output / "reference_manifest.json").write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
