Vito Sansevero
4548beb723
Phase 4 structural refactoring: - Split monolithic py/api.py (5.3k lines) into domain mixins: prompts.py, images.py, admin.py, logging_routes.py, autotag_routes.py - Extract inline JS from admin.html into web/js/admin.js - Extract inline JS from gallery.html into web/js/gallery.js - Add gzip compression middleware to API responses (Phase 5.4) - Standardize API error envelope with success: false (Phase 5.3) - Add filmstrip image enrichment to prompt list responses (Phase 5.2)
225 lines
7.4 KiB
Python
225 lines
7.4 KiB
Python
"""Logging API routes for PromptManager."""
|
|
|
|
import os
|
|
|
|
from aiohttp import web
|
|
|
|
|
|
class LoggingRoutesMixin:
|
|
"""Mixin providing logging-related API endpoints."""
|
|
|
|
def _register_logging_routes(self, routes):
|
|
@routes.get("/prompt_manager/logs")
|
|
async def get_logs_route(request):
|
|
return await self.get_logs(request)
|
|
|
|
@routes.get("/prompt_manager/logs/files")
|
|
async def get_log_files_route(request):
|
|
return await self.get_log_files(request)
|
|
|
|
@routes.get("/prompt_manager/logs/download/{filename}")
|
|
async def download_log_route(request):
|
|
return await self.download_log_file(request)
|
|
|
|
@routes.post("/prompt_manager/logs/truncate")
|
|
async def truncate_logs_route(request):
|
|
return await self.truncate_logs(request)
|
|
|
|
@routes.get("/prompt_manager/logs/config")
|
|
async def get_log_config_route(request):
|
|
return await self.get_log_config(request)
|
|
|
|
@routes.post("/prompt_manager/logs/config")
|
|
async def update_log_config_route(request):
|
|
return await self.update_log_config(request)
|
|
|
|
@routes.get("/prompt_manager/logs/stats")
|
|
async def get_log_stats_route(request):
|
|
return await self.get_log_stats(request)
|
|
|
|
def _get_logger_manager(self):
|
|
"""Get the logger manager instance."""
|
|
try:
|
|
from ...utils.logging_config import get_logger_manager
|
|
except ImportError:
|
|
import sys
|
|
current_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
sys.path.insert(0, current_dir)
|
|
from utils.logging_config import get_logger_manager
|
|
return get_logger_manager()
|
|
|
|
async def get_logs(self, request):
|
|
"""Get recent log entries."""
|
|
try:
|
|
logger_manager = self._get_logger_manager()
|
|
|
|
limit = int(request.query.get('limit', 100))
|
|
level = request.query.get('level', None)
|
|
|
|
if limit > 1000:
|
|
limit = 1000
|
|
elif limit < 1:
|
|
limit = 1
|
|
|
|
logs = logger_manager.get_recent_logs(limit=limit, level=level)
|
|
|
|
return web.json_response({
|
|
'success': True,
|
|
'logs': logs,
|
|
'count': len(logs),
|
|
'level_filter': level,
|
|
'limit': limit
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Get logs error: {e}")
|
|
return web.json_response({
|
|
'success': False,
|
|
'error': str(e),
|
|
'logs': []
|
|
}, status=500)
|
|
|
|
async def get_log_files(self, request):
|
|
"""Get information about available log files."""
|
|
try:
|
|
logger_manager = self._get_logger_manager()
|
|
log_files = logger_manager.get_log_files()
|
|
|
|
return web.json_response({
|
|
'success': True,
|
|
'files': log_files,
|
|
'count': len(log_files)
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Get log files error: {e}")
|
|
return web.json_response({
|
|
'success': False,
|
|
'error': str(e),
|
|
'files': []
|
|
}, status=500)
|
|
|
|
async def download_log_file(self, request):
|
|
"""Download a specific log file."""
|
|
try:
|
|
filename = request.match_info['filename']
|
|
logger_manager = self._get_logger_manager()
|
|
|
|
if not filename or '..' in filename or '/' in filename or '\\' in filename:
|
|
return web.json_response({
|
|
'success': False,
|
|
'error': 'Invalid filename'
|
|
}, status=400)
|
|
|
|
log_file_path = logger_manager.log_dir / filename
|
|
|
|
if not log_file_path.exists():
|
|
return web.json_response({
|
|
'success': False,
|
|
'error': 'Log file not found'
|
|
}, status=404)
|
|
|
|
with open(log_file_path, 'rb') as f:
|
|
file_content = f.read()
|
|
|
|
return web.Response(
|
|
body=file_content,
|
|
content_type='text/plain',
|
|
headers={
|
|
'Content-Disposition': f'attachment; filename="{filename}"',
|
|
'Content-Length': str(len(file_content))
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Download log file error: {e}")
|
|
return web.json_response({
|
|
'success': False,
|
|
'error': str(e)
|
|
}, status=500)
|
|
|
|
async def truncate_logs(self, request):
|
|
"""Truncate all log files."""
|
|
try:
|
|
logger_manager = self._get_logger_manager()
|
|
results = logger_manager.truncate_logs()
|
|
|
|
return web.json_response({
|
|
'success': True,
|
|
'message': f"Truncated {len(results['truncated'])} log files",
|
|
'results': results
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Truncate logs error: {e}")
|
|
return web.json_response({
|
|
'success': False,
|
|
'error': str(e)
|
|
}, status=500)
|
|
|
|
async def get_log_config(self, request):
|
|
"""Get current logging configuration."""
|
|
try:
|
|
logger_manager = self._get_logger_manager()
|
|
config = logger_manager.get_config()
|
|
|
|
return web.json_response({
|
|
'success': True,
|
|
'config': config
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Get log config error: {e}")
|
|
return web.json_response({
|
|
'success': False,
|
|
'error': str(e)
|
|
}, status=500)
|
|
|
|
async def update_log_config(self, request):
|
|
"""Update logging configuration."""
|
|
try:
|
|
data = await request.json()
|
|
logger_manager = self._get_logger_manager()
|
|
|
|
if 'level' in data:
|
|
valid_levels = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']
|
|
if data['level'].upper() not in valid_levels:
|
|
return web.json_response({
|
|
'success': False,
|
|
'error': f'Invalid log level. Must be one of: {valid_levels}'
|
|
}, status=400)
|
|
data['level'] = data['level'].upper()
|
|
|
|
logger_manager.update_config(data)
|
|
|
|
return web.json_response({
|
|
'success': True,
|
|
'message': 'Logging configuration updated',
|
|
'config': logger_manager.get_config()
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Update log config error: {e}")
|
|
return web.json_response({
|
|
'success': False,
|
|
'error': str(e)
|
|
}, status=500)
|
|
|
|
async def get_log_stats(self, request):
|
|
"""Get logging statistics."""
|
|
try:
|
|
logger_manager = self._get_logger_manager()
|
|
stats = logger_manager.get_log_stats()
|
|
|
|
return web.json_response({
|
|
'success': True,
|
|
'stats': stats
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Get log stats error: {e}")
|
|
return web.json_response({
|
|
'success': False,
|
|
'error': str(e)
|
|
}, status=500)
|