| """Dataset hygiene shared by collection, synthetic generation, and training. |
| |
| Only the standard library is required. Reference .gd files are deliberately a |
| different schema from supervised whole-project examples. |
| """ |
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| import re |
| from pathlib import Path, PurePosixPath |
| from typing import Iterable |
|
|
| ALLOWED_LICENSES = frozenset({"MIT", "Apache-2.0", "CC0-1.0"}) |
| FORBIDDEN_LICENSE = re.compile( |
| r"\b(?:A?GPL|LGPL|AGPL|GFDL|MPL|EUPL)(?:[- ]?v?\d)?\b|" |
| r"GNU (?:GENERAL|LESSER|AFFERO)|CC[- ]BY|non.?commercial|" |
| r"all rights reserved|proprietary|not for redistribution", re.I |
| ) |
|
|
|
|
| def sha256(text: str) -> str: |
| return hashlib.sha256(text.encode("utf-8")).hexdigest() |
|
|
|
|
| def canonical_output(output: str) -> tuple[str, dict[str, str]]: |
| """Reject malformed or unsafe file bundles before invoking an engine.""" |
| if not isinstance(output, str): |
| raise ValueError("output must be a JSON string") |
| parsed = json.loads(output) |
| if not isinstance(parsed, dict) or set(parsed) != {"files"}: |
| raise ValueError("output must contain exactly one 'files' mapping") |
| files = parsed["files"] |
| if not isinstance(files, dict) or not files or len(files) > 32: |
| raise ValueError("files must be a nonempty mapping of at most 32 files") |
| if "project.godot" not in files or not any(p.endswith(".tscn") for p in files): |
| raise ValueError("project.godot and a .tscn scene are required") |
| if not any(p.endswith(".gd") for p in files): |
| raise ValueError("at least one .gd script is required") |
| for path, content in files.items(): |
| parts = PurePosixPath(path) |
| if (not isinstance(content, str) or not path or "\\" in path or ":" in path |
| or parts.is_absolute() or ".." in parts.parts or path != str(parts) |
| or any(p.startswith(".") for p in parts.parts)): |
| raise ValueError(f"unsafe filename or non-text content: {path!r}") |
| if parts.suffix not in {".godot", ".gd", ".tscn", ".tres"}: |
| raise ValueError(f"external assets are not allowed: {path}") |
| if len(content.encode("utf-8")) > 200_000: |
| raise ValueError("file exceeds 200 KB") |
| if sum(len(x.encode("utf-8")) for x in files.values()) > 500_000: |
| raise ValueError("project exceeds 500 KB") |
| return json.dumps({"files": files}, ensure_ascii=False, sort_keys=True), files |
|
|
|
|
| def normalized_text(text: str) -> str: |
| return " ".join(re.findall(r"[a-z_][a-z_0-9]*|\d+", text.lower())) |
|
|
|
|
| def shingles(text: str, width: int = 5) -> frozenset[tuple[str, ...]]: |
| tokens = normalized_text(text).split() |
| if len(tokens) < width: |
| return frozenset({tuple(tokens)}) |
| return frozenset(tuple(tokens[i:i + width]) for i in range(len(tokens) - width + 1)) |
|
|
|
|
| def similarity(left: frozenset, right: frozenset) -> float: |
| union = left | right |
| return len(left & right) / len(union) if union else 1.0 |
|
|
|
|
| class Deduper: |
| """Exact and token-shingle dedupe, with a conservative near-duplicate gate.""" |
| def __init__(self, threshold: float = 0.82, width: int = 5): |
| self.threshold = threshold |
| self.width = width |
| self.entries: list[tuple[str, str, frozenset]] = [] |
|
|
| def match(self, text: str) -> str | None: |
| fingerprint = sha256(normalized_text(text)) |
| token_set = shingles(text, self.width) |
| for identifier, previous, previous_tokens in self.entries: |
| if fingerprint == previous or similarity(token_set, previous_tokens) >= self.threshold: |
| return identifier |
| return None |
|
|
| def add(self, identifier: str, text: str) -> None: |
| self.entries.append((identifier, sha256(normalized_text(text)), shingles(text, self.width))) |
|
|
|
|
| def read_jsonl(path: Path) -> Iterable[dict]: |
| with path.open(encoding="utf-8") as stream: |
| for number, line in enumerate(stream, 1): |
| if not line.strip(): |
| continue |
| try: |
| row = json.loads(line) |
| except json.JSONDecodeError as exc: |
| raise ValueError(f"{path}:{number}: invalid JSON") from exc |
| if not isinstance(row, dict): |
| raise ValueError(f"{path}:{number}: each row must be an object") |
| yield row |
|
|
|
|
| def write_jsonl(path: Path, rows: Iterable[dict]) -> None: |
| path.parent.mkdir(parents=True, exist_ok=True) |
| path.write_text("".join(json.dumps(row, ensure_ascii=False, sort_keys=True) + "\n" for row in rows), encoding="utf-8") |
|
|
|
|
| def identify_license(text: str) -> str | None: |
| """Strict signatures; ambiguous/mixed notices require manual exclusion.""" |
| normalized = " ".join(text.lower().split()) |
| if FORBIDDEN_LICENSE.search(text): |
| return None |
| if ("permission is hereby granted, free of charge" in normalized |
| and "the above copyright notice and this permission notice" in normalized |
| and "the software is provided" in normalized): |
| return "MIT" |
| if ("apache license" in normalized and "version 2.0, january 2004" in normalized |
| and "end of terms and conditions" in normalized): |
| return "Apache-2.0" |
| if ("cc0 1.0 universal" in normalized and "waiver" in normalized |
| and "public license fallback" in normalized): |
| return "CC0-1.0" |
| return None |
|
|
|
|
| def has_conflicting_header(code: str, expected_license: str | None = None) -> bool: |
| """Scan comments only; a GPL string inside gameplay text is not a license.""" |
| comments = "\n".join(line.lstrip()[1:] for line in code.splitlines() if line.lstrip().startswith("#")) |
| if FORBIDDEN_LICENSE.search(comments): |
| return True |
| for expression in re.findall(r"SPDX-License-Identifier:\s*([^\n]+)", comments, re.I): |
| if (expression.strip() not in ALLOWED_LICENSES |
| or (expected_license and expression.strip() != expected_license)): |
| return True |
| if expected_license: |
| named = set() |
| if re.search(r"\bMIT license\b", comments, re.I): |
| named.add("MIT") |
| if re.search(r"\bApache (?:software )?license\b", comments, re.I): |
| named.add("Apache-2.0") |
| if re.search(r"\bCC0\b", comments, re.I): |
| named.add("CC0-1.0") |
| if named - {expected_license}: |
| return True |
| return False |
|
|
