fotohost/app/storage_service.py

import io
import os
from ftplib import FTP, FTP_TLS

from flask import current_app

from app.settings_service import get_settings


def save_photo_file(file_storage, stored_name):
    settings = get_settings()
    upload_dir = current_app.config["UPLOAD_FOLDER"]
    os.makedirs(upload_dir, exist_ok=True)
    local_path = os.path.join(upload_dir, stored_name)

    file_storage.save(local_path)
    file_size = os.path.getsize(local_path)
    storage_backend = "local"

    errors = []

    if settings.s3_enabled:
        ok, err = _upload_s3(local_path, stored_name, settings)
        if ok:
            storage_backend = "s3"
        elif err:
            errors.append(f"S3: {err}")

    if settings.sftp_enabled:
        ok, err = _upload_sftp(local_path, stored_name, settings)
        if err:
            errors.append(f"SFTP: {err}")

    if settings.ftp_enabled:
        ok, err = _upload_ftp(local_path, stored_name, settings)
        if err:
            errors.append(f"FTP: {err}")

    return local_path, file_size, storage_backend, errors


def delete_photo_file(stored_name, storage_backend="local"):
    settings = get_settings()
    upload_dir = current_app.config["UPLOAD_FOLDER"]
    local_path = os.path.join(upload_dir, stored_name)

    if os.path.exists(local_path):
        os.remove(local_path)

    if storage_backend == "s3" and settings.s3_enabled:
        _delete_s3(stored_name, settings)

    if settings.sftp_enabled:
        _delete_sftp(stored_name, settings)

    if settings.ftp_enabled:
        _delete_ftp(stored_name, settings)


def get_photo_stream(stored_name, storage_backend="local"):
    settings = get_settings()
    upload_dir = current_app.config["UPLOAD_FOLDER"]
    local_path = os.path.join(upload_dir, stored_name)

    if os.path.exists(local_path):
        return open(local_path, "rb")

    if storage_backend == "s3" and settings.s3_enabled:
        data = _download_s3(stored_name, settings)
        if data:
            return io.BytesIO(data)
    return None


def _upload_s3(local_path, key, settings):
    try:
        import boto3
        from botocore.config import Config

        kwargs = {
            "aws_access_key_id": settings.s3_access_key,
            "aws_secret_access_key": settings.s3_secret_key,
            "region_name": settings.s3_region or "us-east-1",
        }
        if settings.s3_endpoint:
            kwargs["endpoint_url"] = settings.s3_endpoint

        client = boto3.client("s3", config=Config(signature_version="s3v4"), **kwargs)
        client.upload_file(local_path, settings.s3_bucket, key)
        return True, None
    except Exception as exc:
        return False, str(exc)


def _delete_s3(key, settings):
    try:
        import boto3

        kwargs = {
            "aws_access_key_id": settings.s3_access_key,
            "aws_secret_access_key": settings.s3_secret_key,
            "region_name": settings.s3_region or "us-east-1",
        }
        if settings.s3_endpoint:
            kwargs["endpoint_url"] = settings.s3_endpoint
        client = boto3.client("s3", **kwargs)
        client.delete_object(Bucket=settings.s3_bucket, Key=key)
    except Exception:
        current_app.logger.exception("S3 delete failed")


def _download_s3(key, settings):
    try:
        import boto3

        kwargs = {
            "aws_access_key_id": settings.s3_access_key,
            "aws_secret_access_key": settings.s3_secret_key,
            "region_name": settings.s3_region or "us-east-1",
        }
        if settings.s3_endpoint:
            kwargs["endpoint_url"] = settings.s3_endpoint
        client = boto3.client("s3", **kwargs)
        obj = client.get_object(Bucket=settings.s3_bucket, Key=key)
        return obj["Body"].read()
    except Exception:
        current_app.logger.exception("S3 download failed")
        return None


def _upload_sftp(local_path, remote_name, settings):
    try:
        import paramiko

        transport = paramiko.Transport((settings.sftp_host, settings.sftp_port))
        transport.connect(username=settings.sftp_username, password=settings.sftp_password)
        sftp = paramiko.SFTPClient.from_transport(transport)
        remote_dir = settings.sftp_remote_path or "/uploads"
        _sftp_makedirs(sftp, remote_dir)
        sftp.put(local_path, f"{remote_dir.rstrip('/')}/{remote_name}")
        sftp.close()
        transport.close()
        return True, None
    except Exception as exc:
        return False, str(exc)


def _delete_sftp(remote_name, settings):
    try:
        import paramiko

        transport = paramiko.Transport((settings.sftp_host, settings.sftp_port))
        transport.connect(username=settings.sftp_username, password=settings.sftp_password)
        sftp = paramiko.SFTPClient.from_transport(transport)
        remote_path = f"{settings.sftp_remote_path.rstrip('/')}/{remote_name}"
        sftp.remove(remote_path)
        sftp.close()
        transport.close()
    except Exception:
        current_app.logger.exception("SFTP delete failed")


def _sftp_makedirs(sftp, remote_dir):
    parts = remote_dir.strip("/").split("/")
    path = ""
    for part in parts:
        path += f"/{part}"
        try:
            sftp.stat(path)
        except IOError:
            sftp.mkdir(path)


def _upload_ftp(local_path, remote_name, settings):
    try:
        ftp_cls = FTP_TLS if settings.ftp_use_tls else FTP
        ftp = ftp_cls()
        ftp.connect(settings.ftp_host, settings.ftp_port, timeout=30)
        ftp.login(settings.ftp_username, settings.ftp_password)
        if settings.ftp_use_tls:
            ftp.prot_p()
        remote_dir = settings.ftp_remote_path or "/uploads"
        _ftp_makedirs(ftp, remote_dir)
        ftp.cwd(remote_dir)
        with open(local_path, "rb") as f:
            ftp.storbinary(f"STOR {remote_name}", f)
        ftp.quit()
        return True, None
    except Exception as exc:
        return False, str(exc)


def _delete_ftp(remote_name, settings):
    try:
        ftp_cls = FTP_TLS if settings.ftp_use_tls else FTP
        ftp = ftp_cls()
        ftp.connect(settings.ftp_host, settings.ftp_port, timeout=30)
        ftp.login(settings.ftp_username, settings.ftp_password)
        if settings.ftp_use_tls:
            ftp.prot_p()
        ftp.cwd(settings.ftp_remote_path or "/uploads")
        ftp.delete(remote_name)
        ftp.quit()
    except Exception:
        current_app.logger.exception("FTP delete failed")


def _ftp_makedirs(ftp, remote_dir):
    parts = remote_dir.strip("/").split("/")
    path = ""
    for part in parts:
        path += f"/{part}"
        try:
            ftp.cwd(path)
        except Exception:
            ftp.mkd(path)