shopdb-flask/plugins/warranty/api/routes.py

"""Warranty API: manual CRUD now, provider refresh stubbed for later phases.

Coverage status is derived from enddate at read time (see models.derive_status),
never stored. Warranties link to assets many-to-many via warrantyassets, though
the common case is one warranty per asset.
"""

import glob
import os
from datetime import date, datetime, timedelta, timezone

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

from shopdb.api import send_upload
from shopdb.api import (
    db, Asset,
    success_response, error_response, ErrorCodes,
    require_permission,
    DateValueError, parse_date, parse_date_fields,
)

from ..models import Warranty, WarrantyAsset
from ..services import get_provider, ProviderNotConfigured, WarrantyLookupError

warranty_bp = Blueprint('warranty', __name__)

# What a proof of cover actually arrives as: a vendor PDF, a scan, or a
# screenshot of a portal page. Office formats are allowed because purchase
# records often arrive that way.
PROOF_EXTENSIONS = {'.pdf', '.png', '.jpg', '.jpeg', '.gif', '.webp', '.tif',
                    '.tiff', '.msg', '.eml', '.doc', '.docx', '.xls', '.xlsx'}
PROOF_URL_PREFIX = '/api/warranty/proof/'

# A certificate is a document, not a disk image. Anything past this is somebody
# attaching the wrong thing.
MAX_PROOF_BYTES = 25 * 1024 * 1024


def _proof_dir():
    return os.path.join(current_app.instance_path, 'warrantyproofs')


def _vendor_date(value):
    """A date off a VENDOR response, which this code does not control.

    Kept lenient on purpose, unlike a date a person typed: a provider that
    starts returning a shape we do not parse must not fail the refresh of an
    otherwise good record. The field is simply left as it was.
    """
    try:
        return parse_date(value)
    except DateValueError:
        return None


def _related_machine(asset):
    """The machine a covered asset is associated with, or None.

    A shopfloor PC is bought, warranted and replaced as a PC, but it is FOUND
    by the machine it drives - nobody walks the floor looking for an asset
    number. So a warranty row for a PC carries the machine's number, and its
    map position so the row can point at where to go.

    Walks the asset relationship graph in BOTH directions: the canonical edge is
    PC --controls--> machine, but a dual-bay pair carries controls on both bays
    and hand-made links are not guaranteed to be oriented. Returns the first
    active related asset whose type is 'machine'.
    """
    from shopdb.api import Asset as CoreAsset

    related = []
    for rel in (getattr(asset, 'outgoing_relationships', None) or []):
        if rel.isactive and rel.targetasset:
            related.append(rel.targetasset)
    for rel in (getattr(asset, 'incoming_relationships', None) or []):
        if rel.isactive and rel.sourceasset:
            related.append(rel.sourceasset)

    for candidate in related:
        typename = candidate.assettype.assettype if candidate.assettype else None
        if typename != 'machine':
            continue
        return {
            'assetid': candidate.assetid,
            'machinenumber': candidate.assetnumber,
            'name': candidate.name,
            # Map position for the hover preview. Either may be None: a machine
            # that has never been placed on the floor map still has a number
            # worth showing, so the caller decides what to do with a missing
            # position rather than the row being dropped.
            'mapx': candidate.mapx,
            'levelid': candidate.levelid,
            'mapy': candidate.mapy,
            'locationid': candidate.locationid,
            'locationname': (candidate.location.locationname
                             if candidate.location else None),
        }
    return None


def _asset_summary(asset):
    return {
        'assetid': asset.assetid,
        'assetnumber': asset.assetnumber,
        'name': asset.name,
        'assettypename': asset.assettype.assettype if asset.assettype else None,
        'machine': _related_machine(asset),
    }


def _warranty_payload(warranty, today=None, assetmap=None):
    """to_dict plus the linked-asset summaries. Pass assetmap (assetid -> Asset)
    to avoid a per-link query when serializing a list; without it, falls back to
    a per-link get (fine for a single warranty)."""
    data = warranty.to_dict(today)
    assets = []
    for link in warranty.links:
        asset = assetmap.get(link.assetid) if assetmap is not None \
            else db.session.get(Asset, link.assetid)
        if asset:
            assets.append(_asset_summary(asset))
    data['assets'] = assets
    return data


def _apply_links(warranty, assetids):
    """Replace a warranty's asset links with the given asset id list."""
    if assetids is None:
        return
    wanted = {int(a) for a in assetids if str(a).strip()}
    existing = {link.assetid: link for link in warranty.links}
    for assetid in wanted - set(existing):
        if db.session.get(Asset, assetid):
            warranty.links.append(WarrantyAsset(assetid=assetid))
    for assetid in set(existing) - wanted:
        warranty.links.remove(existing[assetid])


# =============================================================================
# CRUD
# =============================================================================

@warranty_bp.route('', methods=['GET'])
@require_permission('warranty.view')
def list_warranties():
    """List warranties. Filters: ?status=, ?assetid=, ?active=false."""
    query = Warranty.query
    if request.args.get('active', 'true').lower() != 'false':
        query = query.filter_by(isactive=True)
    # Exact-match natural-key lookup for idempotent import (servicetag + vendor).
    if exactservicetag := request.args.get('servicetag'):
        query = query.filter(Warranty.servicetag == exactservicetag)
    if exactvendor := request.args.get('vendor'):
        query = query.filter(Warranty.vendor == exactvendor)
    assetid = request.args.get('assetid', type=int)
    if assetid:
        query = (query.join(WarrantyAsset, WarrantyAsset.warrantyid == Warranty.warrantyid)
                      .filter(WarrantyAsset.assetid == assetid))
    warranties = (query.options(joinedload(Warranty.links))
                  .order_by(Warranty.enddate.is_(None), Warranty.enddate).all())

    # Batch-fetch every linked asset in ONE query (was N+1: a db.session.get per
    # link per warranty, ~1.8s for the full list).
    assetids = {link.assetid for w in warranties for link in w.links}
    assetmap = ({a.assetid: a for a in Asset.query.filter(Asset.assetid.in_(assetids)).all()}
                if assetids else {})

    today = date.today()
    items = [_warranty_payload(w, today, assetmap) for w in warranties]

    status_filter = request.args.get('status')
    if status_filter:
        items = [i for i in items if i['status'] == status_filter]
    return success_response(items)


@warranty_bp.route('/asset/<int:assetid>', methods=['GET'])
@require_permission('warranty.view')
def warranties_for_asset(assetid):
    """Warranties covering one asset (for the asset-detail panel)."""
    links = WarrantyAsset.query.filter_by(assetid=assetid).all()
    today = date.today()
    items = []
    for link in links:
        w = db.session.get(Warranty, link.warrantyid)
        if w and w.isactive:
            items.append(_warranty_payload(w, today))
    return success_response(items)


@warranty_bp.route('/<int:warrantyid>', methods=['GET'])
@require_permission('warranty.view')
def get_warranty(warrantyid):
    warranty = db.session.get(Warranty, warrantyid)
    if not warranty:
        return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', http_code=404)
    return success_response(_warranty_payload(warranty))


@warranty_bp.route('', methods=['POST'])
@require_permission('warranty.create')
def create_warranty():
    data = request.get_json() or {}
    vendor = (data.get('vendor') or '').strip()
    if not vendor:
        return error_response(ErrorCodes.VALIDATION_ERROR, 'vendor is required')
    # Refused, not nulled. A mistyped end date used to store NULL and answer
    # 200, which dropped the warranty out of every expiry surface on the
    # strength of a typo.
    try:
        dates = parse_date_fields(data, ('startdate', 'enddate'))
    except DateValueError as bad:
        return error_response(ErrorCodes.VALIDATION_ERROR, str(bad))
    warranty = Warranty(
        vendor=vendor,
        servicetag=(data.get('servicetag') or '').strip() or None,
        provider=(data.get('provider') or 'manual').strip().lower(),
        servicelevel=(data.get('servicelevel') or '').strip() or None,
        startdate=dates.get('startdate'),
        enddate=dates.get('enddate'),
        notes=(data.get('notes') or '').strip() or None,
    )
    _apply_links(warranty, data.get('assetids'))
    db.session.add(warranty)
    db.session.commit()
    return success_response(_warranty_payload(warranty), message='Warranty created', http_code=201)


@warranty_bp.route('/<int:warrantyid>', methods=['PUT'])
@require_permission('warranty.edit')
def update_warranty(warrantyid):
    warranty = db.session.get(Warranty, warrantyid)
    if not warranty:
        return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', http_code=404)
    data = request.get_json() or {}
    try:
        dates = parse_date_fields(data, ('startdate', 'enddate'))
    except DateValueError as bad:
        return error_response(ErrorCodes.VALIDATION_ERROR, str(bad))
    if 'vendor' in data:
        warranty.vendor = (data['vendor'] or '').strip() or warranty.vendor
    if 'servicetag' in data:
        warranty.servicetag = (data['servicetag'] or '').strip() or None
    if 'provider' in data:
        warranty.provider = (data['provider'] or 'manual').strip().lower()
    if 'servicelevel' in data:
        warranty.servicelevel = (data['servicelevel'] or '').strip() or None
    for field, parsed in dates.items():
        setattr(warranty, field, parsed)
    if 'notes' in data:
        warranty.notes = (data['notes'] or '').strip() or None
    if 'isactive' in data:
        warranty.isactive = bool(data['isactive'])
    if 'assetids' in data:
        _apply_links(warranty, data['assetids'])
    db.session.commit()
    return success_response(_warranty_payload(warranty), message='Warranty updated')


@warranty_bp.route('/<int:warrantyid>', methods=['DELETE'])
@require_permission('warranty.delete')
def delete_warranty(warrantyid):
    warranty = db.session.get(Warranty, warrantyid)
    if not warranty:
        return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', http_code=404)
    db.session.delete(warranty)
    db.session.commit()
    return success_response(message='Warranty deleted')


# =============================================================================
# Provider refresh (phase 1: manual only; API providers report not-configured)
# =============================================================================

@warranty_bp.route('/<int:warrantyid>/refresh', methods=['POST'])
@require_permission('warranty.edit')
def refresh_warranty(warrantyid):
    warranty = db.session.get(Warranty, warrantyid)
    if not warranty:
        return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', http_code=404)
    provider = get_provider(warranty.provider)
    try:
        result = provider.lookup(warranty.servicetag, warranty.vendor)
    except (ProviderNotConfigured, WarrantyLookupError) as exc:
        return error_response(ErrorCodes.VALIDATION_ERROR, str(exc), http_code=400)
    if not result:
        return error_response(ErrorCodes.VALIDATION_ERROR,
                              'This warranty is manual - nothing to refresh.', http_code=400)
    if result.get('servicelevel'):
        warranty.servicelevel = result['servicelevel']
    if result.get('startdate'):
        warranty.startdate = _vendor_date(result['startdate'])
    if result.get('enddate'):
        warranty.enddate = _vendor_date(result['enddate'])
    warranty.lastcheckeddate = datetime.now(timezone.utc).replace(tzinfo=None)
    db.session.commit()
    return success_response(_warranty_payload(warranty), message='Warranty refreshed')


# =============================================================================
# Bulk Dell sync - auto-detect Dell hardware by service tag and pull coverage
# =============================================================================

@warranty_bp.route('/sync/dell', methods=['POST'])
@require_permission('warranty.edit')
def sync_dell():
    """Look up every asset serial (= Dell service tag) that has no warranty yet
    and create warranties for the ones Dell recognizes.

    Only assets that lack a dated warranty are looked up, so re-running is cheap
    and non-Dell serials are filtered out by Dell (they return no coverage).
    """
    provider = get_provider('dell')
    # ?all=true re-checks assets that already have a dated warranty too.
    recheck_all = request.args.get('all', 'false').lower() == 'true'

    # Assets already covered by a dated warranty - skipped unless recheck_all.
    covered = set()
    if not recheck_all:
        for link in WarrantyAsset.query.all():
            w = db.session.get(Warranty, link.warrantyid)
            if w and w.isactive and w.enddate:
                covered.add(link.assetid)

    # Candidate assets: active, have a serial, not already covered.
    candidates = (Asset.query
                  .filter(Asset.isactive.is_(True),
                          Asset.serialnumber.isnot(None),
                          Asset.serialnumber != '')
                  .all())
    by_tag = {}
    for asset in candidates:
        if asset.assetid in covered:
            continue
        by_tag.setdefault(asset.serialnumber.strip().upper(), []).append(asset.assetid)

    if not by_tag:
        return success_response({
            'candidates': 0, 'matched': 0, 'created': 0, 'updated': 0,
        }, message='No PCs need a warranty lookup.')

    try:
        results = provider.bulk_lookup(list(by_tag.keys()))
    except (ProviderNotConfigured, WarrantyLookupError) as exc:
        return error_response(ErrorCodes.VALIDATION_ERROR, str(exc), http_code=400)

    created = updated = matched = 0
    now = datetime.now(timezone.utc).replace(tzinfo=None)
    for tag, assetids in by_tag.items():
        found = results.get(tag)
        if not found:
            continue
        matched += 1
        for assetid in assetids:
            # Reuse an existing Dell warranty for this asset if there is one.
            existing = None
            for link in WarrantyAsset.query.filter_by(assetid=assetid).all():
                candidate = db.session.get(Warranty, link.warrantyid)
                # Reuse a warranty that is Dell by ANY signal, not just
                # provider: a manually-added or imported Dell warranty carries
                # provider 'manual' (create default) but the same service tag or
                # a "Dell" vendor. Matching only provider=='dell' made re-check
                # duplicate every one of those instead of updating it.
                if candidate and (
                        candidate.provider == 'dell'
                        or (candidate.servicetag or '').strip().upper() == tag
                        or (candidate.vendor or '').strip().lower() == 'dell'):
                    existing = candidate
                    break
            if existing:
                # Canonicalize to Dell so later re-checks match by provider and
                # never fall through to a duplicate.
                existing.provider = 'dell'
                existing.vendor = existing.vendor or 'Dell'
                existing.servicelevel = found.get('servicelevel')
                existing.startdate = _vendor_date(found.get('startdate'))
                existing.enddate = _vendor_date(found.get('enddate'))
                existing.servicetag = tag
                existing.lastcheckeddate = now
                updated += 1
            else:
                warranty = Warranty(
                    vendor='Dell', provider='dell', servicetag=tag,
                    servicelevel=found.get('servicelevel'),
                    startdate=_vendor_date(found.get('startdate')),
                    enddate=_vendor_date(found.get('enddate')),
                    lastcheckeddate=now,
                )
                warranty.links.append(WarrantyAsset(assetid=assetid))
                db.session.add(warranty)
                created += 1

    db.session.commit()
    return success_response({
        'candidates': sum(len(v) for v in by_tag.values()),
        'tags': len(by_tag),
        'matched': matched,
        'created': created,
        'updated': updated,
    }, message=f'Dell sync: {created} added, {updated} updated, {matched} tags matched.')


# =============================================================================
# Report buckets (for the Reports hub)
# =============================================================================

@warranty_bp.route('/report', methods=['GET'])
@require_permission('warranty.view')
def warranty_report():
    """Counts + lists bucketed by derived status."""
    today = date.today()
    buckets = {'expired': [], 'expiring': [], 'active': [], 'unknown': []}
    for w in Warranty.query.filter_by(isactive=True).all():
        buckets.setdefault(w.status(today), []).append(_warranty_payload(w, today))
    return success_response({
        'counts': {k: len(v) for k, v in buckets.items()},
        'buckets': buckets,
    })


@warranty_bp.route('/dashboard/expiring', methods=['GET'])
@require_permission('warranty.view')
def dashboard_expiring():
    """Warranties running out, and ones that already have.

    Identified the way the floor identifies them: the PC's hostname and the
    MACHINE it drives. Nobody walks out looking for an asset number, and a
    warranty row for a bay PC is only actionable if you know which bay.

    No dates. "Expired" or "expiring" is the whole decision - the exact day
    matters when you are placing the order, which is what the warranty report
    is for, not when you are scanning a board.

    Already-expired stays listed rather than dropping off the day it lapses,
    which is how one gets missed entirely.
    """
    from shopdb.api import Setting

    days = 90
    setting = Setting.query.filter_by(key='warranty_expiringdays').first()
    if setting and (setting.value or '').strip():
        try:
            days = int(setting.value)
        except (TypeError, ValueError):
            pass

    today = date.today()
    horizon = today + timedelta(days=days)

    rows = []
    query = (db.session.query(Warranty, Asset)
             .join(WarrantyAsset, WarrantyAsset.warrantyid == Warranty.warrantyid)
             .join(Asset, Asset.assetid == WarrantyAsset.assetid)
             .filter(Warranty.isactive.is_(True),
                     Warranty.enddate.isnot(None),
                     Warranty.enddate <= horizon,
                     Asset.isactive.is_(True)))

    for warranty, asset in query.all():
        remaining = (warranty.enddate - today).days
        # Reuses the machine lookup behind the warranty page's machine column,
        # so the board and the report agree about which bay a PC belongs to.
        machine = _related_machine(asset)
        rows.append({
            'assetid': asset.assetid,
            # Hostname, else the asset number - never asset.name. The name
            # of a covered PC's ASSET is often the machine's descriptive name
            # ("Haas VF-2"), so falling back to it put a machine name in a
            # column that is supposed to identify the PC.
            'hostname': _hostname(asset) or asset.assetnumber or str(asset.assetid),
            'machinenumber': machine['machinenumber'] if machine else None,
            'state': 'expired' if remaining < 0 else 'expiring',
            'daysleft': remaining,
        })

    rows.sort(key=lambda r: r['daysleft'])
    return success_response(rows[:50])


def _hostname(asset):
    """The PC's hostname, when the covered asset is a PC.

    Optional import: a lean site may not install the computers plugin, and a
    warranty on a printer or a machine has no hostname at all. Falls back to
    the asset number, which is what the row shows either way.
    """
    try:
        from plugins.computers.models import Computer
    except ImportError:
        return None
    computer = Computer.query.filter_by(assetid=asset.assetid).first()
    return computer.hostname if computer else None


# =============================================================================
# Proof of cover
#
# Authenticated on the way in AND on the way out: an invoice carries prices and
# a service tag, so it is not something to serve openly the way an asset photo
# is. Stored as warranty-<id><ext>, one per warranty, with the vendor's own
# filename kept alongside so a person recognises it later.
# =============================================================================

@warranty_bp.route('/<int:warrantyid>/proof', methods=['POST'])
@require_permission('warranty.edit')
def upload_proof(warrantyid):
    """Upload (or replace) the proof-of-cover document for a warranty."""
    warranty = db.session.get(Warranty, warrantyid)
    if not warranty:
        return error_response(ErrorCodes.NOT_FOUND, 'Warranty 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 PROOF_EXTENSIONS:
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            'Unsupported document type {}. Allowed: {}'.format(
                ext, ', '.join(sorted(PROOF_EXTENSIONS))))

    # Seek rather than trust Content-Length: a chunked upload has no length
    # header, and a client can understate the one it sends.
    upload.stream.seek(0, os.SEEK_END)
    size = upload.stream.tell()
    upload.stream.seek(0)
    if size > MAX_PROOF_BYTES:
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            'Document is {:.0f}MB; the limit is {:.0f}MB.'.format(
                size / 1048576, MAX_PROOF_BYTES / 1048576))

    proofdir = _proof_dir()
    os.makedirs(proofdir, exist_ok=True)
    # One proof per warranty: clear any prior extension so a re-upload as .pdf
    # does not leave the old .png behind claiming to be current.
    for old in glob.glob(os.path.join(proofdir,
                                      secure_filename(f'warranty-{warrantyid}') + '.*')):
        os.remove(old)

    filename = secure_filename(f'warranty-{warrantyid}{ext}')
    upload.save(os.path.join(proofdir, filename))

    warranty.proofurl = f'{PROOF_URL_PREFIX}{filename}'
    warranty.prooffilename = upload.filename
    db.session.commit()

    return success_response(warranty.to_dict(), message='Proof uploaded')


@warranty_bp.route('/proof/<path:filename>', methods=['GET'])
@require_permission('warranty.view')
def serve_proof(filename):
    """Download a proof document.

    Authenticated: an invoice carries pricing and a service tag. Sent as an
    attachment under the vendor's original filename where we still have it, so
    a download is recognisable rather than 'warranty-12.pdf'.
    """
    warranty = Warranty.query.filter(
        Warranty.proofurl == f'{PROOF_URL_PREFIX}{filename}').first()
    downloadname = (warranty.prooffilename if warranty and warranty.prooffilename
                    else filename)
    return send_upload(_proof_dir(), filename, as_attachment=True,
                               download_name=downloadname)


@warranty_bp.route('/<int:warrantyid>/proof', methods=['DELETE'])
@require_permission('warranty.edit')
def delete_proof(warrantyid):
    """Remove a warranty's proof document."""
    warranty = db.session.get(Warranty, warrantyid)
    if not warranty:
        return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found',
                              http_code=404)

    for old in glob.glob(os.path.join(_proof_dir(),
                                      secure_filename(f'warranty-{warrantyid}') + '.*')):
        os.remove(old)
    warranty.proofurl = None
    warranty.prooffilename = None
    db.session.commit()

    return success_response(warranty.to_dict(), message='Proof removed')