adding authentication and image support

This commit is contained in:
2025-06-19 11:57:47 -04:00
parent f78f5ce9f9
commit f041266ac3
4 changed files with 468 additions and 138 deletions

1
.gitignore vendored
View File

@@ -5,3 +5,4 @@ __pycache__/
*.pyc *.pyc
*.pyo *.pyo
*.pyd *.pyd
.claude/

View File

@@ -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 os
import json import base64
from routes.admin import admin_bp from routes.admin import admin_bp
from routes.api import api_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__) app = Flask(__name__)
# Set DATA_FOLDER to the absolute path of the data directory inside src # 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(admin_bp)
app.register_blueprint(api_bp) app.register_blueprint(api_bp)
# Load product data from products.json # Load product data from database
def load_products(): def load_products():
try: return database.get_all_products()
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 {}
@app.route('/') @app.route('/')
def index(): def index():
return render_template('index.html') 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']) @app.route('/login', methods=['GET'])
def login(): def login():
# Just return 200 for this example, assuming no auth needed auth_header = request.headers.get('Authorization')
return '', 200
if check_basic_auth(auth_header):
return jsonify({"status": "success", "message": "Authentication successful"}), 200
return jsonify({"error": "Unauthorized"}), 401
@app.route('/arcontentfields', methods=['GET']) @app.route('/arcontentfields', methods=['GET'])
def get_ar_content_fields(): def get_ar_content_fields():
@@ -45,45 +69,129 @@ def get_ar_content_fields():
def get_ar_info(): def get_ar_info():
barcode = request.args.get('barcode') barcode = request.args.get('barcode')
products = load_products() products = load_products()
if not barcode or barcode not in products:
return jsonify({"error": "Item not found"}), 404 # Helper function to convert relative image paths to absolute URLs
return jsonify(products[barcode]), 200 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/<path:filename>', methods=['GET']) @app.route('/images/<path:filename>', methods=['GET'])
def serve_image(filename): 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) image_path = os.path.join(app.config['DATA_FOLDER'], 'images', filename)
if os.path.exists(image_path): 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 return jsonify({"error": "Image not found"}), 404
@app.route('/barcodes/<path:filename>', methods=['GET']) @app.route('/barcodes/<path:filename>', methods=['GET'])
def serve_barcode(filename): def serve_barcode(filename):
barcode_path = os.path.join(app.config['DATA_FOLDER'], 'barcodes', filename) # Parse filename to extract product_id and barcode type
if os.path.exists(barcode_path): # Expected format: {product_id}_{type}.png
return send_file(barcode_path) name_parts = os.path.splitext(filename)[0].split('_')
return jsonify({"error": "Barcode not found"}), 404 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__': if __name__ == '__main__':
# Ensure the required directories exist # Initialize database
os.makedirs(os.path.join(app.config['DATA_FOLDER'], 'images'), exist_ok=True) database.init_database()
os.makedirs(os.path.join(app.config['DATA_FOLDER'], 'barcodes'), exist_ok=True)
# 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') products_file = os.path.join(app.config['DATA_FOLDER'], 'products.json')
if not os.path.exists(products_file): if os.path.exists(products_file):
initial_data = { print("Migrating data from products.json to SQLite...")
"123456": [ database.migrate_from_json()
{"fieldName": "_id", "label": "Item ID", "value": "123456", "editable": "false", "fieldType": "TEXT"}, # Optionally rename the JSON file to indicate it's been migrated
{"fieldName": "_price", "label": "Sale Price", "value": "$49.99", "editable": "true", "fieldType": "TEXT"}, shutil.move(products_file, products_file + '.migrated')
{"fieldName": "_image", "label": "Image", "value": "/images/123456.png", "editable": "false", "fieldType": "IMAGE_URI"} print("Migration complete.")
],
"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)
app.run(port=5555, host="0.0.0.0", debug=True) app.run(port=5555, host="0.0.0.0", debug=True)

268
src/database.py Normal file
View File

@@ -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

View File

@@ -4,6 +4,9 @@ import json
import uuid import uuid
import io import io
from werkzeug.utils import secure_filename 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 # Import barcode generator class, but handle the case if it fails
try: try:
@@ -28,15 +31,11 @@ def allowed_file(filename):
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
def load_products(): def load_products():
try: return database.get_all_products()
with open(PRODUCTS_FILE, 'r') as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return {}
def save_products(products): def save_products(products):
with open(PRODUCTS_FILE, 'w') as f: # This is now handled by database operations
json.dump(products, f, indent=2) pass
@admin_bp.route('/') @admin_bp.route('/')
def index(): def index():
@@ -63,16 +62,27 @@ def add_product():
return render_template('admin/add_product.html') return render_template('admin/add_product.html')
# Handle image upload # Handle image upload
image_filename = f"{product_id}.png" # Default name image_data = None
image_path = f"/images/{image_filename}" image_mime_type = None
image_path = f"/images/{product_id}.png" # Default path for display
if 'image' in request.files: if 'image' in request.files:
file = request.files['image'] file = request.files['image']
if file and file.filename and allowed_file(file.filename): if file and file.filename and allowed_file(file.filename):
extension = file.filename.rsplit('.', 1)[1].lower() extension = file.filename.rsplit('.', 1)[1].lower()
image_filename = f"{product_id}.{extension}" image_path = f"/images/{product_id}.{extension}"
image_path = f"/images/{image_filename}"
file.save(os.path.join(UPLOAD_FOLDER, image_filename)) # 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 # Create product data structure
product_data = [ product_data = [
@@ -82,9 +92,8 @@ def add_product():
{"fieldName": "_image", "label": "Image", "value": image_path, "editable": "false", "fieldType": "IMAGE_URI"} {"fieldName": "_image", "label": "Image", "value": image_path, "editable": "false", "fieldType": "IMAGE_URI"}
] ]
# Save to products.json # Save to database
products[product_id] = product_data database.save_product(product_id, product_data, image_data, image_mime_type)
save_products(products)
flash('Product added successfully!') flash('Product added successfully!')
return redirect(url_for('admin.index')) return redirect(url_for('admin.index'))
@@ -117,36 +126,33 @@ def edit_product(product_id):
field["value"] = f"${price}" field["value"] = f"${price}"
# Handle image upload # Handle image upload
image_data = None
image_mime_type = None
if 'image' in request.files and request.files['image'].filename: if 'image' in request.files and request.files['image'].filename:
file = request.files['image'] file = request.files['image']
if allowed_file(file.filename): if allowed_file(file.filename):
# Find current image path # Read new image data
current_image = None image_data = file.read()
for field in products[product_id]:
if field["fieldName"] == "_image":
current_image = field["value"]
if current_image: # Determine mime type
# 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
extension = file.filename.rsplit('.', 1)[1].lower() extension = file.filename.rsplit('.', 1)[1].lower()
image_filename = f"{product_id}.{extension}" mime_types = {
image_path = f"/images/{image_filename}" 'jpg': 'image/jpeg',
file.save(os.path.join(UPLOAD_FOLDER, image_filename)) '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]: for field in products[product_id]:
if field["fieldName"] == "_image": if field["fieldName"] == "_image":
field["value"] = image_path field["value"] = image_path
# Save to products.json # Save to database
save_products(products) database.save_product(product_id, products[product_id], image_data, image_mime_type)
flash('Product updated successfully!') flash('Product updated successfully!')
return redirect(url_for('admin.index')) return redirect(url_for('admin.index'))
@@ -161,24 +167,8 @@ def delete_product(product_id):
flash('Product not found.') flash('Product not found.')
return redirect(url_for('admin.index')) return redirect(url_for('admin.index'))
# Find image path # Delete product from database (image is stored in DB)
image_path = None database.delete_product(product_id)
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)
flash('Product deleted successfully!') flash('Product deleted successfully!')
return redirect(url_for('admin.index')) return redirect(url_for('admin.index'))
@@ -195,59 +185,22 @@ def view_product(product_id):
@admin_bp.route('/generate_barcode/<product_id>/<code_type>') @admin_bp.route('/generate_barcode/<product_id>/<code_type>')
def generate_barcode(product_id, code_type): def generate_barcode(product_id, code_type):
"""Generate and serve a barcode or QR code for a product""" """Redirect to the main barcode generation endpoint"""
# Make sure the barcodes directory exists
os.makedirs(BARCODE_FOLDER, exist_ok=True)
products = load_products() products = load_products()
if product_id not in products: if product_id not in products:
return jsonify({"error": "Product not found"}), 404 return jsonify({"error": "Product not found"}), 404
# Check if barcode generator is available # Map code_type to the expected format for the main endpoint
if not BARCODE_GENERATOR_AVAILABLE: type_mapping = {
# Create a simple error image 'qrcode': 'qr',
from PIL import Image, ImageDraw, ImageFont 'ean13': 'ean13',
img = Image.new('RGB', (300, 150), color=(255, 255, 255)) 'code128': 'code128'
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')
generator = BarcodeGenerator() barcode_type = type_mapping.get(code_type, code_type)
# Content for the code - use the product ID # Redirect to the main barcode endpoint which generates dynamically
data = product_id return redirect(f'/barcodes/{product_id}_{barcode_type}.png')
# 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')
@admin_bp.route('/generate_barcode_page/<product_id>') @admin_bp.route('/generate_barcode_page/<product_id>')
def generate_barcode_page(product_id): def generate_barcode_page(product_id):