#!/usr/bin/env python3 """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") # Only ordinary Git blobs; never symlinks or submodules. 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() # A root notice may add conditions; do not silently ignore such a notice. 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"]} # Even an MIT nested notice is excluded by default: no automatic inheritance # across a licensing boundary, and no accidentally stripping local notices. 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) # Spread selection across the tree instead of taking only alphabetic early demos. 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] # Repositories sometimes put licensing exceptions only in README/CREDITS. # Inspect relevant ancestor documents too; conservatively exclude their # entire subtree even when the restriction appears to describe assets. 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: # Attribution-only mentions of GPL are still conservatively excluded. 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()