history: add modified time check for file existence

Signed-off-by:  Eric Callahan <arksine.code@gmail.com>
This commit is contained in:
Eric Callahan 2023-06-01 05:50:31 -04:00
parent c41c5881c8
commit d6231634db
No known key found for this signature in database
GPG Key ID: 5A1EB336DFB4C71B
2 changed files with 20 additions and 5 deletions

View File

@ -15,6 +15,7 @@ import tempfile
import asyncio import asyncio
import zipfile import zipfile
import time import time
import math
from copy import deepcopy from copy import deepcopy
from inotify_simple import INotify from inotify_simple import INotify
from inotify_simple import flags as iFlags from inotify_simple import flags as iFlags
@ -346,10 +347,22 @@ class FileManager:
def get_metadata_storage(self) -> MetadataStorage: def get_metadata_storage(self) -> MetadataStorage:
return self.gcode_metadata return self.gcode_metadata
def check_file_exists(self, root: str, filename: str) -> bool: def check_file_exists(
root_dir = self.file_paths.get(root, "") self,
file_path = os.path.join(root_dir, filename) root: str,
return os.path.exists(file_path) filename: str,
modified: Optional[float] = None
) -> bool:
if root not in self.file_paths:
return False
root_dir = pathlib.Path(self.file_paths[root])
file_path = root_dir.joinpath(filename)
if file_path.is_file():
if modified is None:
return True
fstat = file_path.stat()
return math.isclose(fstat.st_mtime, modified)
return False
def can_access_path(self, path: StrOrPath) -> bool: def can_access_path(self, path: StrOrPath) -> bool:
if isinstance(path, str): if isinstance(path, str):

View File

@ -349,9 +349,11 @@ class History:
job: Dict[str, Any], job: Dict[str, Any],
job_id: str job_id: str
) -> Dict[str, Any]: ) -> Dict[str, Any]:
mtime = job.get("metadata", {}).get("modified", None)
job['job_id'] = job_id job['job_id'] = job_id
job['exists'] = self.file_manager.check_file_exists( job['exists'] = self.file_manager.check_file_exists(
"gcodes", job['filename']) "gcodes", job['filename'], mtime
)
return job return job
def on_exit(self) -> None: def on_exit(self) -> None: