mirror of
https://github.com/sfiorini/iptv-server.git
synced 2026-04-11 08:40:44 +00:00
129 lines
4.0 KiB
Python
129 lines
4.0 KiB
Python
import re
|
|
from pathlib import PurePath
|
|
from fastapi import HTTPException, UploadFile
|
|
import vercel_blob
|
|
from .database import get_blob_by_path
|
|
import logging
|
|
|
|
# Configure logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ALLOWED_EXTENSIONS = {
|
|
'playlists': ('.m3u', '.m3u8'),
|
|
'epgs': ('.xml', '.xml.gz')
|
|
}
|
|
|
|
async def upload_blob_file(folder: str, file: UploadFile) -> dict:
|
|
"""
|
|
Upload a file to Vercel Blob storage with validation
|
|
|
|
Args:
|
|
folder (str): The folder name (e.g. 'playlists' or 'epgs')
|
|
file (UploadFile): The file to upload
|
|
|
|
Returns:
|
|
dict: Information about the uploaded blob
|
|
|
|
Raises:
|
|
HTTPException: If validation fails or upload fails
|
|
"""
|
|
try:
|
|
# Validate filename
|
|
if not file.filename or not file.filename.strip():
|
|
raise HTTPException(status_code=400, detail="Invalid filename provided")
|
|
|
|
# Sanitize filename
|
|
sanitized_filename = PurePath(file.filename).name
|
|
sanitized_filename = re.sub(r'[^a-zA-Z0-9_.-]', '_', sanitized_filename)
|
|
|
|
# Validate file extension
|
|
if not sanitized_filename.lower().endswith(ALLOWED_EXTENSIONS[folder]):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Only {', '.join(ALLOWED_EXTENSIONS[folder])} files are allowed"
|
|
)
|
|
|
|
content = await file.read()
|
|
|
|
# Validate content size (10MB limit)
|
|
max_size = 10 * 1024 * 1024 # 10MB
|
|
if len(content) > max_size:
|
|
raise HTTPException(
|
|
status_code=413,
|
|
detail=f"File size exceeds {max_size//1024//1024}MB limit"
|
|
)
|
|
|
|
blob_info = vercel_blob.put(
|
|
path=f"{folder}/{sanitized_filename}",
|
|
data=content,
|
|
options={
|
|
"contentType": file.content_type or "application/octet-stream",
|
|
"access": 'public',
|
|
"addRandomSuffix": False
|
|
}
|
|
)
|
|
|
|
return {
|
|
"status": "uploaded",
|
|
"filename": sanitized_filename,
|
|
"url": blob_info['url'],
|
|
"size": len(content),
|
|
"content_type": blob_info.get('contentType')
|
|
}
|
|
|
|
except KeyError as e:
|
|
logger.error(f"Unexpected blob storage response: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Unexpected response format from blob storage: {str(e)}"
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"File upload failed: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"File upload failed: {str(e)}"
|
|
)
|
|
|
|
def delete_blob_file(folder: str, filename: str) -> dict:
|
|
"""
|
|
Delete a file from Vercel Blob storage
|
|
|
|
Args:
|
|
folder (str): The folder name (e.g. 'playlists' or 'epgs')
|
|
filename (str): The filename to delete
|
|
|
|
Returns:
|
|
dict: Information about the deleted blob
|
|
|
|
Raises:
|
|
HTTPException: If the file is not found or deletion fails
|
|
"""
|
|
try:
|
|
# Sanitize filename for safety
|
|
sanitized_filename = PurePath(filename).name
|
|
sanitized_filename = re.sub(r'[^a-zA-Z0-9_.-]', '_', sanitized_filename)
|
|
|
|
# Get the blob information first to get the URL
|
|
blob_info = get_blob_by_path(folder, sanitized_filename)
|
|
|
|
# Delete using the blob URL
|
|
vercel_blob.delete(blob_info['url'])
|
|
|
|
return {
|
|
"status": "deleted",
|
|
"filename": sanitized_filename,
|
|
"url": blob_info['url']
|
|
}
|
|
|
|
except HTTPException as he:
|
|
# Re-raise HTTP exceptions (like 404 from get_blob_by_path)
|
|
logger.error(f"HTTP error while deleting blob: {str(he)}")
|
|
raise he
|
|
except Exception as e:
|
|
logger.error(f"Failed to delete blob: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to delete file: {str(e)}"
|
|
) |