File size: 4,436 Bytes
afeb845 21c194a 58bb62c afeb845 | 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 | #!/usr/bin/env python3
"""Hash-verified deploy to the HF Space (abc1181/Jaarvis).
Usage:
python3 scripts/deploy_hf.py path/one.py path/two.html ... # files to sync
python3 scripts/deploy_hf.py --check path/one.py ... # diff only, no upload
Rules:
* Token comes from the HF_TOKEN env var or the repo-local `.hf_token` file
(gitignored). Never hardcoded, never printed.
* Only the listed files are uploaded — `config/` must never appear here.
* sha256 local-vs-remote decides what uploads; every uploaded file is
re-fetched after the commit and re-verified before we call it done.
"""
import argparse
import fnmatch
import hashlib
import os
import sys
import time
from pathlib import Path
REPO_ID = "abc1181/Jaarvis"
ROOT = Path(__file__).resolve().parent.parent
# Belt-and-suspenders: these never leave the machine, tracked or not.
# (config/__init__.py, config/*.ico etc. are plain source and deploy fine.)
DENY_GLOBS = ("*api_keys.json", "config/certs/*", "config/whatsapp_web/*",
"*credentials*", "*client_secret*", "*token*.json",
".env", ".env.*", "*.env", ".hf_token", "*.pem", "*.key",
"jarvisd.json", "*agents.json", "dashboard_auth.json")
def _token() -> str:
tok = os.environ.get("HF_TOKEN", "").strip()
if not tok:
f = ROOT / ".hf_token"
if f.exists():
tok = f.read_text().strip()
if not tok:
sys.exit("no HF token: set HF_TOKEN or create .hf_token")
return tok
def _sha256(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def _remote_bytes(path: str, token: str) -> bytes | None:
"""Raw file from the Space's main branch, cache-busted. None if absent."""
import requests
url = f"https://huggingface.co/spaces/{REPO_ID}/resolve/main/{path}"
r = requests.get(url, headers={"Authorization": f"Bearer {token}"},
params={"download": "true"}, timeout=60)
if r.status_code == 404:
return None
r.raise_for_status()
return r.content
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("files", nargs="+", help="repo-relative paths")
ap.add_argument("--check", action="store_true", help="diff only, no upload")
ap.add_argument("--message", default=None, help="commit message")
args = ap.parse_args()
token = _token()
from huggingface_hub import CommitOperationAdd, HfApi
api = HfApi(token=token)
ops, verified = [], []
for rel in args.files:
local = ROOT / rel
if not local.is_file():
sys.exit(f"missing local file: {rel}")
if any(fnmatch.fnmatch(rel, g) for g in DENY_GLOBS):
sys.exit(f"refusing to deploy {rel} (matches secret denylist)")
want = _sha256(local.read_bytes())
try:
have = _sha256(_remote_bytes(rel, token) or b"")
except Exception as e:
sys.exit(f"remote check failed for {rel}: {e}")
if want == have:
print(f" = {rel} (up to date)")
else:
print(f" + {rel} ({want[:12]})")
ops.append(CommitOperationAdd(path_in_repo=rel, path_or_fileobj=str(local)))
verified.append((rel, want))
if not ops:
print("nothing to deploy")
return 0
if args.check:
print(f"check: {len(ops)} file(s) differ")
return 0
msg = args.message or ("deploy: " + ", ".join(r for r, _ in verified))
print(f"committing {len(ops)} file(s)…")
api.create_commit(repo_id=REPO_ID, repo_type="space", operations=ops,
commit_message=msg)
# Post-commit verification: every uploaded file must match byte-for-byte.
for rel, want in verified:
for attempt in range(5):
got = _remote_bytes(rel, token)
if got is not None and _sha256(got) == want:
print(f" ok {rel}")
break
time.sleep(2)
else:
sys.exit(f"VERIFY FAILED: {rel} does not match local after commit")
try:
rt = api.get_space_runtime(REPO_ID)
print(f"space: {rt.stage} sha={rt.raw.get('sha')}")
if rt.stage != "RUNNING":
print("note: Space is rebuilding — watch logs before traffic")
except Exception as e:
print(f"(runtime status unavailable: {e})")
print("DEPLOY OK")
return 0
if __name__ == "__main__":
sys.exit(main())
|