feat(attachments): sha-addressed AttachmentStore
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,27 @@
|
|||||||
|
"""sha-addressed store for preserved non-art post files.
|
||||||
|
|
||||||
|
Arbitrary binaries — never opened or validated as images. Parallels the
|
||||||
|
thumbnail store layout (/images/attachments/<sha[:3]>/<sha><ext>).
|
||||||
|
"""
|
||||||
|
|
||||||
|
import shutil
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
|
class AttachmentStore:
|
||||||
|
def __init__(self, images_root: Path):
|
||||||
|
self.root = Path(images_root) / "attachments"
|
||||||
|
|
||||||
|
def store(self, src: Path, sha256: str) -> str:
|
||||||
|
"""Copy src into the sha-addressed store; idempotent on sha.
|
||||||
|
Returns the stored absolute path as a string."""
|
||||||
|
ext = Path(src).suffix.lower()
|
||||||
|
dest_dir = self.root / sha256[:3]
|
||||||
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
dest = dest_dir / f"{sha256}{ext}"
|
||||||
|
if dest.exists():
|
||||||
|
return str(dest)
|
||||||
|
partial = dest.with_suffix(dest.suffix + ".partial")
|
||||||
|
shutil.copy2(src, partial)
|
||||||
|
partial.rename(dest)
|
||||||
|
return str(dest)
|
||||||
@@ -0,0 +1,28 @@
|
|||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from backend.app.services.attachment_store import AttachmentStore
|
||||||
|
|
||||||
|
|
||||||
|
def test_store_layout_and_dedup(tmp_path):
|
||||||
|
src = tmp_path / "pack.zip"
|
||||||
|
src.write_bytes(b"\x00\x01binary\xff")
|
||||||
|
store = AttachmentStore(images_root=tmp_path / "images")
|
||||||
|
sha = "ab" + "c" * 62
|
||||||
|
|
||||||
|
p1 = store.store(src, sha)
|
||||||
|
assert p1.endswith(f"attachments/{sha[:3]}/{sha}.zip")
|
||||||
|
assert Path(p1).read_bytes() == b"\x00\x01binary\xff"
|
||||||
|
|
||||||
|
# second store of same sha: returns same path, no rewrite
|
||||||
|
before = Path(p1).stat().st_mtime_ns
|
||||||
|
p2 = store.store(src, sha)
|
||||||
|
assert p2 == p1
|
||||||
|
assert Path(p1).stat().st_mtime_ns == before
|
||||||
|
|
||||||
|
|
||||||
|
def test_store_preserves_extension_case_insensitive(tmp_path):
|
||||||
|
src = tmp_path / "Doc.PDF"
|
||||||
|
src.write_bytes(b"%PDF-1.4")
|
||||||
|
store = AttachmentStore(images_root=tmp_path / "images")
|
||||||
|
p = store.store(src, "f" * 64)
|
||||||
|
assert p.endswith(".pdf")
|
||||||
Reference in New Issue
Block a user