80 lines
2.7 KiB
Python
80 lines
2.7 KiB
Python
import os
|
|
import shutil
|
|
from uuid import uuid4
|
|
from fastapi import UploadFile
|
|
from typing import Optional
|
|
from datetime import datetime
|
|
from sqlmodel import Session, select
|
|
from app.models.coupon import Coupon
|
|
|
|
# Use absolute path relative to THIS file's location (app/services/image_handler.py)
|
|
# Going up 3 levels: services -> app -> backend -> uploads
|
|
_BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
UPLOAD_BASE = os.path.join(_BASE_DIR, "uploads")
|
|
|
|
|
|
def save_image(image: UploadFile, subfolder: str) -> str:
|
|
"""Saves an image to the local filesystem and returns its relative URL."""
|
|
upload_dir = os.path.join(UPLOAD_BASE, subfolder)
|
|
os.makedirs(upload_dir, exist_ok=True)
|
|
|
|
# Generate unique filename
|
|
ext = os.path.splitext(image.filename or "")[1].lower()
|
|
if not ext or ext not in ['.jpg', '.jpeg', '.png', '.gif', '.webp', '.avif']:
|
|
ext = ".jpg" # Default extension if missing or invalid
|
|
|
|
filename = f"{uuid4()}{ext}"
|
|
path = os.path.join(upload_dir, filename)
|
|
|
|
# Reset file pointer to start before reading
|
|
image.file.seek(0)
|
|
|
|
# Save file
|
|
with open(path, "wb") as buffer:
|
|
shutil.copyfileobj(image.file, buffer)
|
|
|
|
print(f"DEBUG: Saved image to {path}")
|
|
return f"/uploads/{subfolder}/{filename}"
|
|
|
|
|
|
def delete_image(image_url: Optional[str]):
|
|
"""Deletes an image from the filesystem if it's stored locally."""
|
|
if not image_url:
|
|
return
|
|
|
|
# We only delete if it points to our local uploads folder
|
|
if image_url.startswith("/uploads/"):
|
|
# Build absolute path
|
|
relative = image_url.lstrip("/") # e.g. "uploads/businesses/uuid.jpg"
|
|
file_path = os.path.join(_BASE_DIR, relative)
|
|
if os.path.exists(file_path):
|
|
try:
|
|
os.remove(file_path)
|
|
print(f"DEBUG: Deleted file {file_path}")
|
|
except Exception as e:
|
|
print(f"ERROR: Failed to delete {file_path}: {e}")
|
|
else:
|
|
print(f"DEBUG: File not found for deletion: {file_path}")
|
|
|
|
|
|
def cleanup_expired_coupons(session: Session):
|
|
"""Finds and deletes expired coupons and their associated images."""
|
|
try:
|
|
now = datetime.now()
|
|
# Coupons where valid_until is passed
|
|
statement = select(Coupon).where(Coupon.valid_until < now)
|
|
expired = session.exec(statement).all()
|
|
|
|
count = 0
|
|
for coupon in expired:
|
|
if coupon.image_url:
|
|
delete_image(coupon.image_url)
|
|
session.delete(coupon)
|
|
count += 1
|
|
|
|
if count > 0:
|
|
session.commit()
|
|
print(f"DEBUG: Cleaned up {count} expired coupons and their images.")
|
|
except Exception as e:
|
|
print(f"ERROR: Cleanup failed: {e}")
|