PARA.AI / api /routers /files.py
Carlex22's picture
Revert "ParaAIV3.1"
1f24745
"""
Router de testes de gestão de arquivos
Endpoints para testar criação, listagem e download de arquivos
"""
from fastapi import APIRouter, HTTPException
from typing import List
from pathlib import Path
from api.models.requests import FileCreateRequest
from api.models.responses import FileInfoResponse, MessageResponse
from api.utils.logger import setup_logger
router = APIRouter()
logger = setup_logger(__name__)
@router.post("/create", response_model=MessageResponse)
async def create_test_file(request: FileCreateRequest):
"""
**Cria arquivo de teste.**
Útil para testar sistema de arquivos e permissões.
"""
try:
from files.files_manager import FilesManager
fm = FilesManager()
# Criar arquivo
filepath = fm.save_text_file(
content=request.content,
filename=request.filename,
acordao_id=request.acordao_id
)
logger.info(f"✅ Test file created: {filepath}")
return MessageResponse(
message="File created successfully",
details={
"filepath": str(filepath),
"filename": request.filename,
"size": len(request.content)
}
)
except Exception as e:
logger.error(f"Error creating file: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/list", response_model=List[FileInfoResponse])
async def list_files(acordao_id: str = None, limit: int = 100):
"""
**Lista arquivos.**
Args:
acordao_id: Filtrar por ID de acórdão (opcional)
limit: Limite de resultados
"""
try:
from files.files_manager import FilesManager
fm = FilesManager()
files = fm.list_files(
acordao_id=acordao_id,
limit=limit
)
return [
FileInfoResponse(
filename=f["filename"],
size_bytes=f["size_bytes"],
hash_sha256=f.get("hash", "unknown"),
created_at=f.get("created_at")
)
for f in files
]
except Exception as e:
logger.error(f"Error listing files: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/info/{filename}")
async def get_file_info(filename: str):
"""
**Informações sobre um arquivo específico.**
"""
try:
from files.files_manager import FilesManager
fm = FilesManager()
info = fm.get_file_info(filename)
if not info:
raise HTTPException(status_code=404, detail="File not found")
return info
except HTTPException:
raise
except Exception as e:
logger.error(f"Error getting file info: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.delete("/{filename}")
async def delete_file(filename: str):
"""
**Deleta arquivo.**
Args:
filename: Nome do arquivo a deletar
"""
try:
from files.files_manager import FilesManager
fm = FilesManager()
success = fm.delete_file(filename)
if success:
logger.info(f"🗑️ File deleted: {filename}")
return MessageResponse(
message="File deleted successfully",
details={"filename": filename}
)
else:
raise HTTPException(status_code=404, detail="File not found")
except HTTPException:
raise
except Exception as e:
logger.error(f"Error deleting file: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/storage/info")
async def storage_info():
"""
**Informações sobre armazenamento.**
Retorna uso de disco e estatísticas.
"""
try:
from files.files_manager import FilesManager
import shutil
fm = FilesManager()
# Pegar info de disco
total, used, free = shutil.disk_usage(fm.base_path)
return {
"base_path": str(fm.base_path),
"storage": {
"total_gb": total / (1024**3),
"used_gb": used / (1024**3),
"free_gb": free / (1024**3),
"percent_used": (used / total) * 100
},
"total_files": len(list(Path(fm.base_path).rglob("*.*")))
}
except Exception as e:
logger.error(f"Error getting storage info: {e}")
raise HTTPException(status_code=500, detail=str(e))