Upload Backblaze_PNG_Upload.py

Backblaze_PNG_Upload.py

@@ -1,45 +1,63 @@
-"""Single-file ComfyUI node: public Hugging Face PNG -> private Backblaze B2.
-
-INSTALL: Put this file directly in ComfyUI/custom_nodes/, then restart ComfyUI.
-NODE: Backblaze PNG Upload (Auto), category Salia/Backblaze.
-
-INPUTS:
-  master_key: The application key SECRET. Legacy KEY_ID:SECRET also works.
-  png_url:    Public Hugging Face /resolve/ URL. /blob/ URLs also work.
-  key_id:     Master key ID (or account ID). When blank, extract the ID from
-              KEY_ID:SECRET or reuse an ID verified previously with this secret.
-              Backblaze has no API to discover the ID from a bare secret alone.
-              On the first run, copy it once from your Backblaze App Keys page.
-OUTPUTS: png_url (STRING) and STATUS (STRING). The URL is valid for seven days.
-STATUS shows the selected key ID, its source, and success/failure. It never shows
-the secret. Errors return an empty URL and an explanatory STATUS, not a traceback.
-
-Automatically creates/reuses its own private bucket. Existing public buckets
-are never made private or used. PNG bytes are preserved, including transparency
-and metadata. Content-hash filenames reuse identical uploads on subsequent runs.
-Running again refreshes the link; the stored image itself does not expire.
-Anyone holding the returned link can download that image until the link expires.
-
-No SDK, pip install, manual config file, or __init__.py is required.
-Verified key IDs are saved automatically under the ComfyUI user directory
-(fallback: ~/.cache/comfyui-backblaze-png). Only ID + SHA-256(secret) are saved;
-the master secret and authorization tokens are never written to this cache.
-Uses Python's standard library. Source PNG limit: 128 MiB. Public HF URLs only.
-B2 Cloud Storage must already be enabled on the account. The master key has the
-needed permissions; a normal unrestricted key also works if granted listBuckets,
-writeBuckets, listFiles, writeFiles, and shareFiles.
-
-Credentials entered in widgets can be saved inside ComfyUI workflows/prompts;
-remove them before sharing workflows. This node never prints credentials.
-
-API references:
+"""Single-file ComfyUI node: resumable Hugging Face district -> Backblaze mirror.
+
+INSTALL: Replace this file in ComfyUI/custom_nodes/, restart ComfyUI, then re-add
+"Backblaze District Mirror (Auto / Resume)". No __init__.py or packages needed.
+Node class ID remains BackblazePNGUpload. The old png_url input is replaced.
+
+REQUIRED INPUTS:
+  master_key:   Backblaze master secret, or legacy KEY_ID:SECRET.
+  District_Name: e.g. District_19_K. Recursively scans that folder in the model
+                 repository saliacoel/map_imgs_gpt, branch main.
+  hf_key:       Hugging Face token with WRITE access to that repository.
+OPTIONAL INPUT:
+  key_id:       Backblaze key ID/account ID. Blank uses the ID in KEY_ID:SECRET
+                or the ID remembered after a previous successful authentication.
+                Backblaze cannot discover a first-time ID from the secret alone.
+OUTPUTS: dictionary_url, STATUS, dictionary_json (all STRING).
+
+The node publishes DICT_BLAZE_<District_Name>.json at the repository root.
+Its "dictionary" maps complete Hugging Face main-branch image URLs to Backblaze
+URLs. "files" contains source hashes, pinned source URLs/commits, B2 object IDs,
+checksums, URL expiration timestamps, statuses, attempts and errors. Other fields
+record source inventory/commit history, added/changed/removed paths, counts and
+run/checkpoint history. Source deletions never delete Backblaze objects.
+
+Every scan/download is pinned to one source commit. Unchanged source hashes reuse
+existing Backblaze objects. Re-running also refreshes private links without
+redownloading unchanged images. Links are valid for at most seven days; stored
+objects do not expire. Anyone who can read the dictionary can use its bearer URLs
+until expiration. API secrets and account-wide authorization tokens are NOT put
+in the dictionary. One scoped download token serves only this district's prefix.
+
+Transient requests get 8 attempts total, waiting 3,5,7,9,11,13,15 seconds. A longer
+Retry-After is respected. Exhausted transient failures stop the run and trigger
+a final partial dictionary save. Permanent bad source files are recorded and
+skipped so later images can still upload. Run again to resume failed/pending work.
+
+Progress is written atomically locally after each image, before requesting its
+link, and after completion. Hugging Face checkpoints occur before image transfer,
+every 25 files or 60 seconds, and on completion/failure/cancellation. Local state
+survives a failed remote publication and retains source variants if scanning
+fails. Local process/OS locks and remote parent-commit checks prevent conflicting
+writers from silently overwriting progress. If HF rejects the final save, STATUS
+says so, and dictionary_json plus the local checkpoint retain the result.
+
+Runtime data lives under the ComfyUI user directory/backblaze_png_upload (fallback
+~/.cache/comfyui-backblaze-png). Key cache stores only key ID + SHA-256(secret).
+No manual configuration required. Limits: 128 MiB per image, 32 MiB per dictionary;
+last 100 run records and up to 100 recent source commit summaries are retained.
+Credentials in ComfyUI widgets may be saved in workflows; remove before sharing.
+
+Verified API contracts:
 https://www.backblaze.com/apidocs/b2-authorize-account
-https://www.backblaze.com/apidocs/b2-create-bucket
 https://www.backblaze.com/apidocs/b2-upload-file
 https://www.backblaze.com/apidocs/b2-get-download-authorization
+https://huggingface.co/docs/huggingface_hub/package_reference/hf_api
+https://github.com/huggingface/huggingface_hub/blob/main/src/huggingface_hub/_commit_api.py
 """
 
 import base64
+import copy
 import hashlib
 import json
 import os
@@ -48,7 +66,8 @@ import socket
 import tempfile
 import threading
 import time
-from datetime import timezone
+import uuid
+from datetime import datetime, timezone
 from email.utils import parsedate_to_datetime
 from http.client import HTTPException
 from pathlib import Path
@@ -66,7 +85,7 @@ _DEFAULT_PNG = (
 _MAX_PNG_BYTES = 128 * 1024 * 1024
 _MAX_JSON_BYTES = 4 * 1024 * 1024
 _LINK_SECONDS = 7 * 24 * 60 * 60
-_ATTEMPTS = 5
+_ATTEMPTS = 8
 _TIMEOUT_SECONDS = 120
 _TOKEN_ERRORS = {"bad_auth_token", "expired_auth_token"}
 _KEY_ID_CACHE = {}
@@ -140,6 +159,7 @@ class _RequestError(RuntimeError):
         self.status = status
         self.code = code if re.fullmatch(r"[A-Za-z0-9_]{1,80}", code) else "http_error"
         self.retry_after = retry_after
+        self.attempts = 1
         super().__init__("HTTP {} ({})".format(status, self.code))
 
     @property
@@ -161,7 +181,7 @@ class _HTTPSRedirect(HTTPRedirectHandler):
 
 
 def _delay(attempt, error):
-    seconds = min(2 ** attempt, 30)
+    seconds = 3 + 2 * attempt
     value = error.retry_after
     if value:
         try:
@@ -174,13 +194,12 @@ def _delay(attempt, error):
                 seconds = max(seconds, date.timestamp() - time.time())
             except (TypeError, ValueError, OverflowError):
                 pass
-    if seconds > 60:
-        raise RuntimeError("The server requested a longer retry delay. Run again later.")
-    time.sleep(seconds)
+    _interruptible_sleep(seconds)
 
 
 def _read_url(url, *, data=None, headers=None, limit=_MAX_JSON_BYTES,
               source=False, timeout=_TIMEOUT_SECONDS):
+    _check_cancel()
     request_headers = {"User-Agent": "ComfyUI-Backblaze-PNG/1.0"}
     request_headers.update(headers or {})
     opener = build_opener(_HTTPSRedirect() if source else _NoRedirect())
@@ -192,6 +211,7 @@ def _read_url(url, *, data=None, headers=None, limit=_MAX_JSON_BYTES,
                 raise RuntimeError("The response exceeds the allowed size (PNG limit: 128 MiB).")
             chunks, total = [], 0
             while True:
+                _check_cancel()
                 chunk = response.read(min(1024 * 1024, limit + 1 - total))
                 if not chunk:
                     break
@@ -234,6 +254,7 @@ def _json_request(url, *, payload=None, headers=None, retries=True):
                 raise _RequestError(502, "invalid_json_response")
             return result
         except _RequestError as exc:
+            exc.attempts = attempt + 1
             if not retries or not exc.temporary or attempt == _ATTEMPTS - 1:
                 raise
             _delay(attempt, exc)
@@ -372,27 +393,30 @@ class _B2Client:
                     return owned[name]
         raise RuntimeError("Could not choose an available private bucket name. Run again later.")
 
-    def file_exists(self, bucket_id, name, sha1, length):
+    def find_file(self, bucket_id, name, sha1, length):
         result = self.call("b2_list_file_names", {
             "bucketId": bucket_id, "prefix": name, "startFileName": name, "maxFileCount": 1})
-        return any(item.get("fileName") == name and item.get("action") == "upload"
-                   and item.get("contentSha1") == sha1 and item.get("contentLength") == length
-                   for item in result["files"])
+        return next((item for item in result["files"]
+                     if item.get("fileName") == name and item.get("action") == "upload"
+                     and item.get("contentSha1") == sha1 and item.get("contentLength") == length), None)
 
-    def ensure_file(self, bucket, name, data):
+    def file_exists(self, bucket_id, name, sha1, length):
+        return self.find_file(bucket_id, name, sha1, length) is not None
+
+    def ensure_file(self, bucket, name, data, content_type="image/png"):
         sha1 = hashlib.sha1(data).hexdigest()
         for attempt in range(_ATTEMPTS):
             # Also covers a timed-out upload which actually succeeded on the server.
-            if self.file_exists(bucket["bucketId"], name, sha1, len(data)):
-                print("[Backblaze PNG] Reusing an identical uploaded PNG.")
-                return
+            existing = self.find_file(bucket["bucketId"], name, sha1, len(data))
+            if existing:
+                return existing
             upload = self.call("b2_get_upload_url", {"bucketId": bucket["bucketId"]})
             upload_url = _b2_base_url(upload["uploadUrl"], upload=True)
             try:
                 raw = _read_url(upload_url, data=data, headers={
                     "Authorization": upload["authorizationToken"],
                     "X-Bz-File-Name": quote(name, safe=""),
-                    "Content-Type": "image/png",
+                    "Content-Type": content_type,
                     "Content-Length": str(len(data)),
                     "X-Bz-Content-Sha1": sha1,
                 })
@@ -404,9 +428,9 @@ class _B2Client:
                         or result.get("contentSha1") != sha1
                         or result.get("contentLength") != len(data)):
                     raise _RequestError(502, "invalid_upload_response")
-                print("[Backblaze PNG] PNG uploaded and checksum confirmed.")
-                return
+                return result
             except _RequestError as exc:
+                exc.attempts = attempt + 1
                 retryable = exc.temporary or exc.code in _TOKEN_ERRORS or exc.code == "auth_token_limit"
                 if not retryable or attempt == _ATTEMPTS - 1:
                     raise
@@ -422,82 +446,903 @@ class _B2Client:
                 + urlencode({"Authorization": result["authorizationToken"]}))
 
 
+
+
+import base64
+import json
+import math
+import re
+import socket
+import time
+from datetime import timezone
+from email.utils import parsedate_to_datetime
+from http.client import HTTPException
+from urllib.error import HTTPError, URLError
+from urllib.parse import quote, urljoin, urlsplit
+from urllib.request import HTTPRedirectHandler, Request, build_opener
+
+
+_HF_REPO = "saliacoel/map_imgs_gpt"
+_HF_ROOT = "https://huggingface.co"
+_HF_API = _HF_ROOT + "/api/models/" + _HF_REPO
+_HF_METADATA_LIMIT = 32 * 1024 * 1024
+
+
+class _HFError(RuntimeError):
+    """Safe for STATUS: never retain server bodies, URLs or credentials."""
+
+    def __init__(self, status, code="http_error", retry_after=None):
+        self.status = int(status)
+        self.code = code if re.fullmatch(r"[a-z_]{1,64}", str(code)) else "http_error"
+        self.retry_after = retry_after
+        self.attempts = 1
+        super().__init__("Hugging Face HTTP {} ({})".format(self.status, self.code))
+
+    @property
+    def temporary(self):
+        return (self.status in (0, 408, 429) or 500 <= self.status <= 599
+                or self.code == "rate_limited")
+
+    @property
+    def conflict(self):
+        return self.status in (409, 412) or self.code == "parent_commit_conflict"
+
+
+class _HFSafeRedirect(HTTPRedirectHandler):
+    """HF resolves large files to a signed CDN URL; do not send HF tokens there."""
+
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        old = urlsplit(req.full_url)
+        new = urlsplit(newurl)
+        if new.scheme.lower() != "https" or new.username or new.password:
+            raise _HFError(400, "unsafe_redirect")
+        # Commits must never be redirected/replayed to another endpoint.
+        if req.get_method() not in ("GET", "HEAD"):
+            raise _HFError(400, "unexpected_commit_redirect")
+        redirected = super().redirect_request(req, fp, code, msg, headers, newurl)
+        if redirected is not None and (old.scheme, old.netloc) != (new.scheme, new.netloc):
+            for key in list(redirected.headers):
+                if key.lower() in ("authorization", "cookie", "proxy-authorization"):
+                    del redirected.headers[key]
+            for key in list(redirected.unredirected_hdrs):
+                if key.lower() in ("authorization", "cookie", "proxy-authorization"):
+                    del redirected.unredirected_hdrs[key]
+        return redirected
+
+
+def _hf_delay_seconds(attempt, retry_after):
+    """attempt=0 after the first failure => 3,5,7,9,11,13,15 seconds."""
+    seconds = float(3 + 2 * attempt)
+    if retry_after:
+        try:
+            supplied = float(retry_after)
+        except (TypeError, ValueError):
+            try:
+                date = parsedate_to_datetime(str(retry_after))
+                if date.tzinfo is None:
+                    date = date.replace(tzinfo=timezone.utc)
+                supplied = date.timestamp() - time.time()
+            except (TypeError, ValueError, OverflowError):
+                supplied = 0
+        if math.isfinite(supplied):
+            seconds = max(seconds, supplied)
+    return seconds
+
+
+class _HFClient:
+    def __init__(self, token, check_cancel=None, sleep=None, on_retry=None):
+        token = str(token or "").strip()
+        if not token or any(c.isspace() for c in token):
+            raise ValueError("hf_key must be a Hugging Face token with write access to the repository.")
+        self._token = token
+        self._check_cancel = check_cancel or (lambda: None)
+        self._sleep = sleep or time.sleep
+        self._on_retry = on_retry or (lambda attempt, seconds, error: None)
+        self._opener = build_opener(_HFSafeRedirect())
+
+    def _wait(self, seconds):
+        # Long Retry-After values remain cancellable, with no busy waiting.
+        remaining = seconds
+        while remaining > 0:
+            self._check_cancel()
+            interval = min(1.0, remaining)
+            self._sleep(interval)
+            remaining -= interval
+        self._check_cancel()
+
+    @staticmethod
+    def _safe_url(url):
+        parts = urlsplit(url)
+        if (parts.scheme != "https" or parts.hostname != "huggingface.co"
+                or parts.port not in (None, 443) or parts.username or parts.password):
+            raise _HFError(400, "unsafe_api_url")
+
+    def _once(self, url, data=None, content_type=None, limit=_HF_METADATA_LIMIT):
+        self._safe_url(url)
+        self._check_cancel()
+        headers = {"User-Agent": "ComfyUI-Backblaze-District/3.0",
+                   "Authorization": "Bearer " + self._token,
+                   "Accept-Encoding": "identity"}
+        if content_type:
+            headers["Content-Type"] = content_type
+        request = Request(url, data=data, headers=headers)
+        try:
+            with self._opener.open(request, timeout=120) as response:
+                length = response.headers.get("Content-Length", "")
+                if length.isdigit() and int(length) > limit:
+                    raise _HFError(413, "response_too_large")
+                chunks, total = [], 0
+                while True:
+                    self._check_cancel()
+                    chunk = response.read(min(1024 * 1024, limit + 1 - total))
+                    if not chunk:
+                        break
+                    total += len(chunk)
+                    if total > limit:
+                        raise _HFError(413, "response_too_large")
+                    chunks.append(chunk)
+                if length.isdigit() and total != int(length):
+                    raise _HFError(0, "incomplete_response")
+                return b"".join(chunks), dict(response.headers.items())
+        except HTTPError as error:
+            status = error.code
+            retry_after = error.headers.get("Retry-After")
+            code = "http_error"
+            try:
+                # Only classify recognized server phrases. Never expose the body.
+                body = error.read(65536).decode("utf-8", "replace").lower()
+                if status in (401, 403, 429) and any(phrase in body for phrase in
+                        ("rate limit", "rate-limit", "rate_limit", "too many requests")):
+                    code = "rate_limited"
+                elif status in (400, 409, 412) and any(phrase in body for phrase in
+                        ("parentcommit", "parent commit", "commit has changed", "concurrent")):
+                    code = "parent_commit_conflict"
+                elif status in (401, 403):
+                    code = "authentication_or_permission_denied"
+                elif status == 404:
+                    code = "not_found"
+            except (OSError, HTTPException):
+                pass
+            finally:
+                error.close()
+            raise _HFError(status, code, retry_after) from None
+        except (URLError, TimeoutError, socket.timeout, ConnectionError, OSError, HTTPException):
+            raise _HFError(0, "connection_or_tls_error") from None
+
+    def _request(self, url, data=None, content_type=None, limit=_HF_METADATA_LIMIT, decode_json=False):
+        for attempt in range(8):
+            try:
+                raw, headers = self._once(url, data, content_type, limit)
+                if decode_json:
+                    try:
+                        return json.loads(raw), headers
+                    except (ValueError, UnicodeError):
+                        raise _HFError(502, "invalid_json_response") from None
+                return raw, headers
+            except _HFError as error:
+                error.attempts = attempt + 1
+                if not error.temporary or attempt == 7:
+                    raise
+                delay = _hf_delay_seconds(attempt, error.retry_after)
+                self._on_retry(attempt + 1, delay, error)
+                self._wait(delay)
+
+    def _json(self, url, data=None, content_type=None):
+        return self._request(url, data, content_type, decode_json=True)
+
+    @staticmethod
+    def _next_link(headers, current_url):
+        # RFC Link URLs can include commas; match bracketed targets first.
+        value = next((v for k, v in headers.items() if k.lower() == "link"), "")
+        for match in re.finditer(r"<([^<>]+)>\s*;([^<>]*)(?=,\s*<|$)", value):
+            relation = re.search(r'\brel\s*=\s*(?:"([^"]*)"|([^;,\s]+))', match.group(2))
+            if relation and "next" in (relation.group(1) or relation.group(2)).split():
+                return urljoin(current_url, match.group(1))
+        return None
+
+    def info(self):
+        data, _ = self._json(_HF_API + "/revision/main")
+        if not isinstance(data, dict) or not re.fullmatch(r"[0-9a-fA-F]{40,64}", str(data.get("sha", ""))):
+            raise _HFError(502, "invalid_repository_info")
+        return data
+
+    def tree(self, revision, district):
+        """Return every file (all extensions) within the recursively listed district."""
+        prefix = district.rstrip("/") + "/"
+        url = (_HF_API + "/tree/" + quote(revision, safe="") + "/" +
+               quote(district, safe="/") + "?recursive=true&expand=false&limit=1000")
+        files, seen_urls, seen_paths = [], set(), set()
+        expected_path = urlsplit(url).path
+        while url:
+            if url in seen_urls:
+                raise _HFError(502, "pagination_loop")
+            self._safe_url(url)
+            if urlsplit(url).path != expected_path:
+                raise _HFError(502, "unexpected_pagination_path")
+            seen_urls.add(url)
+            data, headers = self._json(url)
+            if not isinstance(data, list):
+                raise _HFError(502, "invalid_tree_response")
+            for item in data:
+                if not isinstance(item, dict):
+                    raise _HFError(502, "invalid_tree_entry")
+                if item.get("type") != "file":
+                    continue
+                path = item.get("path")
+                if (not isinstance(path, str) or not path.startswith(prefix)
+                        or any(part in (".", "..", "") for part in path.split("/"))):
+                    raise _HFError(502, "invalid_tree_path")
+                if path not in seen_paths:
+                    seen_paths.add(path)
+                    files.append(item)
+            url = self._next_link(headers, url)
+        return sorted(files, key=lambda item: item["path"])
+
+    def history(self, revision):
+        # Recent history is diagnostic; file blob/LFS identities determine changes.
+        data, _ = self._json(_HF_API + "/commits/" + quote(revision, safe="") + "?limit=100")
+        if not isinstance(data, list):
+            raise _HFError(502, "invalid_commit_history")
+        return [item for item in data[:100] if isinstance(item, dict)]
+
+    def read_file(self, path, revision="main", max_bytes=_HF_METADATA_LIMIT):
+        url = (_HF_ROOT + "/" + _HF_REPO + "/resolve/" +
+               quote(revision, safe="") + "/" + quote(path, safe="/"))
+        return self._request(url, limit=max_bytes)[0]
+
+    def commit_json(self, path, content_bytes, parent_sha, message):
+        """POST one regular JSON blob; optimistic concurrency is mandatory.
+
+        The caller owns 409/412 reconciliation. In particular, after a response
+        was lost, reread the remote dictionary and compare before recommitting.
+        """
+        if not re.fullmatch(r"[0-9a-fA-F]{40,64}", str(parent_sha)):
+            raise ValueError("A valid parent commit SHA is required for dictionary upload.")
+        if not isinstance(content_bytes, bytes):
+            raise TypeError("Dictionary content must be UTF-8 JSON bytes.")
+        # Explicit local bound, well below the documented regular-commit payload limit.
+        if len(content_bytes) > _HF_METADATA_LIMIT:
+            raise _HFError(413, "dictionary_exceeds_thirty_two_mib")
+        lines = [
+            {"key": "header", "value": {"summary": message, "description": "",
+                                         "parentCommit": parent_sha}},
+            {"key": "file", "value": {"path": path, "encoding": "base64",
+                                       "content": base64.b64encode(content_bytes).decode("ascii")}},
+        ]
+        payload = ("\n".join(json.dumps(line, separators=(",", ":"), ensure_ascii=True)
+                             for line in lines) + "\n").encode("utf-8")
+        result, _ = self._json(_HF_API + "/commit/main", payload, "application/x-ndjson")
+        if isinstance(result, dict) and result.get("success") is False:
+            raise _HFError(400, "commit_rejected")
+        if (not isinstance(result, dict) or not
+                re.fullmatch(r"[0-9a-fA-F]{40,64}", str(result.get("commitOid", "")))):
+            raise _HFError(502, "invalid_commit_response")
+        return result
+
+
+# District mirroring and durable dictionary checkpoints.
+_REPO_ID = "saliacoel/map_imgs_gpt"
+_MANIFEST_LIMIT = 32 * 1024 * 1024
+_CHECKPOINT_FILES = 25
+_CHECKPOINT_SECONDS = 60
+_IMAGE_TYPES = {
+    ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
+    ".webp": "image/webp", ".gif": "image/gif", ".bmp": "image/bmp",
+    ".tif": "image/tiff", ".tiff": "image/tiff", ".avif": "image/avif",
+    ".heic": "image/heic", ".heif": "image/heif", ".ico": "image/x-icon",
+    ".svg": "image/svg+xml", ".exr": "image/x-exr",
+}
+_DISTRICT_LOCKS = {}
+_DISTRICT_LOCKS_GUARD = threading.Lock()
+
+
+class _MirrorInterrupted(Exception):
+    pass
+
+
+class _ConcurrentMirror(RuntimeError):
+    pass
+
+
+def _check_cancel():
+    try:
+        import comfy.model_management as management
+    except ImportError:
+        return
+    if getattr(management, "processing_interrupted", lambda: False)():
+        raise _MirrorInterrupted("Cancelled by the user; completed uploads are saved.")
+
+
+def _interruptible_sleep(seconds):
+    remaining = max(0.0, float(seconds))
+    while remaining > 0:
+        _check_cancel()
+        part = min(1.0, remaining)
+        time.sleep(part)
+        remaining -= part
+
+
+def _utc_now():
+    return datetime.now(timezone.utc).isoformat(timespec="milliseconds")
+
+
+def _resolve_url(path, revision="main"):
+    return "https://huggingface.co/" + _REPO_ID + "/resolve/" + quote(revision, safe="") + "/" + quote(path, safe="/")
+
+
+def _json_bytes(document):
+    return (json.dumps(document, ensure_ascii=False, sort_keys=True, indent=2) + "\n").encode("utf-8")
+
+
+def _atomic_json(path, document):
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = None
+    try:
+        with tempfile.NamedTemporaryFile(mode="wb", dir=str(path.parent), prefix=path.name + ".", delete=False) as stream:
+            temporary = stream.name
+            stream.write(_json_bytes(document))
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary, path)
+    finally:
+        if temporary:
+            try:
+                os.unlink(temporary)
+            except OSError:
+                pass
+
+
+def _source_identity(item):
+    lfs = item.get("lfs") or {}
+    sha256 = str(lfs.get("oid") or lfs.get("sha256") or "").removeprefix("sha256:")
+    oid = item.get("oid") or item.get("blob_id") or ""
+    size = lfs.get("size", item.get("size"))
+    if not isinstance(size, int) or size < 0:
+        raise ValueError("Hugging Face returned an invalid image size.")
+    if sha256 and re.fullmatch(r"[0-9a-fA-F]{64}", sha256):
+        identity = "sha256:" + sha256.lower()
+    elif re.fullmatch(r"[0-9a-fA-F]{40}", str(oid)):
+        identity = "git:" + oid.lower()
+    else:
+        raise ValueError("Hugging Face did not return a usable source hash for an image.")
+    return identity, size
+
+
+def _verify_source_bytes(data, record):
+    if len(data) != record["size"]:
+        raise ValueError("Downloaded image size does not match the pinned Hugging Face file.")
+    fingerprint = record["source_fingerprint"]
+    if fingerprint.startswith("sha256:"):
+        actual = "sha256:" + hashlib.sha256(data).hexdigest()
+    else:
+        actual = "git:" + hashlib.sha1(b"blob " + str(len(data)).encode("ascii") + b"\0" + data).hexdigest()
+    if actual != fingerprint:
+        raise ValueError("Downloaded image checksum does not match the pinned Hugging Face file.")
+
+
+def _validate_manifest(doc, district):
+    if (not isinstance(doc, dict) or doc.get("schema_version") != 1
+            or doc.get("repo_id") != _REPO_ID or doc.get("district") != district
+            or not isinstance(doc.get("files"), dict) or not isinstance(doc.get("dictionary"), dict)):
+        raise ValueError("The existing district dictionary has an incompatible format; it was not overwritten.")
+    for path, record in doc["files"].items():
+        if not isinstance(path, str) or not path.startswith(district + "/") or not isinstance(record, dict):
+            raise ValueError("The existing district dictionary contains invalid file records; it was not overwritten.")
+    return doc
+
+
+def _inventory_hash(files):
+    pairs = [(path, entry["source_fingerprint"]) for path, entry in sorted(files.items()) if entry.get("status") != "removed"]
+    return hashlib.sha256(json.dumps(pairs, separators=(",", ":")).encode()).hexdigest()
+
+
+def _merge_successes(target, other):
+    """Resume only verified upload records for the SAME source content/destination."""
+    if target.get("backblaze") != other.get("backblaze"):
+        return
+    for path, candidate in other.get("files", {}).items():
+        entry = target["files"].get(path)
+        if (not entry or entry.get("status") == "removed" or candidate.get("status") == "removed"
+                or entry.get("source_fingerprint") != candidate.get("source_fingerprint")
+                or not candidate.get("b2", {}).get("file_name")):
+            continue
+        mine = entry.get("b2", {})
+        theirs = candidate["b2"]
+        if (not mine or (candidate.get("status") == "uploaded" and entry.get("status") != "uploaded")
+                or theirs.get("url_expires_at_unix", 0) > mine.get("url_expires_at_unix", 0)):
+            entry["b2"] = copy.deepcopy(theirs)
+            entry["status"] = candidate.get("status", "stored")
+            entry["last_error"] = candidate.get("last_error")
+            entry["updated_at"] = candidate.get("updated_at")
+            entry["processing_attempts"] = max(entry.get("processing_attempts", 0), candidate.get("processing_attempts", 0))
+
+
+def _recovery_records(document):
+    if not document:
+        return []
+    records = list(document.get("recovery_records", []))
+    for path, record in document.get("files", {}).items():
+        if record.get("b2", {}).get("file_name"):
+            records.append({"path": path, "record": copy.deepcopy(record),
+                            "backblaze": copy.deepcopy(document.get("backblaze", {}))})
+    unique = {}
+    for item in records:
+        if not isinstance(item, dict) or not isinstance(item.get("record"), dict):
+            continue
+        key = (item.get("path"), item["record"].get("source_fingerprint"),
+               item.get("backblaze", {}).get("bucket_id"))
+        unique[key] = item
+    return list(unique.values())
+
+
+class _CheckpointLock:
+    """OS locks survive no crashes: the OS releases them when the process exits."""
+    def __init__(self, checkpoint_path):
+        self.path = Path(str(checkpoint_path) + ".lock")
+        self.stream = None
+
+    def acquire(self):
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self.stream = self.path.open("a+b")
+        try:
+            if os.name == "nt":
+                import msvcrt
+                if self.path.stat().st_size == 0:
+                    self.stream.write(b"0")
+                    self.stream.flush()
+                self.stream.seek(0)
+                msvcrt.locking(self.stream.fileno(), msvcrt.LK_NBLCK, 1)
+            else:
+                import fcntl
+                fcntl.flock(self.stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError:
+            self.stream.close()
+            self.stream = None
+            raise RuntimeError("Another ComfyUI process is using this district checkpoint. Let it finish first.") from None
+
+    def release(self):
+        if self.stream is not None:
+            self.stream.close()
+            self.stream = None
+
+
+class _DistrictMirror:
+    def __init__(self, hf, b2, district, secrets=()):
+        self.hf, self.b2, self.district = hf, b2, district
+        self.secrets = tuple(value for value in secrets if value)
+        self.filename = "DICT_BLAZE_" + district + ".json"
+        self.url = _resolve_url(self.filename)
+        self.run_id = uuid.uuid4().hex
+        self.doc = {
+            "schema_version": 1, "repo_id": _REPO_ID, "district": district,
+            "dictionary": {}, "files": {}, "history": [], "source": {},
+            "backblaze": {}, "status": "starting", "created_at": _utc_now(),
+            "retry_policy": {"attempts": 8, "wait_seconds": [3, 5, 7, 9, 11, 13, 15], "honor_retry_after": True},
+            "links": {"private_bucket": True, "valid_seconds": _LINK_SECONDS,
+                      "scope": "district", "note": "Bearer URLs expire. Re-run this node to refresh. Stored objects do not expire."},
+        }
+        self.path = _key_id_cache_path().parent / "districts" / (
+            hashlib.sha256((_REPO_ID + ":" + b2.account_id).encode()).hexdigest()[:24] + "_" + self.filename)
+        self.remote_loaded = False
+        self.remote_digest = None
+        self.published = False
+        self.remote_existed = False
+        self.last_commit = None
+        self.last_checkpoint = 0.0
+        self.completed_since_checkpoint = 0
+        self.bucket = None
+        self.link_token = None
+        self.link_issued = 0.0
+        self.error = None
+        self.publish_error = None
+        self.local_error = None
+        self.run = {"run_id": self.run_id, "started_at": _utc_now(), "status": "starting",
+                    "added": [], "changed": [], "removed": [], "reused": 0, "processed": 0,
+                    "dictionary_commits": [], "inventory_complete": False}
+
+    def safe_error(self, exc):
+        if isinstance(exc, (_HFError, _RequestError, ValueError, RuntimeError, _MirrorInterrupted)):
+            text = str(exc)
+        else:
+            text = "Unexpected " + type(exc).__name__
+        for secret in sorted(self.secrets, key=len, reverse=True):
+            text = text.replace(secret, "[REDACTED]")
+        return text[:1500]
+
+    def read_remote(self, revision):
+        try:
+            data = self.hf.read_file(self.filename, revision, _MANIFEST_LIMIT)
+        except _HFError as exc:
+            if exc.status == 404:
+                return None, None
+            raise
+        try:
+            doc = json.loads(data)
+        except (ValueError, UnicodeError):
+            raise ValueError("The remote district dictionary is not valid JSON; it was not overwritten.") from None
+        return _validate_manifest(doc, self.district), hashlib.sha256(data).hexdigest()
+
+    def rebuild(self):
+        mapping = {}
+        counts = {"total": 0, "uploaded": 0, "stored": 0, "failed": 0, "pending": 0, "removed": 0}
+        now = time.time()
+        for path, entry in sorted(self.doc["files"].items()):
+            state = entry.get("status", "pending")
+            if state == "removed":
+                counts["removed"] += 1
+                continue
+            counts["total"] += 1
+            counts[state if state in counts else "pending"] += 1
+            b2 = entry.get("b2", {})
+            if (state == "uploaded" and b2.get("download_url")
+                    and b2.get("url_expires_at_unix", 0) > now):
+                mapping[entry["source_url"]] = b2["download_url"]
+        counts["mapped"] = len(mapping)
+        self.doc["dictionary"] = mapping
+        self.doc["counts"] = counts
+        self.doc["updated_at"] = _utc_now()
+        self.doc["run_id"] = self.run_id
+        history = [item for item in self.doc.get("history", []) if item.get("run_id") != self.run_id]
+        self.doc["history"] = (history + [copy.deepcopy(self.run)])[-100:]
+
+    def save_local(self, strict=False):
+        self.rebuild()
+        try:
+            _atomic_json(self.path, self.doc)
+            self.local_error = None
+        except OSError as exc:
+            self.local_error = "Local checkpoint could not be saved: " + type(exc).__name__
+            if strict:
+                raise RuntimeError(self.local_error) from None
+
+    def publish(self):
+        """Compare-and-swap; merge concurrent same-inventory successes, never overwrite a newer inventory."""
+        if not self.remote_loaded:
+            raise RuntimeError("Remote state was not read successfully; refusing to overwrite it.")
+        self.doc.pop("last_publication_error", None)
+        checkpoint_id = uuid.uuid4().hex
+        for conflict_attempt in range(8):
+            info = self.hf.info()
+            head = info["sha"]
+            remote, digest = self.read_remote(head)
+            if remote and remote.get("checkpoint_id") == checkpoint_id:
+                # The previous POST committed but its response was lost.
+                self.remote_digest = digest
+                self._published(head)
+                return
+            if digest != self.remote_digest:
+                if (not remote or remote.get("inventory_hash") != self.doc.get("inventory_hash")
+                        or remote.get("backblaze") != self.doc.get("backblaze")):
+                    raise _ConcurrentMirror("Another run changed this dictionary's source inventory or destination. Local progress is saved; run again to merge safely.")
+                _merge_successes(self.doc, remote)
+                known = {item.get("run_id") for item in self.doc.get("history", [])}
+                self.doc.setdefault("history", []).extend(item for item in remote.get("history", []) if item.get("run_id") not in known)
+            self.doc["checkpoint_id"] = checkpoint_id
+            self.doc["hf_head_before_checkpoint"] = head
+            self.rebuild()
+            payload = _json_bytes(self.doc)
+            try:
+                result = self.hf.commit_json(self.filename, payload, head,
+                    "Backblaze mirror " + self.district + ": " + self.doc["status"])
+            except _HFError as exc:
+                if exc.conflict and conflict_attempt < 7:
+                    continue
+                raise
+            commit = result.get("commitOid") or result.get("oid") or result.get("commit_id")
+            if not commit:
+                # Verify content using the same checkpoint ID rather than claiming success.
+                head = self.hf.info()["sha"]
+                check, _ = self.read_remote(head)
+                if not check or check.get("checkpoint_id") != checkpoint_id:
+                    raise RuntimeError("Hugging Face did not confirm the dictionary commit.")
+                commit = head
+            self.remote_digest = hashlib.sha256(payload).hexdigest()
+            self._published(commit)
+            return
+        raise RuntimeError("The dictionary changed repeatedly during publication. Local progress is saved.")
+
+    def _published(self, commit):
+        self.published = True
+        self.remote_existed = True
+        self.last_commit = commit
+        self.publish_error = None
+        self.run["dictionary_commits"].append(commit)
+        self.doc["last_confirmed_dictionary_commit"] = commit
+        self.last_checkpoint = time.monotonic()
+        self.completed_since_checkpoint = 0
+        self.save_local()
+
+    def reconcile(self, items, revision):
+        previous = self.doc["files"]
+        current = {}
+        for item in items:
+            path = item.get("path", "")
+            if not path.startswith(self.district + "/") or item.get("type") != "file":
+                continue
+            suffix = Path(path).suffix.lower()
+            if suffix not in _IMAGE_TYPES:
+                continue
+            fingerprint, size = _source_identity(item)
+            old = previous.get(path)
+            if old and old.get("source_fingerprint") == fingerprint and old.get("status") != "removed":
+                entry = copy.deepcopy(old)
+            else:
+                entry = {"status": "pending", "processing_attempts": 0, "last_error": None}
+                self.run["changed" if old else "added"].append(path)
+            entry.update({"source_fingerprint": fingerprint, "size": size,
+                          "source_url": _resolve_url(path), "source_pinned_url": _resolve_url(path, revision),
+                          "seen_in_commit": revision, "content_type": _IMAGE_TYPES[suffix]})
+            current[path] = entry
+        for path, entry in previous.items():
+            if path not in current:
+                removed = copy.deepcopy(entry)
+                if removed.get("status") != "removed":
+                    self.run["removed"].append(path)
+                removed.update({"status": "removed", "removed_in_commit": revision})
+                current[path] = removed
+        self.doc["files"] = current
+        self.doc["inventory_hash"] = _inventory_hash(current)
+        self.doc["source"].update({"snapshot_commit": revision, "inventory_complete": True,
+                                   "scanned_at": _utc_now(), "branch": "main"})
+        self.run["source_commit"] = revision
+        self.run["inventory_complete"] = True
+
+    def ensure_link(self, record):
+        if not self.link_token or time.time() >= self.link_issued + _LINK_SECONDS - 3600:
+            issued_before_request = time.time()
+            grant = self.b2.call("b2_get_download_authorization", {
+                "bucketId": self.bucket["bucketId"], "fileNamePrefix": "districts/" + self.district + "/",
+                "validDurationInSeconds": _LINK_SECONDS})
+            self.link_token = grant["authorizationToken"]
+            self.link_issued = issued_before_request
+        b2 = record["b2"]
+        base = self.b2.download_url + "/file/" + quote(self.bucket["bucketName"], safe="") + "/" + quote(b2["file_name"], safe="/")
+        b2.update({"download_url_base": base, "download_url": base + "?" + urlencode({"Authorization": self.link_token}),
+                   "url_issued_at_unix": self.link_issued, "url_expires_at_unix": self.link_issued + _LINK_SECONDS})
+
+    def process(self, path, entry, revision):
+        _check_cancel()
+        entry["processing_attempts"] = entry.get("processing_attempts", 0) + 1
+        entry["last_error"] = None
+        old = entry.get("b2", {})
+        valid = (old.get("bucket_id") == self.bucket["bucketId"] and
+                 str(old.get("file_name", "")).startswith("districts/" + self.district + "/") and
+                 old.get("sha1") and old.get("size") == entry["size"])
+        if valid and self.b2.file_exists(self.bucket["bucketId"], old["file_name"], old["sha1"], old["size"]):
+            self.run["reused"] += 1
+        else:
+            if entry["size"] > _MAX_PNG_BYTES:
+                raise ValueError("Image exceeds the supported 128 MiB per-file limit.")
+            data = self.hf.read_file(path, revision, _MAX_PNG_BYTES)
+            _verify_source_bytes(data, entry)
+            sha256 = hashlib.sha256(data).hexdigest()
+            name = "districts/" + self.district + "/" + sha256 + Path(path).suffix.lower()
+            uploaded = self.b2.ensure_file(self.bucket, name, data, entry["content_type"])
+            entry["b2"] = {"bucket_id": self.bucket["bucketId"], "bucket_name": self.bucket["bucketName"],
+                           "file_name": name, "file_id": (uploaded or {}).get("fileId"),
+                           "sha1": hashlib.sha1(data).hexdigest(), "sha256": sha256, "size": len(data)}
+            entry["uploaded_from_commit"] = revision
+        # Record successful storage BEFORE any link/token request can fail.
+        entry.update({"status": "stored", "updated_at": _utc_now()})
+        self.save_local(strict=True)
+        self.ensure_link(entry)
+        entry.update({"status": "uploaded", "last_error": None, "updated_at": _utc_now()})
+        self.run["processed"] += 1
+        self.completed_since_checkpoint += 1
+        self.save_local(strict=True)
+
+    def execute(self):
+        local = getattr(self, "local_resume", None)
+        recovery = _recovery_records(local)
+        if local and local.get("backblaze", {}).get("account_id") == self.b2.account_id:
+            self.doc = copy.deepcopy(local)
+        try:
+            info = self.hf.info()
+            revision = info["sha"]
+            remote, self.remote_digest = self.read_remote(revision)
+            self.remote_loaded = True
+            self.remote_existed = remote is not None
+            if remote:
+                self.doc = copy.deepcopy(remote)
+                if local:
+                    _merge_successes(self.doc, local)
+                    known_runs = {run.get("run_id") for run in self.doc.get("history", [])}
+                    self.doc.setdefault("history", []).extend(run for run in local.get("history", [])
+                                                              if run.get("run_id") not in known_runs)
+            # Retain local-only/newer-content uploads even if listing now fails.
+            # These are recovery candidates, never active mappings until inventory validation.
+            self.doc["recovery_records"] = recovery + self.doc.get("recovery_records", [])
+            self.run["attempted_source_commit"] = revision
+            self.doc["status"] = "scanning"
+            self.save_local()
+            # Listing must finish before files are added/changed/marked removed.
+            items = self.hf.tree(revision, self.district)
+            self.reconcile(items, revision)
+            try:
+                history = self.hf.history(revision)
+                self.doc["source"]["recent_commits"] = [
+                    {"id": item.get("id") or item.get("commit_id"), "title": item.get("title", ""),
+                     "date": item.get("date") or item.get("created_at")} for item in history[:100]]
+            except _HFError as exc:
+                self.run["history_warning"] = self.safe_error(exc)
+            self.bucket = self.b2.ensure_bucket()
+            destination = {"account_id": self.b2.account_id, "bucket_id": self.bucket["bucketId"],
+                           "bucket_name": self.bucket["bucketName"], "private": True}
+            if self.doc.get("backblaze") != destination:
+                for entry in self.doc["files"].values():
+                    entry.pop("b2", None)
+                    if entry.get("status") != "removed":
+                        entry["status"] = "pending"
+            self.doc["backblaze"] = destination
+            # Prefer additional successful local progress after a failed HF commit.
+            # The pre-run local state is retained separately, never read after overwriting.
+            if getattr(self, "local_resume", None):
+                _merge_successes(self.doc, self.local_resume)
+            for candidate in self.doc.get("recovery_records", []):
+                _merge_successes(self.doc, {"backblaze": candidate.get("backblaze"),
+                    "files": {candidate["path"]: candidate["record"]}})
+            self.doc.pop("recovery_records", None)
+            self.doc["status"] = self.run["status"] = "running"
+            self.save_local(strict=True)
+            # Test actual HF write permission before transferring any images.
+            self.publish()
+            paths = [path for path in sorted(self.doc["files"]) if self.doc["files"][path].get("status") != "removed"]
+            try:
+                from comfy.utils import ProgressBar
+                progress = ProgressBar(len(paths)) if paths else None
+            except ImportError:
+                progress = None
+            print("[Backblaze District] {} images found in {} at commit {}.".format(len(paths), self.district, revision))
+            for index, path in enumerate(paths, 1):
+                entry = self.doc["files"][path]
+                if progress is not None:
+                    progress.update_absolute(index - 1, len(paths))
+                if index == 1 or index % 10 == 0:
+                    print("[Backblaze District] Processing image {}/{}.".format(index, len(paths)))
+                try:
+                    self.process(path, entry, revision)
+                except _MirrorInterrupted:
+                    raise
+                except Exception as exc:
+                    entry.update({"status": "failed", "last_error": {"message": self.safe_error(exc),
+                                  "attempts": getattr(exc, "attempts", None), "at": _utc_now()}, "updated_at": _utc_now()})
+                    self.save_local()
+                    if isinstance(exc, ValueError) or (isinstance(exc, _HFError) and exc.status in (404, 413)):
+                        # One permanently bad/oversized source must not block the rest.
+                        self.run.setdefault("permanent_file_errors", []).append(path)
+                        self.completed_since_checkpoint += 1
+                        if (self.completed_since_checkpoint >= _CHECKPOINT_FILES or
+                                time.monotonic() - self.last_checkpoint >= _CHECKPOINT_SECONDS):
+                            self.publish()
+                        continue
+                    raise
+                if (self.completed_since_checkpoint >= _CHECKPOINT_FILES or
+                        time.monotonic() - self.last_checkpoint >= _CHECKPOINT_SECONDS):
+                    self.publish()
+            if progress is not None:
+                progress.update_absolute(len(paths), len(paths))
+            self.rebuild()
+            self.doc["status"] = self.run["status"] = (
+                "complete" if self.doc["counts"]["mapped"] == self.doc["counts"]["total"] else "partial")
+        except _MirrorInterrupted as exc:
+            self.error = self.safe_error(exc)
+            self.doc["status"] = self.run["status"] = "interrupted"
+        except _ConcurrentMirror as exc:
+            self.error = self.safe_error(exc)
+            self.doc["status"] = self.run["status"] = "conflict"
+        except Exception as exc:
+            self.error = self.safe_error(exc)
+            self.doc["status"] = self.run["status"] = "partial"
+        finally:
+            self.run["finished_at"] = _utc_now()
+            self.run["error"] = self.error
+            self.save_local()
+            if self.remote_loaded and self.doc["status"] != "conflict":
+                # A final dictionary save is attempted even after transfer errors/cancellation.
+                saved_check = self.hf._check_cancel
+                self.hf._check_cancel = lambda: None
+                try:
+                    self.publish()
+                except Exception as exc:
+                    self.publish_error = self.safe_error(exc)
+                    self.doc["last_publication_error"] = self.publish_error
+                    self.save_local()
+                finally:
+                    self.hf._check_cancel = saved_check
+        counts = self.doc.get("counts", {})
+        status = "{}: {}/{} image URLs ready; {} reused, {} failed, {} pending.".format(
+            self.doc["status"].upper(), counts.get("mapped", 0), counts.get("total", 0),
+            self.run["reused"], counts.get("failed", 0), counts.get("pending", 0))
+        if self.error:
+            status += "\n" + self.error
+        if self.publish_error or not self.published or self.doc["status"] == "conflict":
+            status += "\nLATEST DICTIONARY WAS NOT PUBLISHED. " + (self.publish_error or "Hugging Face state was unavailable.")
+        else:
+            status += "\nDictionary saved to Hugging Face. Commit: " + str(self.last_commit)
+        if self.local_error:
+            status += "\n" + self.local_error
+        else:
+            status += "\nLocal checkpoint: " + str(self.path)
+        status += "\nPrivate links expire after seven days; run again to refresh them."
+        self.rebuild()
+        return (self.url if self.remote_existed else "", status, _json_bytes(self.doc).decode("utf-8"))
+
+
 class BackblazePNGUpload:
     @classmethod
     def INPUT_TYPES(cls):
         return {"required": {
             "master_key": ("STRING", {"default": "", "multiline": False,
-                "tooltip": "Master application key SECRET. Legacy KEY_ID:SECRET also works. "
-                           "Remove this credential before sharing a saved workflow."}),
-            "png_url": ("STRING", {"default": _DEFAULT_PNG, "multiline": False,
-                "tooltip": "Public Hugging Face PNG URL. Bucket and upload setup are automatic."}),
+                "tooltip": "Backblaze master secret. KEY_ID:SECRET is also accepted."}),
+            "District_Name": ("STRING", {"default": "District_19_K", "multiline": False,
+                "tooltip": "Folder inside saliacoel/map_imgs_gpt. All image subfolders are included."}),
+            "hf_key": ("STRING", {"default": "", "multiline": False,
+                "tooltip": "Hugging Face token with write access to saliacoel/map_imgs_gpt."}),
         }, "optional": {
             "key_id": ("STRING", {"default": "", "multiline": False,
-                "tooltip": "Key ID (or account ID for a master key). Blank = use a previously "
-                           "verified saved ID, or the ID from KEY_ID:SECRET. First use of a bare "
-                           "secret requires this ID; Backblaze cannot retrieve it from the secret."}),
+                "tooltip": "Backblaze key ID. Blank uses ID:SECRET or a previously verified saved ID."}),
         }}
 
-    RETURN_TYPES = ("STRING", "STRING")
-    RETURN_NAMES = ("png_url", "STATUS")
+    RETURN_TYPES = ("STRING", "STRING", "STRING")
+    RETURN_NAMES = ("dictionary_url", "STATUS", "dictionary_json")
     FUNCTION = "upload"
     CATEGORY = "Salia/Backblaze"
     OUTPUT_NODE = True
-    DESCRIPTION = ("Uploads a Hugging Face PNG to an automatically created private Backblaze bucket. "
-                   "Returns a seven-day download URL and STATUS with the key ID. "
-                   "Blank key_id uses a saved ID; first-time bare secrets still require an ID.")
+    DESCRIPTION = "Mirror every district image to Backblaze and publish a resumable DICT_BLAZE JSON to Hugging Face."
 
     @classmethod
-    def IS_CHANGED(cls, master_key, png_url, key_id=""):
-        # A source URL can change; private download URLs must never stay cached forever.
+    def IS_CHANGED(cls, master_key, District_Name, hf_key, key_id=""):
         return float("nan")
 
-    def upload(self, master_key, png_url, key_id=""):
-        resolved_id = ""
-        source = ""
+    def upload(self, master_key, District_Name, hf_key, key_id=""):
+        mirror = None
+        acquired = False
+        lock = None
+        file_lock = None
         secret = ""
-        cache_note = ""
-        link = ""
         try:
+            district = District_Name.strip()
+            if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}", district) or district in (".", ".."):
+                raise ValueError("District_Name must be a single folder name, for example District_19_K.")
+            if not isinstance(hf_key, str) or not hf_key.strip() or any(c.isspace() for c in hf_key.strip()):
+                raise ValueError("hf_key must contain a Hugging Face token with write access to the repository.")
             resolved_id, secret, source = _credential_parts(master_key, key_id)
-            basic_auth = base64.b64encode((resolved_id + ":" + secret).encode("utf-8")).decode("ascii")
-            print("[Backblaze PNG] Downloading source PNG.")
-            data = _download_png(png_url)
-            name = "images/" + hashlib.sha256(data).hexdigest() + ".png"
-            client = _B2Client(basic_auth)
+            basic = base64.b64encode((resolved_id + ":" + secret).encode()).decode("ascii")
+            b2 = _B2Client(basic)
             cache_note = _remember_key_id(secret, resolved_id)
-            bucket = client.ensure_bucket()
-            client.ensure_file(bucket, name, data)
-            link = client.download_link(bucket, name)
-            status = "OK. Private download link is valid for seven days."
-        except _RequestError as exc:
-            hints = {
-                0: "Check your internet connection and TLS certificates.",
-                401: "Check the key ID, application key, and key permissions.",
-                403: "Check B2 account status, storage/transaction caps, and key permissions.",
-                429: "Backblaze is rate limiting requests. Run again later.",
-            }
-            status = "ERROR. Backblaze failed: {}. {}".format(
-                exc, hints.get(exc.status, "Check that B2 Cloud Storage is enabled, then try again."))
-        except (ValueError, RuntimeError) as exc:
-            status = "ERROR. " + str(exc)
+            with _DISTRICT_LOCKS_GUARD:
+                lock = _DISTRICT_LOCKS.setdefault((_REPO_ID, district), threading.Lock())
+            acquired = lock.acquire(blocking=False)
+            if not acquired:
+                raise RuntimeError("This district is already running in this ComfyUI process. Let it finish first.")
+            hf = _HFClient(hf_key.strip(), check_cancel=_check_cancel,
+                           on_retry=lambda attempt, seconds, error: print(
+                               "[Backblaze District] HF retry {}/8 in {} seconds ({}).".format(attempt + 1, seconds, error.code)))
+            mirror = _DistrictMirror(hf, b2, district, secrets=(master_key, secret, hf_key.strip()))
+            file_lock = _CheckpointLock(mirror.path)
+            file_lock.acquire()
+            try:
+                with mirror.path.open("r", encoding="utf-8") as stream:
+                    mirror.local_resume = _validate_manifest(json.load(stream), district)
+            except (OSError, ValueError, TypeError):
+                mirror.local_resume = None
+            url, status, json_text = mirror.execute()
+            status += "\nKEY_ID: {} ({})\n{}".format(resolved_id, source, cache_note)
         except Exception as exc:
-            # Do not expose unexpected exception payloads, which may contain tokens.
-            status = "ERROR. Unexpected {}. Check the inputs and try again.".format(type(exc).__name__)
-        if resolved_id:
-            status += "\nKEY_ID: {}\nID source: {}".format(resolved_id, source)
-        if cache_note:
-            status += "\n" + cache_note
-        # Defense in depth: status may include a provider error or an ID pasted by mistake.
-        sensitive = [secret]
-        if isinstance(master_key, str):
-            sensitive += [master_key.strip(), master_key.strip().split(":", 1)[-1]]
-        for value in sorted(set(sensitive), key=len, reverse=True):
+            url, json_text = "", _json_bytes(mirror.doc).decode() if mirror else "{}"
+            status = "ERROR. " + (str(exc) if isinstance(exc, (ValueError, RuntimeError, _HFError, _RequestError, _MirrorInterrupted)) else type(exc).__name__)
+        finally:
+            if file_lock is not None:
+                file_lock.release()
+            if acquired:
+                lock.release()
+        for value in sorted({str(master_key).strip(), secret, str(hf_key).strip()}, key=len, reverse=True):
             if value:
                 status = status.replace(value, "[REDACTED]")
-        print("[Backblaze PNG] " + status)
-        return {"ui": {"text": [status] + ([link] if link else [])}, "result": (link, status)}
+        print("[Backblaze District] " + status)
+        return {"ui": {"text": [status] + ([url] if url else [])}, "result": (url, status, json_text)}
 
 
 NODE_CLASS_MAPPINGS = {"BackblazePNGUpload": BackblazePNGUpload}
-NODE_DISPLAY_NAME_MAPPINGS = {"BackblazePNGUpload": "Backblaze PNG Upload (Auto)"}
+NODE_DISPLAY_NAME_MAPPINGS = {"BackblazePNGUpload": "Backblaze District Mirror (Auto / Resume)"}