Spaces:
Sleeping
Sleeping
File size: 3,184 Bytes
8ce1429 | 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 | """
Validation utilities for file upload pipeline.
"""
import os
from fastapi import UploadFile, HTTPException
STRUCTURED_EXTENSIONS = {"csv", "xlsx"}
UNSTRUCTURED_EXTENSIONS = {"pdf", "txt", "doc", "docx", "ppt", "pptx", "jpg", "jpeg", "png"}
ALLOWED_EXTENSIONS = STRUCTURED_EXTENSIONS | UNSTRUCTURED_EXTENSIONS
MIME_TYPE_MAP = {
"csv": ["text/csv", "application/csv", "text/plain"],
"xlsx": ["application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"],
"pdf": ["application/pdf"],
"txt": ["text/plain"],
"doc": ["application/msword"],
"docx": ["application/vnd.openxmlformats-officedocument.wordprocessingml.document"],
"ppt": ["application/vnd.ms-powerpoint"],
"pptx": ["application/vnd.openxmlformats-officedocument.presentationml.presentation"],
"jpg": ["image/jpeg"],
"jpeg": ["image/jpeg"],
"png": ["image/png"],
}
def get_file_extension(filename: str) -> str:
"""Extract and normalize file extension."""
if not filename or "." not in filename:
return ""
return filename.rsplit(".", 1)[-1].lower()
def validate_file_extension(filename: str) -> str:
"""
Validate that file has an allowed extension.
Returns the extension if valid, raises HTTPException otherwise.
"""
ext = get_file_extension(filename)
if not ext:
raise HTTPException(
status_code=400,
detail="File must have a valid extension."
)
if ext not in ALLOWED_EXTENSIONS:
raise HTTPException(
status_code=400,
detail=f"Unsupported file type: .{ext}. Allowed types: {', '.join(sorted(ALLOWED_EXTENSIONS))}"
)
return ext
def validate_file_size(file: UploadFile) -> None:
"""
Validate that file is not empty.
Raises HTTPException if file size is 0.
"""
file.file.seek(0, os.SEEK_END)
size = file.file.tell()
file.file.seek(0)
if size == 0:
raise HTTPException(
status_code=400,
detail="File is empty. Please upload a valid file."
)
def validate_mime_type(file: UploadFile, extension: str) -> None:
"""
Light MIME type sanity check.
Raises HTTPException if MIME type doesn't match extension.
"""
content_type = file.content_type or ""
expected_mimes = MIME_TYPE_MAP.get(extension, [])
if not expected_mimes:
return
if content_type and content_type not in expected_mimes:
if content_type != "application/octet-stream":
raise HTTPException(
status_code=400,
detail=f"MIME type mismatch. Expected one of {expected_mimes} for .{extension}, got {content_type}"
)
async def validate_upload_file(file: UploadFile) -> str:
"""
Run all validations on uploaded file.
Returns the validated file extension.
"""
if not file or not file.filename:
raise HTTPException(
status_code=400,
detail="No file provided or file has no name."
)
extension = validate_file_extension(file.filename)
validate_file_size(file)
validate_mime_type(file, extension)
return extension
|