143 lines
5.7 KiB
Python
143 lines
5.7 KiB
Python
from datetime import datetime
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db import (
|
|
AuditLogORM,
|
|
AuthorizedUserORM,
|
|
MemoryItemORM,
|
|
SecretORM,
|
|
SettingORM,
|
|
ToolStateORM,
|
|
list_recent_logs,
|
|
)
|
|
from app.config import get_settings
|
|
from app.models import DashboardSnapshot, MemoryRecord, RuntimeSettings, TelegramStatus, ToolToggle, UserRecord
|
|
|
|
|
|
class AdminService:
|
|
def __init__(self, session: Session) -> None:
|
|
self.session = session
|
|
|
|
def get_runtime_settings(self) -> RuntimeSettings:
|
|
settings = {
|
|
item.key: item.value for item in self.session.scalars(select(SettingORM))
|
|
}
|
|
tools = list(self.session.scalars(select(ToolStateORM).order_by(ToolStateORM.name.asc())))
|
|
return RuntimeSettings(
|
|
terminal_mode=int(settings["terminal_mode"]),
|
|
search_provider=settings["search_provider"],
|
|
ollama_base_url=settings["ollama_base_url"],
|
|
default_model=settings["default_model"],
|
|
tools=[ToolToggle(name=tool.name, enabled=tool.enabled) for tool in tools],
|
|
)
|
|
|
|
def update_runtime_settings(self, payload: RuntimeSettings) -> RuntimeSettings:
|
|
self._save_setting("terminal_mode", str(payload.terminal_mode))
|
|
self._save_setting("search_provider", payload.search_provider)
|
|
self._save_setting("ollama_base_url", payload.ollama_base_url)
|
|
self._save_setting("default_model", payload.default_model)
|
|
|
|
for tool in payload.tools:
|
|
record = self.session.get(ToolStateORM, tool.name)
|
|
if record is None:
|
|
self.session.add(ToolStateORM(name=tool.name, enabled=tool.enabled, updated_at=datetime.utcnow()))
|
|
else:
|
|
record.enabled = tool.enabled
|
|
record.updated_at = datetime.utcnow()
|
|
|
|
self.session.add(AuditLogORM(category="settings", message="settings:runtime-updated"))
|
|
self.session.commit()
|
|
return self.get_runtime_settings()
|
|
|
|
def dashboard(self) -> DashboardSnapshot:
|
|
return DashboardSnapshot(
|
|
settings=self.get_runtime_settings(),
|
|
whitelist_count=self.session.scalar(select(func.count()).select_from(AuthorizedUserORM)) or 0,
|
|
memory_items=self.session.scalar(select(func.count()).select_from(MemoryItemORM)) or 0,
|
|
recent_logs=list_recent_logs(self.session, limit=10),
|
|
)
|
|
|
|
def list_users(self) -> list[UserRecord]:
|
|
stmt = select(AuthorizedUserORM).order_by(AuthorizedUserORM.created_at.desc())
|
|
return [
|
|
UserRecord(
|
|
telegram_user_id=user.telegram_user_id,
|
|
username=user.username,
|
|
display_name=user.display_name,
|
|
is_active=user.is_active,
|
|
)
|
|
for user in self.session.scalars(stmt)
|
|
]
|
|
|
|
def save_user(self, user: UserRecord) -> UserRecord:
|
|
record = self.session.get(AuthorizedUserORM, user.telegram_user_id)
|
|
if record is None:
|
|
record = AuthorizedUserORM(
|
|
telegram_user_id=user.telegram_user_id,
|
|
username=user.username,
|
|
display_name=user.display_name,
|
|
is_active=user.is_active,
|
|
created_at=datetime.utcnow(),
|
|
updated_at=datetime.utcnow(),
|
|
)
|
|
self.session.add(record)
|
|
else:
|
|
record.username = user.username
|
|
record.display_name = user.display_name
|
|
record.is_active = user.is_active
|
|
record.updated_at = datetime.utcnow()
|
|
self.session.add(AuditLogORM(category="users", message=f"users:upsert:{user.telegram_user_id}"))
|
|
self.session.commit()
|
|
return user
|
|
|
|
def list_memory(self) -> list[MemoryRecord]:
|
|
stmt = select(MemoryItemORM).order_by(MemoryItemORM.created_at.desc(), MemoryItemORM.id.desc()).limit(50)
|
|
return [
|
|
MemoryRecord(id=item.id, content=item.content, kind=item.kind, created_at=item.created_at)
|
|
for item in self.session.scalars(stmt)
|
|
]
|
|
|
|
def clear_memory(self) -> None:
|
|
for item in self.session.scalars(select(MemoryItemORM)):
|
|
self.session.delete(item)
|
|
self.session.add(AuditLogORM(category="memory", message="memory:cleared"))
|
|
self.session.commit()
|
|
|
|
def get_secret_mask(self, key: str) -> str:
|
|
record = self.session.get(SecretORM, key)
|
|
value = record.value if record else ""
|
|
if len(value) < 4:
|
|
return ""
|
|
return f"{value[:2]}***{value[-2:]}"
|
|
|
|
def save_secret(self, key: str, value: str) -> None:
|
|
record = self.session.get(SecretORM, key)
|
|
if record is None:
|
|
self.session.add(SecretORM(key=key, value=value, updated_at=datetime.utcnow()))
|
|
else:
|
|
record.value = value
|
|
record.updated_at = datetime.utcnow()
|
|
self.session.add(AuditLogORM(category="secrets", message=f"secrets:updated:{key}"))
|
|
self.session.commit()
|
|
|
|
def _save_setting(self, key: str, value: str) -> None:
|
|
record = self.session.get(SettingORM, key)
|
|
if record is None:
|
|
self.session.add(SettingORM(key=key, value=value, updated_at=datetime.utcnow()))
|
|
else:
|
|
record.value = value
|
|
record.updated_at = datetime.utcnow()
|
|
|
|
def telegram_status(self) -> TelegramStatus:
|
|
settings = get_settings()
|
|
configured = bool(settings.telegram_bot_token)
|
|
return TelegramStatus(
|
|
configured=configured,
|
|
polling_active=False,
|
|
message="Telegram token is configured. Polling starts when the backend boots."
|
|
if configured
|
|
else "Telegram token is not configured.",
|
|
)
|