mirror of
https://github.com/jkingsman/Remote-Terminal-for-MeshCore.git
synced 2026-03-28 17:43:05 +01:00
226 lines
8.2 KiB
Python
226 lines
8.2 KiB
Python
import asyncio
|
|
import logging
|
|
from typing import Literal
|
|
|
|
from fastapi import APIRouter
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.models import AppSettings
|
|
from app.repository import AppSettingsRepository
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/settings", tags=["settings"])
|
|
|
|
|
|
class AppSettingsUpdate(BaseModel):
|
|
max_radio_contacts: int | None = Field(
|
|
default=None,
|
|
ge=1,
|
|
le=1000,
|
|
description=(
|
|
"Maximum contacts to keep on radio (favorites first, then recent non-repeaters)"
|
|
),
|
|
)
|
|
auto_decrypt_dm_on_advert: bool | None = Field(
|
|
default=None,
|
|
description="Whether to attempt historical DM decryption on new contact advertisement",
|
|
)
|
|
sidebar_sort_order: Literal["recent", "alpha"] | None = Field(
|
|
default=None,
|
|
description="Sidebar sort order: 'recent' or 'alpha'",
|
|
)
|
|
advert_interval: int | None = Field(
|
|
default=None,
|
|
ge=0,
|
|
description="Periodic advertisement interval in seconds (0 = disabled, minimum 3600)",
|
|
)
|
|
flood_scope: str | None = Field(
|
|
default=None,
|
|
description="Outbound flood scope / region name (empty = disabled)",
|
|
)
|
|
blocked_keys: list[str] | None = Field(
|
|
default=None,
|
|
description="Public keys whose messages are hidden from the UI",
|
|
)
|
|
blocked_names: list[str] | None = Field(
|
|
default=None,
|
|
description="Display names whose messages are hidden from the UI",
|
|
)
|
|
|
|
|
|
class BlockKeyRequest(BaseModel):
|
|
key: str = Field(description="Public key to toggle block status")
|
|
|
|
|
|
class BlockNameRequest(BaseModel):
|
|
name: str = Field(description="Display name to toggle block status")
|
|
|
|
|
|
class FavoriteRequest(BaseModel):
|
|
type: Literal["channel", "contact"] = Field(description="'channel' or 'contact'")
|
|
id: str = Field(description="Channel key or contact public key")
|
|
|
|
|
|
class MigratePreferencesRequest(BaseModel):
|
|
favorites: list[FavoriteRequest] = Field(
|
|
default_factory=list,
|
|
description="List of favorites from localStorage",
|
|
)
|
|
sort_order: str = Field(
|
|
default="recent",
|
|
description="Sort order preference from localStorage",
|
|
)
|
|
last_message_times: dict[str, int] = Field(
|
|
default_factory=dict,
|
|
description="Map of conversation state keys to timestamps from localStorage",
|
|
)
|
|
|
|
|
|
class MigratePreferencesResponse(BaseModel):
|
|
migrated: bool = Field(description="Whether migration occurred (false if already migrated)")
|
|
settings: AppSettings = Field(description="Current settings after migration attempt")
|
|
|
|
|
|
@router.get("", response_model=AppSettings)
|
|
async def get_settings() -> AppSettings:
|
|
"""Get current application settings."""
|
|
return await AppSettingsRepository.get()
|
|
|
|
|
|
@router.patch("", response_model=AppSettings)
|
|
async def update_settings(update: AppSettingsUpdate) -> AppSettings:
|
|
"""Update application settings.
|
|
|
|
Settings are persisted to the database and survive restarts.
|
|
"""
|
|
kwargs = {}
|
|
if update.max_radio_contacts is not None:
|
|
logger.info("Updating max_radio_contacts to %d", update.max_radio_contacts)
|
|
kwargs["max_radio_contacts"] = update.max_radio_contacts
|
|
|
|
if update.auto_decrypt_dm_on_advert is not None:
|
|
logger.info("Updating auto_decrypt_dm_on_advert to %s", update.auto_decrypt_dm_on_advert)
|
|
kwargs["auto_decrypt_dm_on_advert"] = update.auto_decrypt_dm_on_advert
|
|
|
|
if update.sidebar_sort_order is not None:
|
|
logger.info("Updating sidebar_sort_order to %s", update.sidebar_sort_order)
|
|
kwargs["sidebar_sort_order"] = update.sidebar_sort_order
|
|
|
|
if update.advert_interval is not None:
|
|
# Enforce minimum 1-hour interval; 0 means disabled
|
|
interval = update.advert_interval
|
|
if 0 < interval < 3600:
|
|
interval = 3600
|
|
logger.info("Updating advert_interval to %d", interval)
|
|
kwargs["advert_interval"] = interval
|
|
|
|
# Block lists
|
|
if update.blocked_keys is not None:
|
|
kwargs["blocked_keys"] = [k.lower() for k in update.blocked_keys]
|
|
if update.blocked_names is not None:
|
|
kwargs["blocked_names"] = update.blocked_names
|
|
|
|
# Flood scope
|
|
flood_scope_changed = False
|
|
if update.flood_scope is not None:
|
|
stripped = update.flood_scope.strip()
|
|
kwargs["flood_scope"] = stripped
|
|
flood_scope_changed = True
|
|
|
|
if kwargs:
|
|
result = await AppSettingsRepository.update(**kwargs)
|
|
|
|
# Apply flood scope to radio immediately if changed
|
|
if flood_scope_changed:
|
|
from app.radio import radio_manager
|
|
|
|
if radio_manager.is_connected:
|
|
try:
|
|
scope = result.flood_scope
|
|
async with radio_manager.radio_operation("set_flood_scope") as mc:
|
|
await mc.commands.set_flood_scope(scope if scope else "")
|
|
logger.info("Applied flood_scope=%r to radio", scope or "(disabled)")
|
|
except Exception as e:
|
|
logger.warning("Failed to apply flood_scope to radio: %s", e)
|
|
|
|
return result
|
|
|
|
return await AppSettingsRepository.get()
|
|
|
|
|
|
@router.post("/favorites/toggle", response_model=AppSettings)
|
|
async def toggle_favorite(request: FavoriteRequest) -> AppSettings:
|
|
"""Toggle a conversation's favorite status."""
|
|
settings = await AppSettingsRepository.get()
|
|
is_favorited = any(f.type == request.type and f.id == request.id for f in settings.favorites)
|
|
|
|
if is_favorited:
|
|
logger.info("Removing favorite: %s %s", request.type, request.id[:12])
|
|
result = await AppSettingsRepository.remove_favorite(request.type, request.id)
|
|
else:
|
|
logger.info("Adding favorite: %s %s", request.type, request.id[:12])
|
|
result = await AppSettingsRepository.add_favorite(request.type, request.id)
|
|
|
|
# When a contact favorite changes, sync the radio so the contact is
|
|
# loaded/unloaded immediately rather than waiting for the next advert.
|
|
if request.type == "contact":
|
|
from app.radio_sync import sync_recent_contacts_to_radio
|
|
|
|
asyncio.create_task(sync_recent_contacts_to_radio(force=True))
|
|
|
|
return result
|
|
|
|
|
|
@router.post("/blocked-keys/toggle", response_model=AppSettings)
|
|
async def toggle_blocked_key(request: BlockKeyRequest) -> AppSettings:
|
|
"""Toggle a public key's blocked status."""
|
|
logger.info("Toggling blocked key: %s", request.key[:12])
|
|
return await AppSettingsRepository.toggle_blocked_key(request.key)
|
|
|
|
|
|
@router.post("/blocked-names/toggle", response_model=AppSettings)
|
|
async def toggle_blocked_name(request: BlockNameRequest) -> AppSettings:
|
|
"""Toggle a display name's blocked status."""
|
|
logger.info("Toggling blocked name: %s", request.name)
|
|
return await AppSettingsRepository.toggle_blocked_name(request.name)
|
|
|
|
|
|
@router.post("/migrate", response_model=MigratePreferencesResponse)
|
|
async def migrate_preferences(request: MigratePreferencesRequest) -> MigratePreferencesResponse:
|
|
"""Migrate all preferences from frontend localStorage to database.
|
|
|
|
This is a one-time migration. If preferences have already been migrated,
|
|
this endpoint will not overwrite them and will return migrated=false.
|
|
|
|
Call this on frontend startup to ensure preferences are moved to the database.
|
|
After successful migration, the frontend should clear localStorage preferences.
|
|
|
|
Migrates:
|
|
- favorites (remoteterm-favorites)
|
|
- sort_order (remoteterm-sortOrder)
|
|
- last_message_times (remoteterm-lastMessageTime)
|
|
"""
|
|
# Convert to dict format for the repository method
|
|
frontend_favorites = [{"type": f.type, "id": f.id} for f in request.favorites]
|
|
|
|
settings, did_migrate = await AppSettingsRepository.migrate_preferences_from_frontend(
|
|
favorites=frontend_favorites,
|
|
sort_order=request.sort_order,
|
|
last_message_times=request.last_message_times,
|
|
)
|
|
|
|
if did_migrate:
|
|
logger.info(
|
|
"Migrated preferences from frontend: %d favorites, sort_order=%s, %d message times",
|
|
len(frontend_favorites),
|
|
request.sort_order,
|
|
len(request.last_message_times),
|
|
)
|
|
else:
|
|
logger.debug("Preferences already migrated, skipping")
|
|
|
|
return MigratePreferencesResponse(
|
|
migrated=did_migrate,
|
|
settings=settings,
|
|
)
|