88efa2a6ae
Some checks failed
Build and Test / build (pull_request) Has been cancelled
Two code paths were appending silo/manifest.json to the ZIP without removing the previous entry, causing Python's zipfile module to warn about duplicate names: 1. slotFinishSaveDocument() re-injected the cached manifest from entries, then the modified_at update branch wrote a second copy. 2. update_manifest_fields() opened the ZIP in append mode and wrote an updated manifest without removing the old one. Fix slotFinishSaveDocument() by preparing the final manifest (with updated modified_at) in the entries dict before writing, so only one copy is written to the ZIP. Fix update_manifest_fields() by rewriting the ZIP atomically via a temp file, deduplicating any pre-existing duplicate entries in the process.
243 lines
8.1 KiB
Python
243 lines
8.1 KiB
Python
"""
|
|
kc_format.py — .kc file format support.
|
|
|
|
Handles two responsibilities:
|
|
1. Round-trip preservation: caches silo/ ZIP entries before FreeCAD's C++
|
|
save rewrites the ZIP from scratch, then re-injects them after save.
|
|
2. Manifest auto-creation: ensures every .kc file has silo/manifest.json.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import uuid
|
|
import zipfile
|
|
from datetime import datetime, timezone
|
|
|
|
import FreeCAD
|
|
|
|
# Cache: filepath -> {entry_name: bytes}
|
|
_silo_cache = {}
|
|
|
|
# Pre-reinject hooks: called with (doc, filename, entries) before ZIP write.
|
|
_pre_reinject_hooks = []
|
|
|
|
|
|
def register_pre_reinject(callback):
|
|
"""Register a callback invoked before silo/ entries are written to ZIP.
|
|
|
|
Signature: callback(doc, filename, entries) -> None
|
|
``entries`` is a dict {entry_name: bytes} or None. Mutate in place.
|
|
"""
|
|
_pre_reinject_hooks.append(callback)
|
|
|
|
|
|
def _metadata_save_hook(doc, filename, entries):
|
|
"""Write dirty metadata back to the silo/ cache before ZIP write."""
|
|
obj = doc.getObject("SiloMetadata")
|
|
if obj is None or not hasattr(obj, "Proxy"):
|
|
return
|
|
proxy = obj.Proxy
|
|
if proxy is None or not proxy.is_dirty():
|
|
return
|
|
entries["silo/metadata.json"] = obj.RawContent.encode("utf-8")
|
|
|
|
|
|
register_pre_reinject(_metadata_save_hook)
|
|
|
|
|
|
def _manifest_enrich_hook(doc, filename, entries):
|
|
"""Populate silo_instance and part_uuid from the tracked Silo object."""
|
|
raw = entries.get("silo/manifest.json")
|
|
if raw is None:
|
|
return
|
|
try:
|
|
manifest = json.loads(raw)
|
|
except (json.JSONDecodeError, ValueError):
|
|
return
|
|
|
|
changed = False
|
|
|
|
# Populate part_uuid from SiloItemId if available.
|
|
for obj in doc.Objects:
|
|
if hasattr(obj, "SiloItemId") and obj.SiloItemId:
|
|
if manifest.get("part_uuid") != obj.SiloItemId:
|
|
manifest["part_uuid"] = obj.SiloItemId
|
|
changed = True
|
|
break
|
|
|
|
# Populate silo_instance from Silo settings.
|
|
if not manifest.get("silo_instance"):
|
|
try:
|
|
import silo_commands
|
|
|
|
api_url = silo_commands._get_api_url()
|
|
if api_url:
|
|
# Strip /api suffix to get base instance URL.
|
|
instance = api_url.rstrip("/")
|
|
if instance.endswith("/api"):
|
|
instance = instance[:-4]
|
|
manifest["silo_instance"] = instance
|
|
changed = True
|
|
except Exception:
|
|
pass
|
|
|
|
if changed:
|
|
entries["silo/manifest.json"] = (json.dumps(manifest, indent=2) + "\n").encode(
|
|
"utf-8"
|
|
)
|
|
|
|
|
|
register_pre_reinject(_manifest_enrich_hook)
|
|
|
|
|
|
def _solver_context_hook(doc, filename, entries):
|
|
"""Pack solver context into silo/solver/context.json for assemblies."""
|
|
try:
|
|
for obj in doc.Objects:
|
|
if obj.TypeId == "Assembly::AssemblyObject":
|
|
ctx = obj.getSolveContext()
|
|
if ctx: # non-empty means we have grounded parts
|
|
entries["silo/solver/context.json"] = (
|
|
json.dumps(ctx, indent=2) + "\n"
|
|
).encode("utf-8")
|
|
break # one assembly per document
|
|
except Exception as exc:
|
|
FreeCAD.Console.PrintWarning(f"kc_format: solver context hook failed: {exc}\n")
|
|
|
|
|
|
register_pre_reinject(_solver_context_hook)
|
|
|
|
|
|
KC_VERSION = "1.0"
|
|
|
|
|
|
def _default_manifest():
|
|
"""Generate a default silo/manifest.json for new .kc files."""
|
|
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
username = os.environ.get("USER", os.environ.get("USERNAME", "unknown"))
|
|
return {
|
|
"kc_version": KC_VERSION,
|
|
"silo_instance": None,
|
|
"part_uuid": str(uuid.uuid4()),
|
|
"revision_hash": None,
|
|
"created_at": now,
|
|
"modified_at": now,
|
|
"created_by": username,
|
|
}
|
|
|
|
|
|
class _KcFormatObserver:
|
|
"""Document observer that preserves silo/ entries across saves."""
|
|
|
|
def slotStartSaveDocument(self, doc, filename):
|
|
"""Before save: cache silo/ entries from the existing file."""
|
|
if not filename.lower().endswith(".kc"):
|
|
return
|
|
if not os.path.isfile(filename):
|
|
return
|
|
try:
|
|
with zipfile.ZipFile(filename, "r") as zf:
|
|
entries = {}
|
|
for name in zf.namelist():
|
|
if name.startswith("silo/"):
|
|
entries[name] = zf.read(name)
|
|
if entries:
|
|
_silo_cache[filename] = entries
|
|
except Exception:
|
|
pass
|
|
|
|
def slotFinishSaveDocument(self, doc, filename):
|
|
"""After save: re-inject cached silo/ entries and ensure manifest."""
|
|
if not filename.lower().endswith(".kc"):
|
|
_silo_cache.pop(filename, None)
|
|
return
|
|
entries = _silo_cache.pop(filename, None)
|
|
if entries is None:
|
|
entries = {}
|
|
for _hook in _pre_reinject_hooks:
|
|
try:
|
|
_hook(doc, filename, entries)
|
|
except Exception as exc:
|
|
FreeCAD.Console.PrintWarning(
|
|
f"kc_format: pre_reinject hook failed: {exc}\n"
|
|
)
|
|
try:
|
|
# Ensure silo/manifest.json exists in entries and update modified_at.
|
|
# All manifest mutations happen here so only one copy is written.
|
|
if "silo/manifest.json" in entries:
|
|
try:
|
|
manifest = json.loads(entries["silo/manifest.json"])
|
|
except (json.JSONDecodeError, ValueError):
|
|
manifest = _default_manifest()
|
|
else:
|
|
manifest = _default_manifest()
|
|
manifest["modified_at"] = datetime.now(timezone.utc).strftime(
|
|
"%Y-%m-%dT%H:%M:%SZ"
|
|
)
|
|
entries["silo/manifest.json"] = (
|
|
json.dumps(manifest, indent=2) + "\n"
|
|
).encode("utf-8")
|
|
|
|
with zipfile.ZipFile(filename, "a") as zf:
|
|
existing = set(zf.namelist())
|
|
for name, data in entries.items():
|
|
if name not in existing:
|
|
zf.writestr(name, data)
|
|
existing.add(name)
|
|
except Exception as e:
|
|
FreeCAD.Console.PrintWarning(
|
|
f"kc_format: failed to update .kc silo/ entries: {e}\n"
|
|
)
|
|
|
|
|
|
def update_manifest_fields(filename, updates):
|
|
"""Update fields in an existing .kc manifest after save.
|
|
|
|
*filename*: path to the .kc file.
|
|
*updates*: dict of field_name -> value to merge into the manifest.
|
|
|
|
Used by silo_commands to write ``revision_hash`` after a successful
|
|
upload (which happens after the ZIP has already been written by save).
|
|
"""
|
|
if not filename or not filename.lower().endswith(".kc"):
|
|
return
|
|
if not os.path.isfile(filename):
|
|
return
|
|
import shutil
|
|
import tempfile
|
|
|
|
try:
|
|
fd, tmp = tempfile.mkstemp(suffix=".kc", dir=os.path.dirname(filename))
|
|
os.close(fd)
|
|
try:
|
|
with (
|
|
zipfile.ZipFile(filename, "r") as zin,
|
|
zipfile.ZipFile(tmp, "w", compression=zipfile.ZIP_DEFLATED) as zout,
|
|
):
|
|
found = False
|
|
for item in zin.infolist():
|
|
if item.filename == "silo/manifest.json":
|
|
if found:
|
|
continue # skip duplicate entries
|
|
found = True
|
|
raw = zin.read(item.filename)
|
|
manifest = json.loads(raw)
|
|
manifest.update(updates)
|
|
zout.writestr(
|
|
item.filename,
|
|
json.dumps(manifest, indent=2) + "\n",
|
|
)
|
|
else:
|
|
zout.writestr(item, zin.read(item.filename))
|
|
shutil.move(tmp, filename)
|
|
except BaseException:
|
|
os.unlink(tmp)
|
|
raise
|
|
except Exception as e:
|
|
FreeCAD.Console.PrintWarning(f"kc_format: failed to update manifest: {e}\n")
|
|
|
|
|
|
def register():
|
|
"""Connect to application-level save signals."""
|
|
FreeCAD.addDocumentObserver(_KcFormatObserver())
|