102 lines
3.6 KiB
Python
102 lines
3.6 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
|
|
from app.models.testing import GenerateResult, EvaluationResult
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TestStorageService:
|
|
def __init__(self, results_dir: str, evaluations_dir: str):
|
|
self.results_dir = results_dir
|
|
self.evaluations_dir = evaluations_dir
|
|
Path(results_dir).mkdir(parents=True, exist_ok=True)
|
|
Path(evaluations_dir).mkdir(parents=True, exist_ok=True)
|
|
|
|
# --- Results ---
|
|
|
|
def save_result(self, result: GenerateResult) -> str:
|
|
filepath = os.path.join(self.results_dir, f"{result.result_id}.json")
|
|
with open(filepath, "w", encoding="utf-8") as f:
|
|
f.write(result.model_dump_json(indent=2))
|
|
logger.info("Saved test result: %s", filepath)
|
|
return filepath
|
|
|
|
def load_result(self, result_id: str) -> Optional[GenerateResult]:
|
|
filepath = os.path.join(self.results_dir, f"{result_id}.json")
|
|
if not os.path.isfile(filepath):
|
|
return None
|
|
with open(filepath, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
return GenerateResult.model_validate(data)
|
|
|
|
def list_results(self, limit: int = 50, offset: int = 0) -> List[dict]:
|
|
items = []
|
|
try:
|
|
for entry in sorted(
|
|
Path(self.results_dir).iterdir(),
|
|
key=lambda p: p.stat().st_mtime,
|
|
reverse=True,
|
|
):
|
|
if entry.suffix == ".json":
|
|
stat = entry.stat()
|
|
items.append({
|
|
"result_id": entry.stem,
|
|
"file_size_bytes": stat.st_size,
|
|
})
|
|
except FileNotFoundError:
|
|
return []
|
|
return items[offset : offset + limit]
|
|
|
|
def delete_result(self, result_id: str) -> bool:
|
|
filepath = os.path.join(self.results_dir, f"{result_id}.json")
|
|
if not os.path.isfile(filepath):
|
|
return False
|
|
os.remove(filepath)
|
|
return True
|
|
|
|
# --- Evaluations ---
|
|
|
|
def save_evaluation(self, evaluation: EvaluationResult) -> str:
|
|
filepath = os.path.join(self.evaluations_dir, f"{evaluation.evaluation_id}.json")
|
|
with open(filepath, "w", encoding="utf-8") as f:
|
|
f.write(evaluation.model_dump_json(indent=2))
|
|
logger.info("Saved evaluation: %s", filepath)
|
|
return filepath
|
|
|
|
def load_evaluation(self, eval_id: str) -> Optional[EvaluationResult]:
|
|
filepath = os.path.join(self.evaluations_dir, f"{eval_id}.json")
|
|
if not os.path.isfile(filepath):
|
|
return None
|
|
with open(filepath, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
return EvaluationResult.model_validate(data)
|
|
|
|
def list_evaluations(self, limit: int = 50, offset: int = 0) -> List[dict]:
|
|
items = []
|
|
try:
|
|
for entry in sorted(
|
|
Path(self.evaluations_dir).iterdir(),
|
|
key=lambda p: p.stat().st_mtime,
|
|
reverse=True,
|
|
):
|
|
if entry.suffix == ".json":
|
|
stat = entry.stat()
|
|
items.append({
|
|
"evaluation_id": entry.stem,
|
|
"file_size_bytes": stat.st_size,
|
|
})
|
|
except FileNotFoundError:
|
|
return []
|
|
return items[offset : offset + limit]
|
|
|
|
def delete_evaluation(self, eval_id: str) -> bool:
|
|
filepath = os.path.join(self.evaluations_dir, f"{eval_id}.json")
|
|
if not os.path.isfile(filepath):
|
|
return False
|
|
os.remove(filepath)
|
|
return True
|