khtst-rust / scripts /74_publicar_v090.py
PowerMachine's picture
errata 74: ignore **/ para nomes de estado (fnmatch literal nao casa subdiretorios)
39f2f80 verified
Raw History Blame Contribute Delete
7.6 kB
#!/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()