para i, fila en enumerar(tqdm(ds_test, desc=”inspeccionando estructura”, total=200)): si i >= 200: romper p = parse_task(fila[“task_binary”]) si pag[“format”] in (“tar”, “zip”): para nombre, cuerpo en p[“files”].items(): nombre_archivo_contador[name] += 1 si nombre.endswith(“.json”) e isinstance(body, str): intente: obj = json.loads(body) si isinstance(obj, dict): para k en obj.keys(): all_json_keys[k] += 1 excepto Excepción: pasar si len(samples_for_show) < 2: samples_for_show.append((fila["path"]p)) print("\nNombres de archivos más comunes dentro de los archivos de tareas:") for name, n in filename_counter.most_common(15): print(f" {n:>4} {name}”) print(“\nClaves JSON de nivel superior más comunes (en cualquier *.json):”) for k, n in all_json_keys.most_common(20): print(f” {n:>4} {k}”) if samples_for_show: print(f”\nLista completa de archivos para una tarea de muestra ({samples_for_show[0][0]}):”) para nombre, cuerpo en samples_for_show[0][1][“files”].items(): sz = len(body) if isinstance(body, (str, bytes)) else 0 print(f” {name} ({sz:,} B)”) VERIFIER_FILE_PATTERNS = (“verificador”, “verificar”, “calificador”, “juez”, “puntuación”, “evaluación”) VERIFIER_JSON_KEYS = (“verificador”, “verificador_config”, “juez”, “calificador”, “rúbrica”, “test_patch”, “FAIL_TO_PASS”, “pruebas”) def has_verifier (analizado: Dict[str, Any]) -> bool: “””Detectar verificadores mediante nombre de archivo, contenido JSON o ambos.””” si se analiza[“format”] no en (“tar”, “zip”): c = parsed.get(“content”) if isinstance(c, dict): devuelve cualquiera(k en c para k en VERIFIER_JSON_KEYS) devuelve Archivos falsos = analizado[“files”]
para nombre en archivos: low = name.lower() si existe (pat en low para pat en VERIFIER_FILE_PATTERNS): devuelve True para nombre, cuerpo en files.items(): if name.endswith((“.json”, “.yaml”, “.yml”)) e isinstance(body, str): intente: obj = json.loads(body) if isinstance(obj, dict) and any(k in obj for k in VERIFIER_JSON_KEYS): devuelve Verdadero excepto Excepción: pasa bajo = body.lower() si “verificador” está en nivel bajo o “test_patch” en bajo: devuelve Verdadero devuelve clase Falsa TaskTroveExplorer: “””Interfaz de alto nivel para el conjunto de datos de pensamientos abiertos/TaskTrove.””” def __init__(self, split: str = “test”, dataset_id: str = DATASET_ID): self.dataset_id = dataset_id self.split = split self._ds = load_dataset(dataset_id, split=split, streaming=True) def iter(self, limit: Opcional[int] = Ninguno, filtro_fuente: Opcional[str] = Ninguno) -> Iterador[Dict[str, Any]]: rx = re.compile(source_filter) si source_filter else Ninguno n = 0 para la fila en self._ds: si rx y no rx.search(source_of(row[“path”])): continuar con la fila de rendimiento n += 1 si el límite no es Ninguno y n >= límite: devolver def muestra(self, n: int = 5, source_filter: Opcional[str] = Ninguno) -> Lista[Dict[str, Any]]: fuera = []
para la fila en self.iter(límite=n, source_filter=source_filter): analizado = parse_task(fila[“task_binary”]) analizado[“path”] = fila[“path”]
analizado[“source”] = fuente_de (fila[“path”]) out.append(analizado) devuelve resumen def(self, limit: int = 1000, source_filter: Opcional[str] = Ninguno) -> pd.DataFrame: filas = []
para la fila en self.iter (límite = límite, filtro_fuente = filtro_fuente): analizado = tarea_parse (fila[“task_binary”]) filas.append({ “fuente”: fuente_de(fila[“path”]), “comprimido”: analizado[“compressed_size”]”sin procesar”: analizado[“raw_size”]”formato”: analizado[“format”]”n_files”: len(parsed.get(“files”, {})), “has_verifier”: has_verifier(analizado), }) df = pd.DataFrame(rows) if df.empty: return df return (df.groupby(“source”) .agg(n=(“compressed”, “count”), mean_compressed_kb=(“compressed”, lambda s: s.mean()/1024), mean_raw_kb=(“raw”, lambda s: s.mean()/1024), mean_n_files=(“n_files”, “mean”), verifier_rate=(“has_verifier”, “mean”)) .round(2) .sort_values(“n”, ascending=False)) @staticmethod def has_verifier(analizado: dictar[str, Any]) -> bool: return has_verifier(analizado) def export(self, output_dir: Unión[str, Path]n: int = 10, filtro_fuente: opcional[str] = Ninguno) -> Ruta: salida_dir = Ruta(salida_dir) salida_dir.mkdir(parents=True, exist_ok=True) para analizado en self.sample(n=n, source_filter=source_filter): slug = analizado[“path”].replace(“/”, “_”) tdir = output_dir / slug tdir.mkdir(exist_ok=True) si se analiza[“format”] in (“tar”, “zip”): para nombre, cuerpo analizado[“files”].items(): out = tdir / nombre out.parent.mkdir(parents=True, exist_ok=True) if isinstance(body, str): out.write_text(body, encoding=”utf-8″) else: out.write_bytes(body) else: content = parsed.get(“content”, b””) if isinstance(content, (dict, list)): (tdir / “task.json”).write_text(json.dumps(content, indent=2)) elif isinstance(content, str): (tdir / “task.txt”).write_text(content) else: (tdir / “task.bin”).write_bytes(content) print(f” ✓ tareas exportadas a {output_dir.resolve()}”) return output_dir explorer = TaskTroveExplorer(split=”test”) print(“\nMuestra de 3 tareas analizadas:”) para s en explorer.sample(n=3): print(f”ruta: {s[‘path’]} | fuente: {s[‘source’]} | formato: {s[‘format’]} | “f”archivos: {len(s.get(‘archivos’, {}))} | verificador: {has_verifier(s)}”)