127 lines
4.5 KiB
Python
127 lines
4.5 KiB
Python
"""Public API routes for AnchorHost."""
|
|
|
|
|
|
from database_service import database_connection, dict_from_row
|
|
from datetime import datetime
|
|
from domain_runtime import domain_manager
|
|
from flask import Blueprint, jsonify, request
|
|
import json
|
|
from kubernetes_service import k8s_manager
|
|
from runtime_config import (
|
|
logger,
|
|
)
|
|
from showcase_service import get_homepage_showcase_links
|
|
from user_service import get_pricing, is_demo_enabled
|
|
|
|
__all__ = ['get_announcements', 'get_public_pricing', 'get_public_settings', 'health_check', 'public_homepage_showcase', 'public_stats']
|
|
|
|
|
|
|
|
bp = Blueprint("public", __name__)
|
|
|
|
|
|
@bp.route('/health', methods=['GET'])
|
|
def health_check():
|
|
database_status = 'connected'
|
|
try:
|
|
with database_connection() as conn:
|
|
conn.cursor().execute('SELECT 1')
|
|
except Exception:
|
|
logger.exception("Health check could not reach the database")
|
|
database_status = 'disconnected'
|
|
|
|
return jsonify({
|
|
'status': 'ok' if database_status == 'connected' else 'unavailable',
|
|
'kubernetes': 'connected' if k8s_manager.connected else 'disconnected',
|
|
'database': database_status,
|
|
'domain_manager': 'available' if domain_manager else 'unavailable',
|
|
'timestamp': datetime.now().isoformat()
|
|
}), 200 if database_status == 'connected' else 503
|
|
|
|
@bp.route('/api/announcements', methods=['GET'])
|
|
def get_announcements():
|
|
"""Get public announcements"""
|
|
try:
|
|
limit = request.args.get('limit', default=10, type=int)
|
|
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
SELECT a.*, u.username as author_name
|
|
FROM announcements a
|
|
JOIN users u ON a.author_id = u.id
|
|
ORDER BY a.created_at DESC
|
|
LIMIT ?
|
|
''', (limit,))
|
|
|
|
announcements = [dict_from_row(row) for row in cursor.fetchall()]
|
|
|
|
# Parse tags if they are stored as JSON string or comma-separated
|
|
for ann in announcements:
|
|
if ann.get('tags'):
|
|
try:
|
|
# Try parsing as JSON first
|
|
ann['tags'] = json.loads(ann['tags'])
|
|
except:
|
|
# Fallback to comma-separated if simple string
|
|
if ',' in ann['tags']:
|
|
ann['tags'] = [tag.strip() for tag in ann['tags'].split(',')]
|
|
else:
|
|
ann['tags'] = [ann['tags']]
|
|
else:
|
|
ann['tags'] = []
|
|
|
|
return jsonify({'success': True, 'announcements': announcements})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching announcements: {e}")
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@bp.route('/api/public/homepage-showcase', methods=['GET'])
|
|
def public_homepage_showcase():
|
|
"""Public: three configurable customer-site URLs for the marketing homepage."""
|
|
try:
|
|
links = get_homepage_showcase_links()
|
|
return jsonify({'success': True, 'links': links})
|
|
except Exception as e:
|
|
logger.error(f"public_homepage_showcase: {e}")
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@bp.route('/api/public/stats', methods=['GET'])
|
|
def public_stats():
|
|
"""Return public platform statistics."""
|
|
try:
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
# Use a named column 'count' to ensure compatibility with DictCursor
|
|
cursor.execute("SELECT COUNT(*) as count FROM users")
|
|
row = cursor.fetchone()
|
|
user_count = row['count'] if row else 0
|
|
|
|
# For a more impressive number, add a base number if it's a new installation
|
|
# or just return the real count. User asked for "how much users there are".
|
|
return jsonify({
|
|
'success': True,
|
|
'user_count': user_count,
|
|
# Add other stats if helpful
|
|
'total_customers': user_count
|
|
})
|
|
except Exception as e:
|
|
logger.error(f"public_stats: {e}")
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@bp.route('/api/pricing', methods=['GET'])
|
|
def get_public_pricing():
|
|
"""Get current pricing (publicly accessible)."""
|
|
return jsonify(get_pricing()), 200
|
|
|
|
@bp.route('/api/settings', methods=['GET'])
|
|
def get_public_settings():
|
|
"""Get public panel settings."""
|
|
return jsonify({
|
|
'success': True,
|
|
'settings': {
|
|
'demo_servers_enabled': is_demo_enabled()
|
|
}
|
|
}), 200
|