139 lines
4.6 KiB
Python
139 lines
4.6 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
|
|
from app.config import Settings, get_settings
|
|
from app.runtime_settings import (
|
|
EncryptedSettingsStore,
|
|
RuntimeSettingsResponse,
|
|
RuntimeSettingsTestRequest,
|
|
RuntimeSettingsTestResult,
|
|
RuntimeSettingsUpdate,
|
|
SecretGenerateRequest,
|
|
SecretGenerateResponse,
|
|
SettingsReadiness,
|
|
generate_secret,
|
|
get_runtime_store,
|
|
test_image_generation,
|
|
test_model_pool_item,
|
|
test_runtime_settings,
|
|
)
|
|
|
|
router = APIRouter(prefix="/v1")
|
|
|
|
|
|
@router.get("/settings", response_model=RuntimeSettingsResponse)
|
|
def read_settings(
|
|
store: EncryptedSettingsStore = Depends(get_runtime_store),
|
|
) -> RuntimeSettingsResponse:
|
|
try:
|
|
return store.public_response()
|
|
except (RuntimeError, ValueError) as exc:
|
|
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) from exc
|
|
|
|
|
|
@router.put("/settings", response_model=RuntimeSettingsResponse)
|
|
def update_settings(
|
|
request: RuntimeSettingsUpdate,
|
|
store: EncryptedSettingsStore = Depends(get_runtime_store),
|
|
) -> RuntimeSettingsResponse:
|
|
try:
|
|
store.update(request.values)
|
|
return store.public_response()
|
|
except RuntimeError as exc:
|
|
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) from exc
|
|
|
|
|
|
@router.post("/settings/test", response_model=RuntimeSettingsTestResult)
|
|
async def test_settings(
|
|
request: RuntimeSettingsTestRequest,
|
|
store: EncryptedSettingsStore = Depends(get_runtime_store),
|
|
bootstrap: Settings = Depends(get_settings),
|
|
) -> RuntimeSettingsTestResult:
|
|
try:
|
|
values = store.merged_values(request.values)
|
|
result = await test_runtime_settings(request.target, values, bootstrap)
|
|
if not request.values:
|
|
store.record_test(result, values)
|
|
return result
|
|
except (KeyError, RuntimeError, ValueError) as exc:
|
|
return RuntimeSettingsTestResult(
|
|
target=request.target,
|
|
ok=False,
|
|
message=f"配置不完整：{exc}",
|
|
)
|
|
|
|
|
|
@router.post("/settings/models/{model_id}/test", response_model=RuntimeSettingsTestResult)
|
|
async def test_model(
|
|
model_id: str,
|
|
store: EncryptedSettingsStore = Depends(get_runtime_store),
|
|
) -> RuntimeSettingsTestResult:
|
|
try:
|
|
values = store.merged_values()
|
|
result = await test_model_pool_item(values, model_id)
|
|
store.record_test(result, values)
|
|
return result
|
|
except (RuntimeError, ValueError) as exc:
|
|
return RuntimeSettingsTestResult(
|
|
target=f"model:{model_id}",
|
|
ok=False,
|
|
message=f"模型测试失败：{exc}",
|
|
)
|
|
|
|
|
|
@router.post("/settings/models/{model_id}/test-generation", response_model=RuntimeSettingsTestResult)
|
|
async def test_model_generation(
|
|
model_id: str,
|
|
store: EncryptedSettingsStore = Depends(get_runtime_store),
|
|
) -> RuntimeSettingsTestResult:
|
|
try:
|
|
values = store.merged_values()
|
|
result = await test_image_generation(values, model_id)
|
|
store.record_test(result, values)
|
|
definitive_result = RuntimeSettingsTestResult(
|
|
target=f"model:{model_id}",
|
|
ok=result.ok,
|
|
message=(
|
|
result.message
|
|
if not result.ok
|
|
else "已通过一次真实生图端到端验证；模型、鉴权、请求参数、任务轮询和结果读取均可用。"
|
|
),
|
|
details=result.details,
|
|
)
|
|
store.record_test(definitive_result, values)
|
|
return result
|
|
except (RuntimeError, ValueError) as exc:
|
|
return RuntimeSettingsTestResult(
|
|
target=f"generation:{model_id}",
|
|
ok=False,
|
|
message=f"真实生图失败：{exc}",
|
|
)
|
|
|
|
|
|
@router.post("/settings/generate-secret", response_model=SecretGenerateResponse)
|
|
def create_secret(request: SecretGenerateRequest) -> SecretGenerateResponse:
|
|
return SecretGenerateResponse(value=generate_secret(request.kind))
|
|
|
|
|
|
@router.get("/readiness", response_model=SettingsReadiness)
|
|
def readiness(
|
|
store: EncryptedSettingsStore = Depends(get_runtime_store),
|
|
) -> SettingsReadiness:
|
|
return store.public_response().readiness
|
|
|
|
|
|
def require_workflow_ready(
|
|
store: EncryptedSettingsStore = Depends(get_runtime_store),
|
|
) -> None:
|
|
current = store.public_response().readiness
|
|
if current.ready:
|
|
return
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail={
|
|
"code": "SETTINGS_INCOMPLETE",
|
|
"message": "请先完成系统设置并通过必备连接测试。",
|
|
"missing": current.missing,
|
|
"untested": current.untested,
|
|
},
|
|
)
|