675 lines
26 KiB
Python
675 lines
26 KiB
Python
"""
|
|
Admin Status Routes
|
|
===================
|
|
|
|
System status, health check, and debug panel routes.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import subprocess
|
|
import platform
|
|
from datetime import datetime, timedelta
|
|
|
|
from flask import render_template, request, redirect, url_for, flash, jsonify, Response
|
|
from flask_login import login_required, current_user
|
|
from sqlalchemy import func, text
|
|
|
|
from . import bp
|
|
from database import (
|
|
SessionLocal, Company, User, AuditLog, SecurityAlert,
|
|
CompanySocialMedia, CompanyWebsiteAnalysis
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ============================================================
|
|
# SYSTEM STATUS DASHBOARD
|
|
# ============================================================
|
|
|
|
@bp.route('/status')
|
|
@login_required
|
|
def admin_status():
|
|
"""System status dashboard with real-time metrics"""
|
|
if not current_user.is_admin:
|
|
flash('Brak uprawnień.', 'error')
|
|
return redirect(url_for('public.dashboard'))
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
# Current timestamp
|
|
now = datetime.now()
|
|
|
|
# ===== SYSTEM METRICS =====
|
|
system_metrics = {
|
|
'hostname': platform.node(),
|
|
'os': f"{platform.system()} {platform.release()}",
|
|
'python': platform.python_version(),
|
|
}
|
|
|
|
# CPU usage (via top command)
|
|
try:
|
|
result = subprocess.run(['top', '-l', '1', '-n', '0'], capture_output=True, text=True, timeout=5)
|
|
for line in result.stdout.split('\n'):
|
|
if 'CPU usage' in line:
|
|
# Parse: "CPU usage: 5.88% user, 8.82% sys, 85.29% idle"
|
|
parts = line.split(':')[1].strip().split(',')
|
|
user = float(parts[0].replace('% user', '').strip())
|
|
sys_cpu = float(parts[1].replace('% sys', '').strip())
|
|
idle = float(parts[2].replace('% idle', '').strip())
|
|
system_metrics['cpu_percent'] = round(user + sys_cpu, 1)
|
|
system_metrics['cpu_idle'] = round(idle, 1)
|
|
break
|
|
except Exception:
|
|
# Linux fallback
|
|
try:
|
|
result = subprocess.run(['grep', 'cpu ', '/proc/stat'], capture_output=True, text=True, timeout=5)
|
|
if result.returncode == 0:
|
|
parts = result.stdout.split()
|
|
idle = int(parts[4])
|
|
total = sum(int(x) for x in parts[1:])
|
|
system_metrics['cpu_percent'] = round(100 * (1 - idle / total), 1)
|
|
system_metrics['cpu_idle'] = round(100 * idle / total, 1)
|
|
except Exception:
|
|
system_metrics['cpu_percent'] = None
|
|
system_metrics['cpu_idle'] = None
|
|
|
|
# RAM usage
|
|
try:
|
|
# macOS
|
|
result = subprocess.run(['vm_stat'], capture_output=True, text=True, timeout=5)
|
|
if result.returncode == 0 and 'Pages' in result.stdout:
|
|
lines = result.stdout.strip().split('\n')
|
|
page_size = 16384 # bytes
|
|
stats = {}
|
|
for line in lines[1:]:
|
|
if ':' in line:
|
|
key, val = line.split(':')
|
|
stats[key.strip()] = int(val.strip().rstrip('.'))
|
|
free = stats.get('Pages free', 0) * page_size
|
|
active = stats.get('Pages active', 0) * page_size
|
|
inactive = stats.get('Pages inactive', 0) * page_size
|
|
wired = stats.get('Pages wired down', 0) * page_size
|
|
total_used = active + inactive + wired
|
|
total_mem = total_used + free
|
|
system_metrics['ram_total_gb'] = round(total_mem / (1024**3), 1)
|
|
system_metrics['ram_used_gb'] = round(total_used / (1024**3), 1)
|
|
system_metrics['ram_percent'] = round(100 * total_used / total_mem, 1)
|
|
else:
|
|
raise Exception("Not macOS")
|
|
except Exception:
|
|
# Linux fallback
|
|
try:
|
|
result = subprocess.run(['free', '-b'], capture_output=True, text=True, timeout=5)
|
|
if result.returncode == 0:
|
|
lines = result.stdout.strip().split('\n')
|
|
mem_line = lines[1].split()
|
|
total = int(mem_line[1])
|
|
used = int(mem_line[2])
|
|
system_metrics['ram_total_gb'] = round(total / (1024**3), 1)
|
|
system_metrics['ram_used_gb'] = round(used / (1024**3), 1)
|
|
system_metrics['ram_percent'] = round(100 * used / total, 1)
|
|
except Exception:
|
|
system_metrics['ram_total_gb'] = None
|
|
system_metrics['ram_used_gb'] = None
|
|
system_metrics['ram_percent'] = None
|
|
|
|
# Disk usage
|
|
try:
|
|
result = subprocess.run(['df', '-h', '/'], capture_output=True, text=True, timeout=5)
|
|
if result.returncode == 0:
|
|
lines = result.stdout.strip().split('\n')
|
|
parts = lines[1].split()
|
|
system_metrics['disk_total'] = parts[1]
|
|
system_metrics['disk_used'] = parts[2]
|
|
system_metrics['disk_percent'] = int(parts[4].replace('%', ''))
|
|
except Exception:
|
|
system_metrics['disk_total'] = None
|
|
system_metrics['disk_used'] = None
|
|
system_metrics['disk_percent'] = None
|
|
|
|
# System uptime
|
|
try:
|
|
result = subprocess.run(['uptime'], capture_output=True, text=True, timeout=5)
|
|
if result.returncode == 0:
|
|
system_metrics['uptime'] = result.stdout.strip().split('up')[1].split(',')[0].strip()
|
|
except Exception:
|
|
system_metrics['uptime'] = None
|
|
|
|
# ===== DATABASE METRICS =====
|
|
db_metrics = {}
|
|
|
|
try:
|
|
# PostgreSQL version
|
|
version_result = db.execute(text("SELECT version()")).scalar()
|
|
# Extract just version number: "PostgreSQL 16.11 ..." -> "16.11"
|
|
if version_result:
|
|
import re
|
|
match = re.search(r'PostgreSQL (\d+\.\d+)', version_result)
|
|
db_metrics['version'] = match.group(1) if match else version_result.split()[1]
|
|
|
|
# Database size
|
|
result = db.execute(text("SELECT pg_database_size(current_database())")).scalar()
|
|
db_metrics['size_mb'] = round(result / (1024 * 1024), 2)
|
|
|
|
# Active connections
|
|
result = db.execute(text("SELECT count(*) FROM pg_stat_activity WHERE state = 'active'")).scalar()
|
|
db_metrics['active_connections'] = result
|
|
|
|
# Total connections
|
|
result = db.execute(text("SELECT count(*) FROM pg_stat_activity")).scalar()
|
|
db_metrics['total_connections'] = result
|
|
|
|
# Table counts
|
|
db_metrics['companies'] = db.query(Company).count()
|
|
db_metrics['users'] = db.query(User).count()
|
|
|
|
# Get additional counts if tables exist
|
|
try:
|
|
from database import AIChatMessage
|
|
db_metrics['chat_messages'] = db.query(AIChatMessage).count()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
db_metrics['social_media'] = db.query(CompanySocialMedia).count()
|
|
db_metrics['seo_audits'] = db.query(CompanyWebsiteAnalysis).filter(
|
|
CompanyWebsiteAnalysis.seo_audited_at.isnot(None)
|
|
).count()
|
|
except Exception:
|
|
pass
|
|
|
|
db_metrics['status'] = 'ok'
|
|
except Exception as e:
|
|
db_metrics['status'] = 'error'
|
|
db_metrics['error'] = str(e)[:100]
|
|
|
|
# ===== APPLICATION METRICS =====
|
|
app_metrics = {}
|
|
|
|
# Health check - test key endpoints
|
|
from flask import current_app
|
|
try:
|
|
with current_app.test_client() as client:
|
|
endpoints_ok = 0
|
|
endpoints_total = 5
|
|
test_endpoints = ['/', '/login', '/api/companies', '/health', '/search?q=test']
|
|
for ep in test_endpoints:
|
|
try:
|
|
response = client.get(ep, follow_redirects=False)
|
|
if response.status_code in (200, 302, 304):
|
|
endpoints_ok += 1
|
|
except Exception:
|
|
pass
|
|
app_metrics['endpoints_ok'] = endpoints_ok
|
|
app_metrics['endpoints_total'] = endpoints_total
|
|
app_metrics['endpoints_percent'] = round(100 * endpoints_ok / endpoints_total, 0)
|
|
except Exception:
|
|
app_metrics['endpoints_ok'] = None
|
|
|
|
# Users statistics
|
|
app_metrics['admins'] = db.query(User).filter(User.is_admin == True).count()
|
|
app_metrics['users_with_2fa'] = db.query(User).filter(User.totp_enabled == True).count()
|
|
|
|
# Recent activity (last 24h)
|
|
yesterday = now - timedelta(days=1)
|
|
try:
|
|
app_metrics['logins_24h'] = db.query(AuditLog).filter(
|
|
AuditLog.action == 'login',
|
|
AuditLog.created_at >= yesterday
|
|
).count()
|
|
except Exception:
|
|
app_metrics['logins_24h'] = 0
|
|
|
|
# Security alerts (last 24h)
|
|
try:
|
|
app_metrics['alerts_24h'] = db.query(SecurityAlert).filter(
|
|
SecurityAlert.created_at >= yesterday
|
|
).count()
|
|
except Exception:
|
|
app_metrics['alerts_24h'] = 0
|
|
|
|
# ===== GUNICORN/PROCESS METRICS =====
|
|
process_metrics = {}
|
|
try:
|
|
result = subprocess.run(['/usr/bin/pgrep', '-f', 'gunicorn'], capture_output=True, text=True, timeout=5)
|
|
logger.info(f"pgrep gunicorn: returncode={result.returncode}, stdout={result.stdout.strip()}")
|
|
if result.returncode == 0:
|
|
pids = result.stdout.strip().split('\n')
|
|
process_metrics['gunicorn_workers'] = len(pids) - 1 # -1 for master
|
|
process_metrics['gunicorn_status'] = 'running'
|
|
else:
|
|
process_metrics['gunicorn_status'] = 'not found'
|
|
except Exception as e:
|
|
logger.error(f"pgrep gunicorn exception: {e}")
|
|
process_metrics['gunicorn_status'] = 'unknown'
|
|
|
|
# ===== TECHNOLOGY STACK =====
|
|
import flask
|
|
import sqlalchemy
|
|
# Technology stack - ONLY VERIFIED VERSIONS (checked via SSH 2026-01-14)
|
|
# Dynamic versions are fetched at runtime, static ones were verified manually
|
|
technology_stack = {
|
|
'programming': [
|
|
{'name': 'Python', 'version': platform.python_version(), 'icon': '🐍', 'category': 'Backend'},
|
|
{'name': 'Flask', 'version': flask.__version__, 'icon': '🌶️', 'category': 'Web Framework'},
|
|
{'name': 'SQLAlchemy', 'version': sqlalchemy.__version__, 'icon': '🗃️', 'category': 'ORM'},
|
|
{'name': 'Jinja2', 'version': '3.1.6', 'icon': '📄', 'category': 'Templating'},
|
|
{'name': 'Werkzeug', 'version': '3.1.3', 'icon': '🔧', 'category': 'WSGI Toolkit'},
|
|
],
|
|
'databases': [
|
|
{'name': 'PostgreSQL', 'version': db_metrics.get('version', 'N/A'), 'icon': '🐘', 'category': 'Primary DB'},
|
|
],
|
|
'ai': [
|
|
{'name': 'Google Gemini', 'version': '3 Flash', 'icon': '🤖', 'category': 'AI Chat'},
|
|
{'name': 'Brave Search API', 'version': 'v1', 'icon': '🔍', 'category': 'News Search'},
|
|
{'name': 'Google PageSpeed', 'version': 'v5', 'icon': '⚡', 'category': 'SEO Audit'},
|
|
],
|
|
'infrastructure': [
|
|
{'name': 'Proxmox VE', 'version': '9.1.1', 'icon': '🖥️', 'category': 'Wirtualizacja'},
|
|
{'name': 'Ubuntu Server', 'version': '24.04.3 LTS', 'icon': '🐧', 'category': 'System OS'},
|
|
{'name': 'Nginx', 'version': '1.24.0', 'icon': '🔧', 'category': 'Web Server'},
|
|
],
|
|
'network': [
|
|
{'name': 'Fortigate 500D', 'version': None, 'icon': '🛡️', 'category': 'Firewall/VPN'},
|
|
{'name': 'Nginx Proxy Manager', 'version': '2.12.6', 'icon': '🔀', 'category': 'Reverse Proxy'},
|
|
{'name': 'Docker', 'version': '28.2.2', 'icon': '🐳', 'category': 'Containers'},
|
|
{'name': "Let's Encrypt", 'version': 'ACME v2', 'icon': '🔒', 'category': 'SSL/TLS'},
|
|
],
|
|
'security': [
|
|
{'name': 'Flask-Login', 'version': '0.6.3', 'icon': '🔐', 'category': 'Autentykacja'},
|
|
{'name': 'Flask-WTF', 'version': '1.2.2', 'icon': '🛡️', 'category': 'CSRF Protection'},
|
|
{'name': 'Flask-Limiter', 'version': '4.0.0', 'icon': '⏱️', 'category': 'Rate Limiting'},
|
|
{'name': 'geoip2', 'version': '5.2.0', 'icon': '🌍', 'category': 'GeoIP Blocking'},
|
|
{'name': 'PyOTP', 'version': '2.9.0', 'icon': '📱', 'category': '2FA/TOTP'},
|
|
],
|
|
'devops': [
|
|
{'name': 'Git', 'version': '2.43.0', 'icon': '📦', 'category': 'Version Control'},
|
|
{'name': 'Gitea', 'version': '1.22.6', 'icon': '🍵', 'category': 'Git Server'},
|
|
{'name': 'systemd', 'version': '255', 'icon': '⚙️', 'category': 'Service Manager'},
|
|
],
|
|
'servers': [
|
|
{'name': 'NORDABIZ-01', 'ip': '10.22.68.249', 'icon': '🖥️', 'role': 'App Server (VM 249)'},
|
|
{'name': 'R11-REVPROXY-01', 'ip': '10.22.68.250', 'icon': '🔀', 'role': 'Reverse Proxy (VM 119)'},
|
|
{'name': 'R11-DNS-01', 'ip': '10.22.68.171', 'icon': '📡', 'role': 'DNS Server (VM 122)'},
|
|
{'name': 'R11-GIT-INPI', 'ip': '10.22.68.180', 'icon': '📦', 'role': 'Git Server (VM 180)'},
|
|
],
|
|
}
|
|
|
|
return render_template(
|
|
'admin/status_dashboard.html',
|
|
system_metrics=system_metrics,
|
|
db_metrics=db_metrics,
|
|
app_metrics=app_metrics,
|
|
process_metrics=process_metrics,
|
|
technology_stack=technology_stack,
|
|
generated_at=now
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@bp.route('/api/status')
|
|
@login_required
|
|
def api_admin_status():
|
|
"""API endpoint for status dashboard auto-refresh"""
|
|
if not current_user.is_admin:
|
|
return jsonify({'success': False, 'error': 'Not authorized'}), 403
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
now = datetime.now()
|
|
data = {'timestamp': now.isoformat()}
|
|
|
|
# System metrics
|
|
system = {}
|
|
try:
|
|
# CPU (Linux)
|
|
result = subprocess.run(['grep', 'cpu ', '/proc/stat'], capture_output=True, text=True, timeout=2)
|
|
if result.returncode == 0:
|
|
parts = result.stdout.split()
|
|
idle = int(parts[4])
|
|
total = sum(int(x) for x in parts[1:])
|
|
system['cpu_percent'] = round(100 * (1 - idle / total), 1)
|
|
except Exception:
|
|
system['cpu_percent'] = None
|
|
|
|
try:
|
|
# RAM (Linux)
|
|
result = subprocess.run(['free', '-b'], capture_output=True, text=True, timeout=2)
|
|
if result.returncode == 0:
|
|
lines = result.stdout.strip().split('\n')
|
|
mem_line = lines[1].split()
|
|
total = int(mem_line[1])
|
|
used = int(mem_line[2])
|
|
system['ram_percent'] = round(100 * used / total, 1)
|
|
except Exception:
|
|
system['ram_percent'] = None
|
|
|
|
try:
|
|
# Disk
|
|
result = subprocess.run(['df', '-h', '/'], capture_output=True, text=True, timeout=2)
|
|
if result.returncode == 0:
|
|
lines = result.stdout.strip().split('\n')
|
|
parts = lines[1].split()
|
|
system['disk_percent'] = int(parts[4].replace('%', ''))
|
|
except Exception:
|
|
system['disk_percent'] = None
|
|
|
|
data['system'] = system
|
|
|
|
# Database metrics
|
|
db_data = {}
|
|
try:
|
|
db_data['active_connections'] = db.execute(text("SELECT count(*) FROM pg_stat_activity WHERE state = 'active'")).scalar()
|
|
db_data['status'] = 'ok'
|
|
except Exception as e:
|
|
db_data['status'] = 'error'
|
|
db_data['error'] = str(e)[:50]
|
|
|
|
data['database'] = db_data
|
|
|
|
# App metrics
|
|
yesterday = now - timedelta(days=1)
|
|
app_data = {
|
|
'alerts_24h': db.query(SecurityAlert).filter(SecurityAlert.created_at >= yesterday).count()
|
|
}
|
|
data['app'] = app_data
|
|
|
|
return jsonify(data)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ============================================================
|
|
# HEALTH CHECK DASHBOARD
|
|
# ============================================================
|
|
|
|
@bp.route('/health')
|
|
@login_required
|
|
def admin_health():
|
|
"""
|
|
Graphical health check dashboard.
|
|
Shows status of all critical endpoints with visual indicators.
|
|
"""
|
|
if not current_user.is_admin:
|
|
flash('Brak uprawnień do tej strony.', 'error')
|
|
return redirect(url_for('public.dashboard'))
|
|
|
|
from flask import current_app
|
|
|
|
results = []
|
|
categories = {
|
|
'public': {'name': 'Strony publiczne', 'icon': '🌐', 'endpoints': []},
|
|
'auth': {'name': 'Autentykacja', 'icon': '🔐', 'endpoints': []},
|
|
'api': {'name': 'API', 'icon': '⚡', 'endpoints': []},
|
|
'admin': {'name': 'Panel admina', 'icon': '👨💼', 'endpoints': []},
|
|
'company': {'name': 'Profile firm', 'icon': '🏢', 'endpoints': []},
|
|
}
|
|
|
|
# Endpoints to check (path, name, category)
|
|
endpoints = [
|
|
('/', 'Strona główna', 'public'),
|
|
('/release-notes', 'Historia zmian', 'public'),
|
|
('/search?q=test', 'Wyszukiwarka', 'public'),
|
|
('/chat', 'NordaGPT Chat', 'public'),
|
|
('/raporty/', 'Raporty', 'public'),
|
|
('/login', 'Logowanie', 'auth'),
|
|
('/register', 'Rejestracja', 'auth'),
|
|
('/api/companies', 'Lista firm', 'api'),
|
|
('/health', 'Health check', 'api'),
|
|
('/admin/security', 'Bezpieczeństwo', 'admin'),
|
|
('/admin/seo', 'SEO Audit', 'admin'),
|
|
('/admin/social-media', 'Social Media', 'admin'),
|
|
('/admin/analytics', 'Analityka', 'admin'),
|
|
('/admin/forum', 'Forum', 'admin'),
|
|
('/admin/kalendarz', 'Kalendarz', 'admin'),
|
|
('/admin/status', 'Status systemu', 'admin'),
|
|
('/admin/fees', 'Składki (FIS)', 'admin'),
|
|
('/admin/zopk/news', 'ZOPK News', 'admin'),
|
|
('/admin/recommendations', 'Rekomendacje', 'admin'),
|
|
]
|
|
|
|
# Add company profiles: INPI, Waterm (fixed) + 3 random
|
|
db = SessionLocal()
|
|
try:
|
|
import random as rnd
|
|
|
|
# Fixed companies to always check
|
|
fixed_companies = db.query(Company).filter(
|
|
Company.name.ilike('%INPI%') | Company.name.ilike('%Waterm%')
|
|
).all()
|
|
|
|
for company in fixed_companies:
|
|
endpoints.append((f'/company/{company.slug}', company.name[:30], 'company'))
|
|
|
|
# 3 random companies (excluding fixed ones)
|
|
fixed_ids = [c.id for c in fixed_companies]
|
|
all_other = db.query(Company).filter(~Company.id.in_(fixed_ids)).all()
|
|
random_companies = rnd.sample(all_other, min(3, len(all_other)))
|
|
|
|
for company in random_companies:
|
|
endpoints.append((f'/company/{company.slug}', f'{company.name[:25]}...', 'company'))
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
# Test each endpoint
|
|
with current_app.test_client() as client:
|
|
for path, name, category in endpoints:
|
|
start_time = datetime.now()
|
|
try:
|
|
response = client.get(path, follow_redirects=False)
|
|
status_code = response.status_code
|
|
response_time = (datetime.now() - start_time).total_seconds() * 1000 # ms
|
|
|
|
# Determine status
|
|
# 429 = rate limited (endpoint works, just protected)
|
|
# 403 = forbidden (endpoint works, requires auth)
|
|
if status_code in (200, 302, 304, 429):
|
|
status = 'ok'
|
|
elif status_code == 404:
|
|
status = 'not_found'
|
|
elif status_code >= 500:
|
|
status = 'error'
|
|
else:
|
|
status = 'warning'
|
|
|
|
result = {
|
|
'path': path,
|
|
'name': name,
|
|
'status_code': status_code,
|
|
'status': status,
|
|
'response_time': round(response_time, 1),
|
|
'error': None
|
|
}
|
|
|
|
except Exception as e:
|
|
result = {
|
|
'path': path,
|
|
'name': name,
|
|
'status_code': 500,
|
|
'status': 'error',
|
|
'response_time': None,
|
|
'error': str(e)[:100]
|
|
}
|
|
|
|
categories[category]['endpoints'].append(result)
|
|
results.append(result)
|
|
|
|
# Summary stats
|
|
total = len(results)
|
|
ok_count = sum(1 for r in results if r['status'] == 'ok')
|
|
warning_count = sum(1 for r in results if r['status'] == 'warning')
|
|
error_count = sum(1 for r in results if r['status'] in ('error', 'not_found'))
|
|
avg_response_time = sum(r['response_time'] for r in results if r['response_time']) / total if total else 0
|
|
|
|
summary = {
|
|
'total': total,
|
|
'ok': ok_count,
|
|
'warning': warning_count,
|
|
'error': error_count,
|
|
'health_percent': round(100 * ok_count / total, 1) if total else 0,
|
|
'avg_response_time': round(avg_response_time, 1),
|
|
'overall_status': 'ok' if error_count == 0 else ('degraded' if ok_count > error_count else 'critical')
|
|
}
|
|
|
|
return render_template(
|
|
'admin/health_dashboard.html',
|
|
categories=categories,
|
|
summary=summary,
|
|
generated_at=datetime.now()
|
|
)
|
|
|
|
|
|
@bp.route('/api/health')
|
|
@login_required
|
|
def api_admin_health():
|
|
"""API endpoint for health dashboard auto-refresh"""
|
|
if not current_user.is_admin:
|
|
return jsonify({'success': False, 'error': 'Not authorized'}), 403
|
|
|
|
from flask import current_app
|
|
|
|
# Run the same checks as admin_health but return JSON
|
|
results = []
|
|
|
|
endpoints = [
|
|
('/', 'Strona główna'),
|
|
('/release-notes', 'Historia zmian'),
|
|
('/search?q=test', 'Wyszukiwarka'),
|
|
('/chat', 'NordaGPT Chat'),
|
|
('/login', 'Logowanie'),
|
|
('/api/companies', 'Lista firm'),
|
|
('/health', 'Health check'),
|
|
('/admin/security', 'Bezpieczeństwo'),
|
|
('/admin/status', 'Status systemu'),
|
|
('/admin/fees', 'Składki (FIS)'),
|
|
('/admin/zopk/news', 'ZOPK News'),
|
|
]
|
|
|
|
with current_app.test_client() as client:
|
|
for path, name in endpoints:
|
|
try:
|
|
response = client.get(path, follow_redirects=False)
|
|
status_code = response.status_code
|
|
ok = status_code in (200, 302, 304, 429) # 429 = rate limited, endpoint works
|
|
results.append({'path': path, 'name': name, 'status': status_code, 'ok': ok})
|
|
except Exception as e:
|
|
results.append({'path': path, 'name': name, 'status': 500, 'ok': False, 'error': str(e)[:50]})
|
|
|
|
ok_count = sum(1 for r in results if r['ok'])
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'timestamp': datetime.now().isoformat(),
|
|
'results': results,
|
|
'summary': {
|
|
'total': len(results),
|
|
'ok': ok_count,
|
|
'failed': len(results) - ok_count,
|
|
'health_percent': round(100 * ok_count / len(results), 1)
|
|
}
|
|
})
|
|
|
|
|
|
# ============================================================
|
|
# DEBUG PANEL
|
|
# ============================================================
|
|
|
|
@bp.route('/debug')
|
|
@login_required
|
|
def debug_panel():
|
|
"""Real-time debug panel for monitoring app activity"""
|
|
if not current_user.is_admin:
|
|
flash('Brak uprawnień do tej strony.', 'error')
|
|
return redirect(url_for('public.dashboard'))
|
|
return render_template('admin/debug.html')
|
|
|
|
|
|
@bp.route('/api/logs')
|
|
@login_required
|
|
def api_get_logs():
|
|
"""API: Get recent logs"""
|
|
if not current_user.is_admin:
|
|
return jsonify({'success': False, 'error': 'Not authorized'}), 403
|
|
|
|
# Import debug_handler from main app
|
|
from app import debug_handler
|
|
|
|
# Get optional filters
|
|
level = request.args.get('level', '') # DEBUG, INFO, WARNING, ERROR
|
|
since = request.args.get('since', '') # ISO timestamp
|
|
limit = min(int(request.args.get('limit', 100)), 500)
|
|
|
|
logs = list(debug_handler.logs)
|
|
|
|
# Filter by level
|
|
if level:
|
|
logs = [l for l in logs if l['level'] == level.upper()]
|
|
|
|
# Filter by timestamp
|
|
if since:
|
|
logs = [l for l in logs if l['timestamp'] > since]
|
|
|
|
# Return most recent
|
|
logs = logs[-limit:]
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'logs': logs,
|
|
'total': len(debug_handler.logs)
|
|
})
|
|
|
|
|
|
@bp.route('/api/logs/stream')
|
|
@login_required
|
|
def api_logs_stream():
|
|
"""SSE endpoint for real-time log streaming"""
|
|
if not current_user.is_admin:
|
|
return jsonify({'success': False, 'error': 'Not authorized'}), 403
|
|
|
|
from app import debug_handler
|
|
import time
|
|
|
|
def generate():
|
|
last_count = 0
|
|
while True:
|
|
current_count = len(debug_handler.logs)
|
|
if current_count > last_count:
|
|
# Send new logs
|
|
new_logs = list(debug_handler.logs)[last_count:]
|
|
for log in new_logs:
|
|
yield f"data: {json.dumps(log)}\n\n"
|
|
last_count = current_count
|
|
time.sleep(0.5)
|
|
|
|
return Response(generate(), mimetype='text/event-stream')
|
|
|
|
|
|
@bp.route('/api/logs/clear', methods=['POST'])
|
|
@login_required
|
|
def api_clear_logs():
|
|
"""API: Clear log buffer"""
|
|
if not current_user.is_admin:
|
|
return jsonify({'success': False, 'error': 'Not authorized'}), 403
|
|
|
|
from app import debug_handler
|
|
debug_handler.logs.clear()
|
|
logger.info("Log buffer cleared by admin")
|
|
return jsonify({'success': True})
|
|
|
|
|
|
@bp.route('/api/test-log', methods=['POST'])
|
|
@login_required
|
|
def api_test_log():
|
|
"""API: Generate test log entries"""
|
|
if not current_user.is_admin:
|
|
return jsonify({'success': False, 'error': 'Not authorized'}), 403
|
|
|
|
logger.debug("Test DEBUG message")
|
|
logger.info("Test INFO message")
|
|
logger.warning("Test WARNING message")
|
|
logger.error("Test ERROR message")
|
|
return jsonify({'success': True, 'message': 'Test logs generated'})
|