"""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, send_from_directory from flask_jwt_extended import jwt_required from sqlalchemy.orm import joinedload from werkzeug.utils import secure_filename from shopdb.api import ( db, Asset, success_response, error_response, ErrorCodes, require_permission, ) 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 _parse_date(value): """Accept 'YYYY-MM-DD' (or None/empty) -> date or None.""" if not value: return None try: return datetime.strptime(value[:10], '%Y-%m-%d').date() except (ValueError, TypeError): 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, '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']) @jwt_required(optional=True) 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/', methods=['GET']) @jwt_required(optional=True) 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('/', methods=['GET']) @jwt_required(optional=True) 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']) @jwt_required() @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') 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=_parse_date(data.get('startdate')), enddate=_parse_date(data.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('/', methods=['PUT']) @jwt_required() @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 {} 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 if 'startdate' in data: warranty.startdate = _parse_date(data['startdate']) if 'enddate' in data: warranty.enddate = _parse_date(data['enddate']) 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('/', methods=['DELETE']) @jwt_required() @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('//refresh', methods=['POST']) @jwt_required() @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 = _parse_date(result['startdate']) if result.get('enddate'): warranty.enddate = _parse_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']) @jwt_required() @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 = _parse_date(found.get('startdate')) existing.enddate = _parse_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=_parse_date(found.get('startdate')), enddate=_parse_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']) @jwt_required(optional=True) 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']) @jwt_required() @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-, one per warranty, with the vendor's own # filename kept alongside so a person recognises it later. # ============================================================================= @warranty_bp.route('//proof', methods=['POST']) @jwt_required() @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/', methods=['GET']) @jwt_required() @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_from_directory(_proof_dir(), filename, as_attachment=True, download_name=downloadname) @warranty_bp.route('//proof', methods=['DELETE']) @jwt_required() @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')