Upload folder using huggingface_hub (part 2)

bc50ea0 verified 30 days ago

8.74 kB

"""Recover interrupted PNG publication without touching earlier scene frames.
The journal is written only after a whole scene has been staged and synced.
It grants ownership of exactly the next frame range, never arbitrary PNGs.
All calls are made under the export folder's process lock.
"""

from contextlib import contextmanager
from copy import deepcopy
import json
from pathlib import Path
import re
import shutil
import tempfile
import uuid

from . import processing_persistence as persistence

PENDING = ".png_pending.json"
FORMAT = "h3_png_publication_v1"


def _same_record(left, right):
    # A fallback copy changes stat hints, not scene identity or pixel bytes.
    def identity(record):
        record = deepcopy(record)
        if isinstance(record, dict):
            for clip in record.get("clips", []):
                for item in clip.get("files", []):
                    item.pop("mtime_ns", None)
        return record
    return identity(left) == identity(right)


def _cleanup_stage(chain, stage):
    if any(stage.glob("conflict_*")):
        chain._LOG.warning("H3 PNG recovery preserved conflicting interrupted files in %s", stage)
        return
    shutil.rmtree(stage, ignore_errors=True)


@contextmanager
def staging(chain, directory, index):
    stage = Path(tempfile.mkdtemp(prefix=".png_scene_%04d_" % index, dir=directory))
    try:
        yield stage
    finally:
        # Keep staged pixels if publication or its acknowledgement is uncertain.
        # On restart the journal can finish the scene without decoding again.
        try:
            pending = (directory / PENDING).exists()
        except OSError:
            pending = True
        if not pending:
            _cleanup_stage(chain, stage)


def _matches(chain, path, item):
    return path.is_file() and path.stat().st_size == item["size"] and chain._file_sha256(str(path)) == item["sha256"]


def _read(path):
    return json.loads(path.read_text(encoding="utf-8")) if path.exists() else None


def _finish(chain, root, directory, journal, safe_path, publish_frame, recovering=False):
    stage = safe_path(root, directory / journal["stage"])
    record = deepcopy(journal["record"])
    for item in record["clips"][-1]["files"]:
        chain._png_export_check_interrupted()
        target = safe_path(root, directory / item["file"])
        source = safe_path(root, stage / item["file"])
        if recovering and _matches(chain, target, item):
            pass
        else:
            if not _matches(chain, source, item):
                raise ValueError("Interrupted PNG staging is missing or changed: %s. Saved scenes were kept." % source)
            if target.exists():
                if not recovering or not target.is_file():
                    raise FileExistsError("PNG publication will not overwrite %s" % target)
                # A killed network copy can leave a short final file. Preserve
                # any conflicting bytes (including manual edits), then recover
                # from the verified complete staging file. Never delete them.
                preserved = safe_path(root, stage / ("conflict_%s_%s" % (uuid.uuid4().hex, target.name)))
                target.rename(preserved)
                persistence.sync_directory(stage)
                persistence.sync_directory(directory)
                chain._LOG.warning("H3 PNG recovery preserved an interrupted/conflicting file at %s", preserved)
            publish_frame(source, target)
        item["mtime_ns"] = target.stat().st_mtime_ns
    persistence.sync_directory(directory)
    persistence.atomic_json(directory / "export.json", record)
    (directory / PENDING).unlink()
    persistence.sync_directory(directory)
    _cleanup_stage(chain, stage)
    return record


def publish(chain, root, directory, stage, previous, record, safe_path, publish_frame):
    pending = safe_path(root, directory / PENDING)
    if pending.exists():
        raise ValueError("Recover the pending PNG publication before appending another scene.")
    for item in record["clips"][-1]["files"]:
        if safe_path(root, directory / item["file"]).exists():
            raise FileExistsError("PNG publication will not overwrite %s" % item["file"])
    persistence.sync_directory(stage)
    persistence.sync_directory(directory)
    journal = {"format": FORMAT, "stage": stage.name,
               "previous": previous, "record": record}
    persistence.atomic_json(pending, journal)
    try:
        return _finish(chain, root, directory, journal, safe_path, publish_frame)
    except BaseException:
        # Normal cancellation rolls back only verified files from this attempt.
        # An uncertain index commit / unreachable share retains the journal and
        # media. In particular, never undo a successfully replaced export.json.
        try:
            current = _read(safe_path(root, directory / "export.json"))
            if not _same_record(current, record) and _same_record(current, previous):
                for item in record["clips"][-1]["files"]:
                    target = safe_path(root, directory / item["file"])
                    if target.exists():
                        if not _matches(chain, target, item):
                            raise ValueError("Uncertain interrupted PNG publication")
                        target.unlink()
                persistence.sync_directory(directory)
                pending.unlink(missing_ok=True)
                persistence.sync_directory(directory)
        except (OSError, ValueError):
            chain._LOG.warning("H3 PNG publication will be recovered on retry: %s", pending)
        raise


def recover(chain, root, directory, previous, config, contracts, safe_path, publish_frame):
    pending = safe_path(root, directory / PENDING)
    journal = _read(pending)
    if journal is None:
        return previous
    if (not isinstance(journal, dict) or journal.get("format") != FORMAT
            or not re.fullmatch(r"\.png_scene_[0-9]{4,}_[a-zA-Z0-9_-]+", str(journal.get("stage", "")))):
        raise ValueError("Invalid pending PNG journal; existing files were kept.")
    stage = safe_path(root, directory / journal["stage"])
    record, base = journal.get("record"), journal.get("previous")
    if (not isinstance(record, dict) or record.get("format") != "h3_video_png_sequence_v1"
            or record.get("settings") != config or not isinstance(record.get("clips"), list)
            or not record["clips"] or (base is not None and not isinstance(base, dict))):
        raise ValueError("Pending PNG scene belongs to different settings; existing files were kept.")
    old_clips = (base or {}).get("clips", [])
    if not isinstance(old_clips, list):
        raise ValueError("Invalid pending PNG prefix; existing files were kept.")
    clip = record["clips"][-1]
    first = config["first_frame_number"] + sum(c["delivered_frames"] for c in old_clips)
    if (record["clips"][:-1] != old_clips or not isinstance(clip, dict)
            or clip.get("source_contract") != contracts.get(clip.get("index"))
            or (old_clips and clip["index"] != old_clips[-1]["index"] + 1)
            or clip.get("first_frame_number") != first
            or not isinstance(clip.get("files"), list)
            or len(clip["files"]) != clip.get("delivered_frames")
            or not clip["files"]
            or not journal["stage"].startswith(".png_scene_%04d_" % clip["index"])
            or record.get("frame_count") != first + len(clip["files"]) - config["first_frame_number"]):
        raise ValueError("Pending PNG scene has a different branch/order; existing files were kept.")
    for offset, item in enumerate(clip["files"]):
        if (not isinstance(item, dict) or item.get("file") != "frame_%08d.png" % (first + offset)
                or not re.fullmatch(r"[0-9a-f]{64}", str(item.get("sha256", "")))
                or not isinstance(item.get("size"), int) or item["size"] <= 0):
            raise ValueError("Invalid pending PNG frame address; existing files were kept.")
    if _same_record(previous, record):
        # The index committed before the process stopped; normal validation has
        # already checked every published PNG. Only private staging is obsolete.
        pending.unlink()
        persistence.sync_directory(directory)
        _cleanup_stage(chain, stage)
        return previous
    if not _same_record(previous, base):
        raise ValueError("PNG export changed since the interrupted publication; existing files were kept.")
    result = _finish(chain, root, directory, journal, safe_path, publish_frame, recovering=True)
    chain._LOG.info("H3 PNG recovered committed scene %d from its publication journal", clip["index"])
    return result

	"""Recover interrupted PNG publication without touching earlier scene frames.

	The journal is written only after a whole scene has been staged and synced.
	It grants ownership of exactly the next frame range, never arbitrary PNGs.
	All calls are made under the export folder's process lock.
	"""

	from contextlib import contextmanager
	from copy import deepcopy
	import json
	from pathlib import Path
	import re
	import shutil
	import tempfile
	import uuid

	from . import processing_persistence as persistence

	PENDING = ".png_pending.json"
	FORMAT = "h3_png_publication_v1"


	def _same_record(left, right):
	# A fallback copy changes stat hints, not scene identity or pixel bytes.
	def identity(record):
	record = deepcopy(record)
	if isinstance(record, dict):
	for clip in record.get("clips", []):
	for item in clip.get("files", []):
	item.pop("mtime_ns", None)
	return record
	return identity(left) == identity(right)


	def _cleanup_stage(chain, stage):
	if any(stage.glob("conflict_*")):
	chain._LOG.warning("H3 PNG recovery preserved conflicting interrupted files in %s", stage)
	return
	shutil.rmtree(stage, ignore_errors=True)


	@contextmanager
	def staging(chain, directory, index):
	stage = Path(tempfile.mkdtemp(prefix=".png_scene_%04d_" % index, dir=directory))
	try:
	yield stage
	finally:
	# Keep staged pixels if publication or its acknowledgement is uncertain.
	# On restart the journal can finish the scene without decoding again.
	try:
	pending = (directory / PENDING).exists()
	except OSError:
	pending = True
	if not pending:
	_cleanup_stage(chain, stage)


	def _matches(chain, path, item):
	return path.is_file() and path.stat().st_size == item["size"] and chain._file_sha256(str(path)) == item["sha256"]


	def _read(path):
	return json.loads(path.read_text(encoding="utf-8")) if path.exists() else None


	def _finish(chain, root, directory, journal, safe_path, publish_frame, recovering=False):
	stage = safe_path(root, directory / journal["stage"])
	record = deepcopy(journal["record"])
	for item in record["clips"][-1]["files"]:
	chain._png_export_check_interrupted()
	target = safe_path(root, directory / item["file"])
	source = safe_path(root, stage / item["file"])
	if recovering and _matches(chain, target, item):
	pass
	else:
	if not _matches(chain, source, item):
	raise ValueError("Interrupted PNG staging is missing or changed: %s. Saved scenes were kept." % source)
	if target.exists():
	if not recovering or not target.is_file():
	raise FileExistsError("PNG publication will not overwrite %s" % target)
	# A killed network copy can leave a short final file. Preserve
	# any conflicting bytes (including manual edits), then recover
	# from the verified complete staging file. Never delete them.
	preserved = safe_path(root, stage / ("conflict_%s_%s" % (uuid.uuid4().hex, target.name)))
	target.rename(preserved)
	persistence.sync_directory(stage)
	persistence.sync_directory(directory)
	chain._LOG.warning("H3 PNG recovery preserved an interrupted/conflicting file at %s", preserved)
	publish_frame(source, target)
	item["mtime_ns"] = target.stat().st_mtime_ns
	persistence.sync_directory(directory)
	persistence.atomic_json(directory / "export.json", record)
	(directory / PENDING).unlink()
	persistence.sync_directory(directory)
	_cleanup_stage(chain, stage)
	return record


	def publish(chain, root, directory, stage, previous, record, safe_path, publish_frame):
	pending = safe_path(root, directory / PENDING)
	if pending.exists():
	raise ValueError("Recover the pending PNG publication before appending another scene.")
	for item in record["clips"][-1]["files"]:
	if safe_path(root, directory / item["file"]).exists():
	raise FileExistsError("PNG publication will not overwrite %s" % item["file"])
	persistence.sync_directory(stage)
	persistence.sync_directory(directory)
	journal = {"format": FORMAT, "stage": stage.name,
	"previous": previous, "record": record}
	persistence.atomic_json(pending, journal)
	try:
	return _finish(chain, root, directory, journal, safe_path, publish_frame)
	except BaseException:
	# Normal cancellation rolls back only verified files from this attempt.
	# An uncertain index commit / unreachable share retains the journal and
	# media. In particular, never undo a successfully replaced export.json.
	try:
	current = _read(safe_path(root, directory / "export.json"))
	if not _same_record(current, record) and _same_record(current, previous):
	for item in record["clips"][-1]["files"]:
	target = safe_path(root, directory / item["file"])
	if target.exists():
	if not _matches(chain, target, item):
	raise ValueError("Uncertain interrupted PNG publication")
	target.unlink()
	persistence.sync_directory(directory)
	pending.unlink(missing_ok=True)
	persistence.sync_directory(directory)
	except (OSError, ValueError):
	chain._LOG.warning("H3 PNG publication will be recovered on retry: %s", pending)
	raise


	def recover(chain, root, directory, previous, config, contracts, safe_path, publish_frame):
	pending = safe_path(root, directory / PENDING)
	journal = _read(pending)
	if journal is None:
	return previous
	if (not isinstance(journal, dict) or journal.get("format") != FORMAT
	or not re.fullmatch(r"\.png_scene_[0-9]{4,}_[a-zA-Z0-9_-]+", str(journal.get("stage", "")))):
	raise ValueError("Invalid pending PNG journal; existing files were kept.")
	stage = safe_path(root, directory / journal["stage"])
	record, base = journal.get("record"), journal.get("previous")
	if (not isinstance(record, dict) or record.get("format") != "h3_video_png_sequence_v1"
	or record.get("settings") != config or not isinstance(record.get("clips"), list)
	or not record["clips"] or (base is not None and not isinstance(base, dict))):
	raise ValueError("Pending PNG scene belongs to different settings; existing files were kept.")
	old_clips = (base or {}).get("clips", [])
	if not isinstance(old_clips, list):
	raise ValueError("Invalid pending PNG prefix; existing files were kept.")
	clip = record["clips"][-1]
	first = config["first_frame_number"] + sum(c["delivered_frames"] for c in old_clips)
	if (record["clips"][:-1] != old_clips or not isinstance(clip, dict)
	or clip.get("source_contract") != contracts.get(clip.get("index"))
	or (old_clips and clip["index"] != old_clips[-1]["index"] + 1)
	or clip.get("first_frame_number") != first
	or not isinstance(clip.get("files"), list)
	or len(clip["files"]) != clip.get("delivered_frames")
	or not clip["files"]
	or not journal["stage"].startswith(".png_scene_%04d_" % clip["index"])
	or record.get("frame_count") != first + len(clip["files"]) - config["first_frame_number"]):
	raise ValueError("Pending PNG scene has a different branch/order; existing files were kept.")
	for offset, item in enumerate(clip["files"]):
	if (not isinstance(item, dict) or item.get("file") != "frame_%08d.png" % (first + offset)
	or not re.fullmatch(r"[0-9a-f]{64}", str(item.get("sha256", "")))
	or not isinstance(item.get("size"), int) or item["size"] <= 0):
	raise ValueError("Invalid pending PNG frame address; existing files were kept.")
	if _same_record(previous, record):
	# The index committed before the process stopped; normal validation has
	# already checked every published PNG. Only private staging is obsolete.
	pending.unlink()
	persistence.sync_directory(directory)
	_cleanup_stage(chain, stage)
	return previous
	if not _same_record(previous, base):
	raise ValueError("PNG export changed since the interrupted publication; existing files were kept.")
	result = _finish(chain, root, directory, journal, safe_path, publish_frame, recovering=True)
	chain._LOG.info("H3 PNG recovered committed scene %d from its publication journal", clip["index"])
	return result