File size: 6,353 Bytes
ae36cfb | 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 | """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
|