#!/usr/bin/env python3 """74_publicar_v090.py — Publicação da v0.9.0 no HF (UM ÚNICO upload_folder organizado — o HF bloqueia envios parciais em sequência; upload atômico). Fases: 1. inventário da conta + APAGAMENTO de estados de treino remanescentes; 2. varredura local do VALOR do token (nenhum arquivo pode contê-lo); 3. UM ÚNICO upload_folder com ignore_patterns EXPLÍCITOS; 4. verificação pós-upload (0 estados, módulos-chave presentes); 5. varredura REMOTA do valor do token; 6. apagamento dos estados de treino LOCAIS (checkpoints). Token: EXCLUSIVAMENTE via variável de ambiente HF_TOKEN (nunca em arquivo). """ import os, sys, tempfile ARQUIVOS_EXCLUIR_DIRS = {"target", ".cargo", "__pycache__", ".git", ".venv", "skills", "externos"} ARQUIVOS_EXCLUIR_NOME = {"shard_treino.jsonl", "shard_validacao.jsonl", "bpe.json", "rustup-init.sh"} SUFIXOS_ESTADO = (".pt", ".pth", ".ckpt", ".safetensors", ".bin", ".npz", ".msgpack") NOMES_ESTADO = {"gerador_conceitos.json", "tokenizador_conceitos.json", "gerador_aritmetica.json"} def coletar(pasta): arquivos = {} for raiz, dirs, fs in os.walk(pasta): dirs[:] = [d for d in dirs if d not in ARQUIVOS_EXCLUIR_DIRS] for f in fs: if f in ARQUIVOS_EXCLUIR_NOME or f in NOMES_ESTADO: continue if f.endswith(SUFIXOS_ESTADO) or f.endswith(".estado.json"): continue if f.endswith(".so") or f in (".gitignore", "Cargo.lock.ori"): continue if f.startswith("hf_") or f.startswith("intros_"): continue caminho = os.path.join(raiz, f) arquivos[os.path.relpath(caminho, pasta)] = caminho return arquivos def varre_token_em_arquivos(token, pares): tb = token.encode() achados = [] for rel, p in pares.items(): try: with open(p, "rb") as f: if tb in f.read(): achados.append(rel) except OSError: pass return achados def main(): from huggingface_hub import HfApi token = os.environ.get("HF_TOKEN") if not token: print("ERRO: HF_TOKEN ausente"); sys.exit(1) api = HfApi(token=token) repo = "PowerMachine/khtst-rust" # ---------------- 1) inventário da conta + apaga estados ---------------- print("=== 1) inventário da conta HF (procura estados de treino) ===") repos = list(api.list_models(author="PowerMachine")) print(f"repos: {len(repos)}") total_removidos = 0 for r in repos: try: files = api.list_repo_files(repo_id=r.id, repo_type="model") except Exception as e: print(f" {r.id}: erro ao listar ({e})"); continue estados = [f for f in files if f.endswith(SUFIXOS_ESTADO) or f.split("/")[-1] in NOMES_ESTADO or (".meta.json" in f and "checkpoint" in f)] if estados: print(f" {r.id}: APAGANDO {len(estados)} estados") for f in estados: api.delete_file(path_in_repo=f, repo_id=r.id, repo_type="model") total_removidos += len(estados) else: print(f" {r.id}: {len(files)} arquivos, 0 estados") print(f"estados removidos no total: {total_removidos}") # ---------------- 2) varredura local do token ---------------- pasta = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) arquivos = coletar(pasta) print(f"\n=== 2) varredura local: {len(arquivos)} arquivos ===") achados = varre_token_em_arquivos(token, arquivos) if achados: print(f"ERRO: valor do token encontrado em: {achados}"); sys.exit(1) print("varredura de token local: limpa") # ---------------- 3) UM ÚNICO upload_folder ---------------- ignores = ( [f"{d}/**" for d in ARQUIVOS_EXCLUIR_DIRS] + [f"{d}" for d in ARQUIVOS_EXCLUIR_DIRS] # **/ : fnmatch literal NÃO casa subdiretórios (errata v0.9.0 — os # estados locais vazaram na 1ª publicação em que existiam) + [f"**/{n}" for n in ARQUIVOS_EXCLUIR_NOME] + [f"**/{n}" for n in NOMES_ESTADO] + list(SUFIXOS_ESTADO) + ["*.so", ".gitignore", "hf_*.py", "hf_*.json", "intros_*.json", "*.estado.json"] ) print(f"\n=== 3) upload_folder ÚNICO ({len(arquivos)} arquivos) ===") api.upload_folder( repo_id=repo, repo_type="model", folder_path=pasta, ignore_patterns=ignores, commit_message="v0.9.0: análise/prova/teste " "das críticas de arquitetura Qwen3.8 (r-34) — skew O(n²) via " "index_add (919× n=128, memória 2335×, bit-a-bit F-13/F-14), " "calibrar zero-cópia (1,56×, F-15; premissa SIMD refutada no fonte " "do candle), forward canônico (bug v080-1 fora da API pública, " "oráculos preservados), FFI isolada em feature python (lib.rs " "105kB→1,25kB; 107/107 sem Python; overhead medido 17,6%), " "newtypes T-V1/T-V2 — 97/97 gates, treino real Wikipédia PT-BR", ) print("upload concluído") # ---------------- 4) verificação pós-upload ---------------- print("\n=== 4) verificação pós-upload ===") files = api.list_repo_files(repo_id=repo, repo_type="model") estados = [f for f in files if f.endswith(SUFIXOS_ESTADO) or f.split("/")[-1] in NOMES_ESTADO] assert not estados, f"estados publicados: {estados}" chaves = ["src/python.rs", "src/tipos.rs", "src/fnh.rs", "src/moe_redirecionavel.rs", "src/treino.rs", "scripts/71_treino_v090.py", "scripts/72_suite_v090.py", "scripts/73_telemetria_v090.py", "scripts/74_publicar_v090.py", "docs/memorias/r-34-analise-provas-qwen-v090.md", "resultados/suite_v090_final.json", "resultados/treino_v090.json", "resultados/telemetria_v090.json"] faltando = [c for c in chaves if c not in files] assert not faltando, f"faltando: {faltando}" print(f"repo: {len(files)} arquivos | 0 estados | módulos-chave OK") # ---------------- 5) varredura REMOTA do token ---------------- print("\n=== 5) varredura REMOTA do valor do token ===") with tempfile.TemporaryDirectory() as tmp: achados_remotos = [] for f in files: if f.endswith((".py", ".rs", ".md", ".json", ".toml", ".sh", ".cfg", ".txt", ".jsonl")): try: p = api.hf_hub_download(repo_id=repo, repo_type="model", filename=f, local_dir=tmp) with open(p, "rb") as fh: if token.encode() in fh.read(): achados_remotos.append(f) except Exception: pass if achados_remotos: print(f"ERRO: token remoto em: {achados_remotos}"); sys.exit(1) print("varredura remota: limpa") # ---------------- 6) apaga estados de treino LOCAIS ---------------- print("\n=== 6) estados de treino locais (checkpoints) ===") apagados = 0 for nome in NOMES_ESTADO: p = os.path.join(pasta, "resultados", nome) if os.path.exists(p): os.remove(p) apagados += 1 print(f" apagado: resultados/{nome}") print(f"locais apagados: {apagados}") print("\nPUBLICAÇÃO v0.9.0 CONCLUÍDA") if __name__ == "__main__": main()