Spaces:
Running
Running
File size: 7,811 Bytes
6918275 | 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 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 | #!/usr/bin/env python3
"""Deterministic deployment script to synchronize a green commit to Hugging Face Space."""
from __future__ import annotations
import argparse
import json
import os
import subprocess
import sys
import tempfile
import urllib.request
from pathlib import Path
def deployment_identity(root, requested, space):
head = subprocess.check_output(
["git", "rev-parse", "HEAD"], cwd=root, text=True
).strip()
dirty = bool(
subprocess.check_output(
["git", "status", "--porcelain", "--untracked-files=all"],
cwd=root,
text=True,
)
)
sha = subprocess.check_output(
[
"git",
"rev-parse",
"--verify",
"--end-of-options",
(requested or head) + "^{commit}",
],
cwd=root,
text=True,
).strip()
print(
f"Local HEAD: {head}\nDeployment SHA: {sha}\nWorking tree clean: {not dirty}\nTarget HF Space: {space}"
)
if dirty:
raise RuntimeError(
"Refusing deployment: working tree contains uncommitted changes. "
"The HF deployment is built from a committed Git SHA. "
"Commit/stash your changes before deploying."
)
if sha != head:
print(
"Explicit historical snapshot selected; local HEAD is not being deployed."
)
return sha
def archive_snapshot(root, sha, snapshot):
with tempfile.TemporaryFile() as archive:
subprocess.run(
["git", "archive", "--format=tar", sha],
cwd=root,
stdout=archive,
check=True,
)
archive.seek(0)
subprocess.run(
["tar", "-xf", "-", "-C", str(snapshot)], stdin=archive, check=True
)
(snapshot / "openoperator-build.json").write_text(
json.dumps({"source_sha": sha}) + "\n", encoding="utf-8"
)
def verify_health(url, sha):
try:
with urllib.request.urlopen(url, timeout=15) as response:
data = json.loads(response.read(16384))
return response.status == 200 and data.get("sha") == sha
except (OSError, ValueError, AttributeError):
return False
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--space", default="Leon4gr45/openoperator", help="Target HF Space"
)
parser.add_argument("--sha", default="", help="Expected GitHub commit SHA")
parser.add_argument(
"--verify",
action="store_true",
help="Require a live /health SHA match after pushing",
)
parser.add_argument(
"--health-url", default="https://leon4gr45-openoperator.hf.space/health"
)
args = parser.parse_args()
root = Path(__file__).resolve().parents[1]
try:
sha = deployment_identity(root, args.sha.strip(), args.space)
except (RuntimeError, subprocess.CalledProcessError) as error:
print(
str(error)
if isinstance(error, RuntimeError)
else "Invalid committed deployment SHA",
file=sys.stderr,
)
return 1
hf_token = os.environ.get("HF_TOKEN", "").strip()
if not hf_token:
print(
"Error: HF_TOKEN environment variable is required to push to Hugging Face Space.",
file=sys.stderr,
)
return 1
with tempfile.TemporaryDirectory() as temp_dir:
snapshot = Path(temp_dir) / "hf_snapshot"
snapshot.mkdir(parents=True, exist_ok=True)
archive_snapshot(root, sha, snapshot)
readme_path = snapshot / "README.md"
content = (
readme_path.read_text(encoding="utf-8") if readme_path.exists() else ""
)
metadata = """---
title: OpenOperator
emoji: 🤖
colorFrom: indigo
colorTo: blue
sdk: docker
app_port: 7860
pinned: false
---
"""
if not content.startswith("---"):
readme_path.write_text(metadata + content, encoding="utf-8")
build_json = snapshot / "openoperator-build.json"
build_json.write_text(json.dumps({"source_sha": sha}) + "\n", encoding="utf-8")
# Create .gitattributes to track large binary extensions via LFS / ignore if needed
gitattributes = snapshot / ".gitattributes"
gitattributes.write_text(
"*.png filter=lfs diff=lfs merge=lfs -text\n*.jpg filter=lfs diff=lfs merge=lfs -text\n*.jpeg filter=lfs diff=lfs merge=lfs -text\n*.gif filter=lfs diff=lfs merge=lfs -text\n*.webm filter=lfs diff=lfs merge=lfs -text\n*.ttf filter=lfs diff=lfs merge=lfs -text\n*.woff filter=lfs diff=lfs merge=lfs -text\n*.woff2 filter=lfs diff=lfs merge=lfs -text\n",
encoding="utf-8",
)
subprocess.check_call(["git", "init", "--initial-branch=main"], cwd=snapshot)
subprocess.check_call(
["git", "config", "user.name", "OpenOperator Deployer"], cwd=snapshot
)
subprocess.check_call(
["git", "config", "user.email", "deploy@openoperator.local"], cwd=snapshot
)
# Remove offending documentation and raw binary files (like .png, .jpg, .gif, .webm, .ttf, .woff2) from snapshot
for pattern in ["docs/res", "plugins/_whats_new/webui/assets"]:
path = snapshot / pattern
if path.exists():
import shutil
shutil.rmtree(path, ignore_errors=True)
for binary_file in snapshot.glob("**/*"):
if binary_file.is_file() and binary_file.suffix in [
".png",
".jpg",
".jpeg",
".gif",
".webm",
".ttf",
".woff",
".woff2",
]:
try:
binary_file.unlink()
except OSError:
print("Snapshot cleanup failed; refusing upload.", file=sys.stderr)
return 1
subprocess.check_call(["git", "add", "--all"], cwd=snapshot)
subprocess.check_call(
["git", "commit", "-m", f"Deploy source revision {sha}"], cwd=snapshot
)
remote_url = f"https://huggingface.co/spaces/{args.space}"
subprocess.check_call(
["git", "remote", "add", "huggingface", remote_url], cwd=snapshot
)
print(f"Pushing snapshot to Hugging Face Space {args.space}...")
askpass = Path(temp_dir) / "askpass.py"
askpass.write_text(
"#!/usr/bin/env python3\nimport os,sys\nprint('oauth2' if 'username' in sys.argv[1].lower() else os.environ['HF_TOKEN'])\n"
)
askpass.chmod(0o700)
push_env = dict(os.environ, GIT_ASKPASS=str(askpass), GIT_TERMINAL_PROMPT="0")
result = subprocess.run(
["git", "push", "--force", "huggingface", "main:main"],
cwd=snapshot,
env=push_env,
capture_output=True,
check=False,
)
if result.returncode:
print(
"HF push failed; credential-bearing transport output withheld.",
file=sys.stderr,
)
return 1
print(
f"Snapshot push completed for {args.space} at SHA {sha}; push alone is not deployment qualification."
)
print(f"Verify {args.health_url}: health SHA must equal {sha}.")
if args.verify:
if not verify_health(args.health_url, sha):
print(
"Deployment unqualified: /health SHA does not match the requested SHA.",
file=sys.stderr,
)
return 2
print(
"Deployment qualified: live /health SHA matches requested deployment SHA."
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
|