File size: 5,791 Bytes
46cf14b a295921 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 |
import os
import io
import json
import shutil
import sqlite3
from pathlib import Path
from fastapi import APIRouter, UploadFile, File, Query, HTTPException
from fastapi.responses import FileResponse, JSONResponse
from storage.files.file_manager import FileManager
from storage.common import validate_token
router = APIRouter(prefix="/data", tags=["Data Manager"])
DATA_ROOT = Path("/data")
HF_TOKEN = os.getenv("HF_TOKEN")
@router.get("/data_tree", tags=["Data Manager"])
def get_data_tree(
token: str = Query(..., description="Token required for authorization")
):
"""
Return a formatted tree of folders and files under /data.
Behavior:
- Validate token.
- Walk the /data directory and build a recursive tree.
- Each entry includes: name, type (file/directory), and children if directory.
"""
validate_token(token)
def build_tree(path: Path):
"""
Build a tree representation of directories and files.
Prevents errors by checking if the path is a directory before iterating.
"""
if not path.exists():
return {"name": path.name, "type": "missing"}
# Si es un fichero → devolvemos un nodo simple
if path.is_file():
return {
"name": path.name,
"type": "file"
}
# Si es un directorio → construimos sus hijos
children = []
try:
entries = sorted(path.iterdir(), key=lambda p: p.name.lower())
except Exception:
# Si por cualquier razón no podemos listar, lo tratamos como file
return {
"name": path.name,
"type": "file"
}
for child in entries:
children.append(build_tree(child))
return {
"name": path.name,
"type": "directory",
"children": children
}
if not DATA_ROOT.exists():
return {"error": "/data does not exist"}
return build_tree(DATA_ROOT)
@router.post("/create_data_item", tags=["Data Manager"])
def create_data_item(
path: str = Query(..., description="Full path starting with /data/"),
item_type: str = Query(..., description="directory or file"),
token: str = Query(..., description="Token required for authorization")
):
"""
Create a directory or file under /data.
Restrictions:
- Path must start with /data/.
- Writing to /data/db or /data/media (or any of their subpaths) is forbidden.
- item_type must be 'directory' or 'file'.
Behavior:
- Validate token.
- Check path validity and protection.
- Create directory or empty file.
- Raise error if path already exists.
"""
validate_token(token)
target = Path(path)
# Validación básica
if not path.startswith("/data/"):
raise HTTPException(status_code=400, detail="Path must start with /data/")
if item_type not in ("directory", "file"):
raise HTTPException(status_code=400, detail="item_type must be 'directory' or 'file'")
# Protección de carpetas sensibles
protected = ["/data/db", "/data/media"]
for p in protected:
if path == p or path.startswith(p + "/"):
raise HTTPException(
status_code=403,
detail=f"Access to protected path '{p}' is not allowed"
)
# No permitir sobrescritura
if target.exists():
raise HTTPException(status_code=409, detail="Path already exists")
try:
if item_type == "directory":
target.mkdir(parents=True, exist_ok=False)
else:
target.parent.mkdir(parents=True, exist_ok=True)
with open(target, "wb") as f:
f.write(b"")
except Exception as exc:
raise HTTPException(status_code=500, detail=f"Failed to create item: {exc}")
return {
"status": "ok",
"created": str(target),
"type": item_type
}
@router.delete("/delete_path", tags=["Data Manager"])
def delete_path(
target: str = Query(..., description="Ruta absoluta dentro de /data a borrar"),
token: str = Query(..., description="Token requerido para autorización")
):
"""
Delete a file or directory recursively inside /data, except protected folders.
Behavior:
- Validate token.
- Validate path starts with /data/.
- Deny deletion of /data/db and /data/media (and anything inside them).
- If target is a file → delete file.
- If target is a directory → delete recursively.
"""
validate_token(token)
# Normalizar ruta
path = Path(target).resolve()
# Debe estar dentro de /data/
if not str(path).startswith(str(DATA_ROOT)):
raise HTTPException(status_code=400, detail="Path must be inside /data/")
# Carpetas protegidas
protected = [
DATA_ROOT / "db",
DATA_ROOT / "media",
]
for p in protected:
if path == p or str(path).startswith(str(p) + "/"):
raise HTTPException(
status_code=403,
detail=f"Deletion forbidden: {p} is protected"
)
# Verificar existencia
if not path.exists():
raise HTTPException(status_code=404, detail="Target path does not exist")
try:
if path.is_file():
path.unlink()
else:
shutil.rmtree(path)
except Exception as exc:
raise HTTPException(status_code=500, detail=f"Failed to delete: {exc}")
return {
"status": "ok",
"deleted": str(path)
} |