50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
import logging
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Dict, List
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class DirectoryMonitor:
|
|
dir: Path
|
|
glob: str = '*'
|
|
times: Dict[Path, datetime] = field(default_factory=dict, repr=False)
|
|
changed: List[Path] = field(init=False, repr=False)
|
|
new: List[Path] = field(init=False, repr=False)
|
|
deleted: List[Path] = field(init=False, repr=False)
|
|
last_checked: datetime = field(default_factory=datetime.now)
|
|
|
|
def __post_init__(self):
|
|
self.update()
|
|
|
|
def update(self):
|
|
self.last_checked = datetime.now()
|
|
self.new, self.changed, self.deleted = [], [], []
|
|
|
|
# iterate through the directory and update the stored times
|
|
for file in self.dir.rglob(self.glob):
|
|
if file.is_dir():
|
|
logger.debug(f'Skipping directory: {file}')
|
|
continue
|
|
|
|
modified_time = datetime.fromtimestamp(file.stat().st_mtime)
|
|
if file in self.times:
|
|
if modified_time > self.times[file]:
|
|
logger.info(f'Updated file: {file}')
|
|
self.changed.append(file)
|
|
else:
|
|
logger.debug(f'New file: {file}')
|
|
self.new.append(file)
|
|
|
|
self.times[file] = modified_time
|
|
|
|
# remove any entries for files that no longer exist
|
|
for file in list(self.times.keys()):
|
|
if not file.exists():
|
|
logger.warning(f'Detected deleted file: {file}')
|
|
del self.times[file]
|
|
self.deleted.append(file)
|