""" PartsInventory blueprint: Flask port of the PartsInventory server APIs. Serves API endpoints and a built React SPA under /partsinventory. """ import csv import glob import io import json import os import re import shutil import urllib.parse import urllib.request from datetime import datetime import pymysql from flask import Blueprint, Response, jsonify, redirect, request, send_file, send_from_directory from werkzeug.utils import secure_filename try: import fitz # PyMuPDF except Exception: # optional dependency for datasheet thumbnails fitz = None from config import ( PARTSINV_CLIENT_DIST, PARTSINV_DB_HOST, PARTSINV_DB_NAME, PARTSINV_DB_PASSWORD, PARTSINV_DB_PORT, PARTSINV_DB_USER, PARTSINV_IMAGE_SEARCH_API_KEY, PARTSINV_IMAGE_SEARCH_CX, PARTSINV_UPLOAD_DIR, ) bp = Blueprint("partsinventory", __name__) DB_CONFIG = { "host": PARTSINV_DB_HOST, "port": PARTSINV_DB_PORT, "user": PARTSINV_DB_USER, "password": PARTSINV_DB_PASSWORD, "database": PARTSINV_DB_NAME, "cursorclass": pymysql.cursors.DictCursor, "autocommit": False, } _DB_OVERRIDE_FILE = os.path.join(os.path.dirname(os.path.dirname(__file__)), "partsinventory", "data", "db-config.json") _UPLOAD_OVERRIDE_FILE = _DB_OVERRIDE_FILE _KICAD_TREE_API_URL = "https://api.github.com/repos/KiCad/kicad-footprints/git/trees/master?recursive=1" _DEFAULT_FOOTPRINTS = [ "0402", "0603", "0805", "1206", "1210", "1812", "2010", "2512", "SOT-23", "SOT-223", "SOT-89", "SOT-143", "SOT-323", "SOT-363", "SOD-123", "SOD-323", "SOD-523", "DO-214AA (SMB)", "DO-214AB (SMC)", "DO-214AC (SMA)", "TO-220", "TO-92", "QFN-16", "QFN-24", "QFN-32", "QFN-48", "QFP-32", "QFP-44", "QFP-64", "TQFP-32", "TQFP-44", "TQFP-64", "SOIC-8", "SOIC-14", "SOIC-16", "SOP-8", "TSSOP-8", "TSSOP-14", "TSSOP-16", "SSOP-16", "MSOP-8", "DFN-8", "DIP-8", "DIP-14", "DIP-16", "LQFP-64", "LQFP-100", "BGA-64", "BGA-100", "USB-C Receptacle", ] def _fetch_kicad_footprints(): req = urllib.request.Request(_KICAD_TREE_API_URL, headers={"User-Agent": "PartsInventory/1.0"}) with urllib.request.urlopen(req, timeout=15) as response: payload = json.loads(response.read().decode("utf-8")) names = [] for item in payload.get("tree", []): path = item.get("path") or "" if not path.endswith(".kicad_mod"): continue if ".pretty/" not in path: continue lib, mod = path.split(".pretty/", 1) fp_name = mod.rsplit(".kicad_mod", 1)[0] if not lib or not fp_name: continue names.append(f"{lib}:{fp_name}") # Preserve order but de-duplicate in case upstream paths ever overlap. return list(dict.fromkeys(names)) def _sync_kicad_footprints(cur): kicad_names = _fetch_kicad_footprints() if not kicad_names: return {"added": 0, "reactivated": 0, "total": 0} cur.execute("SELECT name, COALESCE(is_active, 0) AS is_active FROM footprints WHERE source = 'kicad'") existing_rows = cur.fetchall() existing = {r["name"]: int(r.get("is_active") or 0) for r in existing_rows} added = 0 reactivated = 0 payload = [] base_sort = len(_DEFAULT_FOOTPRINTS) for i, name in enumerate(kicad_names): was_active = existing.get(name) if was_active is None: added += 1 elif was_active == 0: reactivated += 1 payload.append((name, base_sort + i)) if payload: cur.executemany( """ INSERT INTO footprints (name, source, sort_order, is_active) VALUES (%s, 'kicad', %s, 1) ON DUPLICATE KEY UPDATE source = VALUES(source), sort_order = VALUES(sort_order), is_active = VALUES(is_active) """, payload, ) return {"added": added, "reactivated": reactivated, "total": len(kicad_names)} def _effective_ui_settings(): override = _load_override() return { "showQrCode": bool(override.get("showQrCode", True)), } def _load_override(): if not os.path.exists(_DB_OVERRIDE_FILE): return {} try: with open(_DB_OVERRIDE_FILE, "r", encoding="utf-8") as f: return json.load(f) except Exception: return {} def _save_override(data): os.makedirs(os.path.dirname(_DB_OVERRIDE_FILE), exist_ok=True) with open(_DB_OVERRIDE_FILE, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) def _effective_db_config(): override = _load_override() return { "host": override.get("host", PARTSINV_DB_HOST), "port": int(override.get("port", PARTSINV_DB_PORT)), "user": override.get("user", PARTSINV_DB_USER), "password": override.get("password", PARTSINV_DB_PASSWORD), "database": override.get("database", PARTSINV_DB_NAME), "cursorclass": pymysql.cursors.DictCursor, "autocommit": False, } def _effective_upload_dir(): override = _load_override() upload_dir = (override.get("uploadDir") or PARTSINV_UPLOAD_DIR or "partsinventory_uploads").strip() if os.path.isabs(upload_dir): return upload_dir base = os.path.dirname(os.path.dirname(__file__)) return os.path.abspath(os.path.join(base, upload_dir)) def _resolve_existing_upload_file(upload_root, stored_path): """Find an on-disk upload file for a DB path across legacy path variants.""" raw = (stored_path or "").replace("\\", "/").strip() if not raw: return None variants = [] seen = set() def add(rel): rel_norm = (rel or "").replace("\\", "/").strip().lstrip("/") if not rel_norm or rel_norm in seen: return seen.add(rel_norm) variants.append(rel_norm) add(raw) if raw.startswith("uploads/"): add(raw[len("uploads/"):]) else: add(f"uploads/{raw}") for rel in variants: candidate = os.path.join(upload_root, rel) if os.path.isfile(candidate): return candidate # Legacy edge case: stored stem exists but extension differs/missing. for rel in variants: rel_dir, rel_name = os.path.split(rel) stem, _ext = os.path.splitext(rel_name) if not stem: continue abs_dir = os.path.join(upload_root, rel_dir) if not os.path.isdir(abs_dir): continue for match in glob.glob(os.path.join(abs_dir, f"{stem}.*")): if os.path.isfile(match): return match return None def _get_db(): return pymysql.connect(**_effective_db_config()) def _db_health_payload(): cfg = _effective_db_config() return { "host": cfg["host"], "port": cfg["port"], "user": cfg["user"], "database": cfg["database"], "password": "****" if cfg["password"] else None, } def _sql_dump_for_current_db(conn): out = io.StringIO() out.write("-- PartsInventory full backup\n") out.write(f"-- Generated at {datetime.now().isoformat()}\n\n") out.write("SET FOREIGN_KEY_CHECKS = 0;\n\n") with conn.cursor() as cur: cur.execute( """ SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_TYPE = 'BASE TABLE' ORDER BY TABLE_NAME """ ) table_rows = cur.fetchall() table_names = [r["TABLE_NAME"] for r in table_rows] for table_name in table_names: cur.execute(f"SHOW CREATE TABLE `{table_name}`") create_row = cur.fetchone() create_sql = create_row.get("Create Table") if create_row else None if not create_sql: continue out.write(f"-- Table: {table_name}\n") out.write(f"DROP TABLE IF EXISTS `{table_name}`;\n") out.write(f"{create_sql};\n\n") cur.execute(f"SELECT * FROM `{table_name}`") rows = cur.fetchall() if not rows: continue columns = list(rows[0].keys()) col_sql = ", ".join([f"`{c}`" for c in columns]) for row in rows: vals = ", ".join([conn.literal(row.get(c)) for c in columns]) out.write(f"INSERT INTO `{table_name}` ({col_sql}) VALUES ({vals});\n") out.write("\n") out.write("SET FOREIGN_KEY_CHECKS = 1;\n") return out.getvalue() def _ensure_tables(conn): with conn.cursor() as cur: cur.execute( """ CREATE TABLE IF NOT EXISTS footprints ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255) NOT NULL, source VARCHAR(20) NOT NULL DEFAULT 'custom', sort_order INT NOT NULL DEFAULT 0, is_active TINYINT(1) NOT NULL DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uq_footprints_name (name), INDEX idx_footprints_active_sort (is_active, sort_order, name) ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS categories ( id INT AUTO_INCREMENT PRIMARY KEY, parent_id INT NULL, source_partkeepr_id INT NULL, name VARCHAR(255) NOT NULL, description TEXT, sort_order INT DEFAULT 0, FOREIGN KEY (parent_id) REFERENCES categories(id) ON DELETE SET NULL, INDEX idx_parent (parent_id), UNIQUE KEY uq_categories_source_partkeepr_id (source_partkeepr_id) ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS storage_locations ( id INT AUTO_INCREMENT PRIMARY KEY, parent_id INT NULL, source_partkeepr_id INT NULL, name VARCHAR(255) NOT NULL, description TEXT, sort_order INT DEFAULT 0, FOREIGN KEY (parent_id) REFERENCES storage_locations(id) ON DELETE SET NULL, INDEX idx_parent (parent_id), UNIQUE KEY uq_storage_locations_source_partkeepr_id (source_partkeepr_id) ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS parts ( id INT AUTO_INCREMENT PRIMARY KEY, category_id INT NULL, storage_location_id INT NULL, overstock_location_id INT NULL, source_partkeepr_id INT NULL, name VARCHAR(255) NOT NULL DEFAULT '', description TEXT, part_number VARCHAR(255) NOT NULL DEFAULT '', quantity INT NOT NULL DEFAULT 0, overstock_quantity INT NOT NULL DEFAULT 0, unit VARCHAR(50) DEFAULT '', average_price DECIMAL(12,4) NOT NULL DEFAULT 0, minimum_stock_level INT NOT NULL DEFAULT 0, comment TEXT, barcode VARCHAR(255) NULL, datasheet_url VARCHAR(1024) NULL, datasheet_file_path VARCHAR(1024) NULL, datasheet_thumbnail_path VARCHAR(1024) NULL, manufacturer VARCHAR(255) NULL, footprint VARCHAR(255) NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, FOREIGN KEY (category_id) REFERENCES categories(id) ON DELETE SET NULL, FOREIGN KEY (storage_location_id) REFERENCES storage_locations(id) ON DELETE SET NULL, FOREIGN KEY (overstock_location_id) REFERENCES storage_locations(id) ON DELETE SET NULL, INDEX idx_category (category_id), INDEX idx_storage (storage_location_id), INDEX idx_overstock_storage (overstock_location_id), INDEX idx_barcode (barcode), UNIQUE KEY uq_parts_source_partkeepr_id (source_partkeepr_id), UNIQUE KEY unique_barcode (barcode) ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS part_images ( id INT AUTO_INCREMENT PRIMARY KEY, part_id INT NOT NULL, file_path VARCHAR(1024) NULL, original_filename VARCHAR(255) NULL, url VARCHAR(1024) NULL, source ENUM('uploaded', 'internet') NOT NULL, is_primary TINYINT(1) NOT NULL DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (part_id) REFERENCES parts(id) ON DELETE CASCADE, INDEX idx_part (part_id) ) """ ) try: cur.execute("ALTER TABLE part_images ADD COLUMN original_filename VARCHAR(255) NULL AFTER file_path") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE categories ADD COLUMN source_partkeepr_id INT NULL") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE categories ADD UNIQUE KEY uq_categories_source_partkeepr_id (source_partkeepr_id)") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE storage_locations ADD COLUMN source_partkeepr_id INT NULL") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE storage_locations ADD UNIQUE KEY uq_storage_locations_source_partkeepr_id (source_partkeepr_id)") except pymysql.OperationalError: pass try: cur.execute( "ALTER TABLE storage_locations ADD COLUMN slot_unusable TINYINT(1) NOT NULL DEFAULT 0 AFTER sort_order" ) except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE parts ADD COLUMN source_partkeepr_id INT NULL") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE parts ADD COLUMN overstock_location_id INT NULL AFTER storage_location_id") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE parts ADD COLUMN overstock_quantity INT NOT NULL DEFAULT 0 AFTER quantity") except pymysql.OperationalError: pass try: cur.execute( "ALTER TABLE parts ADD CONSTRAINT fk_parts_overstock_location FOREIGN KEY (overstock_location_id) REFERENCES storage_locations(id) ON DELETE SET NULL" ) except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE parts ADD INDEX idx_overstock_storage (overstock_location_id)") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE parts ADD UNIQUE KEY uq_parts_source_partkeepr_id (source_partkeepr_id)") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE parts ADD COLUMN average_price DECIMAL(12,4) NOT NULL DEFAULT 0 AFTER unit") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE parts ADD COLUMN minimum_stock_level INT NOT NULL DEFAULT 0 AFTER average_price") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE parts ADD COLUMN comment TEXT NULL AFTER minimum_stock_level") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE parts ADD COLUMN datasheet_thumbnail_path VARCHAR(1024) NULL AFTER datasheet_file_path") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE parts ADD COLUMN footprint VARCHAR(255) NULL AFTER manufacturer") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE parts MODIFY COLUMN footprint VARCHAR(255) NULL") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE footprints ADD COLUMN source VARCHAR(20) NOT NULL DEFAULT 'custom' AFTER name") except pymysql.OperationalError: pass try: cur.execute("ALTER TABLE footprints MODIFY COLUMN name VARCHAR(255) NOT NULL") except pymysql.OperationalError: pass for i, name in enumerate(_DEFAULT_FOOTPRINTS): cur.execute( """ INSERT INTO footprints (name, source, sort_order, is_active) VALUES (%s, 'default', %s, 1) ON DUPLICATE KEY UPDATE sort_order = VALUES(sort_order), is_active = VALUES(is_active) """, (name, i), ) conn.commit() def _dateify(row): out = dict(row) for key, value in list(out.items()): if isinstance(value, datetime): out[key] = value.isoformat() return out @bp.route("/api/health", methods=["GET"]) def health(): return jsonify({"ok": True}), 200 @bp.route("/api/categories", methods=["GET", "POST"]) def categories_root(): conn = _get_db() try: _ensure_tables(conn) with conn.cursor() as cur: if request.method == "GET": cur.execute("SELECT id, parent_id, name, description, sort_order FROM categories ORDER BY sort_order, name") rows = cur.fetchall() return jsonify(rows) payload = request.get_json(silent=True) or {} cur.execute( "INSERT INTO categories (parent_id, name, description, sort_order) VALUES (%s, %s, %s, %s)", (payload.get("parent_id"), payload.get("name", ""), payload.get("description", ""), payload.get("sort_order", 0)), ) cur.execute("SELECT * FROM categories WHERE id = %s", (cur.lastrowid,)) conn.commit() return jsonify(cur.fetchone()), 201 except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/categories/", methods=["PUT", "DELETE"]) def categories_id(item_id): conn = _get_db() try: _ensure_tables(conn) with conn.cursor() as cur: if request.method == "PUT": payload = request.get_json(silent=True) or {} cur.execute( "UPDATE categories SET parent_id = %s, name = %s, description = %s, sort_order = %s WHERE id = %s", (payload.get("parent_id"), payload.get("name"), payload.get("description"), payload.get("sort_order", 0), item_id), ) cur.execute("SELECT * FROM categories WHERE id = %s", (item_id,)) row = cur.fetchone() if not row: conn.rollback() return jsonify({"error": "Not found"}), 404 conn.commit() return jsonify(row) cur.execute("DELETE FROM categories WHERE id = %s", (item_id,)) if cur.rowcount == 0: conn.rollback() return jsonify({"error": "Not found"}), 404 conn.commit() return Response(status=204) except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() def _storage_location_slot_unusable(cur, location_id): if location_id is None: return False cur.execute( "SELECT COALESCE(slot_unusable, 0) AS u FROM storage_locations WHERE id = %s", (int(location_id),), ) row = cur.fetchone() return bool(row and row.get("u")) @bp.route("/api/storage-locations", methods=["GET", "POST"]) def storage_locations_root(): conn = _get_db() try: _ensure_tables(conn) with conn.cursor() as cur: if request.method == "GET": cur.execute( "SELECT id, parent_id, name, description, sort_order, slot_unusable FROM storage_locations ORDER BY sort_order, name" ) rows = cur.fetchall() return jsonify(rows) payload = request.get_json(silent=True) or {} cur.execute( "INSERT INTO storage_locations (parent_id, name, description, sort_order) VALUES (%s, %s, %s, %s)", (payload.get("parent_id"), payload.get("name", ""), payload.get("description", ""), payload.get("sort_order", 0)), ) cur.execute("SELECT * FROM storage_locations WHERE id = %s", (cur.lastrowid,)) conn.commit() return jsonify(cur.fetchone()), 201 except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() _BATCH_BIN_SPECS = { "B": {"prefix": "B", "num_digits": 4, "rows": 4, "cols": 6}, "BA": {"prefix": "BA", "num_digits": 3, "rows": 3, "cols": 6}, "BC": {"prefix": "BC", "num_digits": 3, "rows": 5, "cols": 7}, } def _batch_bin_max_serial(names, bin_type): spec = _BATCH_BIN_SPECS[bin_type] prefix = spec["prefix"] pattern = re.compile(rf"^{re.escape(prefix)}(\d+)-") max_n = 0 for raw in names: name = (raw or "").strip() m = pattern.match(name) if m: max_n = max(max_n, int(m.group(1), 10)) return max_n def _batch_bin_slot_names(bin_base, rows, cols): out = [] for r in range(rows): row_letter = chr(ord("A") + r) for c in range(1, cols + 1): out.append(f"{bin_base}-{row_letter}{c}") return out @bp.route("/api/storage-locations/batch-bins", methods=["POST"]) def storage_locations_batch_bins(): """Create one or more standard bins (grid of pocket locations). Types: B 4×6, BA 3×6, BC 5×7.""" conn = _get_db() try: _ensure_tables(conn) payload = request.get_json(silent=True) or {} bin_type = (payload.get("bin_type") or "").strip().upper() try: count = int(payload.get("count", 1)) except (TypeError, ValueError): return jsonify({"error": "count must be an integer"}), 400 if bin_type not in _BATCH_BIN_SPECS: return jsonify({"error": 'bin_type must be "B", "BA", or "BC"'}), 400 if count < 1 or count > 50: return jsonify({"error": "count must be between 1 and 50"}), 400 spec = _BATCH_BIN_SPECS[bin_type] prefix = spec["prefix"] num_digits = spec["num_digits"] rows = spec["rows"] cols = spec["cols"] with conn.cursor() as cur: cur.execute("SELECT name FROM storage_locations") existing_rows = cur.fetchall() existing_names = {(r.get("name") or "").strip() for r in existing_rows} start_serial = _batch_bin_max_serial(existing_names, bin_type) + 1 names_to_insert = [] bin_bases = [] for i in range(count): serial = start_serial + i serial_str = str(serial) if len(serial_str) > num_digits: return jsonify( {"error": f"Next bin number {serial} exceeds {num_digits} digit width for type {bin_type}"} ), 400 bin_base = prefix + serial_str.zfill(num_digits) bin_bases.append(bin_base) for slot in _batch_bin_slot_names(bin_base, rows, cols): if slot in existing_names: return jsonify({"error": f"Location already exists: {slot}"}), 400 names_to_insert.append(slot) dup_check = set() for n in names_to_insert: if n in dup_check: return jsonify({"error": f"Internal duplicate slot name: {n}"}), 400 dup_check.add(n) insert_rows = [(None, name, "", 0) for name in names_to_insert] cur.executemany( "INSERT INTO storage_locations (parent_id, name, description, sort_order) VALUES (%s, %s, %s, %s)", insert_rows, ) conn.commit() return ( jsonify( { "created": len(names_to_insert), "bin_type": bin_type, "bins": bin_bases, "slots_per_bin": rows * cols, } ), 201, ) except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/storage-locations/", methods=["PUT", "DELETE"]) def storage_locations_id(item_id): conn = _get_db() try: _ensure_tables(conn) with conn.cursor() as cur: if request.method == "PUT": payload = request.get_json(silent=True) or {} cur.execute( "UPDATE storage_locations SET parent_id = %s, name = %s, description = %s, sort_order = %s WHERE id = %s", (payload.get("parent_id"), payload.get("name"), payload.get("description"), payload.get("sort_order", 0), item_id), ) cur.execute("SELECT * FROM storage_locations WHERE id = %s", (item_id,)) row = cur.fetchone() if not row: conn.rollback() return jsonify({"error": "Not found"}), 404 conn.commit() return jsonify(row) cur.execute("DELETE FROM storage_locations WHERE id = %s", (item_id,)) if cur.rowcount == 0: conn.rollback() return jsonify({"error": "Not found"}), 404 conn.commit() return Response(status=204) except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/storage-locations//slot-unusable", methods=["PATCH"]) def storage_location_slot_unusable_patch(item_id): """Mark a bin pocket unusable (empty slots only) or usable again.""" conn = _get_db() try: _ensure_tables(conn) payload = request.get_json(silent=True) or {} if "slot_unusable" not in payload: return jsonify({"error": "slot_unusable required"}), 400 want = bool(payload.get("slot_unusable")) with conn.cursor() as cur: cur.execute("SELECT id FROM storage_locations WHERE id = %s", (item_id,)) if not cur.fetchone(): return jsonify({"error": "Not found"}), 404 if want: cur.execute( "SELECT id FROM parts WHERE storage_location_id = %s LIMIT 1", (item_id,), ) if cur.fetchone(): return jsonify({"error": "Cannot mark unusable while a part occupies this slot"}), 400 cur.execute( "UPDATE storage_locations SET slot_unusable = %s WHERE id = %s", (1 if want else 0, item_id), ) cur.execute("SELECT * FROM storage_locations WHERE id = %s", (item_id,)) row = cur.fetchone() conn.commit() return jsonify(row) except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/footprints", methods=["GET"]) def footprints_root(): conn = _get_db() try: _ensure_tables(conn) with conn.cursor() as cur: q = (request.args.get("q") or "").strip() try: limit = int(request.args.get("limit", 100)) except (TypeError, ValueError): limit = 100 limit = max(1, min(limit, 500)) if q: cur.execute( """ SELECT id, name, sort_order, is_active FROM footprints WHERE is_active = 1 AND name LIKE %s ORDER BY CASE WHEN LOWER(name) = LOWER(%s) THEN 0 ELSE 1 END, CASE WHEN LOWER(name) LIKE LOWER(%s) THEN 0 ELSE 1 END, sort_order, name LIMIT %s """, (f"%{q}%", q, f"{q}%", limit), ) else: cur.execute( """ SELECT id, name, sort_order, is_active FROM footprints WHERE is_active = 1 ORDER BY sort_order, name LIMIT %s """, (limit,), ) return jsonify(cur.fetchall()) except Exception as e: return jsonify({"error": str(e)}), 500 finally: conn.close() def _normalize_footprint(cur, raw_value): raw = (raw_value or "").strip() if not raw: return None cur.execute("SELECT name FROM footprints WHERE LOWER(name) = LOWER(%s) LIMIT 1", (raw,)) hit = cur.fetchone() if hit and hit.get("name"): return hit["name"] return raw def _category_self_and_descendant_ids(cur, root_id): """IDs of root category and all descendants (empty if root_id is missing from DB).""" try: rid = int(root_id) except (TypeError, ValueError): return [] cur.execute("SELECT id, parent_id FROM categories") rows = cur.fetchall() known = {r["id"] for r in rows} if rid not in known: return [] children = {} for r in rows: cid = r["id"] pid = r.get("parent_id") if pid not in children: children[pid] = [] children[pid].append(cid) out = [] stack = [rid] seen = set() while stack: cid = stack.pop() if cid in seen: continue seen.add(cid) out.append(cid) for ch in children.get(cid, ()): stack.append(ch) return out @bp.route("/api/parts/stock-value", methods=["GET"]) def parts_stock_value(): """Aggregate inventory value: sum of (quantity + overstock_quantity) × average_price for all parts.""" conn = _get_db() try: _ensure_tables(conn) with conn.cursor() as cur: cur.execute( "SELECT COALESCE(SUM((quantity + COALESCE(overstock_quantity, 0)) * average_price), 0) AS total_stock_value FROM parts" ) row = cur.fetchone() raw = row.get("total_stock_value") if row else 0 if raw is None: raw = 0 return jsonify({"totalStockValue": float(raw)}) except Exception as e: return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/parts", methods=["GET", "POST"]) def parts_root(): conn = _get_db() try: _ensure_tables(conn) with conn.cursor() as cur: if request.method == "GET": category_id = request.args.get("categoryId") storage_location_id = request.args.get("storageLocationId") storage_location_prefix = request.args.get("storageLocationPrefix") q = request.args.get("q") sql = """ SELECT p.id, p.category_id, p.storage_location_id, p.name, p.description, p.part_number, p.quantity, p.overstock_location_id, p.overstock_quantity, p.unit, p.average_price, p.minimum_stock_level, p.comment, p.barcode, p.datasheet_url, p.datasheet_file_path, p.manufacturer, p.footprint, p.created_at, p.updated_at, c.name AS category_name, s.name AS storage_location_name, so.name AS overstock_location_name, pi.url AS primary_image_url, pi.file_path AS primary_image_file_path, pi2.url AS secondary_image_url, pi2.file_path AS secondary_image_file_path FROM parts p LEFT JOIN categories c ON p.category_id = c.id LEFT JOIN storage_locations s ON p.storage_location_id = s.id LEFT JOIN storage_locations so ON p.overstock_location_id = so.id LEFT JOIN part_images pi ON pi.id = ( SELECT i.id FROM part_images i WHERE i.part_id = p.id ORDER BY i.is_primary DESC, i.id LIMIT 1 ) LEFT JOIN part_images pi2 ON pi2.id = ( SELECT i2.id FROM part_images i2 WHERE i2.part_id = p.id ORDER BY i2.is_primary DESC, i2.id LIMIT 1 OFFSET 1 ) WHERE 1=1 """ params = [] cat_key = (category_id or "").strip().lower() if cat_key == "unassigned": sql += " AND p.category_id IS NULL" elif category_id: cat_ids = _category_self_and_descendant_ids(cur, category_id) if not cat_ids: sql += " AND 1=0" else: sql += " AND p.category_id IN (" + ",".join(["%s"] * len(cat_ids)) + ")" params.extend(cat_ids) if storage_location_id: sql += " AND p.storage_location_id = %s" params.append(storage_location_id) if storage_location_prefix: sql += " AND (s.name = %s OR s.name LIKE %s)" params.extend([storage_location_prefix, f"{storage_location_prefix}-%"]) if q: sql += " AND (p.name LIKE %s OR p.part_number LIKE %s OR p.description LIKE %s OR p.barcode LIKE %s)" like = f"%{q}%" params.extend([like, like, like, like]) sql += " ORDER BY p.name" cur.execute(sql, params) return jsonify([_dateify(r) for r in cur.fetchall()]) payload = request.get_json(silent=True) or {} footprint = _normalize_footprint(cur, payload.get("footprint")) new_loc = payload.get("storage_location_id") overstock_location_id = payload.get("overstock_location_id") if overstock_location_id in ("", None): overstock_location_id = None else: try: overstock_location_id = int(overstock_location_id) except (TypeError, ValueError): conn.rollback() return jsonify({"error": "overstock_location_id must be an integer or null"}), 400 try: overstock_quantity = int(payload.get("overstock_quantity", 0) or 0) except (TypeError, ValueError): conn.rollback() return jsonify({"error": "overstock_quantity must be a non-negative integer"}), 400 if new_loc is not None and _storage_location_slot_unusable(cur, new_loc): conn.rollback() return jsonify({"error": "Cannot assign part to a storage slot marked unusable"}), 400 if overstock_quantity < 0: conn.rollback() return jsonify({"error": "overstock_quantity must be a non-negative integer"}), 400 if overstock_location_id is not None: cur.execute("SELECT id FROM storage_locations WHERE id = %s", (overstock_location_id,)) if not cur.fetchone(): conn.rollback() return jsonify({"error": "Overstock location not found"}), 400 cur.execute( """ INSERT INTO parts (category_id, storage_location_id, overstock_location_id, name, description, part_number, quantity, overstock_quantity, unit, average_price, minimum_stock_level, comment, barcode, datasheet_url, manufacturer, footprint) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( payload.get("category_id"), payload.get("storage_location_id"), overstock_location_id, payload.get("name", ""), payload.get("description", ""), payload.get("part_number", ""), payload.get("quantity", 0), overstock_quantity, payload.get("unit", ""), payload.get("average_price", 0), payload.get("minimum_stock_level", 0), payload.get("comment", ""), payload.get("barcode"), payload.get("datasheet_url"), payload.get("manufacturer"), footprint, ), ) cur.execute("SELECT * FROM parts WHERE id = %s", (cur.lastrowid,)) conn.commit() return jsonify(_dateify(cur.fetchone())), 201 except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/parts/by-barcode/", methods=["GET"]) def parts_by_barcode(barcode): conn = _get_db() try: with conn.cursor() as cur: cur.execute("SELECT id FROM parts WHERE barcode = %s LIMIT 1", (barcode,)) row = cur.fetchone() if not row: return jsonify({"error": "Not found"}), 404 return jsonify({"id": row["id"]}) except Exception as e: return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/parts/export", methods=["GET"]) def parts_export(): if request.args.get("format") != "csv": return jsonify({"error": "format=csv required"}), 400 conn = _get_db() try: with conn.cursor() as cur: category_id = request.args.get("categoryId") storage_location_id = request.args.get("storageLocationId") storage_location_prefix = request.args.get("storageLocationPrefix") q = request.args.get("q") sql = """ SELECT p.part_number, p.name, p.barcode, p.quantity, p.overstock_quantity, p.unit, p.average_price, p.minimum_stock_level, p.comment, p.description, p.datasheet_url, p.manufacturer, p.footprint, c.name AS category_name, s.name AS storage_location_name, so.name AS overstock_location_name FROM parts p LEFT JOIN categories c ON p.category_id = c.id LEFT JOIN storage_locations s ON p.storage_location_id = s.id LEFT JOIN storage_locations so ON p.overstock_location_id = so.id WHERE 1=1 """ params = [] cat_key = (category_id or "").strip().lower() if cat_key == "unassigned": sql += " AND p.category_id IS NULL" elif category_id: cat_ids = _category_self_and_descendant_ids(cur, category_id) if not cat_ids: sql += " AND 1=0" else: sql += " AND p.category_id IN (" + ",".join(["%s"] * len(cat_ids)) + ")" params.extend(cat_ids) if storage_location_id: sql += " AND p.storage_location_id = %s" params.append(storage_location_id) if storage_location_prefix: sql += " AND (s.name = %s OR s.name LIKE %s)" params.extend([storage_location_prefix, f"{storage_location_prefix}-%"]) if q: sql += " AND (p.name LIKE %s OR p.part_number LIKE %s)" like = f"%{q}%" params.extend([like, like]) cur.execute(sql, params) rows = cur.fetchall() out = io.StringIO() writer = csv.writer(out, lineterminator="\r\n") writer.writerow( [ "part_number", "name", "barcode", "quantity", "overstock_quantity", "unit", "average_price", "minimum_stock_level", "comment", "description", "datasheet_url", "manufacturer", "footprint", "category_name", "storage_location_name", "overstock_location_name", ] ) for r in rows: writer.writerow( [ r.get("part_number"), r.get("name"), r.get("barcode"), r.get("quantity"), r.get("overstock_quantity"), r.get("unit"), r.get("average_price"), r.get("minimum_stock_level"), r.get("comment"), r.get("description"), r.get("datasheet_url"), r.get("manufacturer"), r.get("footprint"), r.get("category_name"), r.get("storage_location_name"), r.get("overstock_location_name"), ] ) return Response( out.getvalue(), mimetype="text/csv", headers={"Content-Disposition": "attachment; filename=parts-export.csv"}, ) except Exception as e: return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/parts/import", methods=["POST"]) def parts_import(): file = request.files.get("file") if not file: return jsonify({"error": "No file uploaded"}), 400 text = file.read().decode("utf-8", errors="ignore") reader = csv.DictReader(io.StringIO(text)) rows = list(reader) if not rows: return jsonify({"created": 0, "updated": 0, "errors": ["No data rows"]}) conn = _get_db() created = 0 updated = 0 errors = [] try: with conn.cursor() as cur: for i, row in enumerate(rows, start=2): part_number = (row.get("part_number") or "").strip() name = (row.get("name") or "").strip() if not part_number and not name: continue barcode = (row.get("barcode") or "").strip() or None try: quantity = int((row.get("quantity") or "0").strip() or "0") except ValueError: quantity = 0 try: overstock_quantity = int((row.get("overstock_quantity") or "0").strip() or "0") except ValueError: overstock_quantity = 0 unit = (row.get("unit") or "").strip() try: average_price = float((row.get("average_price") or "0").strip() or "0") except ValueError: average_price = 0 try: minimum_stock_level = int((row.get("minimum_stock_level") or "0").strip() or "0") except ValueError: minimum_stock_level = 0 comment = (row.get("comment") or "").strip() description = (row.get("description") or "").strip() datasheet_url = (row.get("datasheet_url") or "").strip() or None manufacturer = (row.get("manufacturer") or "").strip() or None footprint = _normalize_footprint(cur, row.get("footprint")) category_name = (row.get("category_name") or "").strip() location_name = (row.get("storage_location_name") or "").strip() overstock_location_name = (row.get("overstock_location_name") or "").strip() category_id = None storage_location_id = None overstock_location_id = None if category_name: cur.execute("SELECT id FROM categories WHERE name = %s LIMIT 1", (category_name,)) cat = cur.fetchone() if cat: category_id = cat["id"] if location_name: cur.execute("SELECT id FROM storage_locations WHERE name = %s LIMIT 1", (location_name,)) loc = cur.fetchone() if loc: storage_location_id = loc["id"] if overstock_location_name: cur.execute("SELECT id FROM storage_locations WHERE name = %s LIMIT 1", (overstock_location_name,)) over_loc = cur.fetchone() if over_loc: overstock_location_id = over_loc["id"] try: cur.execute( "SELECT id FROM parts WHERE part_number = %s OR (part_number = '' AND barcode = %s) LIMIT 1", (part_number, barcode or ""), ) existing = cur.fetchone() if existing: cur.execute( """ UPDATE parts SET category_id = %s, storage_location_id = %s, name = %s, description = %s, overstock_location_id = %s, quantity = %s, overstock_quantity = %s, unit = %s, average_price = %s, minimum_stock_level = %s, comment = %s, barcode = %s, datasheet_url = %s, manufacturer = %s, footprint = %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s """, ( category_id, storage_location_id, name, description, overstock_location_id, quantity, max(0, overstock_quantity), unit, average_price, minimum_stock_level, comment, barcode, datasheet_url, manufacturer, footprint, existing["id"], ), ) updated += 1 else: cur.execute( """ INSERT INTO parts (category_id, storage_location_id, overstock_location_id, name, description, part_number, quantity, overstock_quantity, unit, average_price, minimum_stock_level, comment, barcode, datasheet_url, manufacturer, footprint) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( category_id, storage_location_id, overstock_location_id, name, description, part_number, quantity, max(0, overstock_quantity), unit, average_price, minimum_stock_level, comment, barcode, datasheet_url, manufacturer, footprint, ), ) created += 1 except Exception as row_err: errors.append(f"Row {i}: {row_err}") conn.commit() return jsonify({"created": created, "updated": updated, "errors": errors}) except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/parts/", methods=["GET", "PUT", "DELETE"]) def parts_id(part_id): conn = _get_db() try: _ensure_tables(conn) with conn.cursor() as cur: if request.method == "GET": cur.execute( """ SELECT p.*, c.name AS category_name, s.name AS storage_location_name , so.name AS overstock_location_name FROM parts p LEFT JOIN categories c ON p.category_id = c.id LEFT JOIN storage_locations s ON p.storage_location_id = s.id LEFT JOIN storage_locations so ON p.overstock_location_id = so.id WHERE p.id = %s """, (part_id,), ) part = cur.fetchone() if not part: return jsonify({"error": "Not found"}), 404 cur.execute( """ SELECT id, file_path, original_filename, url, source, is_primary, created_at FROM part_images WHERE part_id = %s ORDER BY is_primary DESC, id """, (part_id,), ) part["images"] = [_dateify(r) for r in cur.fetchall()] return jsonify(_dateify(part)) if request.method == "PUT": payload = request.get_json(silent=True) or {} footprint = _normalize_footprint(cur, payload.get("footprint")) cur.execute("SELECT storage_location_id FROM parts WHERE id = %s", (part_id,)) existing_row = cur.fetchone() if not existing_row: conn.rollback() return jsonify({"error": "Not found"}), 404 old_loc = existing_row.get("storage_location_id") new_loc = payload.get("storage_location_id") overstock_location_id = payload.get("overstock_location_id") if overstock_location_id in ("", None): overstock_location_id = None else: try: overstock_location_id = int(overstock_location_id) except (TypeError, ValueError): conn.rollback() return jsonify({"error": "overstock_location_id must be an integer or null"}), 400 try: overstock_quantity = int(payload.get("overstock_quantity", 0) or 0) except (TypeError, ValueError): conn.rollback() return jsonify({"error": "overstock_quantity must be a non-negative integer"}), 400 if new_loc is not None and _storage_location_slot_unusable(cur, new_loc): same_slot = old_loc is not None and int(new_loc) == int(old_loc) if not same_slot: conn.rollback() return jsonify({"error": "Cannot move part to a storage slot marked unusable"}), 400 if overstock_quantity < 0: conn.rollback() return jsonify({"error": "overstock_quantity must be a non-negative integer"}), 400 if overstock_location_id is not None: cur.execute("SELECT id FROM storage_locations WHERE id = %s", (overstock_location_id,)) if not cur.fetchone(): conn.rollback() return jsonify({"error": "Overstock location not found"}), 400 cur.execute( """ UPDATE parts SET category_id = %s, storage_location_id = %s, name = %s, description = %s, overstock_location_id = %s, part_number = %s, quantity = %s, overstock_quantity = %s, unit = %s, average_price = %s, minimum_stock_level = %s, comment = %s, barcode = %s, datasheet_url = %s, manufacturer = %s, footprint = %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s """, ( payload.get("category_id"), payload.get("storage_location_id"), payload.get("name"), payload.get("description"), overstock_location_id, payload.get("part_number"), payload.get("quantity", 0), overstock_quantity, payload.get("unit"), payload.get("average_price", 0), payload.get("minimum_stock_level", 0), payload.get("comment"), payload.get("barcode"), payload.get("datasheet_url"), payload.get("manufacturer"), footprint, part_id, ), ) cur.execute("SELECT * FROM parts WHERE id = %s", (part_id,)) row = cur.fetchone() if not row: conn.rollback() return jsonify({"error": "Not found"}), 404 conn.commit() return jsonify(_dateify(row)) cur.execute("DELETE FROM parts WHERE id = %s", (part_id,)) if cur.rowcount == 0: conn.rollback() return jsonify({"error": "Not found"}), 404 conn.commit() return Response(status=204) except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() def _image_upload_relative(part_id, filename): return f"parts/{part_id}/{filename}".replace("\\", "/") def _datasheet_upload_relative(part_id): return f"datasheets/{part_id}.pdf".replace("\\", "/") def _datasheet_thumbnail_relative(part_id): return f"datasheets/{part_id}.thumb.jpg".replace("\\", "/") def _generate_datasheet_thumbnail(pdf_path, thumb_path): """Render first PDF page to JPEG thumbnail. Returns True if generated.""" if fitz is None: return False doc = None try: doc = fitz.open(pdf_path) if doc.page_count < 1: return False page = doc.load_page(0) pix = page.get_pixmap(matrix=fitz.Matrix(1.5, 1.5), alpha=False) os.makedirs(os.path.dirname(thumb_path), exist_ok=True) pix.save(thumb_path) return True except Exception: return False finally: if doc is not None: doc.close() @bp.route("/api/parts//duplicate", methods=["POST"]) def part_duplicate(part_id): """Clone a part into another storage location (qty reset to 0); copy average_price and file attachments.""" payload = request.get_json(silent=True) or {} target_storage_location_id = payload.get("storage_location_id") if target_storage_location_id is None: return jsonify({"error": "storage_location_id required"}), 400 try: target_storage_location_id = int(target_storage_location_id) except (TypeError, ValueError): return jsonify({"error": "storage_location_id must be an integer"}), 400 conn = _get_db() upload_root = _effective_upload_dir() try: _ensure_tables(conn) with conn.cursor() as cur: cur.execute("SELECT * FROM parts WHERE id = %s", (part_id,)) src = cur.fetchone() if not src: return jsonify({"error": "Source part not found"}), 404 if src.get("storage_location_id") == target_storage_location_id: return jsonify({"error": "Target location is the same as the source part"}), 400 cur.execute( "SELECT id FROM parts WHERE storage_location_id = %s LIMIT 1", (target_storage_location_id,), ) if cur.fetchone(): return jsonify({"error": "Target storage location is not empty"}), 400 if _storage_location_slot_unusable(cur, target_storage_location_id): return jsonify({"error": "Target storage slot is marked unusable"}), 400 cur.execute( """ INSERT INTO parts (category_id, storage_location_id, overstock_location_id, name, description, part_number, quantity, overstock_quantity, unit, average_price, minimum_stock_level, comment, barcode, datasheet_url, datasheet_file_path, datasheet_thumbnail_path, manufacturer, footprint) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( src["category_id"], target_storage_location_id, None, src["name"], src["description"], src["part_number"], 0, 0, src["unit"], src.get("average_price") or 0, src["minimum_stock_level"], src["comment"], None, src["datasheet_url"], None, None, src["manufacturer"], _normalize_footprint(cur, src.get("footprint")), ), ) new_id = cur.lastrowid cur.execute( """ SELECT id, file_path, original_filename, url, source, is_primary FROM part_images WHERE part_id = %s ORDER BY is_primary DESC, id """, (part_id,), ) for img in cur.fetchall(): if img.get("url"): cur.execute( "INSERT INTO part_images (part_id, url, source, is_primary) VALUES (%s, %s, %s, %s)", (new_id, img["url"], img["source"], img["is_primary"]), ) elif img.get("file_path"): src_path = _resolve_existing_upload_file(upload_root, img["file_path"]) if not src_path: continue orig = img.get("original_filename") or os.path.basename(img["file_path"]) ext = os.path.splitext(orig)[1] or os.path.splitext(img["file_path"])[1] or ".bin" rand = os.urandom(16).hex() new_fn = f"{rand}{ext}" rel = _image_upload_relative(new_id, new_fn) dst_path = os.path.join(upload_root, rel) os.makedirs(os.path.dirname(dst_path), exist_ok=True) shutil.copy2(src_path, dst_path) cur.execute( "INSERT INTO part_images (part_id, file_path, original_filename, source, is_primary) VALUES (%s, %s, %s, %s, %s)", (new_id, rel, orig, img["source"], img["is_primary"]), ) ds = src.get("datasheet_file_path") if ds: src_pdf = os.path.join(upload_root, ds) if os.path.isfile(src_pdf): new_rel = _datasheet_upload_relative(new_id) dst_pdf = os.path.join(upload_root, new_rel) os.makedirs(os.path.dirname(dst_pdf), exist_ok=True) shutil.copy2(src_pdf, dst_pdf) cur.execute( "UPDATE parts SET datasheet_file_path = %s WHERE id = %s", (new_rel, new_id), ) ds_thumb = src.get("datasheet_thumbnail_path") if ds_thumb: src_thumb = os.path.join(upload_root, ds_thumb) if os.path.isfile(src_thumb): new_thumb_rel = _datasheet_thumbnail_relative(new_id) dst_thumb = os.path.join(upload_root, new_thumb_rel) os.makedirs(os.path.dirname(dst_thumb), exist_ok=True) shutil.copy2(src_thumb, dst_thumb) cur.execute( "UPDATE parts SET datasheet_thumbnail_path = %s WHERE id = %s", (new_thumb_rel, new_id), ) cur.execute( """ SELECT p.*, c.name AS category_name, s.name AS storage_location_name FROM parts p LEFT JOIN categories c ON p.category_id = c.id LEFT JOIN storage_locations s ON p.storage_location_id = s.id WHERE p.id = %s """, (new_id,), ) part = cur.fetchone() cur.execute( """ SELECT id, file_path, original_filename, url, source, is_primary, created_at FROM part_images WHERE part_id = %s ORDER BY is_primary DESC, id """, (new_id,), ) part["images"] = [_dateify(r) for r in cur.fetchall()] conn.commit() return jsonify(_dateify(part)), 201 except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/parts//images", methods=["POST"]) def part_upload_image(part_id): image = request.files.get("image") if not image: return jsonify({"error": "No file uploaded"}), 400 upload_root = _effective_upload_dir() safe_name = secure_filename(image.filename or "image.jpg") or "image.jpg" ext = os.path.splitext(safe_name)[1] or ".jpg" rand = os.urandom(16).hex() filename = f"{rand}{ext}" rel = _image_upload_relative(part_id, filename) full = os.path.join(upload_root, rel) os.makedirs(os.path.dirname(full), exist_ok=True) image.save(full) conn = _get_db() try: with conn.cursor() as cur: cur.execute("SELECT COUNT(*) AS c FROM part_images WHERE part_id = %s", (part_id,)) is_primary = 1 if cur.fetchone()["c"] == 0 else 0 cur.execute( "INSERT INTO part_images (part_id, file_path, original_filename, source, is_primary) VALUES (%s, %s, %s, %s, %s)", (part_id, rel, safe_name, "uploaded", is_primary), ) cur.execute("SELECT * FROM part_images WHERE id = %s", (cur.lastrowid,)) row = cur.fetchone() conn.commit() return jsonify(_dateify(row)), 201 except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/parts//images/url", methods=["POST"]) def part_add_image_url(part_id): payload = request.get_json(silent=True) or {} image_url = payload.get("url") if not image_url: return jsonify({"error": "url required"}), 400 conn = _get_db() try: with conn.cursor() as cur: cur.execute("SELECT COUNT(*) AS c FROM part_images WHERE part_id = %s", (part_id,)) is_primary = 1 if cur.fetchone()["c"] == 0 else 0 cur.execute( "INSERT INTO part_images (part_id, url, source, is_primary) VALUES (%s, %s, %s, %s)", (part_id, image_url, "internet", is_primary), ) cur.execute("SELECT * FROM part_images WHERE id = %s", (cur.lastrowid,)) row = cur.fetchone() conn.commit() return jsonify(_dateify(row)), 201 except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/parts//images//primary", methods=["PATCH"]) def part_set_primary(part_id, image_id): conn = _get_db() try: with conn.cursor() as cur: cur.execute("UPDATE part_images SET is_primary = 0 WHERE part_id = %s", (part_id,)) cur.execute("UPDATE part_images SET is_primary = 1 WHERE id = %s AND part_id = %s", (image_id, part_id)) cur.execute("SELECT * FROM part_images WHERE part_id = %s ORDER BY is_primary DESC, id", (part_id,)) rows = [_dateify(r) for r in cur.fetchall()] conn.commit() return jsonify(rows) except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/parts//images/", methods=["DELETE"]) def part_delete_image(part_id, image_id): conn = _get_db() try: file_path = None with conn.cursor() as cur: cur.execute("SELECT file_path FROM part_images WHERE id = %s AND part_id = %s", (image_id, part_id)) row = cur.fetchone() if not row: return jsonify({"error": "Not found"}), 404 file_path = row.get("file_path") cur.execute("DELETE FROM part_images WHERE id = %s", (image_id,)) conn.commit() if file_path: full = os.path.join(_effective_upload_dir(), file_path) if os.path.exists(full): os.remove(full) return Response(status=204) except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/parts//datasheet", methods=["GET", "POST", "DELETE"]) def part_datasheet(part_id): conn = _get_db() try: with conn.cursor() as cur: if request.method == "POST": pdf = request.files.get("datasheet") if not pdf: return jsonify({"error": "No file uploaded"}), 400 rel = _datasheet_upload_relative(part_id) full = os.path.join(_effective_upload_dir(), rel) thumb_rel = _datasheet_thumbnail_relative(part_id) thumb_full = os.path.join(_effective_upload_dir(), thumb_rel) os.makedirs(os.path.dirname(full), exist_ok=True) pdf.save(full) cur.execute("SELECT datasheet_file_path FROM parts WHERE id = %s", (part_id,)) existing = cur.fetchone() if not existing: return jsonify({"error": "Part not found"}), 404 old = existing.get("datasheet_file_path") if old and old != rel: old_full = os.path.join(_effective_upload_dir(), old) if os.path.exists(old_full): os.remove(old_full) has_thumb = _generate_datasheet_thumbnail(full, thumb_full) if not has_thumb and os.path.exists(thumb_full): os.remove(thumb_full) cur.execute( "UPDATE parts SET datasheet_file_path = %s, datasheet_thumbnail_path = %s WHERE id = %s", (rel, thumb_rel if has_thumb else None, part_id), ) cur.execute("SELECT * FROM parts WHERE id = %s", (part_id,)) row = cur.fetchone() conn.commit() return jsonify(_dateify(row)) cur.execute("SELECT datasheet_file_path, datasheet_thumbnail_path, datasheet_url FROM parts WHERE id = %s", (part_id,)) row = cur.fetchone() if not row: return jsonify({"error": "Not found"}), 404 if request.method == "GET": datasheet_file_path = row.get("datasheet_file_path") if datasheet_file_path: full = os.path.join(_effective_upload_dir(), datasheet_file_path) if os.path.exists(full): return send_file(full) if row.get("datasheet_url"): return redirect(row["datasheet_url"], code=302) return jsonify({"error": "No datasheet"}), 404 datasheet_file_path = row.get("datasheet_file_path") if datasheet_file_path: full = os.path.join(_effective_upload_dir(), datasheet_file_path) if os.path.exists(full): os.remove(full) datasheet_thumbnail_path = row.get("datasheet_thumbnail_path") if datasheet_thumbnail_path: thumb = os.path.join(_effective_upload_dir(), datasheet_thumbnail_path) if os.path.exists(thumb): os.remove(thumb) cur.execute( "UPDATE parts SET datasheet_file_path = NULL, datasheet_thumbnail_path = NULL WHERE id = %s", (part_id,), ) conn.commit() return Response(status=204) except Exception as e: conn.rollback() return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/api/images/search", methods=["GET"]) def image_search(): q = request.args.get("q") if not q: return jsonify({"error": "q required"}), 400 if not PARTSINV_IMAGE_SEARCH_API_KEY or not PARTSINV_IMAGE_SEARCH_CX: return jsonify({"items": [], "message": "Image search not configured. Set PARTSINV_IMAGE_SEARCH_API_KEY and PARTSINV_IMAGE_SEARCH_CX."}) try: params = urllib.parse.urlencode( { "key": PARTSINV_IMAGE_SEARCH_API_KEY, "cx": PARTSINV_IMAGE_SEARCH_CX, "q": q, "searchType": "image", "num": 10, } ) with urllib.request.urlopen(f"https://www.googleapis.com/customsearch/v1?{params}", timeout=10) as response: payload = json.loads(response.read().decode("utf-8")) items = [] for item in payload.get("items", []): items.append( { "link": item.get("link"), "thumbnail": (item.get("image") or {}).get("thumbnailLink") or item.get("link"), "title": item.get("title"), } ) return jsonify({"items": items}) except Exception as e: return jsonify({"error": str(e), "items": []}), 500 @bp.route("/api/settings/upload-dir", methods=["GET", "PUT"]) def settings_upload_dir(): data = _load_override() if request.method == "GET": return jsonify({"uploadDir": data.get("uploadDir", PARTSINV_UPLOAD_DIR)}) payload = request.get_json(silent=True) or {} upload_dir = str(payload.get("uploadDir") or "partsinventory_uploads").strip() or "partsinventory_uploads" data["uploadDir"] = upload_dir _save_override(data) return jsonify({"uploadDir": upload_dir}) @bp.route("/api/settings/database", methods=["GET", "PUT"]) def settings_database(): if request.method == "GET": return jsonify(_db_health_payload()) payload = request.get_json(silent=True) or {} data = _load_override() current = _effective_db_config() data["host"] = payload.get("host", current["host"]) data["port"] = payload.get("port", current["port"]) data["user"] = payload.get("user", current["user"]) data["database"] = payload.get("database", current["database"]) if "password" in payload: data["password"] = payload.get("password", "") _save_override(data) return jsonify(_db_health_payload()) @bp.route("/api/settings/database/test", methods=["POST"]) def settings_database_test(): payload = request.get_json(silent=True) or {} config = { "host": payload.get("host") or "localhost", "port": int(payload.get("port") or 3306), "user": payload.get("user"), "password": payload.get("password", ""), "database": payload.get("database"), "cursorclass": pymysql.cursors.DictCursor, } conn = None try: conn = pymysql.connect(**config) with conn.cursor() as cur: cur.execute("SELECT 1") cur.fetchone() return jsonify({"success": True}) except Exception as e: return jsonify({"success": False, "message": str(e)}) finally: if conn: conn.close() @bp.route("/api/settings/ui", methods=["GET", "PUT"]) def settings_ui(): data = _load_override() if request.method == "GET": return jsonify(_effective_ui_settings()) payload = request.get_json(silent=True) or {} if "showQrCode" in payload: data["showQrCode"] = bool(payload.get("showQrCode")) _save_override(data) return jsonify(_effective_ui_settings()) @bp.route("/api/settings/footprints/sync-kicad", methods=["POST"]) def settings_footprints_sync_kicad(): conn = _get_db() try: _ensure_tables(conn) with conn.cursor() as cur: result = _sync_kicad_footprints(cur) conn.commit() return jsonify({"success": True, **result}) except Exception as e: conn.rollback() return jsonify({"success": False, "error": str(e)}), 500 finally: conn.close() @bp.route("/api/settings/backup/full", methods=["GET"]) def settings_backup_full(): conn = _get_db() try: _ensure_tables(conn) dump_sql = _sql_dump_for_current_db(conn) filename = f"parts_inventory_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.sql" return Response( dump_sql, mimetype="application/sql", headers={"Content-Disposition": f'attachment; filename="{filename}"'}, ) except Exception as e: return jsonify({"error": str(e)}), 500 finally: conn.close() @bp.route("/uploads/", methods=["GET"]) def uploads(filename): upload_root = _effective_upload_dir() normalized = (filename or "").replace("\\", "/") candidate = os.path.join(upload_root, normalized) if os.path.isfile(candidate): return send_from_directory(upload_root, normalized) rel_dir, rel_name = os.path.split(normalized) stem, ext = os.path.splitext(rel_name) abs_dir = os.path.join(upload_root, rel_dir) # Fallback for legacy rows where extension may be missing or different. if stem and os.path.isdir(abs_dir): if ext: no_ext_rel = os.path.join(rel_dir, stem).replace("\\", "/") if rel_dir else stem if os.path.isfile(os.path.join(upload_root, no_ext_rel)): return send_from_directory(upload_root, no_ext_rel) pattern = os.path.join(abs_dir, f"{stem}.*") matches = [m for m in glob.glob(pattern) if os.path.isfile(m)] if matches: rel_match = os.path.relpath(matches[0], upload_root).replace("\\", "/") return send_from_directory(upload_root, rel_match) return jsonify({"error": "Not found"}), 404 @bp.route("/", defaults={"path": ""}) @bp.route("/") def spa(path): dist_dir = PARTSINV_CLIENT_DIST if path.startswith("api/") or path.startswith("uploads/"): return jsonify({"error": "Not found"}), 404 if path and os.path.exists(os.path.join(dist_dir, path)): return send_from_directory(dist_dir, path) index_path = os.path.join(dist_dir, "index.html") if os.path.exists(index_path): return send_from_directory(dist_dir, "index.html") return jsonify( { "error": "PartsInventory frontend not built", "message": "Build the React client and place output in PARTSINV_CLIENT_DIST.", "dist": dist_dir, } ), 503