feat: 支持 ONLYOFFICE 持久化配置管理

- 添加 SettingsRenderForm schema 和 renderForm 字段
- 实现数据库 schema 自动迁移(onlyoffice_enabled, onlyoffice_public_url, onlyoffice_jwt_secret_encrypted)
- 新增 resolve_onlyoffice_settings() 函数支持运行时配置解析
- 知识库服务改用数据库配置替代运行时配置
- 前端添加文件渲染配置页面,支持 JWT 密钥管理
- 完善相关测试覆盖
This commit is contained in:
caoxiaozhu
2026-05-09 08:02:01 +00:00
parent 94122fd34b
commit 4fbd313f35
14 changed files with 735 additions and 314 deletions

View File

@@ -39,6 +39,8 @@ class SystemSetting(Base):
embedding_provider: Mapped[str] = mapped_column(String(64), default="GLM")
embedding_model: Mapped[str] = mapped_column(String(255), default="Embedding-3")
embedding_endpoint: Mapped[str] = mapped_column(String(512), default="https://open.bigmodel.cn/api/paas/v4/")
onlyoffice_enabled: Mapped[bool] = mapped_column(Boolean, default=False)
onlyoffice_public_url: Mapped[str] = mapped_column(String(512), default="")
log_level: Mapped[str] = mapped_column(String(16), default="INFO")
retention_days: Mapped[int] = mapped_column(Integer, default=180)

View File

@@ -18,6 +18,7 @@ class SystemSettingSecret(Base):
backup_api_key_encrypted: Mapped[str] = mapped_column(Text, default="")
vlm_api_key_encrypted: Mapped[str] = mapped_column(Text, default="")
embedding_api_key_encrypted: Mapped[str] = mapped_column(Text, default="")
onlyoffice_jwt_secret_encrypted: Mapped[str] = mapped_column(Text, default="")
smtp_password_encrypted: Mapped[str] = mapped_column(Text, default="")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

View File

@@ -110,6 +110,20 @@ class SettingsLogForm(BaseModel):
return value.strip()
class SettingsRenderForm(BaseModel):
enabled: bool = False
publicUrl: str = Field(default="", max_length=512)
jwtSecret: str = Field(default="", max_length=1024)
jwtSecretConfigured: bool = False
@field_validator("publicUrl", "jwtSecret", mode="before")
@classmethod
def strip_string(cls, value: str | None) -> str | None:
if value is None:
return None
return value.strip()
class SettingsMailForm(BaseModel):
smtpHost: str = Field(min_length=1, max_length=255)
port: int = Field(default=465, ge=1, le=65535)
@@ -146,6 +160,7 @@ class SettingsRead(BaseModel):
companyForm: SettingsCompanyForm
adminForm: SettingsAdminForm
llmForm: SettingsLlmForm
renderForm: SettingsRenderForm
logForm: SettingsLogForm
mailForm: SettingsMailForm
@@ -154,6 +169,7 @@ class SettingsWrite(BaseModel):
companyForm: SettingsCompanyForm
adminForm: SettingsAdminForm
llmForm: SettingsLlmForm
renderForm: SettingsRenderForm
logForm: SettingsLogForm
mailForm: SettingsMailForm

View File

@@ -18,16 +18,17 @@ import jwt
from app.api.deps import CurrentUserContext
from app.core.config import get_settings
from app.core.logging import get_logger
from app.schemas.knowledge import (
KnowledgeDocumentDetailRead,
from app.schemas.knowledge import (
KnowledgeDocumentDetailRead,
KnowledgeDocumentRead,
KnowledgeFolderRead,
KnowledgeLibraryRead,
KnowledgeOnlyOfficeConfigRead,
KnowledgePreviewBlockRead,
KnowledgePreviewPageRead,
KnowledgePreviewStatRead,
)
KnowledgePreviewPageRead,
KnowledgePreviewStatRead,
)
from app.services.settings import resolve_onlyoffice_settings
logger = get_logger("app.services.knowledge")
@@ -239,51 +240,52 @@ class KnowledgeService:
) -> KnowledgeOnlyOfficeConfigRead:
self.ensure_library_ready()
settings = get_settings()
if not settings.onlyoffice_enabled:
onlyoffice_settings = resolve_onlyoffice_settings()
if not onlyoffice_settings.enabled:
logger.warning(
"ONLYOFFICE disabled in runtime config doc=%s enabled=%s public_url=%s backend_url=%s jwt_set=%s",
document_id,
settings.onlyoffice_enabled,
settings.onlyoffice_public_url,
settings.onlyoffice_backend_url,
bool(settings.onlyoffice_jwt_secret),
onlyoffice_settings.enabled,
onlyoffice_settings.public_url,
onlyoffice_settings.backend_url,
bool(onlyoffice_settings.jwt_secret),
)
raise ValueError("ONLYOFFICE 预览未启用。")
if not settings.onlyoffice_public_url or not settings.onlyoffice_backend_url:
if not onlyoffice_settings.public_url or not onlyoffice_settings.backend_url:
logger.warning(
"ONLYOFFICE config incomplete doc=%s enabled=%s public_url=%s backend_url=%s jwt_set=%s",
document_id,
settings.onlyoffice_enabled,
settings.onlyoffice_public_url,
settings.onlyoffice_backend_url,
bool(settings.onlyoffice_jwt_secret),
onlyoffice_settings.enabled,
onlyoffice_settings.public_url,
onlyoffice_settings.backend_url,
bool(onlyoffice_settings.jwt_secret),
)
raise ValueError("ONLYOFFICE 地址配置不完整。")
if not settings.onlyoffice_jwt_secret:
if not onlyoffice_settings.jwt_secret:
logger.warning(
"ONLYOFFICE JWT missing doc=%s enabled=%s public_url=%s backend_url=%s jwt_set=%s",
document_id,
settings.onlyoffice_enabled,
settings.onlyoffice_public_url,
settings.onlyoffice_backend_url,
bool(settings.onlyoffice_jwt_secret),
onlyoffice_settings.enabled,
onlyoffice_settings.public_url,
onlyoffice_settings.backend_url,
bool(onlyoffice_settings.jwt_secret),
)
raise ValueError("ONLYOFFICE JWT 密钥未配置。")
index = self._load_index()
entry = self._require_entry(index, document_id)
extension = self._extract_extension(entry["original_name"])
if extension not in ONLYOFFICE_EDITABLE_EXTENSIONS:
raise ValueError("当前文件格式不支持 ONLYOFFICE 预览。")
document_type = self._resolve_onlyoffice_document_type(extension)
backend_base_url = settings.onlyoffice_backend_url.rstrip("/")
public_url = settings.onlyoffice_public_url.rstrip("/")
access_token = self._build_onlyoffice_access_token(document_id)
document_url = (
f"{backend_base_url}{settings.api_v1_prefix}/knowledge/documents/{document_id}/onlyoffice/content"
f"?access_token={access_token}"
)
if extension not in ONLYOFFICE_EDITABLE_EXTENSIONS:
raise ValueError("当前文件格式不支持 ONLYOFFICE 预览。")
document_type = self._resolve_onlyoffice_document_type(extension)
backend_base_url = onlyoffice_settings.backend_url.rstrip("/")
public_url = onlyoffice_settings.public_url.rstrip("/")
access_token = self._build_onlyoffice_access_token(document_id)
document_url = (
f"{backend_base_url}{settings.api_v1_prefix}/knowledge/documents/{document_id}/onlyoffice/content"
f"?access_token={access_token}"
)
callback_url = (
f"{backend_base_url}{settings.api_v1_prefix}/knowledge/documents/{document_id}/onlyoffice/callback"
)
@@ -322,23 +324,23 @@ class KnowledgeService:
"width": "100%",
"height": "100%",
}
config["token"] = jwt.encode(config, settings.onlyoffice_jwt_secret, algorithm="HS256")
config["token"] = jwt.encode(config, onlyoffice_settings.jwt_secret, algorithm="HS256")
return KnowledgeOnlyOfficeConfigRead(
documentServerUrl=public_url,
config=config,
)
def validate_onlyoffice_access_token(self, document_id: str, access_token: str) -> None:
settings = get_settings()
try:
payload = jwt.decode(
access_token,
settings.onlyoffice_jwt_secret,
algorithms=["HS256"],
)
except jwt.PyJWTError as exc:
raise ValueError("ONLYOFFICE 文件访问令牌无效。") from exc
def validate_onlyoffice_access_token(self, document_id: str, access_token: str) -> None:
onlyoffice_settings = resolve_onlyoffice_settings()
try:
payload = jwt.decode(
access_token,
onlyoffice_settings.jwt_secret,
algorithms=["HS256"],
)
except jwt.PyJWTError as exc:
raise ValueError("ONLYOFFICE 文件访问令牌无效。") from exc
if payload.get("scope") != "onlyoffice-content" or payload.get("document_id") != document_id:
raise ValueError("ONLYOFFICE 文件访问令牌无效。")
@@ -665,13 +667,13 @@ class KnowledgeService:
checksum = str(entry.get("sha256") or "")[:12]
return f"{entry['id']}-v{version}-{checksum or 'nochecksum'}"
def _build_onlyoffice_access_token(self, document_id: str) -> str:
settings = get_settings()
payload = {
"scope": "onlyoffice-content",
"document_id": document_id,
}
return jwt.encode(payload, settings.onlyoffice_jwt_secret, algorithm="HS256")
def _build_onlyoffice_access_token(self, document_id: str) -> str:
onlyoffice_settings = resolve_onlyoffice_settings()
payload = {
"scope": "onlyoffice-content",
"document_id": document_id,
}
return jwt.encode(payload, onlyoffice_settings.jwt_secret, algorithm="HS256")
@staticmethod
def _resolve_onlyoffice_document_type(extension: str) -> str:

View File

@@ -1,20 +1,22 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from sqlalchemy.orm import Session
from app.core.admin_secret import legacy_admin_secret_to_password_hash, read_admin_secret, verify_admin_secret
from app.core.config import get_settings
from app.core.secret_box import decrypt_secret, encrypt_secret
from app.core.security import hash_password, verify_password
from app.db.base import Base
from app.models.system_model_setting import SystemModelSetting
from app.models.system_setting import SystemSetting
from app.models.system_setting_secret import SystemSettingSecret
from app.repositories.settings import SETTINGS_ROW_ID, SettingsRepository
from app.schemas.settings import SettingsRead, SettingsWrite
from dataclasses import dataclass
from datetime import datetime
from sqlalchemy import inspect, text
from sqlalchemy.orm import Session
from app.core.admin_secret import legacy_admin_secret_to_password_hash, read_admin_secret, verify_admin_secret
from app.core.config import get_settings
from app.core.secret_box import decrypt_secret, encrypt_secret
from app.core.security import hash_password, verify_password
from app.db.base import Base
from app.db.session import get_session_factory
from app.models.system_model_setting import SystemModelSetting
from app.models.system_setting import SystemSetting
from app.models.system_setting_secret import SystemSettingSecret
from app.repositories.settings import SETTINGS_ROW_ID, SettingsRepository
from app.schemas.settings import SettingsRead, SettingsWrite
@dataclass(frozen=True, slots=True)
@@ -79,25 +81,34 @@ MODEL_SLOT_CONFIGS = {
@dataclass(slots=True)
class AdminCredentialRecord:
account: str
email: str
password_hash: str
class SettingsService:
class AdminCredentialRecord:
account: str
email: str
password_hash: str
@dataclass(frozen=True, slots=True)
class OnlyOfficeRuntimeConfig:
enabled: bool
public_url: str
backend_url: str
jwt_secret: str
class SettingsService:
def __init__(self, db: Session) -> None:
self.db = db
self.repository = SettingsRepository(db)
self.runtime_settings = get_settings()
def ensure_settings_ready(self) -> tuple[SystemSetting, SystemSettingSecret]:
Base.metadata.create_all(bind=self.db.get_bind())
settings_row = self.repository.get_settings()
secrets_row = self.repository.get_secrets()
should_commit = False
legacy_admin = read_admin_secret()
def ensure_settings_ready(self) -> tuple[SystemSetting, SystemSettingSecret]:
Base.metadata.create_all(bind=self.db.get_bind())
self._ensure_settings_schema()
settings_row = self.repository.get_settings()
secrets_row = self.repository.get_secrets()
should_commit = False
legacy_admin = read_admin_secret()
if settings_row is None:
settings_row = self._build_default_settings()
@@ -112,14 +123,17 @@ class SettingsService:
if legacy_admin is not None and not secrets_row.admin_password_hash:
secrets_row.admin_password_hash = legacy_admin_secret_to_password_hash(legacy_admin)
admin_username = str(legacy_admin.get("username", "")).strip()
if admin_username and str(settings_row.admin_account or "").strip() in {"", "superadmin"}:
settings_row.admin_account = admin_username
should_commit = True
if should_commit:
self.db.commit()
self.db.refresh(settings_row)
self.db.refresh(secrets_row)
if admin_username and str(settings_row.admin_account or "").strip() in {"", "superadmin"}:
settings_row.admin_account = admin_username
should_commit = True
if self._sync_onlyoffice_defaults(settings_row, secrets_row):
should_commit = True
if should_commit:
self.db.commit()
self.db.refresh(settings_row)
self.db.refresh(secrets_row)
return settings_row, secrets_row
@@ -207,31 +221,42 @@ class SettingsService:
payload.llmForm.vlmEndpoint,
payload.llmForm.vlmApiKey,
)
self._apply_model_setting(
model_rows["embedding"],
payload.llmForm.embeddingProvider,
payload.llmForm.embeddingModel,
payload.llmForm.embeddingEndpoint,
payload.llmForm.embeddingApiKey,
)
settings_row.main_provider = model_rows["main"].provider
settings_row.main_model = model_rows["main"].model_name
settings_row.main_endpoint = model_rows["main"].endpoint
settings_row.backup_provider = model_rows["backup"].provider
self._apply_model_setting(
model_rows["embedding"],
payload.llmForm.embeddingProvider,
payload.llmForm.embeddingModel,
payload.llmForm.embeddingEndpoint,
payload.llmForm.embeddingApiKey,
)
if payload.renderForm.enabled and not payload.renderForm.publicUrl:
raise ValueError("启用 ONLYOFFICE 时必须配置服务地址。")
if (
payload.renderForm.enabled
and not payload.renderForm.jwtSecret
and not secrets_row.onlyoffice_jwt_secret_encrypted
):
raise ValueError("启用 ONLYOFFICE 时必须配置 JWT 密钥。")
settings_row.main_provider = model_rows["main"].provider
settings_row.main_model = model_rows["main"].model_name
settings_row.main_endpoint = model_rows["main"].endpoint
settings_row.backup_provider = model_rows["backup"].provider
settings_row.backup_model = model_rows["backup"].model_name
settings_row.backup_endpoint = model_rows["backup"].endpoint
settings_row.vlm_provider = model_rows["vlm"].provider
settings_row.vlm_model = model_rows["vlm"].model_name
settings_row.vlm_endpoint = model_rows["vlm"].endpoint
settings_row.embedding_provider = model_rows["embedding"].provider
settings_row.embedding_model = model_rows["embedding"].model_name
settings_row.embedding_endpoint = model_rows["embedding"].endpoint
settings_row.log_level = payload.logForm.level
settings_row.retention_days = payload.logForm.retentionDays
settings_row.archive_cycle = payload.logForm.archiveCycle
settings_row.log_path = payload.logForm.logPath
settings_row.vlm_endpoint = model_rows["vlm"].endpoint
settings_row.embedding_provider = model_rows["embedding"].provider
settings_row.embedding_model = model_rows["embedding"].model_name
settings_row.embedding_endpoint = model_rows["embedding"].endpoint
settings_row.onlyoffice_enabled = payload.renderForm.enabled
settings_row.onlyoffice_public_url = payload.renderForm.publicUrl
settings_row.log_level = payload.logForm.level
settings_row.retention_days = payload.logForm.retentionDays
settings_row.archive_cycle = payload.logForm.archiveCycle
settings_row.log_path = payload.logForm.logPath
settings_row.alert_email = payload.logForm.alertEmail
settings_row.operation_audit = payload.logForm.operationAudit
settings_row.login_audit = payload.logForm.loginAudit
@@ -243,12 +268,17 @@ class SettingsService:
settings_row.sender_name = payload.mailForm.senderName
settings_row.sender_address = payload.mailForm.senderAddress
settings_row.smtp_username = payload.mailForm.username
settings_row.alert_enabled = payload.mailForm.alertEnabled
settings_row.digest_enabled = payload.mailForm.digestEnabled
settings_row.digest_time = payload.mailForm.digestTime
settings_row.default_receiver = payload.mailForm.defaultReceiver
self._replace_secret_if_present(secrets_row, "smtp_password_encrypted", payload.mailForm.password)
settings_row.alert_enabled = payload.mailForm.alertEnabled
settings_row.digest_enabled = payload.mailForm.digestEnabled
settings_row.digest_time = payload.mailForm.digestTime
settings_row.default_receiver = payload.mailForm.defaultReceiver
self._replace_secret_if_present(
secrets_row,
"onlyoffice_jwt_secret_encrypted",
payload.renderForm.jwtSecret,
)
self._replace_secret_if_present(secrets_row, "smtp_password_encrypted", payload.mailForm.password)
self.db.add(settings_row)
self.db.add(secrets_row)
@@ -342,11 +372,11 @@ class SettingsService:
return AdminCredentialRecord(account=admin_username, email=admin_email, password_hash="")
def _build_default_settings(self) -> SystemSetting:
current_year = datetime.now().year
company_name = str(self.runtime_settings.company_name or "X-Financial").strip() or "X-Financial"
company_code = str(self.runtime_settings.company_code or "XF-001").strip() or "XF-001"
admin_email = str(self.runtime_settings.admin_email or "").strip()
def _build_default_settings(self) -> SystemSetting:
current_year = datetime.now().year
company_name = str(self.runtime_settings.company_name or "X-Financial").strip() or "X-Financial"
company_code = str(self.runtime_settings.company_code or "XF-001").strip() or "XF-001"
admin_email = str(self.runtime_settings.admin_email or "").strip()
legacy_admin = read_admin_secret() or {}
admin_account = str(legacy_admin.get("username", "")).strip() or "superadmin"
@@ -369,18 +399,20 @@ class SettingsService:
main_endpoint="https://api.openai.com/v1",
backup_provider="GLM",
backup_model="glm-5.1",
backup_endpoint="https://open.bigmodel.cn/api/paas/v4/",
vlm_provider="Gemini",
vlm_model="gemini-2.5-flash",
vlm_endpoint="https://generativelanguage.googleapis.com/v1beta/openai/",
embedding_provider="GLM",
embedding_model="Embedding-3",
embedding_endpoint="https://open.bigmodel.cn/api/paas/v4/",
log_level="INFO",
retention_days=180,
archive_cycle="weekly",
log_path="server/logs/app.log",
alert_email=admin_email,
backup_endpoint="https://open.bigmodel.cn/api/paas/v4/",
vlm_provider="Gemini",
vlm_model="gemini-2.5-flash",
vlm_endpoint="https://generativelanguage.googleapis.com/v1beta/openai/",
embedding_provider="GLM",
embedding_model="Embedding-3",
embedding_endpoint="https://open.bigmodel.cn/api/paas/v4/",
onlyoffice_enabled=bool(self.runtime_settings.onlyoffice_enabled),
onlyoffice_public_url=str(self.runtime_settings.onlyoffice_public_url or "").strip(),
log_level="INFO",
retention_days=180,
archive_cycle="weekly",
log_path="server/logs/app.log",
alert_email=admin_email,
operation_audit=True,
login_audit=True,
mask_sensitive=True,
@@ -405,11 +437,11 @@ class SettingsService:
setattr(secret_row, field_name, encrypt_secret(normalized))
@staticmethod
def _apply_model_setting(
model_row: SystemModelSetting,
provider: str,
model_name: str,
endpoint: str,
def _apply_model_setting(
model_row: SystemModelSetting,
provider: str,
model_name: str,
endpoint: str,
api_key: str,
) -> None:
model_row.provider = provider
@@ -417,13 +449,71 @@ class SettingsService:
model_row.endpoint = endpoint
normalized_api_key = api_key.strip()
if normalized_api_key:
model_row.api_key_encrypted = encrypt_secret(normalized_api_key)
@staticmethod
def _serialize(
settings_row: SystemSetting,
secrets_row: SystemSettingSecret,
if normalized_api_key:
model_row.api_key_encrypted = encrypt_secret(normalized_api_key)
def _ensure_settings_schema(self) -> None:
bind = self.db.get_bind()
inspector = inspect(bind)
table_names = set(inspector.get_table_names())
migration_statements: list[str] = []
if "system_settings" in table_names:
settings_columns = {column["name"] for column in inspector.get_columns("system_settings")}
if "onlyoffice_enabled" not in settings_columns:
migration_statements.append(
"ALTER TABLE system_settings ADD COLUMN onlyoffice_enabled BOOLEAN DEFAULT 0"
)
if "onlyoffice_public_url" not in settings_columns:
migration_statements.append(
"ALTER TABLE system_settings ADD COLUMN onlyoffice_public_url VARCHAR(512) DEFAULT ''"
)
if "system_setting_secrets" in table_names:
secret_columns = {column["name"] for column in inspector.get_columns("system_setting_secrets")}
if "onlyoffice_jwt_secret_encrypted" not in secret_columns:
migration_statements.append(
"ALTER TABLE system_setting_secrets ADD COLUMN onlyoffice_jwt_secret_encrypted TEXT DEFAULT ''"
)
for statement in migration_statements:
self.db.execute(text(statement))
if migration_statements:
self.db.commit()
def _sync_onlyoffice_defaults(
self,
settings_row: SystemSetting,
secrets_row: SystemSettingSecret,
) -> bool:
should_commit = False
runtime_public_url = str(self.runtime_settings.onlyoffice_public_url or "").strip()
runtime_jwt_secret = str(self.runtime_settings.onlyoffice_jwt_secret or "").strip()
if not str(settings_row.onlyoffice_public_url or "").strip() and runtime_public_url:
settings_row.onlyoffice_public_url = runtime_public_url
should_commit = True
if not secrets_row.onlyoffice_jwt_secret_encrypted and runtime_jwt_secret:
secrets_row.onlyoffice_jwt_secret_encrypted = encrypt_secret(runtime_jwt_secret)
should_commit = True
if (
not settings_row.onlyoffice_enabled
and self.runtime_settings.onlyoffice_enabled
and (runtime_public_url or runtime_jwt_secret)
):
settings_row.onlyoffice_enabled = True
should_commit = True
return should_commit
@staticmethod
def _serialize(
settings_row: SystemSetting,
secrets_row: SystemSettingSecret,
model_rows: dict[str, SystemModelSetting],
) -> SettingsRead:
main_model = model_rows["main"]
@@ -469,15 +559,21 @@ class SettingsService:
"vlmApiKeyConfigured": bool(vlm_model.api_key_encrypted),
"embeddingProvider": embedding_model.provider,
"embeddingModel": embedding_model.model_name,
"embeddingEndpoint": embedding_model.endpoint,
"embeddingApiKey": "",
"embeddingApiKeyConfigured": bool(embedding_model.api_key_encrypted),
},
logForm={
"level": settings_row.log_level,
"retentionDays": settings_row.retention_days,
"archiveCycle": settings_row.archive_cycle,
"logPath": settings_row.log_path,
"embeddingEndpoint": embedding_model.endpoint,
"embeddingApiKey": "",
"embeddingApiKeyConfigured": bool(embedding_model.api_key_encrypted),
},
renderForm={
"enabled": settings_row.onlyoffice_enabled,
"publicUrl": settings_row.onlyoffice_public_url,
"jwtSecret": "",
"jwtSecretConfigured": bool(secrets_row.onlyoffice_jwt_secret_encrypted),
},
logForm={
"level": settings_row.log_level,
"retentionDays": settings_row.retention_days,
"archiveCycle": settings_row.archive_cycle,
"logPath": settings_row.log_path,
"alertEmail": settings_row.alert_email,
"operationAudit": settings_row.operation_audit,
"loginAudit": settings_row.login_audit,
@@ -495,6 +591,41 @@ class SettingsService:
"alertEnabled": settings_row.alert_enabled,
"digestEnabled": settings_row.digest_enabled,
"digestTime": settings_row.digest_time,
"defaultReceiver": settings_row.default_receiver,
},
)
"defaultReceiver": settings_row.default_receiver,
},
)
def resolve_onlyoffice_settings(db: Session | None = None) -> OnlyOfficeRuntimeConfig:
runtime_settings = get_settings()
owned_session = False
session = db
try:
if session is None:
session_factory = get_session_factory()
session = session_factory()
owned_session = True
service = SettingsService(session)
settings_row, secrets_row = service.ensure_settings_ready()
jwt_secret = ""
if secrets_row.onlyoffice_jwt_secret_encrypted:
jwt_secret = decrypt_secret(secrets_row.onlyoffice_jwt_secret_encrypted)
return OnlyOfficeRuntimeConfig(
enabled=settings_row.onlyoffice_enabled,
public_url=str(settings_row.onlyoffice_public_url or "").strip(),
backend_url=str(runtime_settings.onlyoffice_backend_url or "").strip(),
jwt_secret=jwt_secret,
)
except Exception:
return OnlyOfficeRuntimeConfig(
enabled=bool(runtime_settings.onlyoffice_enabled),
public_url=str(runtime_settings.onlyoffice_public_url or "").strip(),
backend_url=str(runtime_settings.onlyoffice_backend_url or "").strip(),
jwt_secret=str(runtime_settings.onlyoffice_jwt_secret or "").strip(),
)
finally:
if owned_session and session is not None:
session.close()

View File

@@ -67,4 +67,4 @@ tests/test_employee_service.py
tests/test_imports.py
tests/test_server_start_dependencies.py
tests/test_settings_persistence.py
tests/test_settings_service.py
tests/test_settings_service.py

View File

@@ -1,8 +1,31 @@
from __future__ import annotations
from pathlib import Path
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.api.deps import CurrentUserContext
from app.core.config import Settings, get_settings
from app.core import secret_box
from app.db.base import Base
from app.models.system_model_setting import SystemModelSetting
from app.models.system_setting import SystemSetting
from app.models.system_setting_secret import SystemSettingSecret
from app.schemas.settings import SettingsWrite
from app.services.knowledge import KnowledgeService
from app.services.settings import SettingsService
def build_session_factory(db_file: Path):
engine = create_engine(
f"sqlite+pysqlite:///{db_file.as_posix()}",
connect_args={"check_same_thread": False},
)
SystemSetting.__table__.create(bind=engine)
SystemSettingSecret.__table__.create(bind=engine)
SystemModelSetting.__table__.create(bind=engine)
return sessionmaker(bind=engine, autoflush=False, autocommit=False)
def test_onlyoffice_config_is_read_only_for_admin_users(tmp_path, monkeypatch) -> None:
@@ -53,3 +76,61 @@ def test_onlyoffice_config_is_read_only_for_admin_users(tmp_path, monkeypatch) -
finally:
monkeypatch.setitem(Settings.model_config, "env_file", original_env_file)
get_settings.cache_clear()
def test_onlyoffice_config_prefers_saved_settings_snapshot(tmp_path, monkeypatch) -> None:
env_file = tmp_path / ".env"
env_file.write_text(
"\n".join(
[
"ADMIN_EMAIL=admin@example.com",
"ONLYOFFICE_ENABLED=false",
"ONLYOFFICE_BACKEND_URL=http://main:8000",
]
)
+ "\n",
encoding="utf-8",
)
original_env_file = Settings.model_config.get("env_file")
monkeypatch.setitem(Settings.model_config, "env_file", (env_file,))
monkeypatch.setattr(secret_box, "SECRET_KEY_FILE", tmp_path / "settings.key")
monkeypatch.setattr(Base.metadata, "create_all", lambda *args, **kwargs: None)
get_settings.cache_clear()
session_factory = build_session_factory(tmp_path / "settings.db")
monkeypatch.setattr("app.services.settings.get_session_factory", lambda: session_factory)
try:
with session_factory() as db:
service = SettingsService(db)
payload = service.get_settings_snapshot().model_dump()
payload["renderForm"]["enabled"] = True
payload["renderForm"]["publicUrl"] = "http://10.10.10.122:8082"
payload["renderForm"]["jwtSecret"] = "change-me-onlyoffice"
service.save_settings_snapshot(SettingsWrite(**payload))
service = KnowledgeService(storage_root=tmp_path)
service.ensure_library_ready()
document_id = "db-backed-docx"
folder = "制度政策"
stored_name = f"{document_id}__制度预览.docx"
target_path = tmp_path / "knowledge" / folder / stored_name
target_path.write_bytes(b"fake-docx-content")
current_user = CurrentUserContext(
username="admin",
name="管理员",
role_codes=["manager"],
is_admin=True,
)
config = service.build_onlyoffice_config(document_id, current_user)
assert config.documentServerUrl == "http://10.10.10.122:8082"
assert config.config["document"]["url"].startswith(
"http://main:8000/api/v1/knowledge/documents/db-backed-docx/onlyoffice/content?access_token="
)
finally:
monkeypatch.setitem(Settings.model_config, "env_file", original_env_file)
get_settings.cache_clear()

View File

@@ -35,7 +35,7 @@ def build_temp_secret_dir() -> Path:
return Path(tempfile.mkdtemp(prefix="xf-settings-test-"))
def test_settings_service_persists_non_secret_and_secret_fields(monkeypatch) -> None:
def test_settings_service_persists_non_secret_and_secret_fields(monkeypatch) -> None:
temp_dir = build_temp_secret_dir()
monkeypatch.setattr(secret_box, "SECRET_KEY_FILE", temp_dir / "settings.key")
monkeypatch.setattr(Base.metadata, "create_all", lambda *args, **kwargs: None)
@@ -51,30 +51,44 @@ def test_settings_service_persists_non_secret_and_secret_fields(monkeypatch) ->
payload["adminForm"]["adminEmail"] = "admin@example.com"
payload["adminForm"]["newPassword"] = "54321"
payload["adminForm"]["confirmPassword"] = "54321"
payload["llmForm"]["mainModel"] = "glm-4.5"
payload["llmForm"]["mainApiKey"] = "main-secret"
payload["mailForm"]["password"] = "smtp-secret"
saved_snapshot = service.save_settings_snapshot(SettingsWrite(**payload))
payload["llmForm"]["mainModel"] = "glm-4.5"
payload["llmForm"]["mainApiKey"] = "main-secret"
payload["renderForm"]["enabled"] = True
payload["renderForm"]["publicUrl"] = "http://10.10.10.122:8082"
payload["renderForm"]["jwtSecret"] = "change-me-onlyoffice"
payload["mailForm"]["password"] = "smtp-secret"
saved_snapshot = service.save_settings_snapshot(SettingsWrite(**payload))
assert saved_snapshot.companyForm.companyName == "YGSOFT"
assert saved_snapshot.companyForm.displayName == "云广软件"
assert saved_snapshot.llmForm.mainModel == "glm-4.5"
assert saved_snapshot.llmForm.mainApiKey == ""
assert saved_snapshot.llmForm.mainApiKeyConfigured is True
assert saved_snapshot.mailForm.password == ""
assert saved_snapshot.mailForm.passwordConfigured is True
assert saved_snapshot.adminForm.newPassword == ""
assert saved_snapshot.llmForm.mainModel == "glm-4.5"
assert saved_snapshot.llmForm.mainApiKey == ""
assert saved_snapshot.llmForm.mainApiKeyConfigured is True
assert saved_snapshot.renderForm.enabled is True
assert saved_snapshot.renderForm.publicUrl == "http://10.10.10.122:8082"
assert saved_snapshot.renderForm.jwtSecret == ""
assert saved_snapshot.renderForm.jwtSecretConfigured is True
assert saved_snapshot.mailForm.password == ""
assert saved_snapshot.mailForm.passwordConfigured is True
assert saved_snapshot.adminForm.newPassword == ""
assert saved_snapshot.adminForm.adminPasswordConfigured is True
model_row = db.get(SystemModelSetting, "main")
assert model_row is not None
assert model_row.model_name == "glm-4.5"
assert model_row.api_key_encrypted
assert service.load_saved_model_api_key("main") == "main-secret"
assert service.verify_admin_login("admin-root", "54321") is not None
assert service.verify_admin_login("admin@example.com", "54321") is not None
model_row = db.get(SystemModelSetting, "main")
settings_row = db.get(SystemSetting, "default")
secrets_row = db.get(SystemSettingSecret, "default")
assert model_row is not None
assert model_row.model_name == "glm-4.5"
assert model_row.api_key_encrypted
assert settings_row is not None
assert settings_row.onlyoffice_enabled is True
assert settings_row.onlyoffice_public_url == "http://10.10.10.122:8082"
assert secrets_row is not None
assert secrets_row.onlyoffice_jwt_secret_encrypted
assert service.load_saved_model_api_key("main") == "main-secret"
assert service.verify_admin_login("admin-root", "54321") is not None
assert service.verify_admin_login("admin@example.com", "54321") is not None
def test_blank_secret_input_does_not_clear_saved_secret(monkeypatch) -> None: