shopdb-flask/shopdb/core/api/models.py

"""Models (vendor model catalog) API endpoints - Full CRUD."""

import glob
import os
from urllib.parse import urlparse

from flask import Blueprint, request, current_app
from flask_jwt_extended import jwt_required
from werkzeug.utils import secure_filename

from shopdb.extensions import db
from shopdb.core.models import Model, ModelType
from shopdb.utils.responses import (
    success_response,
    error_response,
    paginated_response,
    ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query

from shopdb.utils.authz import require_role
from shopdb.utils.import_mode import apply_import_timestamps
from shopdb.utils.uploads import send_upload
from shopdb.utils.authz import require_permission

models_bp = Blueprint('models', __name__)

# Uploaded model photos live in the instance dir and are served publicly
# (asset detail pages read the model image without auth). Same image set the
# map/branding uploads accept.
MODEL_IMAGE_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.svg'}

# URL prefix an uploaded image resolves to. imageurl values with this prefix
# are our own files under the instance dir; anything else (external URLs or the
# shipped /images/models/* assets) is left on disk untouched.
MODEL_IMAGE_URL_PREFIX = '/api/models/image/'


def _modelimage_dir():
    return os.path.join(current_app.instance_path, 'modelimages')


def _validate_url_scheme(url):
    """Validate that URL uses safe scheme (prevents XSS via javascript: URIs)."""
    if not url:
        return True
    parsed = urlparse(url)
    if parsed.scheme and parsed.scheme not in ['http', 'https']:
        raise ValueError(f"Invalid URL scheme '{parsed.scheme}'. Only http:// and https:// are allowed.")
    return True


@models_bp.route('', methods=['GET'])
@require_permission('assets.view')
def list_models():
    """List all vendor catalog models."""
    page, per_page = get_pagination_params(request)

    query = Model.query

    if request.args.get('active', 'true').lower() != 'false':
        query = query.filter(Model.isactive == True)

    if vendor_id := request.args.get('vendor', type=int):
        query = query.filter(Model.vendorid == vendor_id)

    if modeltype_id := request.args.get('modeltype', type=int):
        query = query.filter(Model.modeltypeid == modeltype_id)

    # Scope to a model-type category (Equipment, PC, Network, Printer) so a
    # subtype form only offers models that make sense for it. Models with no
    # type are excluded, since a null category cannot satisfy the filter.
    if category := request.args.get('category'):
        query = query.join(ModelType).filter(ModelType.category == category)

    # Exact-match natural-key lookup for idempotent import. Natural key is
    # modelnumber + vendor; pair this with ?vendor=<id> to disambiguate.
    if exactmodelnumber := request.args.get('modelnumber'):
        query = query.filter(Model.modelnumber == exactmodelnumber)

    if search := request.args.get('search'):
        query = query.filter(Model.modelnumber.ilike(f'%{search}%'))

    query = query.order_by(Model.modelnumber)

    items, total = paginate_query(query, page, per_page)

    data = []
    for m in items:
        d = m.to_dict()
        d['vendor'] = m.vendor.vendor if m.vendor else None
        d['modeltype'] = m.modeltype.modeltype if m.modeltype else None
        d['category'] = m.modeltype.category if m.modeltype else None
        data.append(d)

    return paginated_response(data, page, per_page, total)


@models_bp.route('/<int:model_id>', methods=['GET'])
@require_permission('assets.view')
def get_model(model_id: int):
    """Get a single model."""
    m = db.session.get(Model, model_id)

    if not m:
        return error_response(
            ErrorCodes.NOT_FOUND,
            f'Model with ID {model_id} not found',
            http_code=404
        )

    data = m.to_dict()
    data['vendor'] = m.vendor.to_dict() if m.vendor else None
    data['modeltype'] = m.modeltype.to_dict() if m.modeltype else None

    return success_response(data)


@models_bp.route('', methods=['POST'])
@jwt_required()
@require_role('admin')
def create_model():
    """Create a new model."""
    data = request.get_json()

    if not data or not data.get('modelnumber'):
        return error_response(ErrorCodes.VALIDATION_ERROR, 'modelnumber is required')

    # Check duplicate
    existing = Model.query.filter_by(
        modelnumber=data['modelnumber'],
        vendorid=data.get('vendorid')
    ).first()
    if existing:
        return error_response(
            ErrorCodes.CONFLICT,
            f"Model '{data['modelnumber']}' already exists for this vendor",
            http_code=409
        )

    # Validate URL scheme to prevent XSS
    try:
        _validate_url_scheme(data.get('documentationurl'))
        _validate_url_scheme(data.get('imageurl'))
    except ValueError as e:
        return error_response(ErrorCodes.VALIDATION_ERROR, str(e))

    m = Model(
        modelnumber=data['modelnumber'],
        vendorid=data.get('vendorid'),
        modeltypeid=data.get('modeltypeid'),
        description=data.get('description'),
        imageurl=data.get('imageurl'),
        documentationurl=data.get('documentationurl'),
        notes=data.get('notes')
    )

    db.session.add(m)
    apply_import_timestamps(m, data)
    db.session.commit()

    return success_response(m.to_dict(), message='Model created', http_code=201)


@models_bp.route('/<int:model_id>', methods=['PUT'])
@jwt_required()
@require_role('admin')
def update_model(model_id: int):
    """Update a model."""
    m = db.session.get(Model, model_id)

    if not m:
        return error_response(
            ErrorCodes.NOT_FOUND,
            f'Model with ID {model_id} not found',
            http_code=404
        )

    data = request.get_json()
    if not data:
        return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')

    # Validate URL schemes to prevent XSS
    try:
        if 'documentationurl' in data:
            _validate_url_scheme(data['documentationurl'])
        if 'imageurl' in data:
            _validate_url_scheme(data['imageurl'])
    except ValueError as e:
        return error_response(ErrorCodes.VALIDATION_ERROR, str(e))

    for key in ['modelnumber', 'vendorid', 'modeltypeid', 'description', 'imageurl', 'documentationurl', 'notes', 'isactive']:
        if key in data:
            setattr(m, key, data[key])

    apply_import_timestamps(m, data)
    db.session.commit()
    return success_response(m.to_dict(), message='Model updated')


@models_bp.route('/<int:model_id>', methods=['DELETE'])
@jwt_required()
@require_role('admin')
def delete_model(model_id: int):
    """Delete (deactivate) a model."""
    m = db.session.get(Model, model_id)

    if not m:
        return error_response(
            ErrorCodes.NOT_FOUND,
            f'Model with ID {model_id} not found',
            http_code=404
        )

    m.isactive = False
    db.session.commit()

    return success_response(message='Model deleted')


@models_bp.route('/<int:model_id>/image', methods=['POST'])
@jwt_required()
@require_role('admin')
def upload_model_image(model_id: int):
    """Upload (or replace) the photo for a model.

    multipart/form-data: file=<image>. Saves to the instance modelimages dir as
    model-<id><ext> (one image per model) and points model.imageurl at the
    served URL. Re-upload replaces the old file even when the extension changes.
    """
    m = db.session.get(Model, model_id)
    if not m:
        return error_response(
            ErrorCodes.NOT_FOUND,
            f'Model with ID {model_id} not found',
            http_code=404
        )

    upload = request.files.get('file')
    if not upload or not upload.filename:
        return error_response(ErrorCodes.VALIDATION_ERROR, 'No file provided')
    ext = os.path.splitext(upload.filename)[1].lower()
    if ext not in MODEL_IMAGE_EXTENSIONS:
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              f'Unsupported image type {ext}')

    imagedir = _modelimage_dir()
    os.makedirs(imagedir, exist_ok=True)

    # Wipe any prior model-<id>.* file so a new extension does not orphan the old
    # one (one image per model).
    for old in glob.glob(os.path.join(imagedir, secure_filename(f'model-{model_id}') + '.*')):
        os.remove(old)

    filename = secure_filename(f'model-{model_id}{ext}')
    upload.save(os.path.join(imagedir, filename))

    m.imageurl = f'{MODEL_IMAGE_URL_PREFIX}{filename}'
    db.session.commit()

    return success_response(m.to_dict(), message='Model image uploaded')


@models_bp.route('/image/<path:filename>', methods=['GET'])
def serve_model_image(filename):
    """Serve an uploaded model image (public - asset detail pages read it)."""
    return send_upload(_modelimage_dir(), filename)


@models_bp.route('/<int:model_id>/image', methods=['DELETE'])
@jwt_required()
@require_role('admin')
def delete_model_image(model_id: int):
    """Clear a model image and delete the uploaded file if we own it.

    External URLs and the shipped /images/models/* assets are never touched on
    disk - only the imageurl field is cleared.
    """
    m = db.session.get(Model, model_id)
    if not m:
        return error_response(
            ErrorCodes.NOT_FOUND,
            f'Model with ID {model_id} not found',
            http_code=404
        )

    url = m.imageurl or ''
    if url.startswith(MODEL_IMAGE_URL_PREFIX):
        # secure_filename strips any traversal; the file lives in our dir only.
        filename = secure_filename(url[len(MODEL_IMAGE_URL_PREFIX):])
        path = os.path.join(_modelimage_dir(), filename)
        if os.path.exists(path):
            os.remove(path)

    m.imageurl = None
    db.session.commit()

    return success_response(m.to_dict(), message='Model image removed')