Initial commit for finalisation web interface

copienator/evaluation_lock.py

@@ -0,0 +1,67 @@
+"""OS-backed evaluation lease; released automatically on process exit."""
+from __future__ import annotations
+
+import json
+import os
+import secrets
+from pathlib import Path
+
+
+class EvaluationLock:
+    def __init__(self, root: Path):
+        self.path = root / '.copienator-finalisation.lock'
+        self.stream = None
+        self.token = secrets.token_urlsafe(32)
+
+    def __enter__(self):
+        from copienator import CliError
+        self.stream = self.path.open('a+b')
+        try:
+            if os.name == 'nt':
+                import msvcrt
+                self.stream.seek(0)
+                if not self.stream.read(1):
+                    self.stream.write(b' ')
+                    self.stream.flush()
+                self.stream.seek(0)
+                msvcrt.locking(self.stream.fileno(), msvcrt.LK_NBLCK, 1)
+            else:
+                import fcntl
+                fcntl.flock(self.stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError:
+            self.stream.close()
+            self.stream = None
+            raise CliError('Cette évaluation est utilisée par un autre traitement ou la finalisation web.') from None
+        self.stream.seek(0)
+        self.stream.truncate()
+        self.stream.write(json.dumps({'pid': os.getpid(), 'token': self.token}).encode())
+        self.stream.flush()
+        return self
+
+    def __exit__(self, *args):
+        if self.stream:
+            self.stream.close()
+            self.stream = None
+
+
+def command_lock(args):
+    from contextlib import nullcontext
+    from copienator import EvaluationWorkspace
+    root = getattr(args, 'evaluation', None)
+    if root is None and getattr(args, 'target', None):
+        try:
+            root = EvaluationWorkspace.discover(Path(args.target).resolve()).root
+        except Exception:
+            return nullcontext()
+    if root is None or not Path(root).is_dir():
+        return nullcontext()
+    root = Path(root).resolve()
+    token = os.environ.get('COPIENATOR_FINALISATION_TOKEN')
+    if token:
+        try:
+            data = json.loads((root / '.copienator-finalisation.lock').read_text())
+            if secrets.compare_digest(token, data['token']):
+                return nullcontext()
+        except (OSError, ValueError, KeyError):
+            pass
+    return EvaluationLock(root)