423 lines
15 KiB
Python
423 lines
15 KiB
Python
import os
|
|
import requests
|
|
from flask import Flask, render_template, request, redirect, url_for, flash, session, jsonify
|
|
from werkzeug.utils import secure_filename
|
|
import json
|
|
from datetime import datetime
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = 'your-secret-key-change-in-production'
|
|
|
|
# Configuration
|
|
UPLOAD_FOLDER = 'uploads'
|
|
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'bmp', 'webp'}
|
|
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
|
|
app.config['MAX_CONTENT_LENGTH'] = 10 * 1024 * 1024 # 10MB max file size
|
|
|
|
# Ensure upload folder exists
|
|
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
|
|
|
|
def allowed_file(filename):
|
|
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
|
|
|
|
def get_api_base_url():
|
|
return session.get('api_base_url', 'http://localhost:8000')
|
|
|
|
def get_api_key():
|
|
return session.get('api_key')
|
|
|
|
def make_api_request(method, endpoint, data=None, files=None, params=None):
|
|
"""Make API request with proper headers"""
|
|
api_key = get_api_key()
|
|
if not api_key:
|
|
return None, "API key not set"
|
|
|
|
headers = {'X-API-Key': api_key}
|
|
url = f"{get_api_base_url()}/api/v1{endpoint}"
|
|
|
|
try:
|
|
if method == 'GET':
|
|
response = requests.get(url, headers=headers, params=params)
|
|
elif method == 'POST':
|
|
if files:
|
|
response = requests.post(url, headers=headers, data=data, files=files)
|
|
else:
|
|
headers['Content-Type'] = 'application/json'
|
|
response = requests.post(url, headers=headers, json=data)
|
|
elif method == 'PUT':
|
|
headers['Content-Type'] = 'application/json'
|
|
response = requests.put(url, headers=headers, json=data)
|
|
elif method == 'DELETE':
|
|
response = requests.delete(url, headers=headers)
|
|
else:
|
|
return None, f"Unsupported method: {method}"
|
|
|
|
if response.status_code == 204:
|
|
return True, None
|
|
|
|
return response.json(), None if response.status_code < 400 else response.json().get('detail', 'Unknown error')
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return None, f"Request failed: {str(e)}"
|
|
except json.JSONDecodeError:
|
|
return None, "Invalid JSON response"
|
|
|
|
@app.route('/')
|
|
def index():
|
|
return render_template('index.html')
|
|
|
|
@app.route('/config', methods=['GET', 'POST'])
|
|
def config():
|
|
if request.method == 'POST':
|
|
api_base_url = request.form.get('api_base_url', '').strip()
|
|
api_key = request.form.get('api_key', '').strip()
|
|
|
|
if api_base_url:
|
|
session['api_base_url'] = api_base_url.rstrip('/')
|
|
if api_key:
|
|
session['api_key'] = api_key
|
|
|
|
flash('Configuration updated successfully!', 'success')
|
|
return redirect(url_for('index'))
|
|
|
|
return render_template('config.html',
|
|
api_base_url=get_api_base_url(),
|
|
api_key=get_api_key())
|
|
|
|
# Teams CRUD
|
|
@app.route('/teams')
|
|
def teams():
|
|
data, error = make_api_request('GET', '/teams')
|
|
if error:
|
|
flash(f'Error loading teams: {error}', 'error')
|
|
return render_template('teams.html', teams=[])
|
|
|
|
return render_template('teams.html', teams=data.get('teams', []))
|
|
|
|
@app.route('/teams/create', methods=['GET', 'POST'])
|
|
def create_team():
|
|
if request.method == 'POST':
|
|
team_data = {
|
|
'name': request.form.get('name'),
|
|
'description': request.form.get('description', '')
|
|
}
|
|
|
|
data, error = make_api_request('POST', '/teams', team_data)
|
|
if error:
|
|
flash(f'Error creating team: {error}', 'error')
|
|
else:
|
|
flash('Team created successfully!', 'success')
|
|
return redirect(url_for('teams'))
|
|
|
|
return render_template('team_form.html', team=None, action='Create')
|
|
|
|
@app.route('/teams/<team_id>/edit', methods=['GET', 'POST'])
|
|
def edit_team(team_id):
|
|
if request.method == 'POST':
|
|
team_data = {
|
|
'name': request.form.get('name'),
|
|
'description': request.form.get('description', '')
|
|
}
|
|
|
|
data, error = make_api_request('PUT', f'/teams/{team_id}', team_data)
|
|
if error:
|
|
flash(f'Error updating team: {error}', 'error')
|
|
else:
|
|
flash('Team updated successfully!', 'success')
|
|
return redirect(url_for('teams'))
|
|
|
|
# Get team data for editing
|
|
data, error = make_api_request('GET', f'/teams/{team_id}')
|
|
if error:
|
|
flash(f'Error loading team: {error}', 'error')
|
|
return redirect(url_for('teams'))
|
|
|
|
return render_template('team_form.html', team=data, action='Edit')
|
|
|
|
@app.route('/teams/<team_id>/delete', methods=['POST'])
|
|
def delete_team(team_id):
|
|
data, error = make_api_request('DELETE', f'/teams/{team_id}')
|
|
if error:
|
|
flash(f'Error deleting team: {error}', 'error')
|
|
else:
|
|
flash('Team deleted successfully!', 'success')
|
|
|
|
return redirect(url_for('teams'))
|
|
|
|
# Users CRUD
|
|
@app.route('/users')
|
|
def users():
|
|
data, error = make_api_request('GET', '/users')
|
|
if error:
|
|
flash(f'Error loading users: {error}', 'error')
|
|
return render_template('users.html', users=[])
|
|
|
|
return render_template('users.html', users=data.get('users', []))
|
|
|
|
@app.route('/users/create', methods=['GET', 'POST'])
|
|
def create_user():
|
|
if request.method == 'POST':
|
|
user_data = {
|
|
'name': request.form.get('name'),
|
|
'email': request.form.get('email'),
|
|
'team_id': request.form.get('team_id') or None,
|
|
'is_admin': request.form.get('is_admin') == 'on'
|
|
}
|
|
|
|
data, error = make_api_request('POST', '/users', user_data)
|
|
if error:
|
|
flash(f'Error creating user: {error}', 'error')
|
|
else:
|
|
flash('User created successfully!', 'success')
|
|
return redirect(url_for('users'))
|
|
|
|
# Get teams for dropdown
|
|
teams_data, teams_error = make_api_request('GET', '/teams')
|
|
teams = teams_data.get('teams', []) if teams_data else []
|
|
|
|
return render_template('user_form.html', user=None, teams=teams, action='Create')
|
|
|
|
@app.route('/users/<user_id>/edit', methods=['GET', 'POST'])
|
|
def edit_user(user_id):
|
|
if request.method == 'POST':
|
|
user_data = {
|
|
'name': request.form.get('name'),
|
|
'email': request.form.get('email'),
|
|
'is_admin': request.form.get('is_admin') == 'on'
|
|
}
|
|
|
|
data, error = make_api_request('PUT', f'/users/{user_id}', user_data)
|
|
if error:
|
|
flash(f'Error updating user: {error}', 'error')
|
|
else:
|
|
flash('User updated successfully!', 'success')
|
|
return redirect(url_for('users'))
|
|
|
|
# Get user data for editing
|
|
data, error = make_api_request('GET', f'/users/{user_id}')
|
|
if error:
|
|
flash(f'Error loading user: {error}', 'error')
|
|
return redirect(url_for('users'))
|
|
|
|
# Get teams for dropdown
|
|
teams_data, teams_error = make_api_request('GET', '/teams')
|
|
teams = teams_data.get('teams', []) if teams_data else []
|
|
|
|
return render_template('user_form.html', user=data, teams=teams, action='Edit')
|
|
|
|
@app.route('/users/<user_id>/delete', methods=['POST'])
|
|
def delete_user(user_id):
|
|
data, error = make_api_request('DELETE', f'/users/{user_id}')
|
|
if error:
|
|
flash(f'Error deleting user: {error}', 'error')
|
|
else:
|
|
flash('User deleted successfully!', 'success')
|
|
|
|
return redirect(url_for('users'))
|
|
|
|
# API Keys CRUD
|
|
@app.route('/api-keys')
|
|
def api_keys():
|
|
data, error = make_api_request('GET', '/auth/api-keys')
|
|
if error:
|
|
flash(f'Error loading API keys: {error}', 'error')
|
|
return render_template('api_keys.html', api_keys=[])
|
|
|
|
return render_template('api_keys.html', api_keys=data.get('api_keys', []))
|
|
|
|
@app.route('/api-keys/create', methods=['GET', 'POST'])
|
|
def create_api_key():
|
|
if request.method == 'POST':
|
|
key_data = {
|
|
'name': request.form.get('name'),
|
|
'description': request.form.get('description', ''),
|
|
'team_id': request.form.get('team_id') or None,
|
|
'user_id': request.form.get('user_id') or None
|
|
}
|
|
|
|
data, error = make_api_request('POST', '/auth/api-keys', key_data)
|
|
if error:
|
|
flash(f'Error creating API key: {error}', 'error')
|
|
else:
|
|
flash(f'API key created successfully! Key: {data.get("key")}', 'success')
|
|
return redirect(url_for('api_keys'))
|
|
|
|
# Get teams and users for dropdowns
|
|
teams_data, _ = make_api_request('GET', '/teams')
|
|
users_data, _ = make_api_request('GET', '/users')
|
|
teams = teams_data.get('teams', []) if teams_data else []
|
|
users = users_data.get('users', []) if users_data else []
|
|
|
|
return render_template('api_key_form.html', teams=teams, users=users)
|
|
|
|
@app.route('/api-keys/<key_id>/delete', methods=['POST'])
|
|
def delete_api_key(key_id):
|
|
data, error = make_api_request('DELETE', f'/auth/api-keys/{key_id}')
|
|
if error:
|
|
flash(f'Error deleting API key: {error}', 'error')
|
|
else:
|
|
flash('API key deleted successfully!', 'success')
|
|
|
|
return redirect(url_for('api_keys'))
|
|
|
|
# Images CRUD
|
|
@app.route('/images')
|
|
def images():
|
|
page = request.args.get('page', 1, type=int)
|
|
limit = 20
|
|
skip = (page - 1) * limit
|
|
|
|
params = {'skip': skip, 'limit': limit}
|
|
tags = request.args.get('tags')
|
|
if tags:
|
|
params['tags'] = tags
|
|
|
|
data, error = make_api_request('GET', '/images', params=params)
|
|
if error:
|
|
flash(f'Error loading images: {error}', 'error')
|
|
return render_template('images.html', images=[], total=0, page=page, limit=limit)
|
|
|
|
return render_template('images.html',
|
|
images=data.get('images', []),
|
|
total=data.get('total', 0),
|
|
page=page,
|
|
limit=limit)
|
|
|
|
@app.route('/images/upload', methods=['GET', 'POST'])
|
|
def upload_image():
|
|
if request.method == 'POST':
|
|
if 'file' not in request.files:
|
|
flash('No file selected', 'error')
|
|
return redirect(request.url)
|
|
|
|
file = request.files['file']
|
|
if file.filename == '':
|
|
flash('No file selected', 'error')
|
|
return redirect(request.url)
|
|
|
|
if file and allowed_file(file.filename):
|
|
filename = secure_filename(file.filename)
|
|
|
|
# Prepare form data
|
|
form_data = {
|
|
'description': request.form.get('description', ''),
|
|
'tags': request.form.get('tags', ''),
|
|
'collection_id': request.form.get('collection_id') or None
|
|
}
|
|
|
|
# Prepare files
|
|
files = {'file': (filename, file, file.content_type)}
|
|
|
|
data, error = make_api_request('POST', '/images', data=form_data, files=files)
|
|
if error:
|
|
flash(f'Error uploading image: {error}', 'error')
|
|
else:
|
|
flash('Image uploaded successfully!', 'success')
|
|
return redirect(url_for('images'))
|
|
else:
|
|
flash('Invalid file type. Please upload an image file.', 'error')
|
|
|
|
return render_template('image_upload.html')
|
|
|
|
@app.route('/images/<image_id>')
|
|
def view_image(image_id):
|
|
data, error = make_api_request('GET', f'/images/{image_id}')
|
|
if error:
|
|
flash(f'Error loading image: {error}', 'error')
|
|
return redirect(url_for('images'))
|
|
|
|
return render_template('image_detail.html', image=data)
|
|
|
|
@app.route('/images/<image_id>/edit', methods=['GET', 'POST'])
|
|
def edit_image(image_id):
|
|
if request.method == 'POST':
|
|
image_data = {
|
|
'description': request.form.get('description', ''),
|
|
'tags': [tag.strip() for tag in request.form.get('tags', '').split(',') if tag.strip()]
|
|
}
|
|
|
|
data, error = make_api_request('PUT', f'/images/{image_id}', image_data)
|
|
if error:
|
|
flash(f'Error updating image: {error}', 'error')
|
|
else:
|
|
flash('Image updated successfully!', 'success')
|
|
return redirect(url_for('view_image', image_id=image_id))
|
|
|
|
# Get image data for editing
|
|
data, error = make_api_request('GET', f'/images/{image_id}')
|
|
if error:
|
|
flash(f'Error loading image: {error}', 'error')
|
|
return redirect(url_for('images'))
|
|
|
|
return render_template('image_edit.html', image=data)
|
|
|
|
@app.route('/images/<image_id>/delete', methods=['POST'])
|
|
def delete_image(image_id):
|
|
data, error = make_api_request('DELETE', f'/images/{image_id}')
|
|
if error:
|
|
flash(f'Error deleting image: {error}', 'error')
|
|
else:
|
|
flash('Image deleted successfully!', 'success')
|
|
|
|
return redirect(url_for('images'))
|
|
|
|
# Search functionality
|
|
@app.route('/search', methods=['GET', 'POST'])
|
|
def search():
|
|
if request.method == 'POST':
|
|
query = request.form.get('query', '').strip()
|
|
limit = int(request.form.get('limit', 10))
|
|
threshold = float(request.form.get('threshold', 0.7))
|
|
tags = request.form.get('tags', '').strip()
|
|
|
|
params = {
|
|
'q': query,
|
|
'limit': limit,
|
|
'threshold': threshold
|
|
}
|
|
if tags:
|
|
params['tags'] = tags
|
|
|
|
data, error = make_api_request('GET', '/search', params=params)
|
|
if error:
|
|
flash(f'Search error: {error}', 'error')
|
|
return render_template('search.html', results=[], query=query)
|
|
|
|
return render_template('search.html',
|
|
results=data.get('results', []),
|
|
query=query,
|
|
total=data.get('total', 0))
|
|
|
|
return render_template('search.html', results=[], query='')
|
|
|
|
@app.route('/bootstrap', methods=['GET', 'POST'])
|
|
def bootstrap():
|
|
if request.method == 'POST':
|
|
bootstrap_data = {
|
|
'team_name': request.form.get('team_name'),
|
|
'admin_email': request.form.get('admin_email'),
|
|
'admin_name': request.form.get('admin_name'),
|
|
'api_key_name': request.form.get('api_key_name', 'Initial API Key')
|
|
}
|
|
|
|
# Make bootstrap request without API key
|
|
headers = {'Content-Type': 'application/json'}
|
|
url = f"{get_api_base_url()}/api/v1/auth/bootstrap"
|
|
|
|
try:
|
|
response = requests.post(url, headers=headers, json=bootstrap_data)
|
|
if response.status_code == 201:
|
|
data = response.json()
|
|
session['api_key'] = data.get('key')
|
|
flash(f'Bootstrap successful! API Key: {data.get("key")}', 'success')
|
|
return redirect(url_for('index'))
|
|
else:
|
|
error_data = response.json()
|
|
flash(f'Bootstrap error: {error_data.get("detail", "Unknown error")}', 'error')
|
|
except Exception as e:
|
|
flash(f'Bootstrap failed: {str(e)}', 'error')
|
|
|
|
return render_template('bootstrap.html')
|
|
|
|
if __name__ == '__main__':
|
|
app.run(debug=True, port=5000) |