Implement API endpoints and documentation pages

- Modified /stats to /api/stats with authentication
- Updated /api/analyze to fully implement analysis functionality
- Changed /api/browse to /admin/api/browse
- Implemented /admin/api/train, /admin/api/users, /admin/api/cluster_tokens endpoints
- Implemented /api/api_tokens endpoint
- Created /api and /admin/api documentation pages with curl examples
parent e7b79990
{% extends 'base.html' %}
{% block title %}Admin API Documentation{% endblock %}
{% block content %}
<div class="container mt-4">
<h1>Admin API Documentation</h1>
<p class="lead">Administrative API endpoints for system management.</p>
<div class="alert alert-warning">
<strong>Authentication:</strong> All endpoints require admin authentication via login session or admin API token (Bearer token in Authorization header).
</div>
<h2>Endpoints</h2>
<div class="endpoint" id="browse">
<h3>GET /admin/api/browse</h3>
<p>Browse files in the configured server directory.</p>
<h4>Parameters:</h4>
<ul>
<li><code>path</code> (string, optional): Directory path to browse (default: root)</li>
</ul>
<h4>Curl Example:</h4>
<pre><code>curl -H "Authorization: Bearer YOUR_ADMIN_API_TOKEN" "{{ request.host_url }}admin/api/browse?path=/some/path"</code></pre>
<h4>Response:</h4>
<pre><code>{
"current_path": "/some/path",
"items": [
{"name": "file.txt", "path": "/some/path/file.txt", "is_dir": false, "size": 1024}
]
}</code></pre>
</div>
<div class="endpoint" id="train">
<h3>POST /admin/api/train</h3>
<p>Start model training. Costs 100 tokens (free for admins).</p>
<h4>Parameters:</h4>
<ul>
<li><code>output_model</code> (string, optional): Output model name (default: "MyCustomModel")</li>
<li><code>description</code> (string, optional): Model description</li>
<li><code>train_path</code> (string, required): Path to training data directory</li>
</ul>
<h4>Curl Example:</h4>
<pre><code>curl -X POST -H "Authorization: Bearer YOUR_ADMIN_API_TOKEN" -H "Content-Type: application/json" -d '{
"output_model": "MyModel",
"description": "Custom trained model",
"train_path": "/path/to/training/data"
}' {{ request.host_url }}admin/api/train</code></pre>
<h4>Response:</h4>
<pre><code>{
"message": "Training completed",
"tokens_used": 100,
"remaining_tokens": 900
}</code></pre>
</div>
<div class="endpoint" id="users">
<h3>GET /admin/api/users</h3>
<p>List all users in the system.</p>
<h4>Curl Example:</h4>
<pre><code>curl -H "Authorization: Bearer YOUR_ADMIN_API_TOKEN" {{ request.host_url }}admin/api/users</code></pre>
<h4>Response:</h4>
<pre><code>{
"users": [
{"id": 1, "username": "user1", "email": "user1@example.com", "role": "user", "tokens": 100, "active": true, "created_at": "2024-01-01T00:00:00Z"}
]
}</code></pre>
</div>
<div class="endpoint" id="cluster_tokens">
<h3>GET /admin/api/cluster_tokens</h3>
<p>List all cluster tokens.</p>
<h4>Curl Example:</h4>
<pre><code>curl -H "Authorization: Bearer YOUR_ADMIN_API_TOKEN" {{ request.host_url }}admin/api/cluster_tokens</code></pre>
<h4>Response:</h4>
<pre><code>{
"tokens": [
{"id": 1, "name": "Worker Token", "token": "abc123...", "active": true, "created_at": "2024-01-01T00:00:00Z"}
]
}</code></pre>
</div>
</div>
{% endblock %}
\ No newline at end of file
{% extends 'base.html' %}
{% block title %}API Documentation{% endblock %}
{% block content %}
<div class="container mt-4">
<h1>API Documentation</h1>
<p class="lead">Programmatic access to Video AI functionality.</p>
<div class="alert alert-info">
<strong>Authentication:</strong> All endpoints require authentication via login session or API token (Bearer token in Authorization header).
</div>
<h2>Endpoints</h2>
<div class="endpoint" id="stats">
<h3>GET /api/stats</h3>
<p>Get system statistics including CPU, RAM, and GPU information.</p>
<h4>Curl Example:</h4>
<pre><code>curl -H "Authorization: Bearer YOUR_API_TOKEN" {{ request.host_url }}api/stats</code></pre>
<h4>Response:</h4>
<pre><code>{
"status": "Idle",
"gpu_count": 1,
"gpus": [{"name": "NVIDIA RTX 3080", "memory_used": 0.5, "memory_total": 10.0, "utilization": 0}],
"cpu_percent": 15.2,
"ram_used": 4.2,
"ram_total": 16.0
}</code></pre>
</div>
<div class="endpoint" id="analyze">
<h3>POST /api/analyze</h3>
<p>Perform media analysis using AI models. Costs 10 tokens (free for admins).</p>
<h4>Parameters:</h4>
<ul>
<li><code>model_path</code> (string, optional): Model to use (default: "Qwen/Qwen2.5-VL-7B-Instruct")</li>
<li><code>prompt</code> (string, optional): Analysis prompt (default: "Describe this image.")</li>
<li><code>file_path</code> (string, required): Path to media file</li>
<li><code>interval</code> (int, optional): Frame interval for video (default: 10)</li>
</ul>
<h4>Curl Example:</h4>
<pre><code>curl -X POST -H "Authorization: Bearer YOUR_API_TOKEN" -H "Content-Type: application/json" -d '{
"model_path": "Qwen/Qwen2.5-VL-7B-Instruct",
"prompt": "Describe this video",
"file_path": "/path/to/video.mp4",
"interval": 30
}' {{ request.host_url }}api/analyze</code></pre>
<h4>Response:</h4>
<pre><code>{
"result": "Analysis result here...",
"tokens_used": 10,
"remaining_tokens": 90
}</code></pre>
</div>
<div class="endpoint" id="api_tokens">
<h3>GET /api/api_tokens</h3>
<p>List your API tokens.</p>
<h4>Curl Example:</h4>
<pre><code>curl -H "Authorization: Bearer YOUR_API_TOKEN" {{ request.host_url }}api/api_tokens</code></pre>
<h4>Response:</h4>
<pre><code>{
"tokens": [
{"id": 1, "name": "My Token", "created_at": "2024-01-01T00:00:00Z", "last_used": null}
]
}</code></pre>
</div>
</div>
{% endblock %}
\ No newline at end of file
...@@ -81,6 +81,61 @@ def admin_required(f): ...@@ -81,6 +81,61 @@ def admin_required(f):
decorated_function.__name__ = f.__name__ decorated_function.__name__ = f.__name__
return decorated_function return decorated_function
def api_auth_required(f):
"""Decorator to require authentication via session or API token."""
def decorated_function(*args, **kwargs):
# Check for session auth first
user = get_current_user_session()
if user:
request.api_user = user
return f(*args, **kwargs)
# Check for API token auth
auth_header = request.headers.get('Authorization')
if auth_header and auth_header.startswith('Bearer '):
token = auth_header.split(' ')[1]
from .database import validate_user_api_token
user = validate_user_api_token(token)
if user:
request.api_user = user
return f(*args, **kwargs)
# No valid auth
if request.is_json or request.path.startswith('/api/'):
return json.dumps({'error': 'Authentication required'}), 401
else:
return redirect(url_for('login'))
decorated_function.__name__ = f.__name__
return decorated_function
def admin_api_auth_required(f):
"""Decorator to require admin authentication via session or API token."""
def decorated_function(*args, **kwargs):
# Check for session auth first
user = get_current_user_session()
if user and user.get('role') == 'admin':
request.api_user = user
return f(*args, **kwargs)
# Check for API token auth
auth_header = request.headers.get('Authorization')
if auth_header and auth_header.startswith('Bearer '):
token = auth_header.split(' ')[1]
from .database import validate_user_api_token
user = validate_user_api_token(token)
if user and user.get('role') == 'admin':
request.api_user = user
return f(*args, **kwargs)
# No valid auth
if request.is_json or request.path.startswith('/api/'):
return json.dumps({'error': 'Admin authentication required'}), 401
else:
flash('Admin access required', 'error')
return redirect(url_for('dashboard'))
decorated_function.__name__ = f.__name__
return decorated_function
def send_to_backend(msg_type: str, data: dict) -> str: def send_to_backend(msg_type: str, data: dict) -> str:
"""Send message to backend and return message id.""" """Send message to backend and return message id."""
msg_id = str(uuid.uuid4()) msg_id = str(uuid.uuid4())
...@@ -704,9 +759,10 @@ def admin_delete_user(user_id): ...@@ -704,9 +759,10 @@ def admin_delete_user(user_id):
flash('Deletion not confirmed.', 'error') flash('Deletion not confirmed.', 'error')
return redirect(url_for('admin')) return redirect(url_for('admin'))
@app.route('/stats') @app.route('/api/stats')
def stats(): @api_auth_required
"""Get system stats for the sidebar.""" def api_stats():
"""Get system stats for authenticated users."""
import psutil import psutil
import torch import torch
import time import time
...@@ -737,18 +793,10 @@ def stats(): ...@@ -737,18 +793,10 @@ def stats():
return json.dumps(data) return json.dumps(data)
@app.route('/api/analyze', methods=['POST']) @app.route('/api/analyze', methods=['POST'])
@api_auth_required
def api_analyze(): def api_analyze():
"""API endpoint for analysis using user API tokens.""" """API endpoint for analysis using authentication."""
auth_header = request.headers.get('Authorization') user = request.api_user
if not auth_header or not auth_header.startswith('Bearer '):
return json.dumps({'error': 'Missing or invalid authorization header'}), 401
token = auth_header.split(' ')[1]
from .database import validate_user_api_token
user = validate_user_api_token(token)
if not user:
return json.dumps({'error': 'Invalid or expired token'}), 401
# Check token balance (skip for admin users) # Check token balance (skip for admin users)
if user.get('role') != 'admin': if user.get('role') != 'admin':
...@@ -760,22 +808,149 @@ def api_analyze(): ...@@ -760,22 +808,149 @@ def api_analyze():
model_path = request.json.get('model_path', 'Qwen/Qwen2.5-VL-7B-Instruct') model_path = request.json.get('model_path', 'Qwen/Qwen2.5-VL-7B-Instruct')
prompt = request.json.get('prompt', 'Describe this image.') prompt = request.json.get('prompt', 'Describe this image.')
file_path = request.json.get('file_path') file_path = request.json.get('file_path')
interval = request.json.get('interval', 10)
if not file_path: if not file_path:
return json.dumps({'error': 'file_path is required'}), 400 return json.dumps({'error': 'file_path is required'}), 400
# For API, we'll simulate the analysis (in real implementation, send to backend) # Send to backend for processing
result = f"Analysis completed for {file_path} using model {model_path}" data = {
'model_path': model_path,
'prompt': prompt,
'local_path': file_path,
'interval': interval,
'user_id': user['id']
}
msg_id = send_to_backend('analyze_request', data)
result_data = get_result(msg_id)
if 'data' in result_data:
result = result_data['data'].get('result', 'Analysis completed')
# Deduct tokens (skip for admin users)
if user.get('role') != 'admin':
update_user_tokens(user['id'], -10)
return json.dumps({
'result': result,
'tokens_used': 10,
'remaining_tokens': get_user_tokens(user['id'])
})
else:
error = result_data.get('error', 'Analysis failed')
return json.dumps({'error': error}), 500
@app.route('/admin/api/train', methods=['POST'])
@admin_api_auth_required
def admin_api_train():
"""API endpoint for training (admin only)."""
user = request.api_user
# Deduct tokens (skip for admin users) # Check token balance (skip for admin users)
if user.get('role') != 'admin': if user.get('role') != 'admin':
update_user_tokens(user['id'], -10) tokens = get_user_tokens(user['id'])
if tokens < 100:
return json.dumps({'error': 'Insufficient tokens. Training requires 100 tokens.'}), 402
# Process training request
output_model = request.json.get('output_model', 'MyCustomModel')
description = request.json.get('description', '')
train_path = request.json.get('train_path')
if not train_path or not os.path.exists(train_path):
return json.dumps({'error': 'Valid train_path is required'}), 400
# Send to backend for processing
data = {
'output_model': output_model,
'description': description,
'train_path': train_path,
'user_id': user['id']
}
msg_id = send_to_backend('train_request', data)
result_data = get_result(msg_id)
if 'data' in result_data:
message = result_data['data'].get('message', 'Training completed')
# Deduct tokens (skip for admin users)
if user.get('role') != 'admin':
update_user_tokens(user['id'], -100)
return json.dumps({
'message': message,
'tokens_used': 100,
'remaining_tokens': get_user_tokens(user['id'])
})
else:
error = result_data.get('error', 'Training failed')
return json.dumps({'error': error}), 500
@app.route('/admin/api/users')
@admin_api_auth_required
def admin_api_users():
"""API endpoint to list all users (admin only)."""
from .database import get_all_users
users = get_all_users()
# Return user data without sensitive info like passwords
user_list = []
for u in users:
user_list.append({
'id': u['id'],
'username': u['username'],
'email': u['email'],
'role': u['role'],
'tokens': u['tokens'],
'active': u['active'],
'created_at': u['created_at']
})
return json.dumps({'users': user_list})
@app.route('/admin/api/cluster_tokens')
@admin_api_auth_required
def admin_api_cluster_tokens():
"""API endpoint to list cluster tokens (admin only)."""
from .database import get_worker_tokens
worker_tokens = get_worker_tokens()
# Return token data
token_list = []
for t in worker_tokens:
token_list.append({
'id': t['id'],
'name': t['name'],
'token': t['token'], # Note: in real implementation, might not return full token
'active': t['active'],
'created_at': t['created_at']
})
return json.dumps({'tokens': token_list})
@app.route('/api/api_tokens')
@api_auth_required
def api_api_tokens():
"""API endpoint to list user's API tokens."""
user = request.api_user
from .database import get_user_api_tokens
user_tokens = get_user_api_tokens(user['id'])
# Return token data (without full token for security)
token_list = []
for t in user_tokens:
token_list.append({
'id': t['id'],
'name': t['name'],
'created_at': t['created_at'],
'last_used': t.get('last_used')
})
return json.dumps({'tokens': token_list})
return json.dumps({ @app.route('/api')
'result': result, @login_required
'tokens_used': 10, def api_docs():
'remaining_tokens': get_user_tokens(user['id']) """API documentation page."""
}) user = get_current_user_session()
return render_template('api.html', user=user, active_page='api')
@app.route('/admin/api')
@admin_required
def admin_api_docs():
"""Admin API documentation page."""
user = get_current_user_session()
return render_template('admin_api.html', user=user, active_page='admin_api')
@app.route('/static/<path:filename>') @app.route('/static/<path:filename>')
def serve_static(filename): def serve_static(filename):
...@@ -785,9 +960,9 @@ def serve_static(filename): ...@@ -785,9 +960,9 @@ def serve_static(filename):
def serve_logo(): def serve_logo():
return send_from_directory('..', 'image.jpg') return send_from_directory('..', 'image.jpg')
@app.route('/api/browse') @app.route('/admin/api/browse')
@admin_required @admin_api_auth_required
def browse_files(): def admin_api_browse():
"""Browse files in server directory (admin only).""" """Browse files in server directory (admin only)."""
path = request.args.get('path', '') path = request.args.get('path', '')
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment