84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
import os
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime
|
|
from os import DirEntry
|
|
from pathlib import Path
|
|
from stat import S_IFDIR, S_IFLNK, S_IFMT, S_IFREG
|
|
from typing import Literal
|
|
|
|
from . import config
|
|
|
|
_modes = {S_IFDIR: "d", S_IFREG: "f", S_IFLNK: "l"}
|
|
|
|
asdict = asdict
|
|
|
|
StatType = Literal["d", "f", "l", "-"]
|
|
|
|
|
|
@dataclass
|
|
class File:
|
|
id: int
|
|
parent_id: int
|
|
added: datetime
|
|
updated: datetime
|
|
location: str
|
|
hostname: str # XXX should better use a fingerprint/unique-id per host (e.g. `/etc/metadex.hostid`, for disks put it on their /)
|
|
stat_bytes: int
|
|
# stat_changed: datetime # XXX remove? The `ctime` changes not only for content changes but also file attr changes, which we don't track anyway.
|
|
stat_modified: datetime
|
|
stat_type: StatType
|
|
|
|
@classmethod
|
|
def from_direntry(cls, entry: DirEntry):
|
|
now = datetime.now()
|
|
pstat = entry.stat(follow_symlinks=False)
|
|
return cls(
|
|
added=now,
|
|
updated=now,
|
|
location=entry.path,
|
|
hostname=config.hostname,
|
|
stat_bytes=pstat.st_size,
|
|
# stat_changed=datetime.fromtimestamp(pstat.st_ctime),
|
|
stat_modified=datetime.fromtimestamp(pstat.st_mtime),
|
|
stat_type=_modes.get(S_IFMT(pstat.st_mode), "-"), # type: ignore
|
|
)
|
|
|
|
@classmethod
|
|
def from_path(cls, path: Path):
|
|
now = datetime.now()
|
|
pstat = os.stat(path, follow_symlinks=False)
|
|
return cls(
|
|
added=now,
|
|
updated=now,
|
|
location=os.path.abspath(path),
|
|
hostname=config.hostname,
|
|
stat_bytes=pstat.st_size,
|
|
# stat_changed=datetime.fromtimestamp(pstat.st_ctime),
|
|
stat_modified=datetime.fromtimestamp(pstat.st_mtime),
|
|
stat_type=_modes.get(S_IFMT(pstat.st_mode), "-"), # type: ignore
|
|
)
|
|
|
|
@staticmethod
|
|
def dict_from_entry(entry: "DirEntry | Path") -> dict:
|
|
"""Return the File's data structure as dict.
|
|
|
|
This can be useful to skip calling `asdict`, which can be quite slow.
|
|
"""
|
|
# now = datetime.now()
|
|
|
|
if isinstance(entry, Path):
|
|
location = os.path.abspath(entry)
|
|
pstat = os.stat(entry, follow_symlinks=False)
|
|
else:
|
|
location = entry.path.encode(errors="replace").decode()
|
|
pstat = entry.stat(follow_symlinks=False)
|
|
|
|
return dict(
|
|
# added=now,
|
|
# updated=now,
|
|
location=location,
|
|
hostname=config.hostname,
|
|
stat_bytes=pstat.st_size,
|
|
stat_modified=datetime.fromtimestamp(pstat.st_mtime),
|
|
stat_type=_modes.get(S_IFMT(pstat.st_mode), "-"),
|
|
)
|