diff --git a/.gitignore b/.gitignore index 471b22d..e6ea77e 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,4 @@ __pycache__/ *.pyc *.pyo *.pyd +.claude/ \ No newline at end of file diff --git a/src/app.py b/src/app.py index 7ca4a6a..d46944d 100644 --- a/src/app.py +++ b/src/app.py @@ -1,8 +1,14 @@ -from flask import Flask, jsonify, request, send_file, render_template, flash +from flask import Flask, jsonify, request, send_file, render_template, flash, Response import os -import json +import base64 from routes.admin import admin_bp from routes.api import api_bp +import database +import qrcode +import barcode +from barcode.writer import ImageWriter +from io import BytesIO +import shutil app = Flask(__name__) # Set DATA_FOLDER to the absolute path of the data directory inside src @@ -13,23 +19,41 @@ app.config['SECRET_KEY'] = 'dev-key-for-demo-only' app.register_blueprint(admin_bp) app.register_blueprint(api_bp) -# Load product data from products.json +# Load product data from database def load_products(): - try: - products_file = os.path.join(app.config['DATA_FOLDER'], 'products.json') - with open(products_file, 'r') as f: - return json.load(f) - except (FileNotFoundError, json.JSONDecodeError): - return {} + return database.get_all_products() @app.route('/') def index(): return render_template('index.html') +def check_basic_auth(auth_header): + """Validate Basic authentication credentials""" + if not auth_header or not auth_header.startswith('Basic '): + return False + + try: + # Decode the base64 credentials + credentials = base64.b64decode(auth_header[6:]).decode('utf-8') + username, password = credentials.split(':', 1) + + # Simple hardcoded credentials - in production, use secure storage + # You can change these credentials as needed + if username == 'admin' and password == 'knox123': + return True + except Exception: + pass + + return False + @app.route('/login', methods=['GET']) def login(): - # Just return 200 for this example, assuming no auth needed - return '', 200 + auth_header = request.headers.get('Authorization') + + if check_basic_auth(auth_header): + return jsonify({"status": "success", "message": "Authentication successful"}), 200 + + return jsonify({"error": "Unauthorized"}), 401 @app.route('/arcontentfields', methods=['GET']) def get_ar_content_fields(): @@ -45,45 +69,129 @@ def get_ar_content_fields(): def get_ar_info(): barcode = request.args.get('barcode') products = load_products() - if not barcode or barcode not in products: - return jsonify({"error": "Item not found"}), 404 - return jsonify(products[barcode]), 200 + + # Helper function to convert relative image paths to absolute URLs + def make_absolute_urls(product_fields): + # Create absolute URL for image fields + for field in product_fields: + if field['fieldName'] == '_image' and field['value']: + # If it's already an absolute URL, leave it as is + if not field['value'].startswith('http'): + # Build absolute URL using request host + field['value'] = f"{request.url_root.rstrip('/')}{field['value']}" + return product_fields + + # If barcode is provided, return specific product + if barcode: + if barcode in products: + product_data = make_absolute_urls(products[barcode]) + response = jsonify(product_data) + response.headers['Access-Control-Allow-Origin'] = '*' + return response, 200 + return jsonify({"error": "Product not found"}), 404 + + # Return all products if no barcode specified + # Convert all products to have absolute URLs + all_products = {} + for product_id, fields in products.items(): + all_products[product_id] = make_absolute_urls(fields) + response = jsonify(all_products) + response.headers['Access-Control-Allow-Origin'] = '*' + return response, 200 @app.route('/images/', methods=['GET']) def serve_image(filename): + # Log the request for debugging + app.logger.info(f"Image requested: {filename}") + + # Extract product ID from filename (e.g., "123456.jpg" -> "123456") + product_id = os.path.splitext(filename)[0] + + # Get image from database + image_data = database.get_product_image(product_id) + if image_data: + image_bytes, mime_type = image_data + response = Response(image_bytes, mimetype=mime_type) + # Add CORS headers to allow cross-origin requests + response.headers['Access-Control-Allow-Origin'] = '*' + response.headers['Cache-Control'] = 'public, max-age=3600' + app.logger.info(f"Serving image from database: {product_id} ({len(image_bytes)} bytes)") + return response + + # Fallback to file system for backward compatibility image_path = os.path.join(app.config['DATA_FOLDER'], 'images', filename) if os.path.exists(image_path): - return send_file(image_path) + app.logger.info(f"Serving image from filesystem: {image_path}") + response = send_file(image_path) + response.headers['Access-Control-Allow-Origin'] = '*' + return response + + app.logger.warning(f"Image not found: {filename}") return jsonify({"error": "Image not found"}), 404 @app.route('/barcodes/', methods=['GET']) def serve_barcode(filename): - barcode_path = os.path.join(app.config['DATA_FOLDER'], 'barcodes', filename) - if os.path.exists(barcode_path): - return send_file(barcode_path) - return jsonify({"error": "Barcode not found"}), 404 + # Parse filename to extract product_id and barcode type + # Expected format: {product_id}_{type}.png + name_parts = os.path.splitext(filename)[0].split('_') + if len(name_parts) < 2: + return jsonify({"error": "Invalid barcode filename format"}), 400 + + product_id = '_'.join(name_parts[:-1]) # Handle product IDs with underscores + code_type = name_parts[-1].lower() + + # Generate barcode dynamically + try: + buffer = BytesIO() + + if code_type == 'qr': + # Generate QR code + qr = qrcode.QRCode(version=1, box_size=10, border=5) + qr.add_data(f'http://{request.host}/arinfo?barcode={product_id}') + qr.make(fit=True) + img = qr.make_image(fill_color="black", back_color="white") + img.save(buffer, format='PNG') + + elif code_type == 'ean13': + # Generate EAN-13 barcode + # Convert product_id to numeric format if needed + numeric_id = ''.join(filter(str.isdigit, product_id)) + if not numeric_id: + numeric_id = str(abs(hash(product_id)) % 1000000000000)[:12] + else: + numeric_id = numeric_id[:12].zfill(12) + + EAN = barcode.get_barcode_class('ean13') + ean = EAN(numeric_id, writer=ImageWriter()) + ean.write(buffer) + + elif code_type == 'code128': + # Generate Code 128 barcode + CODE128 = barcode.get_barcode_class('code128') + code = CODE128(product_id, writer=ImageWriter()) + code.write(buffer) + + else: + return jsonify({"error": "Unsupported barcode type"}), 400 + + buffer.seek(0) + return Response(buffer.getvalue(), mimetype='image/png') + + except Exception as e: + app.logger.error(f"Barcode generation error: {str(e)}") + return jsonify({"error": "Failed to generate barcode"}), 500 if __name__ == '__main__': - # Ensure the required directories exist - os.makedirs(os.path.join(app.config['DATA_FOLDER'], 'images'), exist_ok=True) - os.makedirs(os.path.join(app.config['DATA_FOLDER'], 'barcodes'), exist_ok=True) + # Initialize database + database.init_database() - # If products.json doesn't exist, create it with initial data + # Migrate existing data from JSON if needed products_file = os.path.join(app.config['DATA_FOLDER'], 'products.json') - if not os.path.exists(products_file): - initial_data = { - "123456": [ - {"fieldName": "_id", "label": "Item ID", "value": "123456", "editable": "false", "fieldType": "TEXT"}, - {"fieldName": "_price", "label": "Sale Price", "value": "$49.99", "editable": "true", "fieldType": "TEXT"}, - {"fieldName": "_image", "label": "Image", "value": "/images/123456.png", "editable": "false", "fieldType": "IMAGE_URI"} - ], - "789012": [ - {"fieldName": "_id", "label": "Item ID", "value": "789012", "editable": "false", "fieldType": "TEXT"}, - {"fieldName": "_price", "label": "Sale Price", "value": "$59.99", "editable": "true", "fieldType": "TEXT"}, - {"fieldName": "_image", "label": "Image", "value": "/images/789012.png", "editable": "false", "fieldType": "IMAGE_URI"} - ] - } - with open(products_file, 'w') as f: - json.dump(initial_data, f, indent=2) + if os.path.exists(products_file): + print("Migrating data from products.json to SQLite...") + database.migrate_from_json() + # Optionally rename the JSON file to indicate it's been migrated + shutil.move(products_file, products_file + '.migrated') + print("Migration complete.") app.run(port=5555, host="0.0.0.0", debug=True) diff --git a/src/database.py b/src/database.py new file mode 100644 index 0000000..cf4d7c4 --- /dev/null +++ b/src/database.py @@ -0,0 +1,268 @@ +import sqlite3 +import json +import os +from contextlib import contextmanager +from typing import Dict, List, Optional, Any +import base64 + +DATABASE_PATH = os.path.join(os.path.dirname(__file__), 'data', 'products.db') + +@contextmanager +def get_db(): + """Context manager for database connections""" + conn = sqlite3.connect(DATABASE_PATH) + conn.row_factory = sqlite3.Row + try: + yield conn + finally: + conn.close() + +def init_database(): + """Initialize the database with required tables""" + os.makedirs(os.path.dirname(DATABASE_PATH), exist_ok=True) + + with get_db() as conn: + cursor = conn.cursor() + + # Create products table with image_data as BLOB + cursor.execute(''' + CREATE TABLE IF NOT EXISTS products ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + price TEXT, + inventory INTEGER, + image_data BLOB, + image_mime_type TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + ''') + + # Create product_fields table for extensible fields + cursor.execute(''' + CREATE TABLE IF NOT EXISTS product_fields ( + product_id TEXT, + field_name TEXT, + label TEXT, + value TEXT, + editable TEXT, + field_type TEXT, + FOREIGN KEY (product_id) REFERENCES products(id) ON DELETE CASCADE, + PRIMARY KEY (product_id, field_name) + ) + ''') + + conn.commit() + +def migrate_from_json(): + """Migrate existing JSON data to SQLite""" + json_path = os.path.join(os.path.dirname(__file__), 'data', 'products.json') + + if not os.path.exists(json_path): + return + + with open(json_path, 'r') as f: + products_data = json.load(f) + + with get_db() as conn: + cursor = conn.cursor() + + for product_id, fields in products_data.items(): + # Extract core fields + name = '' + price = '' + inventory = None + image_path = '' + + for field in fields: + if field['fieldName'] == '_name': + name = field['value'] + elif field['fieldName'] == '_price': + price = field['value'] + elif field['fieldName'] == '_inventory': + inventory = int(field['value']) if field['value'] else None + elif field['fieldName'] == '_image': + image_path = field['value'] + + # Read image data if exists + image_data = None + image_mime_type = None + if image_path: + # Extract filename from path like '/images/filename.jpg' + filename = image_path.split('/')[-1] + full_image_path = os.path.join(os.path.dirname(__file__), 'data', 'images', filename) + + if os.path.exists(full_image_path): + with open(full_image_path, 'rb') as img_file: + image_data = img_file.read() + # Determine mime type from extension + ext = os.path.splitext(filename)[1].lower() + mime_types = { + '.jpg': 'image/jpeg', + '.jpeg': 'image/jpeg', + '.png': 'image/png', + '.gif': 'image/gif', + '.webp': 'image/webp' + } + image_mime_type = mime_types.get(ext, 'image/jpeg') + + # Insert into products table + cursor.execute(''' + INSERT OR REPLACE INTO products (id, name, price, inventory, image_data, image_mime_type) + VALUES (?, ?, ?, ?, ?, ?) + ''', (product_id, name, price, inventory, image_data, image_mime_type)) + + # Insert all fields into product_fields table + for field in fields: + cursor.execute(''' + INSERT OR REPLACE INTO product_fields + (product_id, field_name, label, value, editable, field_type) + VALUES (?, ?, ?, ?, ?, ?) + ''', ( + product_id, + field['fieldName'], + field['label'], + field['value'], + field.get('editable', 'true'), + field.get('fieldType', 'TEXT') + )) + + conn.commit() + +def get_all_products() -> Dict[str, List[Dict[str, Any]]]: + """Get all products in the legacy format""" + with get_db() as conn: + cursor = conn.cursor() + + # Get all product IDs + cursor.execute('SELECT id FROM products') + product_ids = [row['id'] for row in cursor.fetchall()] + + result = {} + for product_id in product_ids: + # Get all fields for this product + cursor.execute(''' + SELECT field_name, label, value, editable, field_type + FROM product_fields + WHERE product_id = ? + ORDER BY field_name + ''', (product_id,)) + + fields = [] + for row in cursor.fetchall(): + fields.append({ + 'fieldName': row['field_name'], + 'label': row['label'], + 'value': row['value'], + 'editable': row['editable'], + 'fieldType': row['field_type'] + }) + + result[product_id] = fields + + return result + +def get_product(product_id: str) -> Optional[List[Dict[str, Any]]]: + """Get a single product by ID""" + with get_db() as conn: + cursor = conn.cursor() + + cursor.execute(''' + SELECT field_name, label, value, editable, field_type + FROM product_fields + WHERE product_id = ? + ORDER BY field_name + ''', (product_id,)) + + fields = [] + for row in cursor.fetchall(): + fields.append({ + 'fieldName': row['field_name'], + 'label': row['label'], + 'value': row['value'], + 'editable': row['editable'], + 'fieldType': row['field_type'] + }) + + return fields if fields else None + +def save_product(product_id: str, fields: List[Dict[str, Any]], image_data: Optional[bytes] = None, image_mime_type: Optional[str] = None): + """Save or update a product""" + with get_db() as conn: + cursor = conn.cursor() + + # Extract core fields + name = '' + price = '' + inventory = None + + for field in fields: + if field['fieldName'] == '_name': + name = field['value'] + elif field['fieldName'] == '_price': + price = field['value'] + elif field['fieldName'] == '_inventory': + inventory = int(field['value']) if field['value'] else None + + # Check if product exists + cursor.execute('SELECT id FROM products WHERE id = ?', (product_id,)) + exists = cursor.fetchone() is not None + + if exists: + # Update existing product + if image_data is not None: + cursor.execute(''' + UPDATE products + SET name = ?, price = ?, inventory = ?, image_data = ?, image_mime_type = ?, updated_at = CURRENT_TIMESTAMP + WHERE id = ? + ''', (name, price, inventory, image_data, image_mime_type, product_id)) + else: + cursor.execute(''' + UPDATE products + SET name = ?, price = ?, inventory = ?, updated_at = CURRENT_TIMESTAMP + WHERE id = ? + ''', (name, price, inventory, product_id)) + else: + # Insert new product + cursor.execute(''' + INSERT INTO products (id, name, price, inventory, image_data, image_mime_type) + VALUES (?, ?, ?, ?, ?, ?) + ''', (product_id, name, price, inventory, image_data, image_mime_type)) + + # Delete existing fields + cursor.execute('DELETE FROM product_fields WHERE product_id = ?', (product_id,)) + + # Insert all fields + for field in fields: + cursor.execute(''' + INSERT INTO product_fields + (product_id, field_name, label, value, editable, field_type) + VALUES (?, ?, ?, ?, ?, ?) + ''', ( + product_id, + field['fieldName'], + field['label'], + field['value'], + field.get('editable', 'true'), + field.get('fieldType', 'TEXT') + )) + + conn.commit() + +def delete_product(product_id: str): + """Delete a product""" + with get_db() as conn: + cursor = conn.cursor() + cursor.execute('DELETE FROM products WHERE id = ?', (product_id,)) + conn.commit() + +def get_product_image(product_id: str) -> Optional[tuple[bytes, str]]: + """Get product image data and mime type""" + with get_db() as conn: + cursor = conn.cursor() + cursor.execute('SELECT image_data, image_mime_type FROM products WHERE id = ?', (product_id,)) + row = cursor.fetchone() + + if row and row['image_data']: + return row['image_data'], row['image_mime_type'] + return None \ No newline at end of file diff --git a/src/routes/admin.py b/src/routes/admin.py index 668be09..fd862e3 100644 --- a/src/routes/admin.py +++ b/src/routes/admin.py @@ -4,6 +4,9 @@ import json import uuid import io from werkzeug.utils import secure_filename +import sys +sys.path.append(os.path.dirname(os.path.dirname(__file__))) +import database # Import barcode generator class, but handle the case if it fails try: @@ -28,15 +31,11 @@ def allowed_file(filename): return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS def load_products(): - try: - with open(PRODUCTS_FILE, 'r') as f: - return json.load(f) - except (FileNotFoundError, json.JSONDecodeError): - return {} + return database.get_all_products() def save_products(products): - with open(PRODUCTS_FILE, 'w') as f: - json.dump(products, f, indent=2) + # This is now handled by database operations + pass @admin_bp.route('/') def index(): @@ -63,16 +62,27 @@ def add_product(): return render_template('admin/add_product.html') # Handle image upload - image_filename = f"{product_id}.png" # Default name - image_path = f"/images/{image_filename}" + image_data = None + image_mime_type = None + image_path = f"/images/{product_id}.png" # Default path for display if 'image' in request.files: file = request.files['image'] if file and file.filename and allowed_file(file.filename): extension = file.filename.rsplit('.', 1)[1].lower() - image_filename = f"{product_id}.{extension}" - image_path = f"/images/{image_filename}" - file.save(os.path.join(UPLOAD_FOLDER, image_filename)) + image_path = f"/images/{product_id}.{extension}" + + # Read image data + image_data = file.read() + + # Determine mime type + mime_types = { + 'jpg': 'image/jpeg', + 'jpeg': 'image/jpeg', + 'png': 'image/png', + 'gif': 'image/gif' + } + image_mime_type = mime_types.get(extension, 'image/jpeg') # Create product data structure product_data = [ @@ -82,9 +92,8 @@ def add_product(): {"fieldName": "_image", "label": "Image", "value": image_path, "editable": "false", "fieldType": "IMAGE_URI"} ] - # Save to products.json - products[product_id] = product_data - save_products(products) + # Save to database + database.save_product(product_id, product_data, image_data, image_mime_type) flash('Product added successfully!') return redirect(url_for('admin.index')) @@ -117,36 +126,33 @@ def edit_product(product_id): field["value"] = f"${price}" # Handle image upload + image_data = None + image_mime_type = None + if 'image' in request.files and request.files['image'].filename: file = request.files['image'] if allowed_file(file.filename): - # Find current image path - current_image = None - for field in products[product_id]: - if field["fieldName"] == "_image": - current_image = field["value"] + # Read new image data + image_data = file.read() - if current_image: - # Get current filename - current_filename = os.path.basename(current_image) - # Delete old file if it exists and is different - old_path = os.path.join(UPLOAD_FOLDER, current_filename) - if os.path.exists(old_path): - os.remove(old_path) - - # Save new image + # Determine mime type extension = file.filename.rsplit('.', 1)[1].lower() - image_filename = f"{product_id}.{extension}" - image_path = f"/images/{image_filename}" - file.save(os.path.join(UPLOAD_FOLDER, image_filename)) + mime_types = { + 'jpg': 'image/jpeg', + 'jpeg': 'image/jpeg', + 'png': 'image/png', + 'gif': 'image/gif' + } + image_mime_type = mime_types.get(extension, 'image/jpeg') - # Update image path + # Update image path in product data + image_path = f"/images/{product_id}.{extension}" for field in products[product_id]: if field["fieldName"] == "_image": field["value"] = image_path - # Save to products.json - save_products(products) + # Save to database + database.save_product(product_id, products[product_id], image_data, image_mime_type) flash('Product updated successfully!') return redirect(url_for('admin.index')) @@ -161,24 +167,8 @@ def delete_product(product_id): flash('Product not found.') return redirect(url_for('admin.index')) - # Find image path - image_path = None - for field in products[product_id]: - if field["fieldName"] == "_image": - image_path = field["value"] - - # Delete image file - if image_path: - image_filename = os.path.basename(image_path) - image_file_path = os.path.join(UPLOAD_FOLDER, image_filename) - if os.path.exists(image_file_path): - os.remove(image_file_path) - - # Delete product from dict - del products[product_id] - - # Save updated products - save_products(products) + # Delete product from database (image is stored in DB) + database.delete_product(product_id) flash('Product deleted successfully!') return redirect(url_for('admin.index')) @@ -195,59 +185,22 @@ def view_product(product_id): @admin_bp.route('/generate_barcode//') def generate_barcode(product_id, code_type): - """Generate and serve a barcode or QR code for a product""" - # Make sure the barcodes directory exists - os.makedirs(BARCODE_FOLDER, exist_ok=True) - + """Redirect to the main barcode generation endpoint""" products = load_products() if product_id not in products: return jsonify({"error": "Product not found"}), 404 - # Check if barcode generator is available - if not BARCODE_GENERATOR_AVAILABLE: - # Create a simple error image - from PIL import Image, ImageDraw, ImageFont - img = Image.new('RGB', (300, 150), color=(255, 255, 255)) - d = ImageDraw.Draw(img) - d.rectangle([0, 0, 299, 149], outline=(0, 0, 0), width=2) - d.text((25, 65), "Barcode generation unavailable", fill=(0, 0, 0)) - - # Save and serve the error image - image_path = os.path.join(BARCODE_FOLDER, f"{product_id}_{code_type}_error.png") - img.save(image_path) - return send_file(image_path, mimetype='image/png') + # Map code_type to the expected format for the main endpoint + type_mapping = { + 'qrcode': 'qr', + 'ean13': 'ean13', + 'code128': 'code128' + } - generator = BarcodeGenerator() + barcode_type = type_mapping.get(code_type, code_type) - # Content for the code - use the product ID - data = product_id - - # Base filename for the saved code - base_filename = f"{product_id}_{code_type}" - image_path = os.path.join(BARCODE_FOLDER, f"{base_filename}.png") - - # Generate the requested code type - if code_type == 'qrcode': - # Generate QR code with product URL - product_url = request.host_url.rstrip('/') + f"/arinfo?barcode={product_id}" - img = generator.generate_qr_code(product_url) - generator.save_image(img, image_path) - - elif code_type == 'ean13': - # For EAN-13, make sure the product ID is numeric - numeric_id = ''.join(filter(str.isdigit, product_id)) - img = generator.generate_ean13_barcode(numeric_id) - generator.save_image(img, image_path) - - elif code_type == 'code128': - img = generator.generate_code128_barcode(data) - generator.save_image(img, image_path) - - else: - return jsonify({"error": "Invalid code type"}), 400 - - # Return the image directly - return send_file(image_path, mimetype='image/png') + # Redirect to the main barcode endpoint which generates dynamically + return redirect(f'/barcodes/{product_id}_{barcode_type}.png') @admin_bp.route('/generate_barcode_page/') def generate_barcode_page(product_id):