58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
import os
|
|
import shutil
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import settings
|
|
from app.models.export_file import ExportFile
|
|
from app.models.recording import Recording
|
|
from app.models.transcription import Transcription
|
|
|
|
|
|
def get_disk_usage() -> dict:
|
|
upload_size = _dir_size(settings.upload_dir)
|
|
export_size = _dir_size(settings.export_dir)
|
|
return {
|
|
"uploads_bytes": upload_size,
|
|
"exports_bytes": export_size,
|
|
"total_bytes": upload_size + export_size,
|
|
"total_gb": round((upload_size + export_size) / (1024 ** 3), 2),
|
|
}
|
|
|
|
|
|
def _dir_size(path: str) -> int:
|
|
total = 0
|
|
if not os.path.exists(path):
|
|
return 0
|
|
for dirpath, _, filenames in os.walk(path):
|
|
for f in filenames:
|
|
fp = os.path.join(dirpath, f)
|
|
total += os.path.getsize(fp)
|
|
return total
|
|
|
|
|
|
def cleanup_old_data(db: Session, retention_days: int | None = None):
|
|
days = retention_days or settings.retention_days
|
|
cutoff = datetime.now(timezone.utc) - timedelta(days=days)
|
|
|
|
old_recordings = db.execute(
|
|
select(Recording).where(Recording.created_at < cutoff)
|
|
).scalars().all()
|
|
|
|
deleted_count = 0
|
|
for recording in old_recordings:
|
|
if os.path.exists(recording.file_path):
|
|
os.remove(recording.file_path)
|
|
if recording.transcription:
|
|
for protocol in recording.transcription.protocols:
|
|
for export_file in protocol.export_files:
|
|
if os.path.exists(export_file.file_path):
|
|
os.remove(export_file.file_path)
|
|
db.delete(recording)
|
|
deleted_count += 1
|
|
|
|
db.commit()
|
|
return {"deleted_recordings": deleted_count}
|