""" Settings and User Management API Router """ from fastapi import APIRouter, BackgroundTasks, HTTPException, Request, Depends, Query from typing import List, Optional, Dict, Literal from pydantic import BaseModel, SecretStr from datetime import datetime, timezone from app.core.database import execute_query, execute_query_single from app.core.config import settings from app.core.auth_dependencies import require_any_permission, require_superadmin from app.core.auth_service import AuthService from app.core.crypto import decrypt_secret, encrypt_secret import argparse import asyncio import httpx import imaplib import time import logging import json import os import re import threading from pathlib import Path from urllib.parse import quote logger = logging.getLogger(__name__) router = APIRouter() _sag_test_lock = threading.Lock() _sag_test_report_dir = Path(__file__).resolve().parents[3] / "data" / "test-results" DEFAULT_EMAIL_SIGNATURE_TEMPLATE = ( "{full_name}\n" "{title}\n" "{company_name}\n" "Telefon: {company_phone}\n" "Email: {email}\n" "Web: {company_website}\n" "Adresse: {company_address}\n" "BMCid: {bmc_id_tag}" ) # Pydantic Models class Setting(BaseModel): id: int key: str value: Optional[str] category: str description: Optional[str] value_type: str is_public: bool class SettingUpdate(BaseModel): value: str class SettingCreate(BaseModel): key: str value: Optional[str] = None category: Optional[str] = "general" description: Optional[str] = None value_type: Optional[str] = "string" is_public: Optional[bool] = False class MattermostTestRequest(BaseModel): message: Optional[str] = "Test fra BMC Hub" class SagTestRunRequest(BaseModel): customer_id: Optional[int] = None class AIBenchmarkRunRequest(BaseModel): models: List[str] test_keys: Optional[List[str]] = None alternative_url: Optional[str] = None alternative_provider: Literal["ollama", "openai"] = "openai" alternative_model: Optional[str] = None alternative_api_key: Optional[SecretStr] = None class EmailConnectionUpdate(BaseModel): provider: Literal["microsoft365_graph", "imap_smtp"] is_enabled: bool = False mailbox_address: str sender_name: str = "BMC Hub" imap_host: Optional[str] = "" imap_port: int = 993 imap_folder: str = "INBOX" imap_use_ssl: bool = True smtp_host: Optional[str] = "" smtp_port: int = 587 smtp_use_tls: bool = True graph_tenant_id: Optional[str] = "" graph_client_id: Optional[str] = "" imap_password: Optional[str] = None smtp_password: Optional[str] = None graph_client_secret: Optional[str] = None def _email_connection_public(row: Optional[Dict]) -> Dict: """Return connection metadata only; encrypted credentials never leave the server.""" if not row: return {"configured": False, "source": "environment"} encrypted = str(row.get("secrets_encrypted") or "").strip() return { "configured": True, "source": "settings", "provider": row["provider"], "is_enabled": bool(row["is_enabled"]), "mailbox_address": row["mailbox_address"], "sender_name": row.get("sender_name") or "BMC Hub", "imap_host": row.get("imap_host") or "", "imap_port": row.get("imap_port") or 993, "imap_folder": row.get("imap_folder") or "INBOX", "imap_use_ssl": bool(row.get("imap_use_ssl")), "smtp_host": row.get("smtp_host") or "", "smtp_port": row.get("smtp_port") or 587, "smtp_use_tls": bool(row.get("smtp_use_tls")), "graph_tenant_id": row.get("graph_tenant_id") or "", "graph_client_id": row.get("graph_client_id") or "", "has_saved_credentials": bool(encrypted), "inbound_processing_enabled": bool(settings.EMAIL_TO_TICKET_ENABLED), "updated_at": row.get("updated_at"), } MATTERMOST_SETTING_DEFAULTS = ( ("mattermost_reminders_enabled", "false", "Send reminders to Mattermost", "boolean"), ("mattermost_webhook_url", "", "Mattermost incoming webhook URL", "string"), ("mattermost_channel", "", "Optional Mattermost channel override", "string"), ) MASKED_SECRET = "********" def _mask_setting(row: Dict) -> Dict: item = dict(row) if item.get("key") == "mattermost_webhook_url": item["value"] = MASKED_SECRET if str(item.get("value") or "").strip() else "" return item class User(BaseModel): id: int username: str email: Optional[str] full_name: Optional[str] is_active: bool last_login: Optional[datetime] created_at: datetime class UserCreate(BaseModel): username: str email: str password: str full_name: Optional[str] = None class UserUpdate(BaseModel): email: Optional[str] = None full_name: Optional[str] = None is_active: Optional[bool] = None def _load_sag_test_reports(limit: int = 20) -> List[Dict]: if not _sag_test_report_dir.exists(): return [] reports = [] for path in sorted(_sag_test_report_dir.glob("sag-e2e-*.json"), reverse=True)[:limit]: try: payload = json.loads(path.read_text(encoding="utf-8")) payload["report_file"] = path.name reports.append(payload) except (OSError, json.JSONDecodeError): logger.warning("Kunne ikke læse Sag-testrapport: %s", path) return reports def _execute_sag_test(current_user: Dict, customer_id: Optional[int]) -> Dict: from app.modules.sag.scripts.sag_module_e2e import Result, SagE2E _sag_test_report_dir.mkdir(parents=True, exist_ok=True) report_path = _sag_test_report_dir / f"sag-e2e-{datetime.now():%Y%m%d-%H%M%S}.json" token = AuthService.create_access_token( user_id=int(current_user["id"]), username=str(current_user["username"]), is_superadmin=bool(current_user.get("is_superadmin")), is_shadow_admin=bool(current_user.get("is_shadow_admin", False)), ) args = argparse.Namespace( base_url=os.getenv("SAG_TEST_BASE_URL", "http://127.0.0.1:8000"), token=token, username=None, password=None, otp=None, customer_id=customer_id, timeout=20.0, keep_data=False, json_report=str(report_path), ) suite = SagE2E(args) try: suite.run() except Exception as exc: suite.results.append(Result("Testkørsel", "FAIL", str(exc))) finally: suite.cleanup_all() suite.write_report() payload = json.loads(report_path.read_text(encoding="utf-8")) payload["report_file"] = report_path.name return payload @router.get("/settings/tests/sag", tags=["Settings Tests"]) async def get_sag_test_reports( current_user: dict = Depends(require_superadmin), ): reports = _load_sag_test_reports() return { "running": _sag_test_lock.locked(), "latest": reports[0] if reports else None, "history": reports, } @router.post("/settings/tests/sag/run", tags=["Settings Tests"]) async def run_sag_test( payload: SagTestRunRequest, current_user: dict = Depends(require_superadmin), ): if not _sag_test_lock.acquire(blocking=False): raise HTTPException(status_code=409, detail="Sag-testen kører allerede") try: return await asyncio.to_thread(_execute_sag_test, current_user, payload.customer_id) finally: _sag_test_lock.release() @router.get("/settings/ai-benchmarks/catalog", tags=["AI Benchmarks"]) async def get_ai_benchmark_catalog( current_user: dict = Depends(require_superadmin), ): from app.settings.backend.ai_benchmark import fetch_models, public_tests try: models = await fetch_models(settings.OLLAMA_ENDPOINT) model_error = None except Exception as exc: logger.warning("Could not load Ollama models for benchmark: %s", exc) models = [] model_error = str(exc) return { "active_model": settings.OLLAMA_MODEL, "endpoint": settings.OLLAMA_ENDPOINT, "models": models, "tests": public_tests(), "model_error": model_error, } @router.get("/settings/ai-benchmarks/runs", tags=["AI Benchmarks"]) async def get_ai_benchmark_runs( limit: int = 30, current_user: dict = Depends(require_superadmin), ): from app.settings.backend.ai_benchmark import list_runs return {"items": list_runs(max(1, min(limit, 100)))} @router.get("/settings/ai-benchmarks/runs/{run_id}", tags=["AI Benchmarks"]) async def get_ai_benchmark_run( run_id: int, current_user: dict = Depends(require_superadmin), ): from app.settings.backend.ai_benchmark import get_run run = get_run(run_id) if not run: raise HTTPException(status_code=404, detail="Testkørslen findes ikke") return run @router.post("/settings/ai-benchmarks/runs", tags=["AI Benchmarks"]) async def create_ai_benchmark_run( payload: AIBenchmarkRunRequest, background_tasks: BackgroundTasks, current_user: dict = Depends(require_superadmin), ): from app.settings.backend.ai_benchmark import BENCHMARK_TESTS, execute_run, fetch_models, validate_endpoint models = list(dict.fromkeys(model.strip() for model in payload.models if model.strip())) alternative = None if payload.alternative_url: endpoint = validate_endpoint(payload.alternative_url) model = str(payload.alternative_model or "").strip() if not model or len(model) > 200: raise HTTPException(400, "Angiv model-ID for den alternative AI") alternative = {"endpoint": endpoint, "provider": payload.alternative_provider, "model": model, "api_key": payload.alternative_api_key.get_secret_value() if payload.alternative_api_key else None} if len(f"{model} · {payload.alternative_provider} · {endpoint}") > 160: raise HTTPException(400, "Model-ID og AI-URL er for lange (maks. 160 tegn samlet inklusive API-type)") if not 1 <= len(models) + bool(alternative) <= 4: raise HTTPException(status_code=400, detail="Vælg mellem 1 og 4 modeller") installed = {item["name"] for item in await fetch_models(settings.OLLAMA_ENDPOINT)} if models else set() missing = [model for model in models if model not in installed] if missing: raise HTTPException(status_code=400, detail=f"Modellen er ikke installeret: {', '.join(missing)}") available_keys = {test["key"] for test in BENCHMARK_TESTS} test_keys = list(dict.fromkeys(payload.test_keys or [test["key"] for test in BENCHMARK_TESTS])) unknown = [key for key in test_keys if key not in available_keys] if unknown or not test_keys: raise HTTPException(status_code=400, detail="Vælg mindst én gyldig test") running = execute_query_single( "SELECT id FROM ai_benchmark_runs WHERE status IN ('queued', 'running') ORDER BY id DESC LIMIT 1" ) if running: raise HTTPException(status_code=409, detail=f"Testkørsel #{running['id']} kører allerede") user_id = current_user.get("id") or current_user.get("user_id") run_models = models + ([f"{alternative['model']} · {alternative['provider']} · {alternative['endpoint']}"] if alternative else []) rows = execute_query( """INSERT INTO ai_benchmark_runs (status, models, test_keys, total_cases, completed_cases, created_by) VALUES ('queued', %s::jsonb, %s::jsonb, %s, 0, %s) RETURNING *""", (json.dumps(run_models), json.dumps(test_keys), len(run_models) * len(test_keys), user_id), ) run = rows[0] background_tasks.add_task(execute_run, run["id"], settings.OLLAMA_ENDPOINT, models, test_keys, alternative) return run # Settings Endpoints @router.get("/settings/email-connection", tags=["Settings"]) async def get_email_connection(current_user: dict = Depends(require_superadmin)): """Read mailbox setup metadata without exposing any password or client secret.""" row = execute_query_single( "SELECT * FROM email_connection_configurations ORDER BY id DESC LIMIT 1" ) if row: return _email_connection_public(row) return { "configured": False, "source": "environment", "provider": "microsoft365_graph" if settings.USE_GRAPH_API else "imap_smtp", "is_enabled": bool( settings.GRAPH_USER_EMAIL if settings.USE_GRAPH_API else settings.IMAP_USERNAME ), "mailbox_address": settings.GRAPH_USER_EMAIL if settings.USE_GRAPH_API else settings.IMAP_USERNAME, "sender_name": settings.EMAIL_SMTP_FROM_NAME or "BMC Hub", "has_saved_credentials": bool( settings.GRAPH_CLIENT_SECRET if settings.USE_GRAPH_API else settings.IMAP_PASSWORD ), "inbound_processing_enabled": bool(settings.EMAIL_TO_TICKET_ENABLED), } @router.put("/settings/email-connection", tags=["Settings"]) async def save_email_connection( payload: EmailConnectionUpdate, current_user: dict = Depends(require_superadmin), ): """Save one encrypted mailbox configuration used for both inbound and outbound mail.""" mailbox = payload.mailbox_address.strip().lower() if not re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", mailbox): raise HTTPException(status_code=400, detail="Indtast en gyldig postkasseadresse") if not (1 <= payload.imap_port <= 65535 and 1 <= payload.smtp_port <= 65535): raise HTTPException(status_code=400, detail="Port skal være mellem 1 og 65535") existing = execute_query_single( "SELECT * FROM email_connection_configurations ORDER BY id DESC LIMIT 1" ) prior_secrets = (existing or {}).get("secrets_encrypted") saved_secrets = {} if prior_secrets: decoded = decrypt_secret(str(prior_secrets)) if decoded: try: saved_secrets = json.loads(decoded) except (TypeError, json.JSONDecodeError): raise HTTPException(status_code=500, detail="De gemte e-mailadgangsoplysninger kan ikke læses. Indtast dem igen.") supplied_secrets = { key: value.strip() for key, value in { "imap_password": payload.imap_password, "smtp_password": payload.smtp_password, "graph_client_secret": payload.graph_client_secret, }.items() if value is not None and value.strip() } saved_secrets.update(supplied_secrets) secrets_encrypted = prior_secrets if supplied_secrets or (prior_secrets and not saved_secrets): try: secrets_encrypted = encrypt_secret(json.dumps(saved_secrets)) except ValueError: raise HTTPException( status_code=503, detail="Sikker lagring er ikke klar. Sæt NEXTCLOUD_ENCRYPTION_KEY på serveren før adgangsoplysninger gemmes.", ) if payload.provider == "microsoft365_graph": if payload.is_enabled and not (payload.graph_tenant_id.strip() and payload.graph_client_id.strip() and saved_secrets.get("graph_client_secret")): raise HTTPException(status_code=400, detail="Microsoft 365 kræver Tenant ID, Application ID og client secret før aktivering") elif payload.is_enabled and not ( payload.imap_host.strip() and payload.smtp_host.strip() and saved_secrets.get("imap_password") and saved_secrets.get("smtp_password") ): raise HTTPException(status_code=400, detail="IMAP/SMTP kræver begge servere og adgangskoder før aktivering") user_id = current_user.get("id") or current_user.get("user_id") values = ( payload.provider, payload.is_enabled, mailbox, payload.sender_name.strip() or "BMC Hub", payload.imap_host.strip() or None, payload.imap_port, payload.imap_folder.strip() or "INBOX", payload.imap_use_ssl, payload.smtp_host.strip() or None, payload.smtp_port, payload.smtp_use_tls, payload.graph_tenant_id.strip() or None, payload.graph_client_id.strip() or None, secrets_encrypted, user_id, ) if existing: query = """ UPDATE email_connection_configurations SET provider=%s, is_enabled=%s, mailbox_address=%s, sender_name=%s, imap_host=%s, imap_port=%s, imap_folder=%s, imap_use_ssl=%s, smtp_host=%s, smtp_port=%s, smtp_use_tls=%s, graph_tenant_id=%s, graph_client_id=%s, secrets_encrypted=%s, updated_by_user_id=%s, updated_at=CURRENT_TIMESTAMP WHERE id=%s RETURNING * """ result = execute_query(query, (*values, existing["id"])) else: query = """ INSERT INTO email_connection_configurations (provider, is_enabled, mailbox_address, sender_name, imap_host, imap_port, imap_folder, imap_use_ssl, smtp_host, smtp_port, smtp_use_tls, graph_tenant_id, graph_client_id, secrets_encrypted, updated_by_user_id) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING * """ result = execute_query(query, values) if not result: raise HTTPException(status_code=500, detail="Kunne ikke gemme e-mailopsætningen") return _email_connection_public(result[0]) @router.post("/settings/email-connection/test", tags=["Settings"]) async def test_email_connection(current_user: dict = Depends(require_superadmin)): """Verify authentication and inbox access without sending or changing any email.""" from app.services.email_service import EmailService, HAS_AIOSMTPLIB, aiosmtplib configured = execute_query_single( "SELECT provider, is_enabled FROM email_connection_configurations ORDER BY id DESC LIMIT 1" ) if not configured or not configured.get("is_enabled"): raise HTTPException(status_code=400, detail="Gem og aktivér en e-mailopsætning før du tester") service = EmailService() if configured["provider"] == "microsoft365_graph": token = await service._get_graph_access_token() if not token: raise HTTPException(status_code=502, detail="Microsoft 365 kunne ikke udstede et adgangstoken. Tjek Tenant ID, Application ID og client secret.") mailbox = service.graph_config.get("user_email") try: async with httpx.AsyncClient(timeout=15) as client: response = await client.get( f"https://graph.microsoft.com/v1.0/users/{quote(mailbox)}/mailFolders/inbox?$top=1", headers={"Authorization": f"Bearer {token}"}, ) if response.status_code >= 400: raise HTTPException(status_code=502, detail="Microsoft 365-token virker, men postkassen kan ikke læses. Tjek Mail.ReadWrite og admin consent.") except httpx.HTTPError as exc: raise HTTPException(status_code=502, detail=f"Kunne ikke kontakte Microsoft Graph: {exc}") return {"ok": True, "provider": "Microsoft 365 / Graph", "message": "Token og læseadgang til Indbakke virker. Der er ikke sendt en e-mail."} def test_imap() -> None: config = service.imap_config connection = imaplib.IMAP4_SSL(config["server"], config["port"], timeout=15) if config["use_ssl"] else imaplib.IMAP4(config["server"], config["port"], timeout=15) try: connection.login(config["username"], config["password"]) status, _ = connection.select(config["folder"], readonly=True) if status != "OK": raise RuntimeError(f"kunne ikke åbne mappen {config['folder']}") finally: try: connection.logout() except Exception: pass try: await asyncio.to_thread(test_imap) except Exception as exc: raise HTTPException(status_code=502, detail=f"IMAP-test fejlede: {exc}") if not HAS_AIOSMTPLIB: raise HTTPException(status_code=503, detail="IMAP virker, men SMTP-test er ikke mulig fordi aiosmtplib mangler på serveren") try: config = service.smtp_config async with aiosmtplib.SMTP( hostname=config["host"], port=config["port"], use_tls=config["use_tls"], timeout=15 ) as smtp: await smtp.login(config["username"], config["password"]) except Exception as exc: raise HTTPException(status_code=502, detail=f"SMTP-test fejlede: {exc}") return {"ok": True, "provider": "IMAP + SMTP", "message": "IMAP-indbakke og SMTP-login virker. Der er ikke sendt en e-mail."} @router.get("/settings", response_model=List[Setting], tags=["Settings"]) async def get_settings(category: Optional[str] = None): """Get all settings or filter by category""" execute_query( """ INSERT INTO settings (key, value, category, description, value_type, is_public) VALUES (%s, %s, %s, %s, %s, %s), (%s, %s, %s, %s, %s, %s) ON CONFLICT (key) DO NOTHING """, ( "email_default_signature_template", DEFAULT_EMAIL_SIGNATURE_TEMPLATE, "email", "Standard signatur skabelon til udgående sagsmails", "text", True, "company_website", "https://bmcnetworks.dk", "company", "Firma website", "string", True, ), ) execute_query( """ INSERT INTO settings (key, value, category, description, value_type, is_public) VALUES (%s, %s, 'notifications', %s, %s, false), (%s, %s, 'notifications', %s, %s, false), (%s, %s, 'notifications', %s, %s, false) ON CONFLICT (key) DO NOTHING """, tuple(value for item in MATTERMOST_SETTING_DEFAULTS for value in item), ) query = "SELECT * FROM settings" params = [] if category: query += " WHERE category = %s" params.append(category) query += " ORDER BY category, key" result = execute_query(query, tuple(params) if params else None) return [_mask_setting(row) for row in (result or [])] @router.post("/settings", response_model=Setting, tags=["Settings"]) async def create_setting(payload: SettingCreate): """Create a setting row if it does not exist yet.""" if not payload.key or not payload.key.strip(): raise HTTPException(status_code=400, detail="Setting key is required") query = """ INSERT INTO settings (key, value, category, description, value_type, is_public) VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, category = EXCLUDED.category, description = EXCLUDED.description, value_type = EXCLUDED.value_type, is_public = EXCLUDED.is_public, updated_at = CURRENT_TIMESTAMP RETURNING * """ result = execute_query( query, ( payload.key.strip(), payload.value if payload.value is not None else "", payload.category or "general", payload.description, payload.value_type or "string", bool(payload.is_public), ), ) if not result: raise HTTPException(status_code=500, detail="Failed to create setting") return _mask_setting(result[0]) @router.get("/settings/{key}", response_model=Setting, tags=["Settings"]) async def get_setting(key: str): """Get a specific setting by key""" query = "SELECT * FROM settings WHERE key = %s" result = execute_query(query, (key,)) if not result and key in {"case_types", "case_type_module_defaults", "case_statuses", "time_multiplier_presets"}: seed_query = """ INSERT INTO settings (key, value, category, description, value_type, is_public) VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (key) DO NOTHING """ if key == "case_types": execute_query( seed_query, ( "case_types", '["ticket", "salg", "bogholderi", "admin", "projekt", "postit", "intern_ticket", "bug_report"]', "system", "Sags-typer", "json", True, ) ) if key == "case_type_module_defaults": execute_query( seed_query, ( "case_type_module_defaults", '{"ticket": ["relations", "call-history", "files", "emails", "hardware", "locations", "contacts", "customers", "wiki", "todo-steps", "time", "solution", "sales", "subscription", "reminders", "calendar"], "opgave": ["relations", "call-history", "files", "emails", "hardware", "locations", "contacts", "customers", "wiki", "todo-steps", "time", "solution", "sales", "subscription", "reminders", "calendar"], "ordre": ["relations", "call-history", "files", "emails", "hardware", "locations", "contacts", "customers", "wiki", "todo-steps", "time", "solution", "sales", "subscription", "reminders", "calendar"], "projekt": ["relations", "call-history", "files", "emails", "hardware", "locations", "contacts", "customers", "wiki", "todo-steps", "time", "solution", "sales", "subscription", "reminders", "calendar"], "service": ["relations", "call-history", "files", "emails", "hardware", "locations", "contacts", "customers", "wiki", "todo-steps", "time", "solution", "sales", "subscription", "reminders", "calendar"]}', "system", "Standard moduler pr. sagstype", "json", True, ) ) if key == "case_statuses": execute_query( seed_query, ( "case_statuses", json.dumps([ {"value": "åben", "is_closed": False}, {"value": "under behandling", "is_closed": False}, {"value": "afventer", "is_closed": False}, {"value": "løst", "is_closed": True}, {"value": "lukket", "is_closed": True}, ], ensure_ascii=False), "system", "Sagsstatus værdier og lukkede markeringer", "json", True, ) ) if key == "time_multiplier_presets": execute_query( seed_query, ( "time_multiplier_presets", '[{"label":"Haster","text":"Haster","multiplier":3},{"label":"Avanceret netværk","text":"Avanceret netværk","multiplier":2},{"label":"Haster + ava. network","text":"Haster + ava. network","multiplier":6}]', "system", "Valgbare multiplikator presets til tidsregistrering", "json", True, ), ) result = execute_query(query, (key,)) if not result and key == "email_default_signature_template": execute_query( """ INSERT INTO settings (key, value, category, description, value_type, is_public) VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (key) DO NOTHING """, ( "email_default_signature_template", DEFAULT_EMAIL_SIGNATURE_TEMPLATE, "email", "Standard signatur skabelon til udgående sagsmails", "text", True, ) ) result = execute_query(query, (key,)) if not result: raise HTTPException(status_code=404, detail="Setting not found") return _mask_setting(result[0]) @router.put("/settings/{key}", response_model=Setting, tags=["Settings"]) async def update_setting(key: str, setting: SettingUpdate): """Update a setting value""" if key == "invoice_fee_product_id" and setting.value.strip(): from app.services.invoice_fee_service import validate_fee_product validate_fee_product(setting.value.strip()) if key == "mattermost_webhook_url" and setting.value == MASKED_SECRET: current = execute_query("SELECT * FROM settings WHERE key = %s", (key,)) if not current: raise HTTPException(status_code=404, detail="Setting not found") return _mask_setting(current[0]) if key == "mattermost_channel": channel = setting.value.strip() if channel and not re.fullmatch(r"[a-z0-9_-]+", channel): raise HTTPException( status_code=400, detail="Brug Mattermost-kanalens tekniske navn uden mellemrum, fx mollypim-logs", ) query = """ UPDATE settings SET value = %s, updated_at = CURRENT_TIMESTAMP WHERE key = %s RETURNING * """ result = execute_query(query, (setting.value, key)) if not result and key == "email_default_signature_template": result = execute_query( """ INSERT INTO settings (key, value, category, description, value_type, is_public) VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = CURRENT_TIMESTAMP RETURNING * """, ( "email_default_signature_template", setting.value, "email", "Standard signatur skabelon til udgående sagsmails", "text", True, ) ) # Mission camera settings may not exist on older hubs before migration. # AnyDesk settings may not exist on older hubs — auto-create on first save _anydesk_keys = { "anydesk_api_token": ("integrations", "AnyDesk API token", "string", False), "anydesk_license_id": ("integrations", "AnyDesk license ID", "string", False), "anydesk_read_only": ("integrations", "AnyDesk read-only mode", "boolean", True), "anydesk_dry_run": ("integrations", "AnyDesk dry-run mode", "boolean", True), } if not result and key in _anydesk_keys: category, description, value_type, is_public = _anydesk_keys[key] result = execute_query( """ INSERT INTO settings (key, value, category, description, value_type, is_public) VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = CURRENT_TIMESTAMP RETURNING * """, (key, setting.value, category, description, value_type, is_public), ) _label_printer_keys = { "label_printer_enabled": ("integrations", "Enable direct label printing", "boolean", True), "label_printer_model": ("integrations", "Brother printer model for direct labels", "string", True), "label_printer_host": ("integrations", "Brother printer host/IP", "string", True), "label_printer_port": ("integrations", "Brother printer TCP port", "integer", True), "label_printer_label_size": ("integrations", "Brother label size code", "string", True), } if not result and key in _label_printer_keys: category, description, value_type, is_public = _label_printer_keys[key] result = execute_query( """ INSERT INTO settings (key, value, category, description, value_type, is_public) VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = CURRENT_TIMESTAMP RETURNING * """, (key, setting.value, category, description, value_type, is_public), ) _drift_connector_keys = { "drift_uptime_kuma_base_url": ("integrations", "Base URL for the Uptime Kuma drift connector", "string", True), "drift_uptime_kuma_api_key": ("integrations", "API token for the Uptime Kuma drift connector", "string", False), "drift_uisp_base_url": ("integrations", "Base URL for the UISP drift connector", "string", True), "drift_uisp_api_token": ("integrations", "API token for the UISP drift connector", "string", False), "drift_device_blacklist": ("integrations", "JSON array of Drift device identifiers/names to ignore during sync", "text", False), } if not result and key in _drift_connector_keys: category, description, value_type, is_public = _drift_connector_keys[key] result = execute_query( """ INSERT INTO settings (key, value, category, description, value_type, is_public) VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = CURRENT_TIMESTAMP RETURNING * """, (key, setting.value, category, description, value_type, is_public), ) if not result and key in _label_printer_keys: category, description, value_type, is_public = _label_printer_keys[key] result = execute_query( """ INSERT INTO settings (key, value, category, description, value_type, is_public) VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = CURRENT_TIMESTAMP RETURNING * """, (key, setting.value, category, description, value_type, is_public), ) # Mission camera settings may not exist on older hubs before migration. if not result and key in {"mission_camera_enabled", "mission_camera_name", "mission_camera_feed_url", "mission_camera_spotlight_seconds", "mission_access_pin"}: defaults = { "mission_camera_enabled": ("mission", "Enable one camera feed in Mission Control", "boolean", True), "mission_camera_name": ("mission", "Display name for Mission Control camera feed", "string", True), "mission_camera_feed_url": ("mission", "Feed URL for Mission Control camera", "string", True), "mission_camera_spotlight_seconds": ("mission", "Camera spotlight duration in seconds for motion events", "integer", True), "mission_access_pin": ("mission", "Access PIN for Mission Control kiosk mode", "string", False), } category, description, value_type, is_public = defaults[key] result = execute_query( """ INSERT INTO settings (key, value, category, description, value_type, is_public) VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = CURRENT_TIMESTAMP RETURNING * """, (key, setting.value, category, description, value_type, is_public), ) if not result and key == "time_multiplier_presets": result = execute_query( """ INSERT INTO settings (key, value, category, description, value_type, is_public) VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = CURRENT_TIMESTAMP RETURNING * """, ( "time_multiplier_presets", setting.value, "system", "Valgbare multiplikator presets til tidsregistrering", "json", True, ), ) if not result: raise HTTPException(status_code=404, detail="Setting not found") logger.info(f"✅ Updated setting: {key}") return _mask_setting(result[0]) @router.post("/settings/mattermost/test", tags=["Settings"]) async def test_mattermost_setting(payload: MattermostTestRequest): rows = execute_query( "SELECT key, value FROM settings WHERE key IN ('mattermost_webhook_url', 'mattermost_channel')" ) or [] values = {row["key"]: str(row.get("value") or "").strip() for row in rows} webhook_url = values.get("mattermost_webhook_url", "") if not webhook_url: raise HTTPException(status_code=400, detail="Mattermost webhook URL mangler") if not webhook_url.startswith(("https://", "http://")): raise HTTPException(status_code=400, detail="Mattermost webhook URL skal starte med http:// eller https://") message = str(payload.message or "Test fra BMC Hub").strip()[:500] mattermost_payload = {"text": f"✅ **{message}**", "username": "BMC Hub"} if values.get("mattermost_channel"): mattermost_payload["channel"] = values["mattermost_channel"] try: async with httpx.AsyncClient(timeout=10.0) as client: response = await client.post(webhook_url, json=mattermost_payload) response_text = response.text if ( response.status_code in (400, 404, 415) and ( "media type application/json" in response_text.lower() or "incoming_webhook.general.app_error" in response_text.lower() ) ): response = await client.post( webhook_url, data={"payload": json.dumps(mattermost_payload, ensure_ascii=False)}, headers={"Content-Type": "application/x-www-form-urlencoded"}, ) if response.status_code not in (200, 201, 204): response_body = response.text[:500] if "incoming_webhook.general.app_error" in response_body: raise HTTPException( status_code=400, detail=( "Webhooken blev afvist af Mattermost. Opret en ny integration under " "Mattermost → Integrations → Incoming Webhooks, og kopiér hele URL'en " "som Mattermost genererer. Et outgoing webhook-ID eller et deaktiveret " "incoming webhook kan ikke bruges." ), ) raise HTTPException( status_code=502, detail=f"Mattermost returnerede HTTP {response.status_code}: {response.text[:200]}", ) except HTTPException: raise except Exception as exc: raise HTTPException(status_code=502, detail=f"Kunne ikke kontakte Mattermost: {exc}") from exc return {"success": True, "message": "Testbesked sendt til Mattermost"} @router.get("/settings/categories/list", tags=["Settings"]) async def get_setting_categories(): """Get list of all setting categories""" query = "SELECT DISTINCT category FROM settings ORDER BY category" result = execute_query(query) return [row['category'] for row in result] if result else [] @router.get("/settings/migrations/status", tags=["Settings"]) async def get_migration_statuses_api(): """Expose migration status via API router (served under /api/v1).""" from app.settings.backend.views import migration_statuses return migration_statuses() @router.post("/settings/migrations/execute", tags=["Settings"]) async def execute_migration_api(payload: dict): """Expose migration execution via API router (served under /api/v1).""" from app.settings.backend.views import execute_migration, MigrationExecution model = MigrationExecution(**payload) return execute_migration(model) @router.post("/settings/migrations/execute-missing", tags=["Settings"]) async def execute_missing_migrations_api(): """Run schema-detected missing migrations via the API namespace.""" from app.settings.backend.views import execute_missing_migrations return execute_missing_migrations() @router.post("/settings/sync-from-env", tags=["Settings"]) async def sync_settings_from_env(): """Sync settings from .env file into database (only updates empty values)""" from app.core.config import settings as env_settings mapping = { 'vtiger_enabled': str(env_settings.VTIGER_ENABLED).lower(), 'vtiger_url': env_settings.VTIGER_URL or '', 'vtiger_username': env_settings.VTIGER_USERNAME or '', 'economic_enabled': str(env_settings.ECONOMIC_ENABLED).lower(), 'economic_app_secret': env_settings.ECONOMIC_APP_SECRET_TOKEN or '', 'economic_agreement_token': env_settings.ECONOMIC_AGREEMENT_GRANT_TOKEN or '', } updated_count = 0 for key, value in mapping.items(): # Only update if current value is empty or NULL query = """ UPDATE settings SET value = %s, updated_at = CURRENT_TIMESTAMP WHERE key = %s AND (value IS NULL OR value = '') RETURNING key """ result = execute_query(query, (value, key)) if result: updated_count += 1 logger.info(f"✅ Synced {key} from .env") return { "message": f"Synced {updated_count} settings from .env file", "updated_count": updated_count } # User Management Endpoints @router.get("/users", response_model=List[User], tags=["Users"]) async def get_users(is_active: Optional[bool] = None): """Get all users""" query = "SELECT user_id as id, username, email, full_name, is_active, last_login_at as last_login, created_at FROM users" params = [] if is_active is not None: query += " WHERE is_active = %s" params.append(is_active) query += " ORDER BY username" result = execute_query(query, tuple(params) if params else None) return result or [] @router.get("/users/{user_id}", response_model=User, tags=["Users"]) async def get_user(user_id: int): """Get user by ID""" query = "SELECT user_id as id, username, email, full_name, is_active, last_login_at as last_login, created_at FROM users WHERE user_id = %s" result = execute_query(query, (user_id,)) if not result: raise HTTPException(status_code=404, detail="User not found") return result[0] @router.post("/users", response_model=User, tags=["Users"]) async def create_user(user: UserCreate): """Create a new user""" # Check if username exists existing = execute_query("SELECT user_id FROM users WHERE username = %s", (user.username,)) if existing: raise HTTPException(status_code=400, detail="Username already exists") # Hash password (simple SHA256 for now - should use bcrypt in production) import hashlib password_hash = hashlib.sha256(user.password.encode()).hexdigest() query = """ INSERT INTO users (username, email, password_hash, full_name, is_active) VALUES (%s, %s, %s, %s, true) RETURNING user_id as id, username, email, full_name, is_active, last_login_at as last_login, created_at """ result = execute_query(query, (user.username, user.email, password_hash, user.full_name)) if not result: raise HTTPException(status_code=500, detail="Failed to create user") logger.info(f"✅ Created user: {user.username}") return result[0] @router.put("/users/{user_id}", response_model=User, tags=["Users"]) async def update_user(user_id: int, user: UserUpdate): """Update user details""" # Check if user exists existing = execute_query("SELECT user_id FROM users WHERE user_id = %s", (user_id,)) if not existing: raise HTTPException(status_code=404, detail="User not found") # Build update query update_fields = [] params = [] if user.email is not None: update_fields.append("email = %s") params.append(user.email) if user.full_name is not None: update_fields.append("full_name = %s") params.append(user.full_name) if user.is_active is not None: update_fields.append("is_active = %s") params.append(user.is_active) if not update_fields: raise HTTPException(status_code=400, detail="No fields to update") params.append(user_id) query = f""" UPDATE users SET {', '.join(update_fields)}, updated_at = CURRENT_TIMESTAMP WHERE user_id = %s RETURNING user_id as id, username, email, full_name, is_active, last_login_at as last_login, created_at """ result = execute_query(query, tuple(params)) if not result: raise HTTPException(status_code=500, detail="Failed to update user") logger.info(f"✅ Updated user: {user_id}") return result[0] @router.delete("/users/{user_id}", tags=["Users"]) async def deactivate_user(user_id: int): """Deactivate a user (soft delete)""" query = """ UPDATE users SET is_active = false, updated_at = CURRENT_TIMESTAMP WHERE user_id = %s RETURNING user_id as id """ result = execute_query(query, (user_id,)) if not result: raise HTTPException(status_code=404, detail="User not found") logger.info(f"✅ Deactivated user: {user_id}") return {"message": "User deactivated successfully"} @router.post("/users/{user_id}/reset-password", tags=["Users"]) async def reset_user_password(user_id: int, new_password: str): """Reset user password""" import hashlib password_hash = hashlib.sha256(new_password.encode()).hexdigest() query = """ UPDATE users SET password_hash = %s, updated_at = CURRENT_TIMESTAMP WHERE user_id = %s RETURNING user_id as id """ result = execute_query(query, (password_hash, user_id)) if not result: raise HTTPException(status_code=404, detail="User not found") logger.info(f"✅ Reset password for user: {user_id}") return {"message": "Password reset successfully"} # AI Prompts Management def _get_default_prompts(): """Helper to get default system prompts""" from app.services.ollama_service import OllamaService ollama_service = OllamaService() return { "invoice_extraction": { "name": "📄 Faktura Udtrækning (Invoice Parser)", "description": "System prompt brugt til at udtrække data fra fakturaer og kreditnotaer via Ollama LLM. Håndterer danske nummerformater, datoer og linjegenkendelse.", "model": ollama_service.model, "endpoint": ollama_service.endpoint, "prompt": ollama_service._build_system_prompt(), "parameters": { "temperature": 0.1, "top_p": 0.9, "num_predict": 2000 } }, "ticket_classification": { "name": "🎫 Ticket Klassificering (Auto-Triage)", "description": "Klassificerer indkomne tickets baseret på emne og indhold. Tildeler kategori, prioritet og ansvarlig team.", "model": ollama_service.model, "endpoint": ollama_service.endpoint, "prompt": """Du er en erfaren IT-supporter der skal klassificere indkomne support-sager. Dine opgaver er: 1. Analyser emne og beskrivelse 2. Bestem Kategori: [Hardware, Software, Netværk, Adgang, Andet] 3. Bestem Prioritet: [Lav, Mellem, Høj, Kritisk] 4. Foreslå handlingsplan (kort punktform) Output skal være gyldig JSON: { "category": "string", "priority": "string", "summary": "string", "suggested_actions": ["string"] }""", "parameters": { "temperature": 0.3, "top_p": 0.95, "num_predict": 1000 } }, "ticket_summary": { "name": "📝 Ticket Summering (Fakturagrundlag)", "description": "Analyserer alle kommentarer og noter i en ticket for at lave et kort, præcist resumé til fakturaen eller kunden.", "model": ollama_service.model, "endpoint": ollama_service.endpoint, "prompt": """Du er en administrativ assistent der skal gøre en it-sag klar til fakturering. Opgave: Læs historikken igennem og skriv et kort resumé af det udførte arbejde. - Fokusér på løsningen, ikke problemet - Brug professionelt sprog - Undlad interne tekniske detaljer (medmindre relevant for faktura) - Sprog: Dansk - Længde: 2-3 sætninger Input: [Liste af kommentarer] Output: [Fakturastekst]""", "parameters": { "temperature": 0.2, "top_p": 0.9, "num_predict": 500 } }, "text_rewrite": { "name": "✍️ Omskrivning (Email + Sagsbeskrivelse)", "description": "Renskriver tekst til professionelt dansk for både e-mails og sagsbeskrivelser, uden at ændre fakta.", "model": ollama_service.model, "endpoint": ollama_service.endpoint, "prompt": """Du er en dansk skriveassistent for IT-support. Din opgave er at renskrive en rå tekst til klart, professionelt og venligt dansk. Teksten kan være enten en e-mail eller en sagsbeskrivelse. Regler: 1. Bevar ALLE fakta, navne, datoer, beløb, ticket/sags-ID og tekniske termer. 2. Ret stavefejl, tegnsætning og grammatik. 3. Gør teksten kortere og mere præcis, men uden at fjerne vigtig information. 4. Fjern fyldord, gentagelser og intern støj. 5. Bevar tone og intention: neutral, serviceminded og professionel. 6. Opfind aldrig nye oplysninger. 7. Hvis input er e-mail: returner i formatet: Emne: Besked: 8. Hvis input er sagsbeskrivelse: returner i formatet: Titel: Beskrivelse: Output: - Returner KUN den renskrevne tekst i korrekt format. - Ingen forklaringer før eller efter. """, "parameters": { "temperature": 0.2, "top_p": 0.9, "num_predict": 800 } }, "kb_generation": { "name": "📚 Vidensbank Generator (Solution to Article)", "description": "Omdanner en løst ticket til en generel vejledning til vidensbanken.", "model": ollama_service.model, "endpoint": ollama_service.endpoint, "prompt": """Du er teknisk forfatter. Din opgave er at omskrive en konkret support-sag til en generel vejledning. Regler: 1. Fjern alle kunde-specifikke data (navne, IP-adresser, passwords) 2. Strukturer som: - Problem - Årsag (hvis kendt) - Løsning (Trin-for-trin guide) 3. Brug letforståeligt dansk 4. Formater med Markdown Input: [Ticket Beskrivelse + Løsning] Output: [Markdown Guide]""", "parameters": { "temperature": 0.4, "top_p": 0.9, "num_predict": 2000 } }, "troubleshooting_assistant": { "name": "🔧 Fejlsøgnings Copilot (Tech Helper)", "description": "Fungerer som en senior-tekniker der giver sparring på en fejlbeskrivelse. Foreslår konkrete fejlsøgningstrin og kommandoer.", "model": ollama_service.model, "endpoint": ollama_service.endpoint, "prompt": """Du er en Senior Systemadministrator med 20 års erfaring. En junior-tekniker spørger om hjælp til et problem. Din opgave: 1. Analyser symptomerne 2. List de 3 mest sandsynlige årsager 3. Foreslå en trin-for-trin fejlsøgningsplan (start med det mest sandsynlige) 4. Nævn relevante værktøjer eller kommandoer (Windows/Linux/Network) Vær kortfattet, teknisk præcis og handlingsorienteret. Sprog: Dansk (men engelske fagtermer er OK). Input: [Fejlbeskrivelse] Output: [Markdown Guide]""", "parameters": { "temperature": 0.3, "top_p": 0.9, "num_predict": 1500 } }, "sentiment_analysis": { "name": "🌡️ Sentiment Analyse (Kunde-Humør)", "description": "Analyserer tonen i en kundehenvendelse for at vurdere hast, frustration og risiko. Bruges til prioritering.", "model": ollama_service.model, "endpoint": ollama_service.endpoint, "prompt": """Analyser tonen i følgende tekst fra en kunde. Bestem følgende: 1. Sentiment: [Positiv, Neutral, Frustreret, Vred] 2. Hastegrad-indikatorer: Er der ord der indikerer panik eller kritisk hast? 3. Risikovurdering (0-10): Hvor stor risiko er der for at kunden forlader os? (10=Høj) Returner resultatet som JSON format. Input: [Kunde Tekst] Output: { "sentiment": "...", "urgency": "...", "risk_score": 0 }""", "parameters": { "temperature": 0.1, "top_p": 0.9, "num_predict": 500 } }, "meeting_action_items": { "name": "📋 Mødenoter til Opgaver (Action Extraction)", "description": "Scanner rå mødereferater eller notater og udtrækker konkrete 'Action Items', deadlines og ansvarlige personer.", "model": ollama_service.model, "endpoint": ollama_service.endpoint, "prompt": """Du er en effektiv projektleder-assistent. Din opgave er at scanne mødereferater og udtrække "Action Items". For hver opgave skal du finde: - Aktivitet (Hvad skal gøres?) - Ansvarlig (Hvem?) - Deadline (Hvornår? Hvis nævnt) Ignorer løs snak og diskussioner. Fokusér kun på beslutninger og opgaver der skal udføres. Outputtet skal være en punktopstilling. Input: [Mødenoter] Output: [Liste af opgaver]""", "parameters": { "temperature": 0.2, "top_p": 0.9, "num_predict": 1000 } } } def _get_prompts_with_overrides() -> Dict: """Get AI prompts with DB overrides applied""" prompts = _get_default_prompts() try: rows = execute_query("SELECT key, prompt_text FROM ai_prompts") if rows: for row in rows: if row['key'] in prompts: prompts[row['key']]['prompt'] = row['prompt_text'] prompts[row['key']]['is_custom'] = True except Exception as e: logger.warning(f"Could not load custom ai prompts: {e}") return prompts def _get_test_input_for_prompt(key: str) -> str: """Default test input per prompt type""" examples = { "invoice_extraction": "FAKTURA 2026-1001 fra Demo A/S. CVR 12345678. Total 1.250,00 DKK inkl moms.", "ticket_classification": "Emne: Kan ikke logge på VPN. Beskrivelse: Flere brugere er ramt siden i morges.", "ticket_summary": "Bruger havde netværksfejl. Router genstartet og DNS opdateret. Forbindelse virker nu stabilt.", "text_rewrite": "emne: printer virker ikke\nhej\nvi har stadig problem med printeren hos norva24 i hellerup, den stopper hele tiden og brugerne kan ikke printe, kan i kigge på det hurtigst muligt\nvh dennis", "kb_generation": "Problem: Outlook åbner ikke. Løsning: Reparer Office installation og nulstil profil.", "troubleshooting_assistant": "Server svarer langsomt efter opdatering. CPU er høj, disk IO er normal.", "sentiment_analysis": "Jeg er meget frustreret, systemet er nede igen og vi mister kunder!", "meeting_action_items": "Peter opdaterer firewall fredag. Anna sender status til kunden mandag.", } return examples.get(key, "Skriv kort: AI test OK") @router.get("/ai-prompts", tags=["Settings"]) async def get_ai_prompts(): """Get all AI prompts (defaults merged with custom overrides)""" return _get_prompts_with_overrides() class PromptUpdate(BaseModel): prompt_text: str class PromptTestRequest(BaseModel): test_input: Optional[str] = None prompt_text: Optional[str] = None timeout_seconds: Optional[int] = None _prompt_test_last_call: Dict[str, float] = {} @router.put("/ai-prompts/{key}", tags=["Settings"]) async def update_ai_prompt(key: str, update: PromptUpdate): """Override a system prompt with a custom one""" defaults = _get_default_prompts() if key not in defaults: raise HTTPException(status_code=404, detail="Unknown prompt key") try: # Upsert query = """ INSERT INTO ai_prompts (key, prompt_text, updated_at) VALUES (%s, %s, CURRENT_TIMESTAMP) ON CONFLICT (key) DO UPDATE SET prompt_text = EXCLUDED.prompt_text, updated_at = CURRENT_TIMESTAMP RETURNING key """ execute_query(query, (key, update.prompt_text)) return {"message": "Prompt updated", "key": key} except Exception as e: logger.error(f"Error saving prompt: {e}") raise HTTPException(status_code=500, detail="Could not save prompt") @router.delete("/ai-prompts/{key}", tags=["Settings"]) async def reset_ai_prompt(key: str): """Reset a prompt to its system default""" try: execute_query("DELETE FROM ai_prompts WHERE key = %s", (key,)) return {"message": "Prompt reset to default"} except Exception as e: logger.error(f"Error resetting prompt: {e}") raise HTTPException(status_code=500, detail="Could not reset prompt") @router.post("/ai-prompts/{key}/test", tags=["Settings"]) async def test_ai_prompt(key: str, payload: PromptTestRequest, http_request: Request): """Run a quick AI test for a specific system prompt""" prompts = _get_prompts_with_overrides() if key not in prompts: raise HTTPException(status_code=404, detail="Unknown prompt key") prompt_cfg = prompts[key] model = prompt_cfg.get("model") or settings.OLLAMA_MODEL endpoint = prompt_cfg.get("endpoint") or settings.OLLAMA_ENDPOINT prompt_text = (payload.prompt_text or prompt_cfg.get("prompt") or "").strip() if not prompt_text: raise HTTPException(status_code=400, detail="Prompt text is empty") test_input = (payload.test_input or _get_test_input_for_prompt(key)).strip() if not test_input: raise HTTPException(status_code=400, detail="Test input is empty") start = time.perf_counter() client_host = (http_request.client.host if http_request.client else "unknown") # Cooldown to prevent hammering external endpoints and getting rate-limited/banned. cooldown_seconds = 2.0 now_monotonic = time.monotonic() last_call = _prompt_test_last_call.get(client_host) if last_call and (now_monotonic - last_call) < cooldown_seconds: wait_for = round(cooldown_seconds - (now_monotonic - last_call), 2) raise HTTPException( status_code=429, detail=f"For mange tests for hurtigt. Vent {wait_for} sekunder og prøv igen.", ) _prompt_test_last_call[client_host] = now_monotonic read_timeout_seconds = payload.timeout_seconds or 90 read_timeout_seconds = max(5, min(int(read_timeout_seconds), 300)) try: model_normalized = (model or "").strip().lower() # qwen models are more reliable with /api/chat than /api/generate. use_chat_api = model_normalized.startswith("qwen") logger.info( "🧪 AI prompt test start key=%s model=%s timeout=%ss client=%s", key, model, read_timeout_seconds, client_host, ) timeout = httpx.Timeout(connect=10.0, read=float(read_timeout_seconds), write=30.0, pool=10.0) async with httpx.AsyncClient(timeout=timeout) as client: if use_chat_api: response = await client.post( f"{endpoint}/api/chat", json={ "model": model, "messages": [ {"role": "system", "content": prompt_text}, {"role": "user", "content": test_input}, ], "stream": False, "options": {"temperature": 0.2, "num_predict": 600}, }, ) else: response = await client.post( f"{endpoint}/api/generate", json={ "model": model, "prompt": f"{prompt_text}\n\nBrugerinput:\n{test_input}", "stream": False, "options": {"temperature": 0.2, "num_predict": 600}, }, ) if response.status_code != 200: raise HTTPException( status_code=502, detail=f"AI endpoint fejl: {response.status_code} - {response.text[:300]}", ) try: data = response.json() except Exception as parse_error: raise HTTPException( status_code=502, detail=f"AI endpoint returnerede ugyldig JSON: {str(parse_error)}", ) if use_chat_api: message_data = data.get("message", {}) ai_response = (message_data.get("content") or message_data.get("thinking") or "").strip() else: ai_response = (data.get("response") or "").strip() if not ai_response: raise HTTPException(status_code=502, detail="AI returnerede tomt svar") latency_ms = int((time.perf_counter() - start) * 1000) return { "ok": True, "key": key, "model": model, "endpoint": endpoint, "test_input": test_input, "ai_response": ai_response, "timeout_seconds": read_timeout_seconds, "latency_ms": latency_ms, } except HTTPException: raise except httpx.TimeoutException as e: logger.error(f"❌ AI prompt test timed out for {key}: {repr(e)}") raise HTTPException(status_code=504, detail="AI test timed out (model svarer for langsomt)") except Exception as e: logger.error(f"❌ AI prompt test failed for {key}: {repr(e)}") err = str(e) or e.__class__.__name__ raise HTTPException(status_code=500, detail=f"Kunne ikke teste AI prompt: {err}") system_event_log_access = require_any_permission("system.event_log.view", "users.manage", "system.admin") def _event_sort_value(event: Dict) -> float: """Return one comparable timestamp for naive and timezone-aware DB values.""" value = event.get("occurred_at") if not value: return 0.0 if isinstance(value, datetime): if value.tzinfo is None: value = value.replace(tzinfo=timezone.utc) return value.timestamp() try: parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.timestamp() except (TypeError, ValueError): return 0.0 @router.get("/settings/system-events/list", dependencies=[Depends(system_event_log_access)], tags=["Settings"]) async def list_system_events( source: Optional[str] = Query(None), event_type: Optional[str] = Query(None), status: Optional[str] = Query(None), q: Optional[str] = Query(None), date_from: Optional[datetime] = Query(None), date_to: Optional[datetime] = Query(None), user_id: Optional[int] = Query(None), limit: int = Query(250, ge=1, le=2000), ): """Unified operational log over persisted system, VoIP and email events.""" clauses, params = [], [] if source: clauses.append("l.source=%s"); params.append(source) if event_type: clauses.append("l.event_type=%s"); params.append(event_type) if status: clauses.append("l.status=%s"); params.append(status) if user_id: clauses.append("l.user_id=%s"); params.append(user_id) if date_from: clauses.append("l.occurred_at>=%s"); params.append(date_from) if date_to: clauses.append("l.occurred_at<=%s"); params.append(date_to) if q: clauses.append("(l.summary ILIKE %s OR l.correlation_id ILIKE %s OR l.details::text ILIKE %s)") params.extend([f"%{q}%"] * 3) where = "WHERE " + " AND ".join(clauses) if clauses else "" events = execute_query( f"""SELECT l.id,l.source,l.event_type,l.status,l.summary,l.correlation_id,l.user_id, l.details,l.occurred_at,COALESCE(u.full_name,u.username) AS user_name FROM system_event_log l LEFT JOIN users u ON u.user_id=l.user_id {where} ORDER BY l.occurred_at DESC LIMIT %s""", tuple(params + [limit]), ) or [] # Existing mail records remain searchable even before producers adopt system_event_log. if source in (None, "email"): email_clauses, email_params = [], [] if date_from: email_clauses.append("em.received_date>=%s"); email_params.append(date_from) if date_to: email_clauses.append("em.received_date<=%s"); email_params.append(date_to) if q: email_clauses.append("(em.subject ILIKE %s OR em.sender_email ILIKE %s OR em.recipient_email ILIKE %s)") email_params.extend([f"%{q}%"] * 3) email_where = "WHERE " + " AND ".join(email_clauses) if email_clauses else "" emails = execute_query( f"""SELECT em.id,'email' AS source, CASE WHEN LOWER(COALESCE(em.folder,'')) LIKE '%%sent%%' THEN 'email.sent' ELSE 'email.received' END AS event_type, 'success'::text AS status,COALESCE(em.subject,'(intet emne)') AS summary, em.message_id AS correlation_id,NULL::integer AS user_id, jsonb_build_object('from',em.sender_email,'to',em.recipient_email,'folder',em.folder) AS details, em.received_date AS occurred_at,NULL::text AS user_name FROM email_messages em {email_where} ORDER BY em.received_date DESC LIMIT %s""", tuple(email_params + [limit]), ) or [] if event_type: emails = [x for x in emails if x["event_type"] == event_type] if status: emails = [x for x in emails if x["status"] == status] events.extend(emails) events.sort(key=_event_sort_value, reverse=True) return {"items": events[:limit], "sources": ["voip", "email", "system"], "count": len(events[:limit])}