mattintech/KCAPDemoServer
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #3 from mattintech/refactor

Browse Source 
Refactor
This commit is contained in:
Matt Hills
2025-10-20 18:21:27 -04:00
committed by GitHub
parent a775270790 0888edaec5
commit 32ce69057b
42 changed files with 1415 additions and 1855 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
PLANNING.md
View File

@@ -1,55 +0,0 @@
# Project Planning and Code Guidelines
## General Code Guidelines
* **Modular Design:** Use Object-Oriented Programming (OOP) where possible to separate concerns and promote code reuse.
* **File Size Limit:** No single Python file should exceed **600 lines** to improve readability and maintainability.
* **Commenting:**
* Use clear, concise comments to explain complex logic.
* Include function and class docstrings where appropriate.
* Use in-line comments sparingly and only when the logic is not immediately obvious.
* **Code Structure:**
* Follow the Model-View-Controller (MVC) pattern where applicable.
* Group related classes and functions into separate modules for clarity.
* **Error Handling:** Implement robust error handling with meaningful messages and fallback behavior where possible.
* **Readability:**
* Use meaningful variable and function names.
* Avoid deeply nested loops and conditionals where possible.
## API Design Guidelines
* **RESTful Endpoints:** Use RESTful conventions for all API endpoints.
* **Status Codes:** Return appropriate HTTP status codes (e.g., 200 for success, 404 for not found, 500 for server error).
* **Error Responses:** Provide consistent and descriptive error messages.
* **Data Validation:** Validate all incoming data to prevent unexpected behavior and security issues.
* **Unified API and UI:** For demo purposes, both the API endpoints and admin UI should be served by the same Flask application.
## File Organization
* **Static Files:** Store images in the `static/images/` directory.
* **Product Data:** Store product metadata in `products.json`.
* **Configuration Files:** Use a dedicated `config/` directory for environment-specific settings.
* **Templates:** Store HTML templates in the `templates/` directory, with admin-specific templates in `templates/admin/`.
## Documentation Guidelines
* **Keep Documentation Updated:** Update TASK.md as features are implemented to track progress.
* **Readme First:** README.md should always reflect the current state of the project and provide clear setup instructions.
* **Code Comments:** Keep code comments in sync with implementation changes.
* **API Documentation:** Document all API endpoints, including request/response formats.
## Security Guidelines
* **No Hardcoded Credentials:** Avoid hardcoding sensitive information like passwords or API keys.
* **Input Sanitization:** Validate and sanitize all user input to prevent injection attacks.
* **Minimal Permissions:** Run the application with the least required privileges.
## Future Considerations
* **Scalability:** Plan for the potential migration to a database if the product catalog grows.
* **API Rate Limiting:** Consider adding rate limiting for public APIs.
* **Monitoring and Logging:** Implement basic logging for audit trails and error diagnosis.
* **Deployment Automation:** Use Docker or other containerization for easy deployment and scaling.

20
RULES.md
View File

@@ -1,20 +0,0 @@
# AI Agent Coding Rules
## 1. Code Length Limits
- **Python files (`.py`)**: Maximum 600 lines per file.
- **JavaScript/HTML files (`.js`, `.html`)**: Maximum 500 lines per file.
## 2. Architecture
- Follow the **MVC (Model-View-Controller)** pattern for all new features and refactoring.
- **Models**: Handle data and business logic.
- **Views**: Manage presentation and user interface.
- **Controllers**: Process input, interact with models, and render views.
## 3. Coding Practices
- No guessing: All code must be based on clear requirements or existing patterns in the codebase.
- Avoid unnecessary complexity and keep code readable.
- Document any non-obvious logic with concise comments.
## 4. General
- Ensure all code changes comply with these rules before merging or deploying.
- If a rule cannot be followed, document the reason clearly in the code and notify the team.

50
TASK.md
View File

@@ -1,50 +0,0 @@
# Task List - Flask AR API Demo
## Phase 1: Basic Admin UI ✅
* [x] Create a simple admin UI (no authentication required for demo purposes).
* [x] Allow adding, updating, and deleting products.
* [x] Store all product data in a `products.json` file for persistence.
* [x] Store images in the `static/images/` directory.
* [x] Implement basic form validation for product fields (ID, price, image).
* [x] Add a file upload feature for product images.
## Phase 2: API Improvements
* [x] Update the `/arinfo` endpoint to read from `products.json` instead of the in-memory dictionary.
* [ ] Add support for updating product attributes via the API.
* [x] Implement error handling and input validation.
* [ ] Optimize image loading for better performance.
## Phase 3: UI Enhancements
* [x] Add a responsive design for mobile and tablet support.
* [x] Include image previews in the admin UI.
* [ ] Add sorting and search capabilities for the product list.
* [x] Add barcode and QR code generation for products.
## Implementation Notes
* The APIs and admin UI have been implemented as a single server for demonstration purposes.
* The API endpoints and admin interface are now served by the same Flask application.
* Bootstrap 5 has been used for responsive UI design.
* Basic validation has been implemented for product forms.
* Flash messages provide user feedback for actions.
* Barcode generation features support three formats:
* QR Code: Links directly to the product's AR information endpoint
* EAN-13: Standard barcode format for retail products
* Code 128: High-density alphanumeric barcode
* Generated barcodes can be previewed, downloaded individually, or printed all at once.
* Barcodes are dynamically generated and stored in the `static/barcodes/` directory.
## Future Ideas
* [ ] Write basic unit tests for the API.
* [ ] Dockerize the application for easier deployment.
* [ ] Add a basic CI/CD pipeline (e.g., GitHub Actions or GitLab CI).
* [ ] Add user authentication for the admin UI.
* [ ] Implement role-based access control (RBAC).
* [ ] Add analytics for product views and updates.
* [ ] Integrate with a database for larger product catalogs.
* [ ] Add bulk import/export functionality for products.
* [ ] Implement a mobile-friendly scanner interface for testing the AR functionality.

287
src/admin.py
View File

@@ -1,287 +0,0 @@
from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify, send_file
import os
import json
import uuid
import io
from werkzeug.utils import secure_filename
# Import barcode generator class, but handle the case if it fails
try:
from barcode_generator import BarcodeGenerator
BARCODE_GENERATOR_AVAILABLE = True
except ImportError:
BARCODE_GENERATOR_AVAILABLE = False
# Create a stub class for graceful degradation
class BarcodeGenerator:
@staticmethod
def check_dependencies():
return {'qrcode': False, 'barcode': False, 'pillow': False}
admin_bp = Blueprint('admin', __name__, url_prefix='/admin')
DATA_FOLDER = os.path.join(os.path.dirname(__file__), 'data')
PRODUCTS_FILE = os.path.join(DATA_FOLDER, 'products.json')
UPLOAD_FOLDER = os.path.join(DATA_FOLDER, 'images')
BARCODE_FOLDER = os.path.join(DATA_FOLDER, 'barcodes')
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif'}
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 {}
def save_products(products):
with open(PRODUCTS_FILE, 'w') as f:
json.dump(products, f, indent=2)
@admin_bp.route('/')
def index():
products = load_products()
return render_template('admin/index.html', products=products)
@admin_bp.route('/add', methods=['GET', 'POST'])
def add_product():
if request.method == 'POST':
# Get form data
product_id = request.form.get('product_id')
name = request.form.get('name')
price = request.form.get('price')
# Basic validation
if not product_id or not name or not price:
flash('Product ID, Name, and Price are required fields.')
return render_template('admin/add_product.html')
# Check if product ID already exists
products = load_products()
if product_id in products:
flash('Product ID already exists.')
return render_template('admin/add_product.html')
# Handle image upload
image_filename = f"{product_id}.png" # Default name
image_path = f"/images/{image_filename}"
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))
# Create product data structure
product_data = [
{"fieldName": "_id", "label": "Item ID", "value": product_id, "editable": "false", "fieldType": "TEXT"},
{"fieldName": "_name", "label": "Product Name", "value": name, "editable": "true", "fieldType": "TEXT"},
{"fieldName": "_price", "label": "Sale Price", "value": f"${price}", "editable": "true", "fieldType": "TEXT"},
{"fieldName": "_image", "label": "Image", "value": image_path, "editable": "false", "fieldType": "IMAGE_URI"}
]
# Save to products.json
products[product_id] = product_data
save_products(products)
flash('Product added successfully!')
return redirect(url_for('admin.index'))
return render_template('admin/add_product.html')
@admin_bp.route('/edit/<product_id>', methods=['GET', 'POST'])
def edit_product(product_id):
products = load_products()
if product_id not in products:
flash('Product not found.')
return redirect(url_for('admin.index'))
if request.method == 'POST':
# Get form data
name = request.form.get('name')
price = request.form.get('price')
# Basic validation
if not name or not price:
flash('Name and Price are required fields.')
return render_template('admin/edit_product.html', product_id=product_id, product=products[product_id])
# Update name and price
for field in products[product_id]:
if field["fieldName"] == "_name":
field["value"] = name
elif field["fieldName"] == "_price":
field["value"] = f"${price}"
# Handle image upload
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"]
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
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))
# Update image path
for field in products[product_id]:
if field["fieldName"] == "_image":
field["value"] = image_path
# Save to products.json
save_products(products)
flash('Product updated successfully!')
return redirect(url_for('admin.index'))
return render_template('admin/edit_product.html', product_id=product_id, product=products[product_id])
@admin_bp.route('/delete/<product_id>', methods=['POST'])
def delete_product(product_id):
products = load_products()
if product_id not in products:
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)
flash('Product deleted successfully!')
return redirect(url_for('admin.index'))
@admin_bp.route('/view/<product_id>')
def view_product(product_id):
products = load_products()
if product_id not in products:
flash('Product not found.')
return redirect(url_for('admin.index'))
return render_template('admin/view_product.html', product_id=product_id, product=products[product_id])
@admin_bp.route('/generate_barcode/<product_id>/<code_type>')
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)
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')
generator = BarcodeGenerator()
# 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')
@admin_bp.route('/generate_barcode_page/<product_id>')
def generate_barcode_page(product_id):
"""Show a page with different barcode options for a product"""
products = load_products()
if product_id not in products:
flash('Product not found.')
return redirect(url_for('admin.index'))
# Extract product data
product_data = {}
for field in products[product_id]:
field_name = field["fieldName"][1:] if field["fieldName"].startswith('_') else field["fieldName"]
product_data[field_name] = field["value"]
# Add product ID
product_data['id'] = product_id
# Check if barcode generation is available
dependency_status = {}
if BARCODE_GENERATOR_AVAILABLE:
dependency_status = BarcodeGenerator.check_dependencies()
else:
dependency_status = {
'qrcode': False,
'barcode': False,
'pillow': False
}
return render_template('admin/generate_barcode.html',
product_id=product_id,
product=product_data,
dependencies=dependency_status)
# Remove or replace this file, as admin.py has been moved to routes/admin.py

441
src/app.py
View File

@@ -1,441 +0,0 @@
from flask import Flask, jsonify, request, send_file, render_template, flash, Response, redirect
import os
import base64
import database
import qrcode
import barcode
from barcode.writer import ImageWriter
from io import BytesIO
import shutil
from werkzeug.routing import BaseConverter
from functools import wraps
app = Flask(__name__)
# Set DATA_FOLDER to the absolute path of the data directory inside src
app.config['DATA_FOLDER'] = os.path.join(os.path.dirname(__file__), 'data')
app.config['SECRET_KEY'] = 'dev-key-for-demo-only'
# Custom converter for tenant IDs
class TenantConverter(BaseConverter):
regex = '[a-zA-Z0-9_-]+'
def to_python(self, value):
# Convert to lowercase when parsing from URL
return value.lower()
def to_url(self, value):
# Convert to lowercase when generating URLs
return value.lower()
app.url_map.converters['tenant'] = TenantConverter
# Load product data from database
def load_products(tenant_id=None):
return database.get_all_products(tenant_id)
@app.route('/')
def index():
# Show a tenant selection page or redirect to default
tenants = database.get_all_tenants()
server_url = database.get_server_url()
return render_template('tenant_selection.html', tenants=tenants, server_url=server_url)
@app.route('/tenant/<tenant:tenant_id>/delete', methods=['POST'])
def delete_tenant(tenant_id):
"""Delete a tenant and all its data"""
database.delete_tenant(tenant_id)
flash(f'Tenant "{tenant_id}" has been deleted successfully.', 'success')
return redirect('/')
@app.route('/settings', methods=['GET', 'POST'])
def settings():
if request.method == 'POST':
server_url = request.form.get('server_url', '').strip()
if server_url:
# Remove trailing slash for consistency
server_url = server_url.rstrip('/')
database.set_setting('server_url', server_url)
flash('Server settings updated successfully.', 'success')
else:
flash('Please provide a valid server URL.', 'error')
return redirect('/settings')
server_url = database.get_server_url()
return render_template('settings.html', server_url=server_url)
@app.route('/<tenant:tenant_id>/')
def tenant_index(tenant_id):
# Auto-create tenant if it doesn't exist
tenant = database.get_or_create_tenant(tenant_id)
if tenant is None:
# Reserved tenant ID or invalid
return jsonify({"error": f"'{tenant_id}' is a reserved name and cannot be used as a tenant ID"}), 404
# Load products for this tenant
products = load_products(tenant_id)
custom_fields = database.get_custom_ar_fields(tenant_id)
return render_template('index.html', tenant=tenant, products=products, custom_fields=custom_fields)
@app.route('/<tenant:tenant_id>/settings')
def tenant_settings(tenant_id):
# Get tenant
tenant = database.get_or_create_tenant(tenant_id)
if tenant is None:
return jsonify({"error": f"'{tenant_id}' is a reserved name and cannot be used as a tenant ID"}), 404
# Get custom AR fields for this tenant
custom_fields = database.get_custom_ar_fields(tenant_id)
# Get server URL for API endpoint display
server_url = database.get_server_url()
return render_template('tenant_settings.html', tenant=tenant, custom_fields=custom_fields, server_url=server_url)
@app.route('/<tenant:tenant_id>/settings/credentials', methods=['POST'])
def update_tenant_credentials(tenant_id):
# Get tenant
tenant = database.get_tenant(tenant_id)
if tenant is None:
return jsonify({"error": "Tenant not found"}), 404
username = request.form.get('username', '').strip()
password = request.form.get('password', '').strip()
if not username:
flash('Username is required.', 'error')
return redirect(f'/{tenant_id}/settings')
# Update credentials
database.update_tenant_credentials(tenant_id, username, password if password else None)
flash('Credentials updated successfully.', 'success')
return redirect(f'/{tenant_id}/settings')
@app.route('/<tenant:tenant_id>/settings/barcode', methods=['POST'])
def update_tenant_barcode_type(tenant_id):
# Get tenant
tenant = database.get_tenant(tenant_id)
if tenant is None:
return jsonify({"error": "Tenant not found"}), 404
barcode_type = request.form.get('barcode_type', '').strip()
if not barcode_type:
flash('Barcode type is required.', 'error')
return redirect(f'/{tenant_id}/settings')
# Update barcode type
database.update_tenant_barcode_type(tenant_id, barcode_type)
flash('Barcode type updated successfully.', 'success')
return redirect(f'/{tenant_id}/settings')
def check_basic_auth(auth_header, tenant_id):
"""Validate Basic authentication credentials for a tenant"""
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)
# Get tenant credentials from database (without creating)
tenant = database.get_tenant(tenant_id)
if tenant and username == tenant['username'] and password == tenant['password']:
return True
except Exception:
pass
return False
@app.route('/<tenant:tenant_id>/login', methods=['GET'])
def login(tenant_id):
auth_header = request.headers.get('Authorization')
if check_basic_auth(auth_header, tenant_id):
return jsonify({"status": "success", "message": "Authentication successful"}), 200
return jsonify({"error": "Unauthorized"}), 401
@app.route('/<tenant:tenant_id>/arcontentfields', methods=['GET'])
def get_ar_content_fields(tenant_id):
# Get custom fields defined for this tenant
fields = database.get_custom_ar_fields(tenant_id)
return jsonify(fields), 200
@app.route('/<tenant:tenant_id>/arinfo', methods=['GET', 'POST'])
def get_ar_info(tenant_id):
barcode = request.args.get('barcode')
products = load_products(tenant_id)
# Get custom AR fields for this tenant
custom_fields = database.get_custom_ar_fields(tenant_id)
custom_field_names = [f['fieldName'] for f in custom_fields]
# Handle POST request - update product fields
if request.method == 'POST':
if not barcode:
return jsonify({"error": "Barcode parameter required"}), 400
if barcode not in products:
return jsonify({"error": "Product not found"}), 404
try:
# Get the updated fields from the request body
updated_fields = request.get_json()
if not isinstance(updated_fields, list):
return jsonify({"error": "Request body must be an array of fields"}), 400
# Get current product data
current_product = products[barcode]
# Update only the editable fields
for updated_field in updated_fields:
field_name = updated_field.get('fieldName')
new_value = updated_field.get('value')
# Find the field in the current product and update it
for field in current_product:
if field['fieldName'] == field_name:
# Check if the field is editable
if field.get('editable') == 'true':
field['value'] = new_value
break
# Save the updated product to database
database.save_product(barcode, tenant_id, current_product)
app.logger.info(f"Updated product {barcode} for tenant {tenant_id}")
return jsonify({"success": True}), 200
except Exception as e:
app.logger.error(f"Error updating product: {str(e)}")
return jsonify({"error": "Failed to update product"}), 500
# Helper function to convert relative image paths to absolute URLs and filter fields
def filter_and_process_fields(product_fields):
# Filter to only include fields defined in custom AR fields
filtered_fields = []
# Get field types for all custom fields
field_types = {f['fieldName']: f['fieldType'] for f in custom_fields}
for field in product_fields:
if field['fieldName'] in custom_field_names:
# Create absolute URL for IMAGE_URI fields
if field_types.get(field['fieldName']) == 'IMAGE_URI' 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 with tenant
field['value'] = f"{request.url_root.rstrip('/')}/{tenant_id}{field['value']}"
filtered_fields.append(field)
return filtered_fields
# Handle GET request - return product data
# If barcode is provided, return specific product
if barcode:
if barcode in products:
product_data = filter_and_process_fields(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 and filter fields
all_products = {}
for product_id, fields in products.items():
all_products[product_id] = filter_and_process_fields(fields)
response = jsonify(all_products)
response.headers['Access-Control-Allow-Origin'] = '*'
return response, 200
@app.route('/<tenant:tenant_id>/images/<path:filename>', methods=['GET'])
def serve_image(tenant_id, filename):
# Log the request for debugging
app.logger.info(f"Image requested for tenant {tenant_id}: {filename}")
# Check if filename contains field name (e.g., "123456_thumbnail.jpg")
base_name = os.path.splitext(filename)[0]
if '_' in base_name:
# Split to get product_id and field_name
parts = base_name.split('_', 1)
product_id = parts[0]
field_name = '_' + parts[1] if len(parts) > 1 else '_image'
# Try to get field-specific image
image_data = database.get_product_image_by_field(product_id, tenant_id, field_name)
if image_data:
image_bytes, mime_type = image_data
response = Response(image_bytes, mimetype=mime_type)
response.headers['Access-Control-Allow-Origin'] = '*'
response.headers['Cache-Control'] = 'public, max-age=3600'
app.logger.info(f"Serving field-specific image: {product_id}/{field_name} ({len(image_bytes)} bytes)")
return response
# Try standard image lookup (backward compatibility)
product_id = os.path.splitext(filename)[0]
# Get image from database for this tenant
image_data = database.get_product_image(product_id, tenant_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):
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('/<tenant:tenant_id>/qrcode/template', methods=['GET'])
def generate_template_qr_code(tenant_id):
"""Generate QR code for the AR Template URL"""
try:
# Get server URL from settings
server_url = database.get_server_url()
template_url = f"{server_url}/{tenant_id}/"
# Generate QR code
buffer = BytesIO()
qr = qrcode.QRCode(version=1, box_size=10, border=5)
qr.add_data(template_url)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
img.save(buffer, format='PNG')
buffer.seek(0)
return Response(buffer.getvalue(), mimetype='image/png')
except Exception as e:
app.logger.error(f"QR code generation error: {str(e)}")
return jsonify({"error": "Failed to generate QR code"}), 500
@app.route('/<tenant:tenant_id>/qrcode/arinfo', methods=['GET'])
def generate_ar_qr_code(tenant_id):
"""Generate QR code for the AR API endpoint"""
try:
# Get server URL from settings
server_url = database.get_server_url()
ar_url = f"{server_url}/{tenant_id}/arinfo"
# Generate QR code
buffer = BytesIO()
qr = qrcode.QRCode(version=1, box_size=10, border=5)
qr.add_data(ar_url)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
img.save(buffer, format='PNG')
buffer.seek(0)
return Response(buffer.getvalue(), mimetype='image/png')
except Exception as e:
app.logger.error(f"QR code generation error: {str(e)}")
return jsonify({"error": "Failed to generate QR code"}), 500
@app.route('/<tenant:tenant_id>/barcodes/<path:filename>', methods=['GET'])
def serve_barcode(tenant_id, filename):
# 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 with tenant in URL
qr = qrcode.QRCode(version=1, box_size=10, border=5)
qr.add_data(f'http://{request.host}/{tenant_id}/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
# Import product management routes
from routes.admin import (add_product, edit_product, delete_product,
generate_barcode, manage_ar_fields, view_all_barcodes)
# Register product management routes (remove /admin/ from paths)
app.add_url_rule('/<tenant:tenant_id>/add', 'admin.add_product', add_product, methods=['GET', 'POST'])
app.add_url_rule('/<tenant:tenant_id>/edit/<product_id>', 'admin.edit_product', edit_product, methods=['GET', 'POST'])
app.add_url_rule('/<tenant:tenant_id>/delete/<product_id>', 'admin.delete_product', delete_product, methods=['POST'])
app.add_url_rule('/<tenant:tenant_id>/generate_barcode/<product_id>/<code_type>', 'admin.generate_barcode', generate_barcode)
app.add_url_rule('/<tenant:tenant_id>/ar_fields', 'admin.manage_ar_fields', manage_ar_fields, methods=['GET', 'POST'])
app.add_url_rule('/<tenant:tenant_id>/barcodes', 'admin.view_all_barcodes', view_all_barcodes, methods=['GET'])
# Register API routes with tenant prefix
from routes.api import api_index
app.add_url_rule('/<tenant:tenant_id>/api/', 'api.api_index', api_index)
# Catch-all route for admin without tenant - redirect to home
@app.route('/admin/')
@app.route('/admin/<path:path>')
def redirect_admin_to_home(path=None):
return redirect('/')
if __name__ == '__main__':
# Initialize database
database.init_database()
# Migrate existing data from JSON if needed
products_file = os.path.join(app.config['DATA_FOLDER'], 'products.json')
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.")
# Clean up any accidentally created reserved tenants
deleted_count = database.cleanup_reserved_tenants()
if deleted_count > 0:
print(f"Cleaned up {deleted_count} reserved tenant(s)")
app.run(port=5555, host="0.0.0.0", debug=True)

49
src/app/__init__.py Normal file
View File

@@ -0,0 +1,49 @@
import os
from flask import Flask, redirect
from werkzeug.routing import BaseConverter
def create_app(config_name='development'):
"""Application factory pattern"""
app = Flask(__name__)
# Load configuration
from app.config import config
app.config.from_object(config[config_name])
# Ensure data folder exists
os.makedirs(app.config['DATA_FOLDER'], exist_ok=True)
# Custom converter for tenant IDs
class TenantConverter(BaseConverter):
regex = '[a-zA-Z0-9_-]+'
def to_python(self, value):
# Convert to lowercase when parsing from URL
return value.lower()
def to_url(self, value):
# Convert to lowercase when generating URLs
return value.lower()
app.url_map.converters['tenant'] = TenantConverter
# Initialize database
with app.app_context():
from app.models.base import init_database
init_database()
# Register blueprints
from app.blueprints import main_bp, tenant_bp, admin_bp, api_bp
app.register_blueprint(main_bp)
app.register_blueprint(tenant_bp)
app.register_blueprint(admin_bp)
app.register_blueprint(api_bp)
# Catch-all route for admin without tenant - redirect to home
@app.route('/admin/')
@app.route('/admin/<path:path>')
def redirect_admin_to_home(path=None):
return redirect('/')
return app

6
src/app/blueprints/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
from .main import main_bp
from .tenant import tenant_bp
from .admin import admin_bp
from .api import api_bp
__all__ = ['main_bp', 'tenant_bp', 'admin_bp', 'api_bp']

5
src/app/blueprints/admin/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
from flask import Blueprint
admin_bp = Blueprint('admin', __name__, url_prefix='/<tenant:tenant_id>')
from . import routes

244
src/routes/admin.py → src/app/blueprints/admin/routes.py
View File

@@ -1,85 +1,60 @@
from flask import render_template, request, redirect, url_for, flash, jsonify
import os
import sys
sys.path.append(os.path.dirname(os.path.dirname(__file__)))
import database
DATA_FOLDER = os.path.join(os.path.dirname(__file__), '../data')
PRODUCTS_FILE = os.path.join(DATA_FOLDER, 'products.json')
UPLOAD_FOLDER = os.path.join(DATA_FOLDER, 'images')
BARCODE_FOLDER = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'data', 'barcodes'))
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif'}
def allowed_file(filename):
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
def load_products(tenant_id):
return database.get_all_products(tenant_id)
def save_products(products):
# This is now handled by database operations
pass
def index(tenant_id):
products = load_products(tenant_id)
tenant = database.get_or_create_tenant(tenant_id)
custom_fields = database.get_custom_ar_fields(tenant_id)
return render_template('admin/index.html', products=products, tenant=tenant, custom_fields=custom_fields)
from flask import render_template, request, redirect, url_for, flash, jsonify, current_app
from . import admin_bp
from app.models import TenantModel, ProductModel, ARFieldModel
from app.services import ProductService
@admin_bp.route('/add', methods=['GET', 'POST'])
def add_product(tenant_id):
# Get custom AR fields for this tenant
custom_fields = database.get_custom_ar_fields(tenant_id)
"""Add a new product"""
custom_fields = ARFieldModel.get_all(tenant_id)
if request.method == 'POST':
# Get form data
product_id = request.form.get('product_id')
# Basic validation
if not product_id:
flash('Product ID is required.')
return render_template('admin/add_product.html', tenant_id=tenant_id, custom_fields=custom_fields)
# Check if product ID already exists
products = load_products(tenant_id)
products = ProductModel.get_all(tenant_id)
if product_id in products:
flash('Product ID already exists.')
return render_template('admin/add_product.html', tenant_id=tenant_id, custom_fields=custom_fields)
# Handle backward compatibility image upload
image_data = None
image_mime_type = None
# Create product data structure based on custom fields
product_data = []
# Always include _id field
product_data.append({
"fieldName": "_id",
"label": "Item ID",
"value": product_id,
"editable": "false",
"fieldName": "_id",
"label": "Item ID",
"value": product_id,
"editable": "false",
"fieldType": "TEXT"
})
# Process images after product is saved
images_to_save = []
# Add other fields based on custom configuration
for field in custom_fields:
field_name = field['fieldName']
if field_name == '_id':
continue # Already added
elif field['fieldType'] == 'IMAGE_URI':
# Check if image was uploaded for this field
continue
if field['fieldType'] == 'IMAGE_URI':
image_field_name = f'image_{field_name}'
if image_field_name in request.files:
file = request.files[image_field_name]
if file and file.filename and allowed_file(file.filename):
# Read image data
allowed_extensions = current_app.config['ALLOWED_EXTENSIONS']
if file and file.filename and ProductService.allowed_file(file.filename, allowed_extensions):
file.seek(0)
img_data = file.read()
# Determine mime type
extension = file.filename.rsplit('.', 1)[1].lower()
mime_types = {
'jpg': 'image/jpeg',
@@ -88,19 +63,16 @@ def add_product(tenant_id):
'gif': 'image/gif'
}
mime_type = mime_types.get(extension, 'image/jpeg')
# Store for later saving
images_to_save.append({
'field_name': field_name,
'data': img_data,
'mime_type': mime_type
})
# Set the image URL path (strip leading underscore from field_name to avoid double underscores)
field_suffix = field_name[1:] if field_name.startswith('_') else field_name
image_path = f"/images/{product_id}_{field_suffix}.{extension}"
# For backward compatibility with _image field
if field_name == '_image':
image_data = img_data
image_mime_type = mime_type
@@ -108,7 +80,7 @@ def add_product(tenant_id):
image_path = ""
else:
image_path = ""
product_data.append({
"fieldName": field_name,
"label": field['label'],
@@ -117,7 +89,6 @@ def add_product(tenant_id):
"fieldType": field['fieldType']
})
else:
# Get value from form
value = request.form.get(f'field_{field_name}', '')
product_data.append({
"fieldName": field_name,
@@ -126,8 +97,8 @@ def add_product(tenant_id):
"editable": field['editable'],
"fieldType": field['fieldType']
})
# Also save fields that aren't in custom fields (for backward compatibility)
# Handle backward compatibility fields
if '_name' not in [f['fieldName'] for f in custom_fields]:
name = request.form.get('name', '')
if name:
@@ -138,7 +109,7 @@ def add_product(tenant_id):
"editable": "true",
"fieldType": "TEXT"
})
if '_price' not in [f['fieldName'] for f in custom_fields]:
price = request.form.get('price', '')
if price:
@@ -149,57 +120,56 @@ def add_product(tenant_id):
"editable": "true",
"fieldType": "TEXT"
})
# Save to database
database.save_product(product_id, tenant_id, product_data, image_data, image_mime_type)
ProductModel.save(product_id, tenant_id, product_data, image_data, image_mime_type)
# Save additional images
for img in images_to_save:
database.save_product_image(product_id, tenant_id, img['field_name'], img['data'], img['mime_type'])
ProductModel.save_image(product_id, tenant_id, img['field_name'], img['data'], img['mime_type'])
flash('Product added successfully!')
return redirect(f'/{tenant_id}/')
return render_template('admin/add_product.html', tenant_id=tenant_id, custom_fields=custom_fields)
@admin_bp.route('/edit/<product_id>', methods=['GET', 'POST'])
def edit_product(tenant_id, product_id):
products = load_products(tenant_id)
custom_fields = database.get_custom_ar_fields(tenant_id)
"""Edit an existing product"""
products = ProductModel.get_all(tenant_id)
custom_fields = ARFieldModel.get_all(tenant_id)
if product_id not in products:
flash('Product not found.')
return redirect(f'/{tenant_id}/')
if request.method == 'POST':
# Handle backward compatibility image upload
image_data = None
image_mime_type = None
# Process images after product is saved
images_to_save = []
# Update fields based on form data
updated_product = []
# Always include _id field
for field in products[product_id]:
if field["fieldName"] == "_id":
updated_product.append(field)
break
# Process each custom field
for custom_field in custom_fields:
field_name = custom_field['fieldName']
if field_name == '_id':
continue
# Find existing field data
existing_field = None
for field in products[product_id]:
if field["fieldName"] == field_name:
existing_field = field.copy()
break
if not existing_field:
existing_field = {
"fieldName": field_name,
@@ -208,18 +178,16 @@ def edit_product(tenant_id, product_id):
"editable": custom_field['editable'],
"fieldType": custom_field['fieldType']
}
if custom_field['fieldType'] == 'IMAGE_URI':
# Check if new image was uploaded
image_field_name = f'image_{field_name}'
if image_field_name in request.files and request.files[image_field_name].filename:
file = request.files[image_field_name]
if allowed_file(file.filename):
# Read image data
allowed_extensions = current_app.config['ALLOWED_EXTENSIONS']
if ProductService.allowed_file(file.filename, allowed_extensions):
file.seek(0)
img_data = file.read()
# Determine mime type
extension = file.filename.rsplit('.', 1)[1].lower()
mime_types = {
'jpg': 'image/jpeg',
@@ -229,31 +197,26 @@ def edit_product(tenant_id, product_id):
}
mime_type = mime_types.get(extension, 'image/jpeg')
# Store for later saving
images_to_save.append({
'field_name': field_name,
'data': img_data,
'mime_type': mime_type
})
# Update image path (strip leading underscore from field_name to avoid double underscores)
field_suffix = field_name[1:] if field_name.startswith('_') else field_name
existing_field["value"] = f"/images/{product_id}_{field_suffix}.{extension}"
# For backward compatibility with _image field
if field_name == '_image':
image_data = img_data
image_mime_type = mime_type
# If no new image uploaded, keep the existing value (already in existing_field)
else:
# Get value from form
value = request.form.get(f'field_{field_name}', '')
if field_name == '_price' and value and not value.startswith('$'):
value = f"${value}"
existing_field["value"] = value
updated_product.append(existing_field)
# Handle backward compatibility fields
if '_name' not in [f['fieldName'] for f in custom_fields]:
name = request.form.get('name', '')
@@ -263,7 +226,7 @@ def edit_product(tenant_id, product_id):
field["value"] = name
updated_product.append(field)
break
if '_price' not in [f['fieldName'] for f in custom_fields]:
price = request.form.get('price', '')
if price:
@@ -272,52 +235,43 @@ def edit_product(tenant_id, product_id):
field["value"] = f"${price}"
updated_product.append(field)
break
# Save to database
database.save_product(product_id, tenant_id, updated_product, image_data, image_mime_type)
ProductModel.save(product_id, tenant_id, updated_product, image_data, image_mime_type)
# Save additional images
for img in images_to_save:
database.save_product_image(product_id, tenant_id, img['field_name'], img['data'], img['mime_type'])
ProductModel.save_image(product_id, tenant_id, img['field_name'], img['data'], img['mime_type'])
flash('Product updated successfully!')
return redirect(f'/{tenant_id}/')
return render_template('admin/edit_product.html',
product_id=product_id,
product=products[product_id],
return render_template('admin/edit_product.html',
product_id=product_id,
product=products[product_id],
tenant_id=tenant_id,
custom_fields=custom_fields)
@admin_bp.route('/delete/<product_id>', methods=['POST'])
def delete_product(tenant_id, product_id):
products = load_products(tenant_id)
"""Delete a product"""
products = ProductModel.get_all(tenant_id)
if product_id not in products:
flash('Product not found.')
return redirect(f'/{tenant_id}/')
# Delete product from database (image is stored in DB)
database.delete_product(product_id, tenant_id)
ProductModel.delete(product_id, tenant_id)
flash('Product deleted successfully!')
return redirect(f'/{tenant_id}/')
def view_product(tenant_id, product_id):
products = load_products(tenant_id)
if product_id not in products:
flash('Product not found.')
return redirect(f'/{tenant_id}/')
return render_template('admin/view_product.html', product_id=product_id, product=products[product_id], tenant_id=tenant_id)
@admin_bp.route('/generate_barcode/<product_id>/<code_type>')
def generate_barcode(tenant_id, product_id, code_type):
"""Redirect to the main barcode generation endpoint"""
products = load_products(tenant_id)
products = ProductModel.get_all(tenant_id)
if product_id not in products:
return jsonify({"error": "Product not found"}), 404
# Map code_type to the expected format for the main endpoint
type_mapping = {
'qrcode': 'qr',
'ean13': 'ean13',
@@ -325,16 +279,14 @@ def generate_barcode(tenant_id, product_id, code_type):
}
barcode_type = type_mapping.get(code_type, code_type)
# Redirect to the main barcode endpoint which generates dynamically
return redirect(f'/{tenant_id}/barcodes/{product_id}_{barcode_type}.png')
@admin_bp.route('/barcodes', methods=['GET'])
def view_all_barcodes(tenant_id):
"""Display all product barcodes for printing"""
products = load_products(tenant_id)
tenant = database.get_or_create_tenant(tenant_id)
products = ProductModel.get_all(tenant_id)
tenant = TenantModel.get_or_create(tenant_id)
# Get barcode type setting for this tenant
barcode_type = tenant.get('barcode_type', 'qr')
return render_template('admin/all_barcodes.html',
@@ -342,30 +294,14 @@ def view_all_barcodes(tenant_id):
tenant=tenant,
barcode_type=barcode_type)
def manage_credentials(tenant_id):
"""Manage tenant login credentials"""
tenant = database.get_or_create_tenant(tenant_id)
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
if username and password:
database.update_tenant_credentials(tenant_id, username, password)
flash('Credentials updated successfully!')
return redirect(f'/{tenant_id}/')
else:
flash('Username and password are required.')
return render_template('admin/credentials.html', tenant=tenant, tenant_id=tenant_id)
@admin_bp.route('/ar_fields', methods=['GET', 'POST'])
def manage_ar_fields(tenant_id):
"""Manage custom AR content fields"""
tenant = database.get_or_create_tenant(tenant_id)
tenant = TenantModel.get_or_create(tenant_id)
if request.method == 'POST':
action = request.form.get('action')
if action == 'add':
field_data = {
'fieldName': request.form.get('fieldName'),
@@ -374,19 +310,19 @@ def manage_ar_fields(tenant_id):
'editable': request.form.get('editable', 'true'),
'displayOrder': int(request.form.get('displayOrder', 0))
}
if field_data['fieldName'] and field_data['label']:
database.save_custom_ar_field(tenant_id, field_data)
ARFieldModel.save(tenant_id, field_data)
flash('Custom field added successfully!')
else:
flash('Field name and label are required.')
elif action == 'delete':
field_id = request.form.get('field_id')
if field_id:
database.delete_custom_ar_field(tenant_id, int(field_id))
ARFieldModel.delete(tenant_id, int(field_id))
flash('Custom field deleted successfully!')
elif action == 'update':
field_data = {
'id': int(request.form.get('field_id')),
@@ -396,26 +332,26 @@ def manage_ar_fields(tenant_id):
'editable': request.form.get('editable', 'true'),
'displayOrder': int(request.form.get('displayOrder', 0))
}
if field_data['fieldName'] and field_data['label']:
database.save_custom_ar_field(tenant_id, field_data)
ARFieldModel.save(tenant_id, field_data)
flash('Custom field updated successfully!')
else:
flash('Field name and label are required.')
return redirect(url_for('admin.manage_ar_fields', tenant_id=tenant_id))
# Get existing custom fields
custom_fields = database.get_custom_ar_fields(tenant_id)
custom_fields = ARFieldModel.get_all(tenant_id)
# Available field types
field_types = [
'TEXT',
'IMAGE_URI'
]
return render_template('admin/ar_fields.html',
tenant=tenant,
return render_template('admin/ar_fields.html',
tenant=tenant,
tenant_id=tenant_id,
custom_fields=custom_fields,
field_types=field_types)

5
src/app/blueprints/api/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
from flask import Blueprint
api_bp = Blueprint('api', __name__, url_prefix='/<tenant:tenant_id>/api')
from . import routes

4
src/routes/api.py → src/app/blueprints/api/routes.py
View File

@@ -1,5 +1,7 @@
# API routes scaffold
from flask import jsonify
from . import api_bp
@api_bp.route('/')
def api_index(tenant_id):
"""API home endpoint"""
return jsonify({'message': 'API Home', 'tenant': tenant_id})

5
src/app/blueprints/main/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
from flask import Blueprint
main_bp = Blueprint('main', __name__)
from . import routes

27
src/app/blueprints/main/routes.py Normal file
View File

@@ -0,0 +1,27 @@
from flask import render_template, request, redirect, flash
from . import main_bp
from app.models import TenantModel, SettingsModel
@main_bp.route('/')
def index():
"""Tenant selection page"""
tenants = TenantModel.get_all()
server_url = SettingsModel.get_server_url()
return render_template('tenant_selection.html', tenants=tenants, server_url=server_url)
@main_bp.route('/settings', methods=['GET', 'POST'])
def settings():
"""Global settings page"""
if request.method == 'POST':
server_url = request.form.get('server_url', '').strip()
if server_url:
# Remove trailing slash for consistency
server_url = server_url.rstrip('/')
SettingsModel.set('server_url', server_url)
flash('Server settings updated successfully.', 'success')
else:
flash('Please provide a valid server URL.', 'error')
return redirect('/settings')
server_url = SettingsModel.get_server_url()
return render_template('settings.html', server_url=server_url)

5
src/app/blueprints/tenant/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
from flask import Blueprint
tenant_bp = Blueprint('tenant', __name__, url_prefix='/<tenant:tenant_id>')
from . import routes

224
src/app/blueprints/tenant/routes.py Normal file
View File

@@ -0,0 +1,224 @@
from flask import render_template, request, redirect, flash, jsonify, Response, current_app
from . import tenant_bp
from app.models import TenantModel, ProductModel, ARFieldModel, SettingsModel
from app.services import AuthService, ProductService, BarcodeService
import os
@tenant_bp.route('/')
def index(tenant_id):
"""Tenant home page - product listing"""
tenant = TenantModel.get_or_create(tenant_id)
if tenant is None:
return jsonify({"error": f"'{tenant_id}' is a reserved name and cannot be used as a tenant ID"}), 404
products = ProductModel.get_all(tenant_id)
custom_fields = ARFieldModel.get_all(tenant_id)
return render_template('index.html', tenant=tenant, products=products, custom_fields=custom_fields)
@tenant_bp.route('/delete', methods=['POST'])
def delete_tenant(tenant_id):
"""Delete a tenant and all its data"""
TenantModel.delete(tenant_id)
flash(f'Tenant "{tenant_id}" has been deleted successfully.', 'success')
return redirect('/')
@tenant_bp.route('/settings')
def settings(tenant_id):
"""Tenant settings page"""
tenant = TenantModel.get_or_create(tenant_id)
if tenant is None:
return jsonify({"error": f"'{tenant_id}' is a reserved name and cannot be used as a tenant ID"}), 404
custom_fields = ARFieldModel.get_all(tenant_id)
server_url = SettingsModel.get_server_url()
return render_template('tenant_settings.html', tenant=tenant, custom_fields=custom_fields, server_url=server_url)
@tenant_bp.route('/settings/credentials', methods=['POST'])
def update_credentials(tenant_id):
"""Update tenant credentials"""
tenant = TenantModel.get_by_id(tenant_id)
if tenant is None:
return jsonify({"error": "Tenant not found"}), 404
username = request.form.get('username', '').strip()
password = request.form.get('password', '').strip()
if not username:
flash('Username is required.', 'error')
return redirect(f'/{tenant_id}/settings')
TenantModel.update_credentials(tenant_id, username, password if password else None)
flash('Credentials updated successfully.', 'success')
return redirect(f'/{tenant_id}/settings')
@tenant_bp.route('/settings/barcode', methods=['POST'])
def update_barcode_type(tenant_id):
"""Update tenant barcode type"""
tenant = TenantModel.get_by_id(tenant_id)
if tenant is None:
return jsonify({"error": "Tenant not found"}), 404
barcode_type = request.form.get('barcode_type', '').strip()
if not barcode_type:
flash('Barcode type is required.', 'error')
return redirect(f'/{tenant_id}/settings')
TenantModel.update_barcode_type(tenant_id, barcode_type)
flash('Barcode type updated successfully.', 'success')
return redirect(f'/{tenant_id}/settings')
@tenant_bp.route('/login', methods=['GET'])
def login(tenant_id):
"""Login endpoint for API authentication"""
auth_header = request.headers.get('Authorization')
if AuthService.check_basic_auth(auth_header, tenant_id):
return jsonify({"status": "success", "message": "Authentication successful"}), 200
return jsonify({"error": "Unauthorized"}), 401
@tenant_bp.route('/arcontentfields', methods=['GET'])
def get_ar_content_fields(tenant_id):
"""Get custom AR fields for tenant"""
fields = ARFieldModel.get_all(tenant_id)
return jsonify(fields), 200
@tenant_bp.route('/arinfo', methods=['GET', 'POST'])
def get_ar_info(tenant_id):
"""Get or update AR product information"""
barcode = request.args.get('barcode')
# Handle POST request - update product fields
if request.method == 'POST':
if not barcode:
return jsonify({"error": "Barcode parameter required"}), 400
product = ProductModel.get_by_id(barcode, tenant_id)
if not product:
return jsonify({"error": "Product not found"}), 404
try:
updated_fields = request.get_json()
if not isinstance(updated_fields, list):
return jsonify({"error": "Request body must be an array of fields"}), 400
# Update only the editable fields
for updated_field in updated_fields:
field_name = updated_field.get('fieldName')
new_value = updated_field.get('value')
for field in product:
if field['fieldName'] == field_name:
if field.get('editable') == 'true':
field['value'] = new_value
break
# Save the updated product
ProductModel.save(barcode, tenant_id, product)
current_app.logger.info(f"Updated product {barcode} for tenant {tenant_id}")
return jsonify({"success": True}), 200
except Exception as e:
current_app.logger.error(f"Error updating product: {str(e)}")
return jsonify({"error": "Failed to update product"}), 500
# Handle GET request - return product data
if barcode:
product_data = ProductService.get_product_filtered(barcode, tenant_id)
if product_data:
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
all_products = ProductService.get_all_products_filtered(tenant_id)
response = jsonify(all_products)
response.headers['Access-Control-Allow-Origin'] = '*'
return response, 200
@tenant_bp.route('/images/<path:filename>', methods=['GET'])
def serve_image(tenant_id, filename):
"""Serve product images"""
current_app.logger.info(f"Image requested for tenant {tenant_id}: {filename}")
# Check if filename contains field name (e.g., "123456_thumbnail.jpg")
base_name = os.path.splitext(filename)[0]
if '_' in base_name:
parts = base_name.split('_', 1)
product_id = parts[0]
field_name = '_' + parts[1] if len(parts) > 1 else '_image'
# Try to get field-specific image
image_data = ProductModel.get_image_by_field(product_id, tenant_id, field_name)
if image_data:
image_bytes, mime_type = image_data
response = Response(image_bytes, mimetype=mime_type)
response.headers['Access-Control-Allow-Origin'] = '*'
response.headers['Cache-Control'] = 'public, max-age=3600'
current_app.logger.info(f"Serving field-specific image: {product_id}/{field_name} ({len(image_bytes)} bytes)")
return response
# Try standard image lookup (backward compatibility)
product_id = os.path.splitext(filename)[0]
image_data = ProductModel.get_image(product_id, tenant_id)
if image_data:
image_bytes, mime_type = image_data
response = Response(image_bytes, mimetype=mime_type)
response.headers['Access-Control-Allow-Origin'] = '*'
response.headers['Cache-Control'] = 'public, max-age=3600'
current_app.logger.info(f"Serving image from database: {product_id} ({len(image_bytes)} bytes)")
return response
current_app.logger.warning(f"Image not found: {filename}")
return jsonify({"error": "Image not found"}), 404
@tenant_bp.route('/qrcode/template', methods=['GET'])
def generate_template_qr_code(tenant_id):
"""Generate QR code for the AR Template URL"""
try:
server_url = SettingsModel.get_server_url()
template_url = f"{server_url}/{tenant_id}/"
buffer = BarcodeService.generate_qr_code(template_url)
return Response(buffer.getvalue(), mimetype='image/png')
except Exception as e:
current_app.logger.error(f"QR code generation error: {str(e)}")
return jsonify({"error": "Failed to generate QR code"}), 500
@tenant_bp.route('/qrcode/arinfo', methods=['GET'])
def generate_ar_qr_code(tenant_id):
"""Generate QR code for the AR API endpoint"""
try:
server_url = SettingsModel.get_server_url()
ar_url = f"{server_url}/{tenant_id}/arinfo"
buffer = BarcodeService.generate_qr_code(ar_url)
return Response(buffer.getvalue(), mimetype='image/png')
except Exception as e:
current_app.logger.error(f"QR code generation error: {str(e)}")
return jsonify({"error": "Failed to generate QR code"}), 500
@tenant_bp.route('/barcodes/<path:filename>', methods=['GET'])
def serve_barcode(tenant_id, filename):
"""Serve dynamically generated barcodes"""
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])
code_type = name_parts[-1].lower()
try:
url = None
if code_type == 'qr':
url = f'http://{request.host}/{tenant_id}/arinfo?barcode={product_id}'
buffer = BarcodeService.generate_barcode(product_id, code_type, url)
return Response(buffer.getvalue(), mimetype='image/png')
except Exception as e:
current_app.logger.error(f"Barcode generation error: {str(e)}")
return jsonify({"error": "Failed to generate barcode"}), 500

43
src/app/config.py Normal file
View File

@@ -0,0 +1,43 @@
import os
class Config:
"""Base configuration"""
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-key-for-demo-only'
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_FOLDER = os.path.join(BASE_DIR, 'data')
DATABASE_PATH = os.path.join(DATA_FOLDER, 'products.db')
# Reserved tenant IDs that cannot be used
RESERVED_TENANT_IDS = {
'admin', 'api', 'login', 'logout', 'arcontentfields', 'arinfo',
'images', 'barcodes', 'static', 'assets', 'js', 'css', 'tenant',
'auth', 'oauth', 'callback', 'webhook', 'health', 'status'
}
# File upload settings
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif'}
MAX_CONTENT_LENGTH = 16 * 1024 * 1024 # 16MB max file size
class DevelopmentConfig(Config):
"""Development configuration"""
DEBUG = True
TESTING = False
class ProductionConfig(Config):
"""Production configuration"""
DEBUG = False
TESTING = False
class TestingConfig(Config):
"""Testing configuration"""
DEBUG = True
TESTING = True
DATABASE_PATH = ':memory:'
# Configuration dictionary
config = {
'development': DevelopmentConfig,
'production': ProductionConfig,
'testing': TestingConfig,
'default': DevelopmentConfig
}

6
src/app/models/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
from .tenant import TenantModel
from .product import ProductModel
from .ar_field import ARFieldModel
from .settings import SettingsModel
__all__ = ['TenantModel', 'ProductModel', 'ARFieldModel', 'SettingsModel']

84
src/app/models/ar_field.py Normal file
View File

@@ -0,0 +1,84 @@
from typing import Dict, List, Any
from .base import get_db
class ARFieldModel:
"""Model for custom AR field operations"""
@staticmethod
def get_all(tenant_id: str) -> List[Dict[str, Any]]:
"""Get all custom AR fields for a tenant"""
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('''
SELECT id, field_name, label, field_type, editable, display_order
FROM custom_ar_fields
WHERE tenant_id = ?
ORDER BY display_order, id
''', (tenant_id,))
fields = []
for row in cursor.fetchall():
fields.append({
'id': row['id'],
'fieldName': row['field_name'],
'label': row['label'],
'fieldType': row['field_type'],
'editable': row['editable'],
'displayOrder': row['display_order']
})
return fields
@staticmethod
def save(tenant_id: str, field_data: Dict[str, Any]):
"""Save or update a custom AR field"""
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
if 'id' in field_data and field_data['id']:
# Update existing field
cursor.execute('''
UPDATE custom_ar_fields
SET field_name = ?, label = ?, field_type = ?, editable = ?,
display_order = ?, updated_at = CURRENT_TIMESTAMP
WHERE id = ? AND tenant_id = ?
''', (
field_data['fieldName'],
field_data['label'],
field_data['fieldType'],
field_data['editable'],
field_data['displayOrder'],
field_data['id'],
tenant_id
))
else:
# Insert new field
cursor.execute('''
INSERT INTO custom_ar_fields
(tenant_id, field_name, label, field_type, editable, display_order)
VALUES (?, ?, ?, ?, ?, ?)
''', (
tenant_id,
field_data['fieldName'],
field_data['label'],
field_data['fieldType'],
field_data['editable'],
field_data['displayOrder']
))
conn.commit()
@staticmethod
def delete(tenant_id: str, field_id: int):
"""Delete a custom AR field"""
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('DELETE FROM custom_ar_fields WHERE id = ? AND tenant_id = ?',
(field_id, tenant_id))
conn.commit()

111
src/app/models/base.py Normal file
View File

@@ -0,0 +1,111 @@
import sqlite3
from contextlib import contextmanager
from flask import current_app
@contextmanager
def get_db():
"""Context manager for database connections"""
db_path = current_app.config['DATABASE_PATH']
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
try:
yield conn
finally:
conn.close()
def init_database():
"""Initialize the database with required tables"""
import os
db_path = current_app.config['DATABASE_PATH']
os.makedirs(os.path.dirname(db_path), exist_ok=True)
with get_db() as conn:
cursor = conn.cursor()
# Create tenants table
cursor.execute('''
CREATE TABLE IF NOT EXISTS tenants (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
username TEXT,
password TEXT,
barcode_type TEXT DEFAULT 'qr',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Create products table with tenant_id
cursor.execute('''
CREATE TABLE IF NOT EXISTS products (
id TEXT,
tenant_id TEXT NOT NULL,
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,
PRIMARY KEY (id, tenant_id),
FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE CASCADE
)
''')
# Create product_fields table with tenant_id
cursor.execute('''
CREATE TABLE IF NOT EXISTS product_fields (
product_id TEXT,
tenant_id TEXT,
field_name TEXT,
label TEXT,
value TEXT,
editable TEXT,
field_type TEXT,
FOREIGN KEY (product_id, tenant_id) REFERENCES products(id, tenant_id) ON DELETE CASCADE,
PRIMARY KEY (product_id, tenant_id, field_name)
)
''')
# Create custom_ar_fields table for tenant-specific AR field definitions
cursor.execute('''
CREATE TABLE IF NOT EXISTS custom_ar_fields (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tenant_id TEXT NOT NULL,
field_name TEXT NOT NULL,
label TEXT NOT NULL,
field_type TEXT NOT NULL,
editable TEXT DEFAULT 'true',
display_order INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE CASCADE,
UNIQUE(tenant_id, field_name)
)
''')
# Create product_images table for storing multiple images per product
cursor.execute('''
CREATE TABLE IF NOT EXISTS product_images (
id INTEGER PRIMARY KEY AUTOINCREMENT,
product_id TEXT NOT NULL,
tenant_id TEXT NOT NULL,
field_name TEXT NOT NULL,
image_data BLOB NOT NULL,
image_mime_type TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (product_id, tenant_id) REFERENCES products(id, tenant_id) ON DELETE CASCADE,
UNIQUE(product_id, tenant_id, field_name)
)
''')
# Create settings table
cursor.execute('''
CREATE TABLE IF NOT EXISTS settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
conn.commit()

192
src/app/models/product.py Normal file
View File

@@ -0,0 +1,192 @@
from typing import Dict, List, Optional, Any, Tuple
from .base import get_db
class ProductModel:
"""Model for product operations"""
@staticmethod
def get_all(tenant_id: str) -> Dict[str, List[Dict[str, Any]]]:
"""Get all products for a tenant in the legacy format"""
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('SELECT id FROM products WHERE tenant_id = ?', (tenant_id,))
product_ids = [row['id'] for row in cursor.fetchall()]
result = {}
for product_id in product_ids:
cursor.execute('''
SELECT field_name, label, value, editable, field_type
FROM product_fields
WHERE product_id = ? AND tenant_id = ?
ORDER BY field_name
''', (product_id, tenant_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
@staticmethod
def get_by_id(product_id: str, tenant_id: str) -> Optional[List[Dict[str, Any]]]:
"""Get a single product by ID and tenant"""
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('''
SELECT field_name, label, value, editable, field_type
FROM product_fields
WHERE product_id = ? AND tenant_id = ?
ORDER BY field_name
''', (product_id, tenant_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
@staticmethod
def save(product_id: str, tenant_id: str, fields: List[Dict[str, Any]],
image_data: Optional[bytes] = None, image_mime_type: Optional[str] = None):
"""Save or update a product for a tenant"""
tenant_id = tenant_id.lower()
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 = ? AND tenant_id = ?', (product_id, tenant_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 = ? AND tenant_id = ?
''', (name, price, inventory, image_data, image_mime_type, product_id, tenant_id))
else:
cursor.execute('''
UPDATE products
SET name = ?, price = ?, inventory = ?, updated_at = CURRENT_TIMESTAMP
WHERE id = ? AND tenant_id = ?
''', (name, price, inventory, product_id, tenant_id))
else:
# Insert new product
cursor.execute('''
INSERT INTO products (id, tenant_id, name, price, inventory, image_data, image_mime_type)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (product_id, tenant_id, name, price, inventory, image_data, image_mime_type))
# Delete existing fields
cursor.execute('DELETE FROM product_fields WHERE product_id = ? AND tenant_id = ?', (product_id, tenant_id))
# Insert all fields
for field in fields:
cursor.execute('''
INSERT INTO product_fields
(product_id, tenant_id, field_name, label, value, editable, field_type)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
product_id,
tenant_id,
field['fieldName'],
field['label'],
field['value'],
field.get('editable', 'true'),
field.get('fieldType', 'TEXT')
))
conn.commit()
@staticmethod
def delete(product_id: str, tenant_id: str):
"""Delete a product for a tenant"""
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('DELETE FROM products WHERE id = ? AND tenant_id = ?', (product_id, tenant_id))
conn.commit()
@staticmethod
def get_image(product_id: str, tenant_id: str) -> Optional[Tuple[bytes, str]]:
"""Get product image data and mime type for a tenant"""
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('SELECT image_data, image_mime_type FROM products WHERE id = ? AND tenant_id = ?',
(product_id, tenant_id))
row = cursor.fetchone()
if row and row['image_data']:
return row['image_data'], row['image_mime_type']
return None
@staticmethod
def save_image(product_id: str, tenant_id: str, field_name: str,
image_data: bytes, image_mime_type: str):
"""Save an image for a specific product field"""
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('''
INSERT OR REPLACE INTO product_images
(product_id, tenant_id, field_name, image_data, image_mime_type)
VALUES (?, ?, ?, ?, ?)
''', (product_id, tenant_id, field_name, image_data, image_mime_type))
conn.commit()
@staticmethod
def get_image_by_field(product_id: str, tenant_id: str, field_name: str) -> Optional[Tuple[bytes, str]]:
"""Get image for a specific product field"""
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('''
SELECT image_data, image_mime_type
FROM product_images
WHERE product_id = ? AND tenant_id = ? AND field_name = ?
''', (product_id, tenant_id, field_name))
row = cursor.fetchone()
if row:
return row['image_data'], row['image_mime_type']
return None

30
src/app/models/settings.py Normal file
View File

@@ -0,0 +1,30 @@
from typing import Optional
from .base import get_db
class SettingsModel:
"""Model for application settings"""
@staticmethod
def get(key: str, default: Optional[str] = None) -> Optional[str]:
"""Get a setting value by key"""
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('SELECT value FROM settings WHERE key = ?', (key,))
row = cursor.fetchone()
return row['value'] if row else default
@staticmethod
def set(key: str, value: str):
"""Set a setting value"""
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('''
INSERT OR REPLACE INTO settings (key, value, updated_at)
VALUES (?, ?, CURRENT_TIMESTAMP)
''', (key, value))
conn.commit()
@staticmethod
def get_server_url() -> str:
"""Get server URL setting"""
return SettingsModel.get('server_url', 'http://localhost:5555')

98
src/app/models/tenant.py Normal file
View File

@@ -0,0 +1,98 @@
from .base import get_db
from flask import current_app
class TenantModel:
"""Model for tenant operations"""
@staticmethod
def get_all():
"""Get all tenants"""
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('SELECT * FROM tenants ORDER BY created_at DESC')
rows = cursor.fetchall()
return [dict(row) for row in rows]
@staticmethod
def get_by_id(tenant_id):
"""Get tenant by ID"""
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('SELECT * FROM tenants WHERE id = ?', (tenant_id,))
row = cursor.fetchone()
return dict(row) if row else None
@staticmethod
def create(tenant_id, name=None):
"""Create a new tenant"""
# Check if tenant_id is reserved
if tenant_id.lower() in current_app.config['RESERVED_TENANT_IDS']:
return None
if name is None:
name = tenant_id
with get_db() as conn:
cursor = conn.cursor()
try:
cursor.execute(
'INSERT INTO tenants (id, name, username, password) VALUES (?, ?, ?, ?)',
(tenant_id, name, 'admin', 'password')
)
conn.commit()
return TenantModel.get_by_id(tenant_id)
except Exception:
return None
@staticmethod
def get_or_create(tenant_id):
"""Get existing tenant or create new one"""
tenant = TenantModel.get_by_id(tenant_id)
if tenant:
return tenant
return TenantModel.create(tenant_id)
@staticmethod
def update_credentials(tenant_id, username, password):
"""Update tenant credentials"""
with get_db() as conn:
cursor = conn.cursor()
cursor.execute(
'UPDATE tenants SET username = ?, password = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?',
(username, password, tenant_id)
)
conn.commit()
@staticmethod
def update_barcode_type(tenant_id, barcode_type):
"""Update tenant barcode type preference"""
with get_db() as conn:
cursor = conn.cursor()
cursor.execute(
'UPDATE tenants SET barcode_type = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?',
(barcode_type, tenant_id)
)
conn.commit()
@staticmethod
def delete(tenant_id):
"""Delete a tenant and all associated data"""
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('DELETE FROM tenants WHERE id = ?', (tenant_id,))
conn.commit()
@staticmethod
def cleanup_reserved():
"""Clean up any accidentally created reserved tenants"""
reserved = current_app.config['RESERVED_TENANT_IDS']
with get_db() as conn:
cursor = conn.cursor()
placeholders = ','.join('?' * len(reserved))
cursor.execute(
f'DELETE FROM tenants WHERE LOWER(id) IN ({placeholders})',
tuple(reserved)
)
deleted_count = cursor.rowcount
conn.commit()
return deleted_count

5
src/app/services/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
from .auth_service import AuthService
from .product_service import ProductService
from .barcode_service import BarcodeService
__all__ = ['AuthService', 'ProductService', 'BarcodeService']

25
src/app/services/auth_service.py Normal file
View File

@@ -0,0 +1,25 @@
import base64
from app.models import TenantModel
class AuthService:
"""Service for authentication logic"""
@staticmethod
def check_basic_auth(auth_header: str, tenant_id: str) -> bool:
"""Validate Basic authentication credentials for a tenant"""
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)
# Get tenant credentials from database (without creating)
tenant = TenantModel.get_by_id(tenant_id)
if tenant and username == tenant['username'] and password == tenant['password']:
return True
except Exception:
pass
return False

58
src/app/services/barcode_service.py Normal file
View File

@@ -0,0 +1,58 @@
import qrcode
import barcode
from barcode.writer import ImageWriter
from io import BytesIO
class BarcodeService:
"""Service for barcode generation"""
@staticmethod
def generate_qr_code(data: str) -> BytesIO:
"""Generate QR code image"""
buffer = BytesIO()
qr = qrcode.QRCode(version=1, box_size=10, border=5)
qr.add_data(data)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
img.save(buffer, format='PNG')
buffer.seek(0)
return buffer
@staticmethod
def generate_ean13(product_id: str) -> BytesIO:
"""Generate EAN-13 barcode"""
buffer = BytesIO()
# 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)
buffer.seek(0)
return buffer
@staticmethod
def generate_code128(product_id: str) -> BytesIO:
"""Generate Code 128 barcode"""
buffer = BytesIO()
CODE128 = barcode.get_barcode_class('code128')
code = CODE128(product_id, writer=ImageWriter())
code.write(buffer)
buffer.seek(0)
return buffer
@staticmethod
def generate_barcode(product_id: str, code_type: str, url: str = None) -> BytesIO:
"""Generate barcode based on type"""
if code_type == 'qr':
return BarcodeService.generate_qr_code(url or product_id)
elif code_type == 'ean13':
return BarcodeService.generate_ean13(product_id)
elif code_type == 'code128':
return BarcodeService.generate_code128(product_id)
else:
raise ValueError(f"Unsupported barcode type: {code_type}")

56
src/app/services/product_service.py Normal file
View File

@@ -0,0 +1,56 @@
from typing import Dict, List, Any
from app.models import ProductModel, ARFieldModel
from flask import request
class ProductService:
"""Service for product business logic"""
@staticmethod
def filter_and_process_fields(product_fields: List[Dict[str, Any]],
tenant_id: str,
custom_fields: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Filter product fields to only include defined AR fields and process image URLs"""
custom_field_names = [f['fieldName'] for f in custom_fields]
field_types = {f['fieldName']: f['fieldType'] for f in custom_fields}
filtered_fields = []
for field in product_fields:
if field['fieldName'] in custom_field_names:
# Create absolute URL for IMAGE_URI fields
if field_types.get(field['fieldName']) == 'IMAGE_URI' 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 with tenant
field['value'] = f"{request.url_root.rstrip('/')}/{tenant_id}{field['value']}"
filtered_fields.append(field)
return filtered_fields
@staticmethod
def get_all_products_filtered(tenant_id: str) -> Dict[str, List[Dict[str, Any]]]:
"""Get all products with filtered fields"""
products = ProductModel.get_all(tenant_id)
custom_fields = ARFieldModel.get_all(tenant_id)
all_products = {}
for product_id, fields in products.items():
all_products[product_id] = ProductService.filter_and_process_fields(
fields, tenant_id, custom_fields
)
return all_products
@staticmethod
def get_product_filtered(product_id: str, tenant_id: str) -> List[Dict[str, Any]]:
"""Get a single product with filtered fields"""
product = ProductModel.get_by_id(product_id, tenant_id)
if not product:
return None
custom_fields = ARFieldModel.get_all(tenant_id)
return ProductService.filter_and_process_fields(product, tenant_id, custom_fields)
@staticmethod
def allowed_file(filename: str, allowed_extensions: set) -> bool:
"""Check if file extension is allowed"""
return '.' in filename and filename.rsplit('.', 1)[1].lower() in allowed_extensions

0
src/templates/admin/add_product.html → src/app/templates/admin/add_product.html
View File

0
src/templates/admin/all_barcodes.html → src/app/templates/admin/all_barcodes.html
View File

0
src/templates/admin/ar_fields.html → src/app/templates/admin/ar_fields.html
View File

0
src/templates/admin/credentials.html → src/app/templates/admin/credentials.html
View File

0
src/templates/admin/edit_product.html → src/app/templates/admin/edit_product.html
View File

0
src/templates/admin/index.html → src/app/templates/admin/index.html
View File

0
src/templates/index.html → src/app/templates/index.html
View File

0
src/templates/layout.html → src/app/templates/layout.html
View File

0
src/templates/settings.html → src/app/templates/settings.html
View File

268
src/app/templates/tenant_selection.html Normal file
View File

@@ -0,0 +1,268 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Select or Create Tenant - KCAP Demo Server</title>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.11.0/font/bootstrap-icons.css">
<style>
.tenant-card {
transition: transform 0.2s, box-shadow 0.2s;
cursor: pointer;
height: 100%;
}
.tenant-card:hover {
transform: translateY(-5px);
box-shadow: 0 4px 8px rgba(0,0,0,0.1);
}
.tenant-card-clickable {
text-decoration: none;
color: inherit;
display: block;
}
.settings-btn {
position: fixed;
top: 20px;
right: 20px;
z-index: 1000;
}
.search-container {
position: sticky;
top: 0;
background: white;
z-index: 100;
padding: 20px 0;
}
.tenant-grid {
display: grid;
grid-template-columns: repeat(auto-fill, minmax(320px, 1fr));
gap: 20px;
}
.create-tenant-card {
border: 2px dashed #dee2e6;
background: #f8f9fa;
}
.create-tenant-card:hover {
border-color: #0d6efd;
background: #e7f1ff;
}
</style>
</head>
<body>
<!-- Top Right Action Buttons -->
<div class="settings-btn">
<button class="btn btn-primary me-2" data-bs-toggle="modal" data-bs-target="#createTenantModal" title="New Tenant">
<i class="bi bi-plus-circle-fill"></i>
</button>
<a href="/settings" class="btn btn-secondary" title="Server Settings">
<i class="bi bi-gear-fill"></i>
</a>
</div>
<div class="container-fluid px-4 py-4">
<h1 class="text-center mb-4">KCAP Demo Server</h1>
{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
{% for category, message in messages %}
<div class="alert alert-{{ 'success' if category == 'success' else 'danger' }} alert-dismissible fade show" role="alert">
{{ message }}
<button type="button" class="btn-close" data-bs-dismiss="alert" aria-label="Close"></button>
</div>
{% endfor %}
{% endif %}
{% endwith %}
<!-- Search and Filter -->
<div class="search-container">
<div class="row mb-3">
<div class="col-md-8 offset-md-2">
<div class="input-group">
<span class="input-group-text"><i class="bi bi-search"></i></span>
<input type="text" id="searchTenants" class="form-control" placeholder="Search tenants...">
<button class="btn btn-outline-secondary" type="button" id="clearSearch">
<i class="bi bi-x-lg"></i>
</button>
</div>
</div>
</div>
</div>
<!-- Tenant Grid -->
<div class="tenant-grid" id="tenantGrid">
{% if tenants %}
{% for tenant in tenants %}
<div class="tenant-item" data-tenant-name="{{ tenant.name|lower }}" data-tenant-id="{{ tenant.id|lower }}">
<div class="card tenant-card">
<a href="/{{ tenant.id }}/" class="tenant-card-clickable">
<div class="card-body">
<div class="d-flex justify-content-between align-items-start mb-3">
<h5 class="card-title mb-0">{{ tenant.name }}</h5>
<span class="badge bg-secondary">{{ tenant.id }}</span>
</div>
<p class="card-text text-muted small mb-2">
<i class="bi bi-person"></i> {{ tenant.username }}
</p>
<p class="card-text text-muted small mb-2">
<i class="bi bi-calendar"></i> {{ tenant.created_at }}
</p>
<div class="mb-2">
<small class="text-muted">AR Template URL:</small>
<div class="input-group input-group-sm">
<input type="text" class="form-control form-control-sm" value="{{ server_url }}/{{ tenant.id }}/" readonly onclick="this.select()">
</div>
</div>
</div>
</a>
<div class="card-footer bg-white d-flex justify-content-between" onclick="event.stopPropagation();">
<button type="button" class="btn btn-sm btn-outline-primary" data-bs-toggle="modal" data-bs-target="#qrModal-{{ tenant.id }}">
<i class="bi bi-qr-code"></i> QR
</button>
<form method="POST" action="/tenant/{{ tenant.id }}/delete" class="d-inline" onsubmit="return confirmDelete('{{ tenant.name }}')">
<button type="submit" class="btn btn-danger btn-sm">
<i class="bi bi-trash"></i> Delete
</button>
</form>
</div>
</div>
</div>
{% endfor %}
{% else %}
<div class="col-12">
<div class="text-center py-5">
<i class="bi bi-inbox" style="font-size: 4rem; color: #dee2e6;"></i>
<p class="text-muted mt-3">No tenants exist yet. Create your first tenant!</p>
</div>
</div>
{% endif %}
</div>
<!-- No Results Message -->
<div id="noResults" class="text-center py-5" style="display: none;">
<i class="bi bi-search" style="font-size: 4rem; color: #dee2e6;"></i>
<p class="text-muted mt-3">No tenants found matching your search.</p>
</div>
</div>
<!-- Create Tenant Modal -->
<div class="modal fade" id="createTenantModal" tabindex="-1" aria-labelledby="createTenantModalLabel" aria-hidden="true">
<div class="modal-dialog modal-dialog-centered">
<div class="modal-content">
<div class="modal-header">
<h5 class="modal-title" id="createTenantModalLabel">Create New Tenant</h5>
<button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="Close"></button>
</div>
<div class="modal-body">
<form id="newTenantForm">
<div class="mb-3">
<label for="tenantId" class="form-label">Tenant ID</label>
<input type="text" class="form-control" id="tenantId" placeholder="e.g., demo-company" required pattern="[a-zA-Z0-9_-]+">
<small class="form-text text-muted">Only letters, numbers, hyphens, and underscores allowed</small>
</div>
<div class="alert alert-info">
<i class="bi bi-info-circle"></i>
<small>Default credentials: <strong>admin</strong> / <strong>admin</strong></small>
</div>
</form>
</div>
<div class="modal-footer">
<button type="button" class="btn btn-secondary" data-bs-dismiss="modal">Cancel</button>
<button type="submit" form="newTenantForm" class="btn btn-primary">Create & Enter</button>
</div>
</div>
</div>
</div>
<!-- QR Code Modals for each tenant -->
{% for tenant in tenants %}
<div class="modal fade" id="qrModal-{{ tenant.id }}" tabindex="-1" aria-labelledby="qrModalLabel-{{ tenant.id }}" aria-hidden="true">
<div class="modal-dialog modal-dialog-centered">
<div class="modal-content">
<div class="modal-header">
<h5 class="modal-title" id="qrModalLabel-{{ tenant.id }}">AR Template URL - {{ tenant.name }}</h5>
<button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="Close"></button>
</div>
<div class="modal-body text-center">
<p class="text-muted mb-3">Scan this QR code for Knox Capture AR:</p>
<div class="mb-3">
<img src="/{{ tenant.id }}/qrcode/template" alt="AR Template QR Code" class="img-fluid" style="max-width: 300px;">
</div>
<code class="d-block mt-2">{{ server_url }}/{{ tenant.id }}/</code>
</div>
<div class="modal-footer">
<button type="button" class="btn btn-secondary" data-bs-dismiss="modal">Close</button>
</div>
</div>
</div>
</div>
{% endfor %}
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/js/bootstrap.bundle.min.js"></script>
<script>
const reservedIds = ['admin', 'api', 'login', 'logout', 'arcontentfields', 'arinfo',
'images', 'barcodes', 'static', 'assets', 'js', 'css', 'tenant',
'auth', 'oauth', 'callback', 'webhook', 'health', 'status'];
// Create tenant form submission
document.getElementById('newTenantForm').addEventListener('submit', function(e) {
e.preventDefault();
const tenantId = document.getElementById('tenantId').value;
if (tenantId) {
const normalizedId = tenantId.toLowerCase();
if (reservedIds.includes(normalizedId)) {
alert(`"${tenantId}" is a reserved name and cannot be used as a tenant ID.`);
return;
}
window.location.href = '/' + normalizedId + '/';
}
});
// Search functionality
const searchInput = document.getElementById('searchTenants');
const clearButton = document.getElementById('clearSearch');
const tenantItems = document.querySelectorAll('.tenant-item');
const noResults = document.getElementById('noResults');
const tenantGrid = document.getElementById('tenantGrid');
function filterTenants() {
const searchTerm = searchInput.value.toLowerCase().trim();
let visibleCount = 0;
tenantItems.forEach(item => {
const tenantName = item.getAttribute('data-tenant-name');
const tenantId = item.getAttribute('data-tenant-id');
if (tenantName.includes(searchTerm) || tenantId.includes(searchTerm)) {
item.style.display = '';
visibleCount++;
} else {
item.style.display = 'none';
}
});
// Show/hide no results message
if (visibleCount === 0 && searchTerm !== '') {
tenantGrid.style.display = 'none';
noResults.style.display = 'block';
} else {
tenantGrid.style.display = 'grid';
noResults.style.display = 'none';
}
}
searchInput.addEventListener('input', filterTenants);
clearButton.addEventListener('click', function() {
searchInput.value = '';
filterTenants();
searchInput.focus();
});
// Delete confirmation
function confirmDelete(tenantName) {
return confirm(`Are you sure you want to delete the tenant "${tenantName}"?\n\nThis will permanently delete all products and data associated with this tenant.`);
}
</script>
</body>
</html>

0
src/templates/tenant_settings.html → src/app/templates/tenant_settings.html
View File

0
src/utils/barcode_generator.py → src/app/utils/barcode_generator.py
View File

706
src/database.py
View File

@@ -1,706 +0,0 @@
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')
# Reserved tenant IDs that cannot be used
RESERVED_TENANT_IDS = {
'admin', 'api', 'login', 'logout', 'arcontentfields', 'arinfo',
'images', 'barcodes', 'static', 'assets', 'js', 'css', 'tenant',
'auth', 'oauth', 'callback', 'webhook', 'health', 'status'
}
@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 tenants table
cursor.execute('''
CREATE TABLE IF NOT EXISTS tenants (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
username TEXT,
password TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Create products table with tenant_id
cursor.execute('''
CREATE TABLE IF NOT EXISTS products (
id TEXT,
tenant_id TEXT NOT NULL,
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,
PRIMARY KEY (id, tenant_id),
FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE CASCADE
)
''')
# Create product_fields table with tenant_id
cursor.execute('''
CREATE TABLE IF NOT EXISTS product_fields (
product_id TEXT,
tenant_id TEXT,
field_name TEXT,
label TEXT,
value TEXT,
editable TEXT,
field_type TEXT,
FOREIGN KEY (product_id, tenant_id) REFERENCES products(id, tenant_id) ON DELETE CASCADE,
PRIMARY KEY (product_id, tenant_id, field_name)
)
''')
# Create custom_ar_fields table for tenant-specific AR field definitions
cursor.execute('''
CREATE TABLE IF NOT EXISTS custom_ar_fields (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tenant_id TEXT NOT NULL,
field_name TEXT NOT NULL,
label TEXT NOT NULL,
field_type TEXT NOT NULL,
editable TEXT DEFAULT 'true',
display_order INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE CASCADE,
UNIQUE(tenant_id, field_name)
)
''')
# Create product_images table for storing multiple images per product
cursor.execute('''
CREATE TABLE IF NOT EXISTS product_images (
id INTEGER PRIMARY KEY AUTOINCREMENT,
product_id TEXT NOT NULL,
tenant_id TEXT NOT NULL,
field_name TEXT NOT NULL,
image_data BLOB,
image_mime_type TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (product_id, tenant_id) REFERENCES products(id, tenant_id) ON DELETE CASCADE,
UNIQUE(product_id, tenant_id, field_name)
)
''')
# Create settings table for server configuration
cursor.execute('''
CREATE TABLE IF NOT EXISTS settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Add barcode_type column to tenants table if it doesn't exist
try:
cursor.execute('ALTER TABLE tenants ADD COLUMN barcode_type TEXT DEFAULT "code128"')
conn.commit()
except sqlite3.OperationalError:
# Column already exists
pass
conn.commit()
def migrate_from_json():
"""Migrate existing JSON data to SQLite with default tenant"""
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()
# Create default tenant if not exists
default_tenant_id = 'default'
cursor.execute('''
INSERT OR IGNORE INTO tenants (id, name, username, password)
VALUES (?, ?, ?, ?)
''', (default_tenant_id, 'Default', 'admin', 'admin'))
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 with tenant_id
cursor.execute('''
INSERT OR REPLACE INTO products (id, tenant_id, name, price, inventory, image_data, image_mime_type)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (product_id, default_tenant_id, name, price, inventory, image_data, image_mime_type))
# Insert all fields into product_fields table with tenant_id
for field in fields:
cursor.execute('''
INSERT OR REPLACE INTO product_fields
(product_id, tenant_id, field_name, label, value, editable, field_type)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
product_id,
default_tenant_id,
field['fieldName'],
field['label'],
field['value'],
field.get('editable', 'true'),
field.get('fieldType', 'TEXT')
))
conn.commit()
def get_all_products(tenant_id: str = None) -> Dict[str, List[Dict[str, Any]]]:
"""Get all products for a tenant in the legacy format"""
# Normalize tenant_id to lowercase if provided
if tenant_id:
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
# Get all product IDs for the tenant
if tenant_id:
cursor.execute('SELECT id FROM products WHERE tenant_id = ?', (tenant_id,))
else:
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
if tenant_id:
cursor.execute('''
SELECT field_name, label, value, editable, field_type
FROM product_fields
WHERE product_id = ? AND tenant_id = ?
ORDER BY field_name
''', (product_id, tenant_id))
else:
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, tenant_id: str) -> Optional[List[Dict[str, Any]]]:
"""Get a single product by ID and tenant"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('''
SELECT field_name, label, value, editable, field_type
FROM product_fields
WHERE product_id = ? AND tenant_id = ?
ORDER BY field_name
''', (product_id, tenant_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, tenant_id: str, fields: List[Dict[str, Any]], image_data: Optional[bytes] = None, image_mime_type: Optional[str] = None):
"""Save or update a product for a tenant"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
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 = ? AND tenant_id = ?', (product_id, tenant_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 = ? AND tenant_id = ?
''', (name, price, inventory, image_data, image_mime_type, product_id, tenant_id))
else:
cursor.execute('''
UPDATE products
SET name = ?, price = ?, inventory = ?, updated_at = CURRENT_TIMESTAMP
WHERE id = ? AND tenant_id = ?
''', (name, price, inventory, product_id, tenant_id))
else:
# Insert new product
cursor.execute('''
INSERT INTO products (id, tenant_id, name, price, inventory, image_data, image_mime_type)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (product_id, tenant_id, name, price, inventory, image_data, image_mime_type))
# Delete existing fields
cursor.execute('DELETE FROM product_fields WHERE product_id = ? AND tenant_id = ?', (product_id, tenant_id))
# Insert all fields
for field in fields:
cursor.execute('''
INSERT INTO product_fields
(product_id, tenant_id, field_name, label, value, editable, field_type)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
product_id,
tenant_id,
field['fieldName'],
field['label'],
field['value'],
field.get('editable', 'true'),
field.get('fieldType', 'TEXT')
))
conn.commit()
def delete_product(product_id: str, tenant_id: str):
"""Delete a product for a tenant"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('DELETE FROM products WHERE id = ? AND tenant_id = ?', (product_id, tenant_id))
conn.commit()
def get_product_image(product_id: str, tenant_id: str) -> Optional[tuple[bytes, str]]:
"""Get product image data and mime type for a tenant"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('SELECT image_data, image_mime_type FROM products WHERE id = ? AND tenant_id = ?', (product_id, tenant_id))
row = cursor.fetchone()
if row and row['image_data']:
return row['image_data'], row['image_mime_type']
return None
def get_tenant(tenant_id: str) -> Optional[Dict[str, Any]]:
"""Get a tenant without creating it"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
# Check if tenant exists
cursor.execute('SELECT id, name, username, password, created_at, barcode_type FROM tenants WHERE id = ?', (tenant_id,))
row = cursor.fetchone()
if row:
return {
'id': row['id'],
'name': row['name'],
'username': row['username'],
'password': row['password'],
'created_at': row['created_at'],
'barcode_type': row['barcode_type'] or 'code128'
}
return None
def get_or_create_tenant(tenant_id: str, username: str = None, password: str = None) -> Optional[Dict[str, Any]]:
"""Get or create a tenant"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
# Check if tenant_id is reserved
if tenant_id in RESERVED_TENANT_IDS:
return None
with get_db() as conn:
cursor = conn.cursor()
# Check if tenant exists
cursor.execute('SELECT id, name, username, password, created_at, barcode_type FROM tenants WHERE id = ?', (tenant_id,))
row = cursor.fetchone()
if row:
return {
'id': row['id'],
'name': row['name'],
'username': row['username'],
'password': row['password'],
'created_at': row['created_at'],
'barcode_type': row['barcode_type'] or 'code128'
}
else:
# Create new tenant with default credentials
default_username = username or 'admin'
default_password = password or 'admin'
# Preserve original casing for display name
display_name = tenant_id.replace('-', ' ').replace('_', ' ').title()
cursor.execute('''
INSERT INTO tenants (id, name, username, password)
VALUES (?, ?, ?, ?)
''', (tenant_id, display_name, default_username, default_password))
conn.commit()
# Initialize default AR fields for the new tenant
init_default_ar_fields(tenant_id)
# Get the created_at timestamp and barcode_type from the database
cursor.execute('SELECT created_at, barcode_type FROM tenants WHERE id = ?', (tenant_id,))
created_row = cursor.fetchone()
return {
'id': tenant_id,
'name': display_name,
'username': default_username,
'password': default_password,
'created_at': created_row['created_at'] if created_row else None,
'barcode_type': created_row['barcode_type'] if created_row else 'code128'
}
def update_tenant_credentials(tenant_id: str, username: str, password: str):
"""Update tenant credentials"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('''
UPDATE tenants
SET username = ?, password = ?, updated_at = CURRENT_TIMESTAMP
WHERE id = ?
''', (username, password, tenant_id))
conn.commit()
def update_tenant_barcode_type(tenant_id: str, barcode_type: str):
"""Update tenant barcode type"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('''
UPDATE tenants
SET barcode_type = ?, updated_at = CURRENT_TIMESTAMP
WHERE id = ?
''', (barcode_type, tenant_id))
conn.commit()
def get_all_tenants() -> List[Dict[str, Any]]:
"""Get all tenants"""
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('SELECT id, name, username, created_at FROM tenants ORDER BY created_at DESC')
tenants = []
for row in cursor.fetchall():
tenants.append({
'id': row['id'],
'name': row['name'],
'username': row['username'],
'created_at': row['created_at']
})
return tenants
def delete_tenant(tenant_id: str):
"""Delete a tenant and all associated data"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
# Due to ON DELETE CASCADE, this will also delete all products and product_fields
cursor.execute('DELETE FROM tenants WHERE id = ?', (tenant_id,))
conn.commit()
def cleanup_reserved_tenants():
"""Remove any tenants that were accidentally created with reserved IDs"""
with get_db() as conn:
cursor = conn.cursor()
# Get all tenants
cursor.execute('SELECT id FROM tenants')
tenants = cursor.fetchall()
deleted_count = 0
for tenant in tenants:
tenant_id = tenant['id']
if tenant_id.lower() in RESERVED_TENANT_IDS:
cursor.execute('DELETE FROM tenants WHERE id = ?', (tenant_id,))
deleted_count += 1
print(f"Deleted reserved tenant: {tenant_id}")
conn.commit()
return deleted_count
def get_custom_ar_fields(tenant_id: str) -> List[Dict[str, Any]]:
"""Get custom AR fields for a tenant"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('''
SELECT id, field_name, label, field_type, editable, display_order
FROM custom_ar_fields
WHERE tenant_id = ?
ORDER BY display_order, field_name
''', (tenant_id,))
fields = []
for row in cursor.fetchall():
fields.append({
'id': row['id'],
'fieldName': row['field_name'],
'label': row['label'],
'fieldType': row['field_type'],
'editable': row['editable'],
'displayOrder': row['display_order']
})
return fields
def save_custom_ar_field(tenant_id: str, field_data: Dict[str, Any]) -> int:
"""Save or update a custom AR field"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
if 'id' in field_data:
# Update existing field
cursor.execute('''
UPDATE custom_ar_fields
SET field_name = ?, label = ?, field_type = ?, editable = ?,
display_order = ?, updated_at = CURRENT_TIMESTAMP
WHERE id = ? AND tenant_id = ?
''', (
field_data['fieldName'],
field_data['label'],
field_data['fieldType'],
field_data.get('editable', 'true'),
field_data.get('displayOrder', 0),
field_data['id'],
tenant_id
))
conn.commit()
return field_data['id']
else:
# Insert new field
cursor.execute('''
INSERT INTO custom_ar_fields
(tenant_id, field_name, label, field_type, editable, display_order)
VALUES (?, ?, ?, ?, ?, ?)
''', (
tenant_id,
field_data['fieldName'],
field_data['label'],
field_data['fieldType'],
field_data.get('editable', 'true'),
field_data.get('displayOrder', 0)
))
conn.commit()
return cursor.lastrowid
def delete_custom_ar_field(tenant_id: str, field_id: int):
"""Delete a custom AR field"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('''
DELETE FROM custom_ar_fields
WHERE id = ? AND tenant_id = ?
''', (field_id, tenant_id))
conn.commit()
def init_default_ar_fields(tenant_id: str):
"""Initialize default AR fields for a new tenant"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
default_fields = [
{"fieldName": "_id", "label": "Item ID", "fieldType": "TEXT", "editable": "false", "displayOrder": 1},
{"fieldName": "_price", "label": "Sale Price", "fieldType": "TEXT", "editable": "true", "displayOrder": 2},
{"fieldName": "_image", "label": "Image", "fieldType": "IMAGE_URI", "editable": "false", "displayOrder": 3}
]
with get_db() as conn:
cursor = conn.cursor()
for field in default_fields:
# Use INSERT OR IGNORE to avoid duplicate key errors
cursor.execute('''
INSERT OR IGNORE INTO custom_ar_fields
(tenant_id, field_name, label, field_type, editable, display_order)
VALUES (?, ?, ?, ?, ?, ?)
''', (
tenant_id,
field['fieldName'],
field['label'],
field['fieldType'],
field.get('editable', 'true'),
field.get('displayOrder', 0)
))
conn.commit()
def save_product_image(product_id: str, tenant_id: str, field_name: str, image_data: bytes, mime_type: str):
"""Save an image for a specific field of a product"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
# Delete existing image for this field if any
cursor.execute('''
DELETE FROM product_images
WHERE product_id = ? AND tenant_id = ? AND field_name = ?
''', (product_id, tenant_id, field_name))
# Insert new image
cursor.execute('''
INSERT INTO product_images (product_id, tenant_id, field_name, image_data, image_mime_type)
VALUES (?, ?, ?, ?, ?)
''', (product_id, tenant_id, field_name, image_data, mime_type))
conn.commit()
def get_product_image_by_field(product_id: str, tenant_id: str, field_name: str) -> Optional[tuple[bytes, str]]:
"""Get image data for a specific field of a product"""
# Normalize tenant_id to lowercase
tenant_id = tenant_id.lower()
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('''
SELECT image_data, image_mime_type
FROM product_images
WHERE product_id = ? AND tenant_id = ? AND field_name = ?
''', (product_id, tenant_id, field_name))
row = cursor.fetchone()
if row and row['image_data']:
return row['image_data'], row['image_mime_type']
return None
def get_setting(key: str, default_value: str = None) -> Optional[str]:
"""Get a setting value by key"""
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('SELECT value FROM settings WHERE key = ?', (key,))
row = cursor.fetchone()
if row:
return row['value']
return default_value
def set_setting(key: str, value: str):
"""Set a setting value"""
with get_db() as conn:
cursor = conn.cursor()
cursor.execute('''
INSERT INTO settings (key, value)
VALUES (?, ?)
ON CONFLICT(key) DO UPDATE SET value = ?, updated_at = CURRENT_TIMESTAMP
''', (key, value, value))
conn.commit()
def get_server_url() -> str:
"""Get the configured server URL or return a default"""
return get_setting('server_url', 'http://localhost:5000')

1
src/routes/__init__.py
View File

@@ -1 +0,0 @@
# This file is intentionally left blank to make 'routes' a package.

20
src/run.py Normal file
View File

@@ -0,0 +1,20 @@
import os
import shutil
from app import create_app
from app.models import TenantModel
# Create application
app = create_app(os.environ.get('FLASK_ENV', 'development'))
if __name__ == '__main__':
with app.app_context():
# Migrate existing data from JSON if needed (backward compatibility)
# Note: This is for the old database.py migration - keeping for reference
# but in the new structure, this would be handled differently
# Clean up any accidentally created reserved tenants
deleted_count = TenantModel.cleanup_reserved()
if deleted_count > 0:
print(f"Cleaned up {deleted_count} reserved tenant(s)")
app.run(port=5555, host="0.0.0.0", debug=True)

140
src/templates/tenant_selection.html
View File

@@ -1,140 +0,0 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Select or Create Tenant - KCAP Demo Server</title>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.11.0/font/bootstrap-icons.css">
</head>
<body>
<div class="container mt-5">
<h1 class="text-center mb-4">KCAP Demo Server - Multi-Tenant</h1>
{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
{% for category, message in messages %}
<div class="alert alert-{{ 'success' if category == 'success' else 'danger' }} alert-dismissible fade show" role="alert">
{{ message }}
<button type="button" class="btn-close" data-bs-dismiss="alert" aria-label="Close"></button>
</div>
{% endfor %}
{% endif %}
{% endwith %}
<div class="row justify-content-center">
<div class="col-md-8">
<div class="card">
<div class="card-header">
<h3>Select an Existing Tenant</h3>
</div>
<div class="card-body">
{% if tenants %}
<div class="list-group">
{% for tenant in tenants %}
<div class="list-group-item d-flex justify-content-between align-items-center">
<a href="/{{ tenant.id }}/" class="text-decoration-none flex-grow-1">
<div class="d-flex w-100 justify-content-between">
<h5 class="mb-1">{{ tenant.name }}</h5>
<small>{{ tenant.created_at }}</small>
</div>
<p class="mb-1">Username: {{ tenant.username }}</p>
<p class="mb-1">
<strong>Knox Capture AR Template URL:</strong>
<code>{{ server_url }}/{{ tenant.id }}/</code>
</p>
<small>ID: {{ tenant.id }}</small>
</a>
<button type="button" class="btn btn-sm btn-outline-primary me-2" data-bs-toggle="modal" data-bs-target="#qrModal-{{ tenant.id }}" onclick="event.stopPropagation();">
<i class="bi bi-qr-code"></i>
</button>
<form method="POST" action="/tenant/{{ tenant.id }}/delete" class="ms-3" onsubmit="return confirmDelete('{{ tenant.name }}')">
<button type="submit" class="btn btn-danger btn-sm">Delete</button>
</form>
</div>
{% endfor %}
</div>
{% else %}
<p class="text-muted">No tenants exist yet. Create one below!</p>
{% endif %}
</div>
</div>
<div class="card mt-4">
<div class="card-header">
<h3>Create a New Tenant</h3>
</div>
<div class="card-body">
<form id="newTenantForm">
<div class="mb-3">
<label for="tenantId" class="form-label">Tenant ID</label>
<input type="text" class="form-control" id="tenantId" placeholder="e.g., demo-company" required pattern="[a-zA-Z0-9_-]+">
<small class="form-text text-muted">Only letters, numbers, hyphens, and underscores allowed</small>
</div>
<button type="submit" class="btn btn-primary">Create & Enter Tenant</button>
</form>
</div>
</div>
<div class="mt-4 text-center">
<p class="text-muted">
<strong>Note:</strong> When you access a tenant URL directly (e.g., /my-tenant/),
it will be automatically created with default credentials (admin/admin).
</p>
<a href="/settings" class="btn btn-secondary mt-3">Server Settings</a>
</div>
</div>
</div>
</div>
<!-- QR Code Modals for each tenant -->
{% for tenant in tenants %}
<div class="modal fade" id="qrModal-{{ tenant.id }}" tabindex="-1" aria-labelledby="qrModalLabel-{{ tenant.id }}" aria-hidden="true">
<div class="modal-dialog modal-dialog-centered">
<div class="modal-content">
<div class="modal-header">
<h5 class="modal-title" id="qrModalLabel-{{ tenant.id }}">AR Template URL QR Code - {{ tenant.name }}</h5>
<button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="Close"></button>
</div>
<div class="modal-body text-center">
<p class="text-muted mb-3">Scan this QR code for the Knox Capture AR Template URL:</p>
<div class="mb-3">
<img src="/{{ tenant.id }}/qrcode/template" alt="AR Template QR Code" class="img-fluid" style="max-width: 300px;">
</div>
<code class="d-block mt-2">{{ server_url }}/{{ tenant.id }}/</code>
</div>
<div class="modal-footer">
<button type="button" class="btn btn-secondary" data-bs-dismiss="modal">Close</button>
</div>
</div>
</div>
</div>
{% endfor %}
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/js/bootstrap.bundle.min.js"></script>
<script>
const reservedIds = ['admin', 'api', 'login', 'logout', 'arcontentfields', 'arinfo',
'images', 'barcodes', 'static', 'assets', 'js', 'css', 'tenant',
'auth', 'oauth', 'callback', 'webhook', 'health', 'status'];
document.getElementById('newTenantForm').addEventListener('submit', function(e) {
e.preventDefault();
const tenantId = document.getElementById('tenantId').value;
if (tenantId) {
// Convert to lowercase for consistency
const normalizedId = tenantId.toLowerCase();
if (reservedIds.includes(normalizedId)) {
alert(`"${tenantId}" is a reserved name and cannot be used as a tenant ID.`);
return;
}
// Use the normalized (lowercase) ID in the URL
window.location.href = '/' + normalizedId + '/';
}
});
function confirmDelete(tenantName) {
return confirm(`Are you sure you want to delete the tenant "${tenantName}"?\n\nThis will permanently delete all products and data associated with this tenant.`);
}
</script>
</body>
</html>