arch_agent
adee5dfc78
- FastAPI server with scan/build/repo endpoints - LLM scanner (Ollama) with regex pre-scan - makepkg/devtools builder with chroot isolation - Scan cache with TTL + PKGBUILD hash - Client installer + safe-yay wrapper - Docs + config example
78 lines
2.2 KiB
Python
78 lines
2.2 KiB
Python
"""Scan cache — stores scan results to avoid re-scanning unchanged packages."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from dataclasses import dataclass, asdict
|
|
from pathlib import Path
|
|
|
|
from .scanner import ScanResult, ScanVerdict
|
|
|
|
|
|
@dataclass
|
|
class CacheEntry:
|
|
package: str
|
|
verdict: str
|
|
confidence: float
|
|
findings: list[str]
|
|
reasoning: str
|
|
timestamp: float
|
|
pkgbuild_hash: str
|
|
|
|
|
|
class ScanCache:
|
|
"""JSON-file based cache for scan results."""
|
|
|
|
def __init__(self, cache_dir: str, ttl_hours: int = 168) -> None:
|
|
self.cache_dir = Path(cache_dir)
|
|
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
|
self.ttl_seconds = ttl_hours * 3600
|
|
|
|
def _cache_path(self, package: str) -> Path:
|
|
return self.cache_dir / f"{package}.json"
|
|
|
|
def get(self, package: str, pkgbuild_hash: str) -> CacheEntry | None:
|
|
"""Get cached scan if it exists and is fresh."""
|
|
path = self._cache_path(package)
|
|
if not path.exists():
|
|
return None
|
|
|
|
try:
|
|
entry = CacheEntry(**json.loads(path.read_text()))
|
|
except Exception:
|
|
return None
|
|
|
|
# Check TTL
|
|
if time.time() - entry.timestamp > self.ttl_seconds:
|
|
return None
|
|
|
|
# Check if PKGBUILD changed
|
|
if entry.pkgbuild_hash != pkgbuild_hash:
|
|
return None
|
|
|
|
return entry
|
|
|
|
def put(self, result: ScanResult, pkgbuild_hash: str) -> None:
|
|
"""Store a scan result."""
|
|
entry = CacheEntry(
|
|
package=result.package,
|
|
verdict=result.verdict.value,
|
|
confidence=result.confidence,
|
|
findings=result.findings,
|
|
reasoning=result.reasoning,
|
|
timestamp=time.time(),
|
|
pkgbuild_hash=pkgbuild_hash,
|
|
)
|
|
path = self._cache_path(result.package)
|
|
path.write_text(json.dumps(asdict(entry), indent=2))
|
|
|
|
def list_all(self) -> list[dict]:
|
|
"""List all cached entries."""
|
|
entries = []
|
|
for path in self.cache_dir.glob("*.json"):
|
|
try:
|
|
entry = CacheEntry(**json.loads(path.read_text()))
|
|
entries.append(asdict(entry))
|
|
except Exception:
|
|
continue
|
|
return entries |