🚧 Add FileSnapShot
This commit is contained in:
@@ -97,8 +97,6 @@ class ApplicationController:
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _records_from_status(self, entries: list[RawStatusEntry]) -> ChangeSet:
|
||||
from git_auto_tracker.git.status import snapshot_file_attributes
|
||||
|
||||
change_set = ChangeSet()
|
||||
root = self.config.repository_path
|
||||
for entry in entries:
|
||||
@@ -108,9 +106,7 @@ class ApplicationController:
|
||||
previous_path=entry.previous_path,
|
||||
)
|
||||
if entry.change_type is not ChangeType.DELETED:
|
||||
record.size, record.file_timestamp = self._file_attrs(
|
||||
entry.path, root, snapshot_file_attributes
|
||||
)
|
||||
record.size, record.file_timestamp = self._file_attrs(entry.path, root)
|
||||
bucket = {
|
||||
ChangeType.ADDED: change_set.added,
|
||||
ChangeType.MODIFIED: change_set.modified,
|
||||
@@ -121,19 +117,14 @@ class ApplicationController:
|
||||
return change_set
|
||||
|
||||
@staticmethod
|
||||
def _file_attrs(
|
||||
path: str,
|
||||
root: Path,
|
||||
snapshot: object,
|
||||
) -> tuple[int | None, dt.datetime | None]:
|
||||
from git_auto_tracker.git.status import snapshot_file_attributes as _snap
|
||||
def _file_attrs(path: str, root: Path) -> tuple[int | None, dt.datetime | None]:
|
||||
from git_auto_tracker.git.status import snapshot_file_attributes
|
||||
|
||||
_ = snapshot
|
||||
attrs = _snap(path, root)
|
||||
if not attrs.get("exists"):
|
||||
attrs = snapshot_file_attributes(path, root)
|
||||
if not attrs.exists or attrs.mtime_ns is None:
|
||||
return None, None
|
||||
ts = dt.datetime.fromtimestamp(attrs["mtime_ns"] / 1e9, tz=dt.UTC)
|
||||
return int(attrs["size"]), ts
|
||||
ts = dt.datetime.fromtimestamp(attrs.mtime_ns / 1e9, tz=dt.UTC)
|
||||
return attrs.size, ts
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Collection (Section 40)
|
||||
|
||||
@@ -105,16 +105,25 @@ class ChangeDetector:
|
||||
return parse_porcelain_z(raw)
|
||||
|
||||
|
||||
def snapshot_file_attributes(record_path: str, repository_root: Path) -> dict[str, object]:
|
||||
@dataclass(frozen=True)
|
||||
class FileSnapshot:
|
||||
"""Point-in-time file attributes used for Section 20 checks."""
|
||||
|
||||
exists: bool
|
||||
size: int | None
|
||||
mtime_ns: int | None
|
||||
|
||||
|
||||
def snapshot_file_attributes(record_path: str, repository_root: Path) -> FileSnapshot:
|
||||
"""Capture size/mtime of a file for concurrent-modification checks (Section 20)."""
|
||||
absolute = repository_root / Path(record_path)
|
||||
try:
|
||||
stat = absolute.stat()
|
||||
except OSError:
|
||||
return {"exists": False}
|
||||
return {"exists": True, "size": stat.st_size, "mtime_ns": stat.st_mtime_ns}
|
||||
return FileSnapshot(exists=False, size=None, mtime_ns=None)
|
||||
return FileSnapshot(exists=True, size=stat.st_size, mtime_ns=stat.st_mtime_ns)
|
||||
|
||||
|
||||
def attributes_changed(before: dict[str, object], after: dict[str, object]) -> bool:
|
||||
"""True if the file was modified between the two snapshots."""
|
||||
def attributes_changed(before: FileSnapshot, after: FileSnapshot) -> bool:
|
||||
"""True if the file was modified between the two snapshots (Section 20)."""
|
||||
return before != after
|
||||
|
||||
@@ -7,7 +7,6 @@ The log file must not live inside the tracked repository (Section 23).
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
import time
|
||||
import logging
|
||||
import logging.handlers
|
||||
import zoneinfo
|
||||
|
||||
Reference in New Issue
Block a user