Spaces:
Running
Running
| """Read management intents and corroborate their existing terminal receipts.""" | |
| import json | |
| from pathlib import Path | |
| import re | |
| TERMINAL_STATUSES = frozenset({'done', 'failed', 'rejected', 'stale'}) | |
| INTENT_FIELDS = ('request_id', 'submission_id', 'task_name', 'track_id', 'action', | |
| 'requested_by', 'hf_subject', 'researcher_identity_id', | |
| 'replace_submission_id', 'publication_sequence', 'requested_at') | |
| def read_request(path: str | Path) -> dict | None: | |
| path = Path(path) | |
| try: | |
| with path.open('rb') as handle: | |
| raw = handle.read(1_000_001) | |
| except FileNotFoundError: | |
| return None | |
| if len(raw) > 1_000_000: | |
| raise ValueError('Management request exceeds size limit') | |
| record = json.loads(raw) | |
| if (not isinstance(record, dict) or not record.get('submission_id') | |
| or record.get('request_id') != path.stem | |
| or not re.fullmatch(r'[A-Za-z0-9_-]+', path.stem)): | |
| raise ValueError('Invalid management request identity') | |
| return record | |
| def matching_terminal(request: dict, receipt: dict | None) -> bool: | |
| return (receipt is not None and receipt.get('status') in TERMINAL_STATUSES | |
| and all(request.get(key) == receipt.get(key) for key in INTENT_FIELDS)) | |
| def active_requests(pending: str | Path, processing: str | Path, complete: str | Path) -> list[tuple[Path, dict]]: | |
| found = {} | |
| for directory in (Path(pending), Path(processing)): | |
| try: | |
| paths = sorted(directory.iterdir()) | |
| except FileNotFoundError: | |
| continue | |
| for path in paths: | |
| if path.suffix != '.json' or path.name.startswith('.'): | |
| continue | |
| record = read_request(path) | |
| if record is None: | |
| continue | |
| receipt = read_request(Path(complete) / path.name) | |
| if matching_terminal(record, receipt): | |
| continue | |
| previous = found.get(record['request_id']) | |
| if previous and any(previous[1].get(key) != record.get(key) for key in INTENT_FIELDS): | |
| raise ValueError('Conflicting copies of a management request') | |
| found[record['request_id']] = (path, record) | |
| return list(found.values()) | |