diff --git a/backend/artifacts_registry.py b/backend/artifacts_registry.py new file mode 100644 index 0000000..9239ecb --- /dev/null +++ b/backend/artifacts_registry.py @@ -0,0 +1,70 @@ +import json +import os +from datetime import datetime +from pathlib import Path + +DATA_DIR = Path(os.getenv('DEVCONSOLE_DATA_DIR', '/var/lib/devconsole')) +ARTIFACTS_FILE = DATA_DIR / 'apk_artifacts.json' +MAX_ARTIFACTS_PER_WORKSPACE = 50 + + +def _read_all() -> list[dict]: + if not ARTIFACTS_FILE.exists(): + return [] + try: + data = json.loads(ARTIFACTS_FILE.read_text(encoding='utf-8')) + return data if isinstance(data, list) else [] + except Exception: + return [] + + +def _write_all(items: list[dict]) -> list[dict]: + ARTIFACTS_FILE.parent.mkdir(parents=True, exist_ok=True) + ARTIFACTS_FILE.write_text(json.dumps(items, ensure_ascii=False, indent=2), encoding='utf-8') + return items + + +def list_artifacts(workspace: str | None = None, limit: int = 20) -> list[dict]: + items = _read_all() + if workspace: + items = [item for item in items if item.get('workspace') == workspace] + return sorted(items, key=lambda item: item.get('created_at', ''), reverse=True)[:limit] + + +def record_apk_artifact( + workspace: str, + apk_path: str, + project_name: str | None = None, + version: str | None = None, + build: str | int | None = None, + ota_result: dict | None = None, + source: str = 'runtime', +) -> dict: + path = Path(apk_path) + stat = path.stat() if path.exists() else None + item = { + 'id': f"{datetime.utcnow().strftime('%Y%m%d%H%M%S')}-{abs(hash((workspace, apk_path))) % 1000000}", + 'workspace': workspace, + 'project_name': project_name or '', + 'apk_path': path.as_posix(), + 'apk_name': path.name, + 'size_bytes': stat.st_size if stat else 0, + 'version': version or '', + 'build': str(build or ''), + 'source': source, + 'created_at': datetime.utcnow().isoformat() + 'Z', + 'ota_success': bool((ota_result or {}).get('success')), + 'ota_latest_json': (ota_result or {}).get('latest_json'), + 'ota_uploaded': (ota_result or {}).get('uploaded') or [], + } + + items = _read_all() + items.append(item) + + workspace_items = [entry for entry in items if entry.get('workspace') == workspace] + workspace_items = sorted(workspace_items, key=lambda entry: entry.get('created_at', ''), reverse=True) + keep_ids = {entry.get('id') for entry in workspace_items[:MAX_ARTIFACTS_PER_WORKSPACE]} + items = [entry for entry in items if entry.get('workspace') != workspace or entry.get('id') in keep_ids] + + _write_all(items) + return item