"""Settings API endpoints.""" import json import os from pathlib import Path from quart import Blueprint, request, jsonify, current_app from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.setting import Setting, DEFAULT_SETTINGS, EXTENSION_API_KEY_SETTING, generate_api_key from app.config import get_settings bp = Blueprint("settings", __name__) def get_storage_stats(): """Calculate storage statistics for the downloads directory.""" config = get_settings() downloads_path = Path(config.downloads_path) if not downloads_path.exists(): return {"total_size": 0, "total_size_formatted": "0 B", "file_count": 0} total_size = 0 file_count = 0 try: for root, dirs, files in os.walk(downloads_path): for f in files: fp = os.path.join(root, f) try: total_size += os.path.getsize(fp) file_count += 1 except OSError: pass except Exception as e: current_app.logger.error(f"Error calculating storage stats: {e}") return {"total_size": 0, "total_size_formatted": "Error", "file_count": 0} # Format size def format_size(size): for unit in ['B', 'KB', 'MB', 'GB', 'TB']: if size < 1024: return f"{size:.1f} {unit}" size /= 1024 return f"{size:.1f} PB" return { "total_size": total_size, "total_size_formatted": format_size(total_size), "file_count": file_count, } async def get_or_create_extension_api_key(session: AsyncSession) -> str: """Get the extension API key, creating one if it doesn't exist.""" result = await session.execute( select(Setting).where(Setting.key == EXTENSION_API_KEY_SETTING) ) setting = result.scalar() if setting: return setting.value # Generate new key new_key = generate_api_key() setting = Setting(key=EXTENSION_API_KEY_SETTING, value=new_key) session.add(setting) await session.commit() return new_key async def get_setting_value(session: AsyncSession, key: str, default=None): """Get a setting value from the database, falling back to defaults. Args: session: Database session key: Setting key (e.g., 'download.schedule_interval') default: Fallback if not in database or DEFAULT_SETTINGS Returns: The setting value """ result = await session.execute( select(Setting).where(Setting.key == key) ) setting = result.scalar() if setting: return setting.value # Fall back to DEFAULT_SETTINGS, then to provided default return DEFAULT_SETTINGS.get(key, default) @bp.route("", methods=["GET"]) async def get_all_settings(): """Get all application settings.""" async with current_app.db_engine.connect() as conn: session = AsyncSession(bind=conn) result = await session.execute(select(Setting)) settings = result.scalars().all() # Build settings dict with defaults settings_dict = dict(DEFAULT_SETTINGS) for s in settings: settings_dict[s.key] = s.value # Ensure extension API key exists api_key = await get_or_create_extension_api_key(session) settings_dict[EXTENSION_API_KEY_SETTING] = api_key # Include storage statistics storage_stats = get_storage_stats() return jsonify({"settings": settings_dict, "storage_stats": storage_stats}) @bp.route("", methods=["PATCH"]) async def update_settings(): """Update application settings.""" data = await request.get_json() if not data: return jsonify({"error": "No settings provided"}), 400 async with current_app.db_engine.connect() as conn: session = AsyncSession(bind=conn) for key, value in data.items(): # Check if setting exists result = await session.execute(select(Setting).where(Setting.key == key)) setting = result.scalar() if setting: setting.value = value else: setting = Setting(key=key, value=value) session.add(setting) await session.commit() # Return updated settings result = await session.execute(select(Setting)) settings = result.scalars().all() settings_dict = dict(DEFAULT_SETTINGS) for s in settings: settings_dict[s.key] = s.value current_app.logger.info(f"Updated settings: {list(data.keys())}") return jsonify({"settings": settings_dict}) @bp.route("/api-key", methods=["GET"]) async def get_extension_api_key(): """Get the extension API key.""" async with current_app.db_engine.connect() as conn: session = AsyncSession(bind=conn) api_key = await get_or_create_extension_api_key(session) return jsonify({"api_key": api_key}) @bp.route("/api-key/regenerate", methods=["POST"]) async def regenerate_extension_api_key(): """Regenerate the extension API key.""" async with current_app.db_engine.connect() as conn: session = AsyncSession(bind=conn) # Find existing key result = await session.execute( select(Setting).where(Setting.key == EXTENSION_API_KEY_SETTING) ) setting = result.scalar() # Generate new key new_key = generate_api_key() if setting: setting.value = new_key else: setting = Setting(key=EXTENSION_API_KEY_SETTING, value=new_key) session.add(setting) await session.commit() current_app.logger.info("Extension API key regenerated") return jsonify({"api_key": new_key, "message": "API key regenerated"}) @bp.route("/gallery-dl", methods=["GET"]) async def get_gallery_dl_config(): """Get gallery-dl configuration.""" config = get_settings() config_path = Path(config.config_path) / "gallery-dl.conf" if not config_path.exists(): return jsonify({"config": {}, "message": "No configuration file found"}) try: with open(config_path) as f: gallery_dl_config = json.load(f) return jsonify({"config": gallery_dl_config}) except json.JSONDecodeError as e: return jsonify({"error": f"Invalid JSON in config file: {e}"}), 500 @bp.route("/gallery-dl", methods=["PUT"]) async def update_gallery_dl_config(): """Update gallery-dl configuration.""" data = await request.get_json() if not data.get("config"): return jsonify({"error": "Config object is required"}), 400 config = get_settings() config_path = Path(config.config_path) / "gallery-dl.conf" # Ensure config directory exists config_path.parent.mkdir(parents=True, exist_ok=True) try: with open(config_path, "w") as f: json.dump(data["config"], f, indent=4) current_app.logger.info("Updated gallery-dl configuration") return jsonify({"message": "Configuration updated", "config": data["config"]}) except Exception as e: current_app.logger.error(f"Failed to write gallery-dl config: {e}") return jsonify({"error": f"Failed to write config: {e}"}), 500 @bp.route("/logs", methods=["GET"]) async def get_logs(): """Get recent download logs from download metadata. Returns the stdout/stderr from recent downloads for debugging. """ from app.models.download import Download from sqlalchemy import desc limit = min(int(request.args.get("limit", 50)), 200) async with current_app.db_engine.connect() as conn: session = AsyncSession(bind=conn) # Get recent downloads with output logs query = select(Download).order_by(desc(Download.created_at)).limit(limit) result = await session.execute(query) downloads = result.scalars().all() logs = [] for dl in downloads: metadata = dl.metadata_ or {} if metadata.get("stdout") or metadata.get("stderr") or dl.error_message: logs.append({ "id": dl.id, "url": dl.url, "status": dl.status, "created_at": dl.created_at.isoformat() + "Z" if dl.created_at else None, "stdout": metadata.get("stdout", ""), "stderr": metadata.get("stderr", ""), "error_message": dl.error_message, }) return jsonify({"logs": logs, "count": len(logs)})