1760 lines
63 KiB
Rust
1760 lines
63 KiB
Rust
use std::{
|
||
collections::BTreeMap,
|
||
convert::Infallible,
|
||
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
|
||
};
|
||
|
||
use axum::{
|
||
Json,
|
||
extract::{Extension, Path, State, rejection::JsonRejection},
|
||
http::StatusCode,
|
||
response::{
|
||
IntoResponse, Response,
|
||
sse::{Event, Sse},
|
||
},
|
||
};
|
||
use module_assets::{
|
||
AssetObjectAccessPolicy, AssetObjectFieldError, build_asset_entity_binding_input,
|
||
build_asset_object_upsert_input, generate_asset_binding_id, generate_asset_object_id,
|
||
};
|
||
use platform_oss::{LegacyAssetPrefix, OssHeadObjectRequest, OssObjectAccess, OssPutObjectRequest};
|
||
use serde_json::{Map, Value, json};
|
||
use shared_contracts::big_fish::{
|
||
BigFishActionResponse, BigFishAgentMessageResponse, BigFishAnchorItemResponse,
|
||
BigFishAnchorPackResponse, BigFishAssetCoverageResponse, BigFishAssetSlotResponse,
|
||
BigFishBackgroundBlueprintResponse, BigFishGameDraftResponse, BigFishLevelBlueprintResponse,
|
||
BigFishRuntimeParamsResponse, BigFishSessionResponse, BigFishSessionSnapshotResponse,
|
||
CreateBigFishSessionRequest, ExecuteBigFishActionRequest, SendBigFishMessageRequest,
|
||
};
|
||
use shared_contracts::big_fish_works::{BigFishWorkSummaryResponse, BigFishWorksResponse};
|
||
use shared_kernel::{build_prefixed_uuid_id, format_timestamp_micros};
|
||
use spacetime_client::{
|
||
BigFishAgentMessageRecord, BigFishAnchorItemRecord, BigFishAnchorPackRecord,
|
||
BigFishAssetCoverageRecord, BigFishAssetGenerateRecordInput, BigFishAssetSlotRecord,
|
||
BigFishBackgroundBlueprintRecord, BigFishDraftCompileRecordInput, BigFishGameDraftRecord,
|
||
BigFishLevelBlueprintRecord, BigFishMessageSubmitRecordInput, BigFishRuntimeParamsRecord,
|
||
BigFishSessionCreateRecordInput, BigFishSessionRecord, BigFishWorkSummaryRecord,
|
||
SpacetimeClientError,
|
||
};
|
||
use tokio::time::sleep;
|
||
|
||
use crate::big_fish_agent_turn::{
|
||
BigFishAgentTurnRequest, build_failed_finalize_record_input, build_finalize_record_input,
|
||
run_big_fish_agent_turn,
|
||
};
|
||
use crate::big_fish_draft_compiler::compile_big_fish_draft_with_fallback;
|
||
use crate::prompt::big_fish::{
|
||
BIG_FISH_DEFAULT_NEGATIVE_PROMPT, BIG_FISH_TRANSPARENT_ASSET_NEGATIVE_PROMPT,
|
||
build_big_fish_level_main_image_prompt, build_big_fish_level_motion_prompt,
|
||
build_big_fish_stage_background_prompt,
|
||
};
|
||
use crate::{
|
||
ai_generation_drafts::{
|
||
AiGenerationDraftContext, AiGenerationDraftSink, AiGenerationDraftWriter,
|
||
},
|
||
api_response::json_success_body,
|
||
asset_billing::{consume_asset_operation_points, refund_asset_operation_points},
|
||
auth::AuthenticatedAccessToken,
|
||
character_visual_assets::try_apply_background_alpha_to_png,
|
||
http_error::AppError,
|
||
request_context::RequestContext,
|
||
state::AppState,
|
||
};
|
||
|
||
pub async fn create_big_fish_session(
|
||
State(state): State<AppState>,
|
||
Extension(request_context): Extension<RequestContext>,
|
||
Extension(authenticated): Extension<AuthenticatedAccessToken>,
|
||
payload: Result<Json<CreateBigFishSessionRequest>, JsonRejection>,
|
||
) -> Result<Json<Value>, Response> {
|
||
let Json(payload) = payload.map_err(|error| {
|
||
big_fish_error_response(
|
||
&request_context,
|
||
AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
|
||
"provider": "big-fish",
|
||
"message": error.body_text(),
|
||
})),
|
||
)
|
||
})?;
|
||
|
||
let seed_text = payload.seed_text.unwrap_or_default().trim().to_string();
|
||
let session = state
|
||
.spacetime_client()
|
||
.create_big_fish_session(BigFishSessionCreateRecordInput {
|
||
session_id: build_prefixed_uuid_id("big-fish-session-"),
|
||
owner_user_id: authenticated.claims().user_id().to_string(),
|
||
seed_text: seed_text.clone(),
|
||
welcome_message_id: build_prefixed_uuid_id("big-fish-message-"),
|
||
welcome_message_text: build_big_fish_welcome_text(&seed_text),
|
||
created_at_micros: current_utc_micros(),
|
||
})
|
||
.await
|
||
.map_err(|error| {
|
||
big_fish_error_response(&request_context, map_big_fish_client_error(error))
|
||
})?;
|
||
|
||
Ok(json_success_body(
|
||
Some(&request_context),
|
||
BigFishSessionResponse {
|
||
session: map_big_fish_session_response(session),
|
||
},
|
||
))
|
||
}
|
||
|
||
pub async fn get_big_fish_session(
|
||
State(state): State<AppState>,
|
||
Path(session_id): Path<String>,
|
||
Extension(request_context): Extension<RequestContext>,
|
||
Extension(authenticated): Extension<AuthenticatedAccessToken>,
|
||
) -> Result<Json<Value>, Response> {
|
||
ensure_non_empty(&request_context, &session_id, "sessionId")?;
|
||
|
||
let session = load_big_fish_session_with_retry(
|
||
&state,
|
||
session_id,
|
||
authenticated.claims().user_id().to_string(),
|
||
)
|
||
.await
|
||
.map_err(|error| big_fish_error_response(&request_context, map_big_fish_client_error(error)))?;
|
||
|
||
Ok(json_success_body(
|
||
Some(&request_context),
|
||
BigFishSessionResponse {
|
||
session: map_big_fish_session_response(session),
|
||
},
|
||
))
|
||
}
|
||
|
||
pub async fn get_big_fish_works(
|
||
State(state): State<AppState>,
|
||
Extension(request_context): Extension<RequestContext>,
|
||
Extension(authenticated): Extension<AuthenticatedAccessToken>,
|
||
) -> Result<Json<Value>, Response> {
|
||
let items = state
|
||
.spacetime_client()
|
||
.list_big_fish_works(authenticated.claims().user_id().to_string())
|
||
.await
|
||
.map_err(|error| {
|
||
big_fish_error_response(&request_context, map_big_fish_client_error(error))
|
||
})?;
|
||
|
||
Ok(json_success_body(
|
||
Some(&request_context),
|
||
BigFishWorksResponse {
|
||
items: items
|
||
.into_iter()
|
||
.map(map_big_fish_work_summary_response)
|
||
.collect(),
|
||
},
|
||
))
|
||
}
|
||
|
||
pub async fn list_big_fish_gallery(
|
||
State(state): State<AppState>,
|
||
Extension(request_context): Extension<RequestContext>,
|
||
) -> Result<Json<Value>, Response> {
|
||
let items = match state.spacetime_client().list_big_fish_gallery().await {
|
||
Ok(items) => items,
|
||
Err(error) if should_soft_fallback_big_fish_gallery(&error) => {
|
||
tracing::warn!(
|
||
error = %error,
|
||
"大鱼吃小鱼公开广场读取失败，已按空广场降级返回"
|
||
);
|
||
Vec::new()
|
||
}
|
||
Err(error) => {
|
||
return Err(big_fish_error_response(
|
||
&request_context,
|
||
map_big_fish_client_error(error),
|
||
));
|
||
}
|
||
};
|
||
|
||
Ok(json_success_body(
|
||
Some(&request_context),
|
||
BigFishWorksResponse {
|
||
items: items
|
||
.into_iter()
|
||
.map(map_big_fish_work_summary_response)
|
||
.collect(),
|
||
},
|
||
))
|
||
}
|
||
|
||
pub async fn delete_big_fish_work(
|
||
State(state): State<AppState>,
|
||
Path(session_id): Path<String>,
|
||
Extension(request_context): Extension<RequestContext>,
|
||
Extension(authenticated): Extension<AuthenticatedAccessToken>,
|
||
) -> Result<Json<Value>, Response> {
|
||
ensure_non_empty(&request_context, &session_id, "sessionId")?;
|
||
|
||
let items = state
|
||
.spacetime_client()
|
||
.delete_big_fish_work(session_id, authenticated.claims().user_id().to_string())
|
||
.await
|
||
.map_err(|error| {
|
||
big_fish_error_response(&request_context, map_big_fish_client_error(error))
|
||
})?;
|
||
|
||
Ok(json_success_body(
|
||
Some(&request_context),
|
||
BigFishWorksResponse {
|
||
items: items
|
||
.into_iter()
|
||
.map(map_big_fish_work_summary_response)
|
||
.collect(),
|
||
},
|
||
))
|
||
}
|
||
|
||
pub async fn submit_big_fish_message(
|
||
State(state): State<AppState>,
|
||
Path(session_id): Path<String>,
|
||
Extension(request_context): Extension<RequestContext>,
|
||
Extension(authenticated): Extension<AuthenticatedAccessToken>,
|
||
payload: Result<Json<SendBigFishMessageRequest>, JsonRejection>,
|
||
) -> Result<Json<Value>, Response> {
|
||
let Json(payload) = payload.map_err(|error| {
|
||
big_fish_error_response(
|
||
&request_context,
|
||
AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
|
||
"provider": "big-fish",
|
||
"message": error.body_text(),
|
||
})),
|
||
)
|
||
})?;
|
||
ensure_non_empty(&request_context, &session_id, "sessionId")?;
|
||
|
||
let client_message_id = payload.client_message_id.trim().to_string();
|
||
let message_text = payload.text.trim().to_string();
|
||
if client_message_id.is_empty() || message_text.is_empty() {
|
||
return Err(big_fish_bad_request(
|
||
&request_context,
|
||
"clientMessageId and text are required",
|
||
));
|
||
}
|
||
|
||
let owner_user_id = authenticated.claims().user_id().to_string();
|
||
let submitted_session = state
|
||
.spacetime_client()
|
||
.submit_big_fish_message(BigFishMessageSubmitRecordInput {
|
||
session_id: session_id.clone(),
|
||
owner_user_id: owner_user_id.clone(),
|
||
user_message_id: client_message_id,
|
||
user_message_text: message_text,
|
||
assistant_message_id: build_prefixed_uuid_id("big-fish-message-"),
|
||
submitted_at_micros: current_utc_micros(),
|
||
})
|
||
.await
|
||
.map_err(|error| {
|
||
big_fish_error_response(&request_context, map_big_fish_client_error(error))
|
||
})?;
|
||
let mut draft_writer = AiGenerationDraftWriter::new(AiGenerationDraftContext::new(
|
||
"big_fish",
|
||
owner_user_id.as_str(),
|
||
session_id.as_str(),
|
||
payload.client_message_id.as_str(),
|
||
"大鱼吃小鱼模板生成草稿",
|
||
));
|
||
if let Err(error) = draft_writer.ensure_started(state.spacetime_client()).await {
|
||
tracing::warn!(error = %error, "大鱼吃小鱼模板生成草稿任务启动失败，主生成流程继续执行");
|
||
}
|
||
let draft_sink = AiGenerationDraftSink::new(
|
||
AiGenerationDraftContext::new(
|
||
"big_fish",
|
||
owner_user_id.as_str(),
|
||
session_id.as_str(),
|
||
payload.client_message_id.as_str(),
|
||
"大鱼吃小鱼模板生成草稿",
|
||
),
|
||
state.spacetime_client().clone(),
|
||
);
|
||
let turn_result = run_big_fish_agent_turn(
|
||
BigFishAgentTurnRequest {
|
||
llm_client: state.llm_client(),
|
||
session: &submitted_session,
|
||
quick_fill_requested: payload.quick_fill_requested.unwrap_or(false),
|
||
enable_web_search: state.config.creation_agent_llm_web_search_enabled,
|
||
},
|
||
move |text| {
|
||
draft_sink.persist_visible_text_async(text);
|
||
},
|
||
)
|
||
.await;
|
||
if let Ok(result) = &turn_result {
|
||
draft_writer
|
||
.persist_visible_text(
|
||
state.spacetime_client(),
|
||
result.assistant_reply_text.as_str(),
|
||
)
|
||
.await;
|
||
}
|
||
let finalize_input = match turn_result {
|
||
Ok(turn_result) => build_finalize_record_input(
|
||
session_id.clone(),
|
||
owner_user_id.clone(),
|
||
build_prefixed_uuid_id("big-fish-message-"),
|
||
turn_result,
|
||
current_utc_micros(),
|
||
),
|
||
Err(error) => build_failed_finalize_record_input(
|
||
session_id.clone(),
|
||
owner_user_id.clone(),
|
||
&submitted_session,
|
||
error.to_string(),
|
||
current_utc_micros(),
|
||
),
|
||
};
|
||
let session = state
|
||
.spacetime_client()
|
||
.finalize_big_fish_agent_message(finalize_input)
|
||
.await
|
||
.map_err(|error| {
|
||
big_fish_error_response(&request_context, map_big_fish_client_error(error))
|
||
})?;
|
||
|
||
Ok(json_success_body(
|
||
Some(&request_context),
|
||
BigFishSessionResponse {
|
||
session: map_big_fish_session_response(session),
|
||
},
|
||
))
|
||
}
|
||
|
||
pub async fn stream_big_fish_message(
|
||
State(state): State<AppState>,
|
||
Path(session_id): Path<String>,
|
||
Extension(request_context): Extension<RequestContext>,
|
||
Extension(authenticated): Extension<AuthenticatedAccessToken>,
|
||
payload: Result<Json<SendBigFishMessageRequest>, JsonRejection>,
|
||
) -> Result<Response, Response> {
|
||
let Json(payload) = payload.map_err(|error| {
|
||
big_fish_error_response(
|
||
&request_context,
|
||
AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
|
||
"provider": "big-fish",
|
||
"message": error.body_text(),
|
||
})),
|
||
)
|
||
})?;
|
||
ensure_non_empty(&request_context, &session_id, "sessionId")?;
|
||
|
||
let client_message_id = payload.client_message_id.trim().to_string();
|
||
let message_text = payload.text.trim().to_string();
|
||
if client_message_id.is_empty() || message_text.is_empty() {
|
||
return Err(big_fish_bad_request(
|
||
&request_context,
|
||
"clientMessageId and text are required",
|
||
));
|
||
}
|
||
|
||
let owner_user_id = authenticated.claims().user_id().to_string();
|
||
let submitted_session = state
|
||
.spacetime_client()
|
||
.submit_big_fish_message(BigFishMessageSubmitRecordInput {
|
||
session_id: session_id.clone(),
|
||
owner_user_id: owner_user_id.clone(),
|
||
user_message_id: client_message_id,
|
||
user_message_text: message_text,
|
||
assistant_message_id: build_prefixed_uuid_id("big-fish-message-"),
|
||
submitted_at_micros: current_utc_micros(),
|
||
})
|
||
.await
|
||
.map_err(|error| {
|
||
big_fish_error_response(&request_context, map_big_fish_client_error(error))
|
||
})?;
|
||
let quick_fill_requested = payload.quick_fill_requested.unwrap_or(false);
|
||
let state = state.clone();
|
||
let session_id_for_stream = session_id.clone();
|
||
let owner_user_id_for_stream = owner_user_id.clone();
|
||
let client_message_id_for_stream = payload.client_message_id.clone();
|
||
let stream = async_stream::stream! {
|
||
let mut draft_writer = AiGenerationDraftWriter::new(AiGenerationDraftContext::new(
|
||
"big_fish",
|
||
owner_user_id_for_stream.as_str(),
|
||
session_id_for_stream.as_str(),
|
||
client_message_id_for_stream.as_str(),
|
||
"大鱼吃小鱼模板生成草稿",
|
||
));
|
||
if let Err(error) = draft_writer.ensure_started(state.spacetime_client()).await {
|
||
tracing::warn!(error = %error, "大鱼吃小鱼模板生成草稿任务启动失败，主生成流程继续执行");
|
||
}
|
||
let (reply_tx, mut reply_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
|
||
// 与 RPG/拼图 Agent 保持同一语义：回复先流式展示，session 真相仍等 finalize 后下发。
|
||
let turn_result = {
|
||
let run_turn = run_big_fish_agent_turn(
|
||
BigFishAgentTurnRequest {
|
||
llm_client: state.llm_client(),
|
||
session: &submitted_session,
|
||
quick_fill_requested,
|
||
enable_web_search: state.config.creation_agent_llm_web_search_enabled,
|
||
},
|
||
move |text| {
|
||
let _ = reply_tx.send(text.to_string());
|
||
},
|
||
);
|
||
tokio::pin!(run_turn);
|
||
|
||
loop {
|
||
// 每个 replyText 增量同时写草稿表并推给 SSE，避免前端等待完整模型响应。
|
||
tokio::select! {
|
||
result = &mut run_turn => break result,
|
||
maybe_text = reply_rx.recv() => {
|
||
if let Some(text) = maybe_text {
|
||
draft_writer
|
||
.persist_visible_text(state.spacetime_client(), text.as_str())
|
||
.await;
|
||
yield Ok::<Event, Infallible>(big_fish_sse_json_event_or_error(
|
||
"reply_delta",
|
||
json!({ "text": text }),
|
||
));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
};
|
||
|
||
while let Some(text) = reply_rx.recv().await {
|
||
draft_writer
|
||
.persist_visible_text(state.spacetime_client(), text.as_str())
|
||
.await;
|
||
yield Ok::<Event, Infallible>(big_fish_sse_json_event_or_error(
|
||
"reply_delta",
|
||
json!({ "text": text }),
|
||
));
|
||
}
|
||
|
||
let finalize_input = match turn_result {
|
||
Ok(turn_result) => build_finalize_record_input(
|
||
session_id_for_stream.clone(),
|
||
owner_user_id_for_stream.clone(),
|
||
build_prefixed_uuid_id("big-fish-message-"),
|
||
turn_result,
|
||
current_utc_micros(),
|
||
),
|
||
Err(error) => build_failed_finalize_record_input(
|
||
session_id_for_stream.clone(),
|
||
owner_user_id_for_stream.clone(),
|
||
&submitted_session,
|
||
error.to_string(),
|
||
current_utc_micros(),
|
||
),
|
||
};
|
||
let session = match state
|
||
.spacetime_client()
|
||
.finalize_big_fish_agent_message(finalize_input)
|
||
.await
|
||
{
|
||
Ok(session) => session,
|
||
Err(error) => {
|
||
yield Ok::<Event, Infallible>(big_fish_sse_json_event_or_error(
|
||
"error",
|
||
json!({ "message": error.to_string() }),
|
||
));
|
||
return;
|
||
}
|
||
};
|
||
|
||
let session_response = map_big_fish_session_response(session);
|
||
yield Ok::<Event, Infallible>(big_fish_sse_json_event_or_error(
|
||
"session",
|
||
json!({ "session": session_response }),
|
||
));
|
||
yield Ok::<Event, Infallible>(big_fish_sse_json_event_or_error(
|
||
"done",
|
||
json!({ "ok": true }),
|
||
));
|
||
};
|
||
Ok(Sse::new(stream).into_response())
|
||
}
|
||
|
||
pub async fn execute_big_fish_action(
|
||
State(state): State<AppState>,
|
||
Path(session_id): Path<String>,
|
||
Extension(request_context): Extension<RequestContext>,
|
||
Extension(authenticated): Extension<AuthenticatedAccessToken>,
|
||
payload: Result<Json<ExecuteBigFishActionRequest>, JsonRejection>,
|
||
) -> Result<Json<Value>, Response> {
|
||
let Json(payload) = payload.map_err(|error| {
|
||
big_fish_error_response(
|
||
&request_context,
|
||
AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
|
||
"provider": "big-fish",
|
||
"message": error.body_text(),
|
||
})),
|
||
)
|
||
})?;
|
||
ensure_non_empty(&request_context, &session_id, "sessionId")?;
|
||
|
||
let owner_user_id = authenticated.claims().user_id().to_string();
|
||
let now = current_utc_micros();
|
||
let action = payload.action.trim().to_string();
|
||
let billed_asset_kind = match action.as_str() {
|
||
"big_fish_generate_level_main_image" => Some("big_fish_level_main_image"),
|
||
"big_fish_generate_level_motion" => Some("big_fish_level_motion"),
|
||
"big_fish_generate_stage_background" => Some("big_fish_stage_background"),
|
||
"big_fish_publish_game" => Some("big_fish_publish_game"),
|
||
_ => None,
|
||
};
|
||
let billing_asset_id = format!("{session_id}:{now}");
|
||
if let Some(asset_kind) = billed_asset_kind {
|
||
consume_asset_operation_points(&state, &owner_user_id, asset_kind, &billing_asset_id)
|
||
.await
|
||
.map_err(|error| big_fish_error_response(&request_context, error))?;
|
||
}
|
||
|
||
let session_result = match action.as_str() {
|
||
"big_fish_compile_draft" => {
|
||
compile_big_fish_draft_only(&state, session_id.clone(), owner_user_id.clone(), now)
|
||
.await
|
||
}
|
||
"big_fish_generate_level_main_image" => {
|
||
let asset_url = generate_big_fish_formal_asset(
|
||
&state,
|
||
&owner_user_id,
|
||
&session_id,
|
||
"level_main_image",
|
||
payload.level,
|
||
None,
|
||
now,
|
||
)
|
||
.await
|
||
.map_err(|error| {
|
||
if let Some(asset_kind) = billed_asset_kind {
|
||
tokio::spawn({
|
||
let state = state.clone();
|
||
let owner_user_id = owner_user_id.clone();
|
||
let billing_asset_id = billing_asset_id.clone();
|
||
async move {
|
||
refund_asset_operation_points(
|
||
&state,
|
||
&owner_user_id,
|
||
asset_kind,
|
||
&billing_asset_id,
|
||
)
|
||
.await;
|
||
}
|
||
});
|
||
}
|
||
big_fish_error_response(&request_context, error)
|
||
})?;
|
||
state
|
||
.spacetime_client()
|
||
.generate_big_fish_asset(BigFishAssetGenerateRecordInput {
|
||
owner_user_id: owner_user_id.clone(),
|
||
session_id: session_id.clone(),
|
||
asset_kind: "level_main_image".to_string(),
|
||
level: payload.level,
|
||
motion_key: None,
|
||
asset_url: Some(asset_url),
|
||
generated_at_micros: now,
|
||
})
|
||
.await
|
||
}
|
||
"big_fish_generate_level_motion" => {
|
||
let asset_url = generate_big_fish_formal_asset(
|
||
&state,
|
||
&owner_user_id,
|
||
&session_id,
|
||
"level_motion",
|
||
payload.level,
|
||
payload.motion_key.as_deref(),
|
||
now,
|
||
)
|
||
.await
|
||
.map_err(|error| {
|
||
if let Some(asset_kind) = billed_asset_kind {
|
||
tokio::spawn({
|
||
let state = state.clone();
|
||
let owner_user_id = owner_user_id.clone();
|
||
let billing_asset_id = billing_asset_id.clone();
|
||
async move {
|
||
refund_asset_operation_points(
|
||
&state,
|
||
&owner_user_id,
|
||
asset_kind,
|
||
&billing_asset_id,
|
||
)
|
||
.await;
|
||
}
|
||
});
|
||
}
|
||
big_fish_error_response(&request_context, error)
|
||
})?;
|
||
state
|
||
.spacetime_client()
|
||
.generate_big_fish_asset(BigFishAssetGenerateRecordInput {
|
||
owner_user_id: owner_user_id.clone(),
|
||
session_id: session_id.clone(),
|
||
asset_kind: "level_motion".to_string(),
|
||
level: payload.level,
|
||
motion_key: payload.motion_key,
|
||
asset_url: Some(asset_url),
|
||
generated_at_micros: now,
|
||
})
|
||
.await
|
||
}
|
||
"big_fish_generate_stage_background" => {
|
||
let asset_url = generate_big_fish_formal_asset(
|
||
&state,
|
||
&owner_user_id,
|
||
&session_id,
|
||
"stage_background",
|
||
None,
|
||
None,
|
||
now,
|
||
)
|
||
.await
|
||
.map_err(|error| {
|
||
if let Some(asset_kind) = billed_asset_kind {
|
||
tokio::spawn({
|
||
let state = state.clone();
|
||
let owner_user_id = owner_user_id.clone();
|
||
let billing_asset_id = billing_asset_id.clone();
|
||
async move {
|
||
refund_asset_operation_points(
|
||
&state,
|
||
&owner_user_id,
|
||
asset_kind,
|
||
&billing_asset_id,
|
||
)
|
||
.await;
|
||
}
|
||
});
|
||
}
|
||
big_fish_error_response(&request_context, error)
|
||
})?;
|
||
state
|
||
.spacetime_client()
|
||
.generate_big_fish_asset(BigFishAssetGenerateRecordInput {
|
||
owner_user_id: owner_user_id.clone(),
|
||
session_id: session_id.clone(),
|
||
asset_kind: "stage_background".to_string(),
|
||
level: None,
|
||
motion_key: None,
|
||
asset_url: Some(asset_url),
|
||
generated_at_micros: now,
|
||
})
|
||
.await
|
||
}
|
||
"big_fish_publish_game" => {
|
||
state
|
||
.spacetime_client()
|
||
.publish_big_fish_game(session_id, owner_user_id.clone(), now)
|
||
.await
|
||
}
|
||
other => {
|
||
return Err(big_fish_bad_request(
|
||
&request_context,
|
||
format!("action `{other}` is not supported").as_str(),
|
||
));
|
||
}
|
||
};
|
||
let session = match session_result {
|
||
Ok(session) => session,
|
||
Err(error) => {
|
||
if let Some(asset_kind) = billed_asset_kind {
|
||
refund_asset_operation_points(
|
||
&state,
|
||
&owner_user_id,
|
||
asset_kind,
|
||
&billing_asset_id,
|
||
)
|
||
.await;
|
||
}
|
||
return Err(big_fish_error_response(
|
||
&request_context,
|
||
map_big_fish_client_error(error),
|
||
));
|
||
}
|
||
};
|
||
|
||
Ok(json_success_body(
|
||
Some(&request_context),
|
||
BigFishActionResponse {
|
||
session: map_big_fish_session_response(session),
|
||
},
|
||
))
|
||
}
|
||
|
||
fn map_big_fish_session_response(session: BigFishSessionRecord) -> BigFishSessionSnapshotResponse {
|
||
BigFishSessionSnapshotResponse {
|
||
session_id: session.session_id,
|
||
current_turn: session.current_turn,
|
||
progress_percent: session.progress_percent,
|
||
stage: session.stage,
|
||
anchor_pack: map_big_fish_anchor_pack_response(session.anchor_pack),
|
||
draft: session.draft.map(map_big_fish_draft_response),
|
||
asset_slots: session
|
||
.asset_slots
|
||
.into_iter()
|
||
.map(map_big_fish_asset_slot_response)
|
||
.collect(),
|
||
asset_coverage: map_big_fish_asset_coverage_response(session.asset_coverage),
|
||
messages: session
|
||
.messages
|
||
.into_iter()
|
||
.map(map_big_fish_agent_message_response)
|
||
.collect(),
|
||
last_assistant_reply: session.last_assistant_reply,
|
||
publish_ready: session.publish_ready,
|
||
updated_at: session.updated_at,
|
||
}
|
||
}
|
||
|
||
fn map_big_fish_anchor_pack_response(
|
||
anchor_pack: BigFishAnchorPackRecord,
|
||
) -> BigFishAnchorPackResponse {
|
||
BigFishAnchorPackResponse {
|
||
gameplay_promise: map_big_fish_anchor_item_response(anchor_pack.gameplay_promise),
|
||
ecology_visual_theme: map_big_fish_anchor_item_response(anchor_pack.ecology_visual_theme),
|
||
growth_ladder: map_big_fish_anchor_item_response(anchor_pack.growth_ladder),
|
||
risk_tempo: map_big_fish_anchor_item_response(anchor_pack.risk_tempo),
|
||
}
|
||
}
|
||
|
||
fn map_big_fish_anchor_item_response(anchor: BigFishAnchorItemRecord) -> BigFishAnchorItemResponse {
|
||
BigFishAnchorItemResponse {
|
||
key: anchor.key,
|
||
label: anchor.label,
|
||
value: anchor.value,
|
||
status: anchor.status,
|
||
}
|
||
}
|
||
|
||
fn map_big_fish_draft_response(draft: BigFishGameDraftRecord) -> BigFishGameDraftResponse {
|
||
BigFishGameDraftResponse {
|
||
title: draft.title,
|
||
subtitle: draft.subtitle,
|
||
core_fun: draft.core_fun,
|
||
ecology_theme: draft.ecology_theme,
|
||
levels: draft
|
||
.levels
|
||
.into_iter()
|
||
.map(map_big_fish_level_response)
|
||
.collect(),
|
||
background: map_big_fish_background_response(draft.background),
|
||
runtime_params: map_big_fish_runtime_params_response(draft.runtime_params),
|
||
}
|
||
}
|
||
|
||
fn map_big_fish_level_response(
|
||
level: BigFishLevelBlueprintRecord,
|
||
) -> BigFishLevelBlueprintResponse {
|
||
BigFishLevelBlueprintResponse {
|
||
level: level.level,
|
||
name: level.name,
|
||
one_line_fantasy: level.one_line_fantasy,
|
||
text_description: level.text_description,
|
||
silhouette_direction: level.silhouette_direction,
|
||
size_ratio: level.size_ratio,
|
||
visual_description: level.visual_description,
|
||
visual_prompt_seed: level.visual_prompt_seed,
|
||
idle_motion_description: level.idle_motion_description,
|
||
move_motion_description: level.move_motion_description,
|
||
motion_prompt_seed: level.motion_prompt_seed,
|
||
merge_source_level: level.merge_source_level,
|
||
prey_window: level.prey_window,
|
||
threat_window: level.threat_window,
|
||
is_final_level: level.is_final_level,
|
||
}
|
||
}
|
||
|
||
fn map_big_fish_background_response(
|
||
background: BigFishBackgroundBlueprintRecord,
|
||
) -> BigFishBackgroundBlueprintResponse {
|
||
BigFishBackgroundBlueprintResponse {
|
||
theme: background.theme,
|
||
color_mood: background.color_mood,
|
||
foreground_hints: background.foreground_hints,
|
||
midground_composition: background.midground_composition,
|
||
background_depth: background.background_depth,
|
||
safe_play_area_hint: background.safe_play_area_hint,
|
||
spawn_edge_hint: background.spawn_edge_hint,
|
||
background_prompt_seed: background.background_prompt_seed,
|
||
}
|
||
}
|
||
|
||
fn map_big_fish_runtime_params_response(
|
||
params: BigFishRuntimeParamsRecord,
|
||
) -> BigFishRuntimeParamsResponse {
|
||
BigFishRuntimeParamsResponse {
|
||
level_count: params.level_count,
|
||
merge_count_per_upgrade: params.merge_count_per_upgrade,
|
||
spawn_target_count: params.spawn_target_count,
|
||
leader_move_speed: params.leader_move_speed,
|
||
follower_catch_up_speed: params.follower_catch_up_speed,
|
||
offscreen_cull_seconds: params.offscreen_cull_seconds,
|
||
prey_spawn_delta_levels: params.prey_spawn_delta_levels,
|
||
threat_spawn_delta_levels: params.threat_spawn_delta_levels,
|
||
win_level: params.win_level,
|
||
}
|
||
}
|
||
|
||
fn map_big_fish_asset_slot_response(slot: BigFishAssetSlotRecord) -> BigFishAssetSlotResponse {
|
||
BigFishAssetSlotResponse {
|
||
slot_id: slot.slot_id,
|
||
asset_kind: slot.asset_kind,
|
||
level: slot.level,
|
||
motion_key: slot.motion_key,
|
||
status: slot.status,
|
||
asset_url: slot.asset_url,
|
||
prompt_snapshot: slot.prompt_snapshot,
|
||
updated_at: slot.updated_at,
|
||
}
|
||
}
|
||
|
||
fn map_big_fish_asset_coverage_response(
|
||
coverage: BigFishAssetCoverageRecord,
|
||
) -> BigFishAssetCoverageResponse {
|
||
BigFishAssetCoverageResponse {
|
||
level_main_image_ready_count: coverage.level_main_image_ready_count,
|
||
level_motion_ready_count: coverage.level_motion_ready_count,
|
||
background_ready: coverage.background_ready,
|
||
required_level_count: coverage.required_level_count,
|
||
publish_ready: coverage.publish_ready,
|
||
blockers: coverage.blockers,
|
||
}
|
||
}
|
||
|
||
async fn compile_big_fish_draft_only(
|
||
state: &AppState,
|
||
session_id: String,
|
||
owner_user_id: String,
|
||
now: i64,
|
||
) -> Result<BigFishSessionRecord, SpacetimeClientError> {
|
||
// 中文注释：大鱼吃小鱼草稿阶段只负责编译结果页草稿，不在这一步串行生成主图、动作或背景。
|
||
// 这些资产统一留在结果页工坊按需触发，避免 compile action 因长耗时资产任务卡在首步等待态。
|
||
let session =
|
||
load_big_fish_session_with_retry(state, session_id.clone(), owner_user_id.clone()).await?;
|
||
let anchor_pack = map_record_anchor_pack_to_domain(&session.anchor_pack);
|
||
let compiled_draft =
|
||
compile_big_fish_draft_with_fallback(state.llm_client(), &anchor_pack).await;
|
||
let draft_json = serde_json::to_string(&compiled_draft).ok();
|
||
|
||
state
|
||
.spacetime_client()
|
||
.compile_big_fish_draft(BigFishDraftCompileRecordInput {
|
||
session_id,
|
||
owner_user_id,
|
||
draft_json,
|
||
compiled_at_micros: now,
|
||
})
|
||
.await
|
||
}
|
||
|
||
async fn load_big_fish_session_with_retry(
|
||
state: &AppState,
|
||
session_id: String,
|
||
owner_user_id: String,
|
||
) -> Result<BigFishSessionRecord, SpacetimeClientError> {
|
||
let mut last_retryable_error = None;
|
||
|
||
for attempt in 0..2 {
|
||
match state
|
||
.spacetime_client()
|
||
.get_big_fish_session(session_id.clone(), owner_user_id.clone())
|
||
.await
|
||
{
|
||
Ok(session) => return Ok(session),
|
||
Err(error @ SpacetimeClientError::Timeout)
|
||
| Err(error @ SpacetimeClientError::ConnectDropped) => {
|
||
last_retryable_error = Some(error);
|
||
if attempt == 0 {
|
||
sleep(Duration::from_millis(250)).await;
|
||
continue;
|
||
}
|
||
}
|
||
Err(error) => return Err(error),
|
||
}
|
||
}
|
||
|
||
Err(last_retryable_error.unwrap_or(SpacetimeClientError::Timeout))
|
||
}
|
||
|
||
fn map_record_anchor_pack_to_domain(
|
||
anchor_pack: &BigFishAnchorPackRecord,
|
||
) -> module_big_fish::BigFishAnchorPack {
|
||
module_big_fish::BigFishAnchorPack {
|
||
gameplay_promise: map_record_anchor_item_to_domain(&anchor_pack.gameplay_promise),
|
||
ecology_visual_theme: map_record_anchor_item_to_domain(&anchor_pack.ecology_visual_theme),
|
||
growth_ladder: map_record_anchor_item_to_domain(&anchor_pack.growth_ladder),
|
||
risk_tempo: map_record_anchor_item_to_domain(&anchor_pack.risk_tempo),
|
||
}
|
||
}
|
||
|
||
fn map_record_anchor_item_to_domain(
|
||
anchor_item: &BigFishAnchorItemRecord,
|
||
) -> module_big_fish::BigFishAnchorItem {
|
||
module_big_fish::BigFishAnchorItem {
|
||
key: anchor_item.key.clone(),
|
||
label: anchor_item.label.clone(),
|
||
value: anchor_item.value.clone(),
|
||
status: match anchor_item.status.as_str() {
|
||
"confirmed" => module_big_fish::BigFishAnchorStatus::Confirmed,
|
||
"locked" => module_big_fish::BigFishAnchorStatus::Locked,
|
||
"inferred" => module_big_fish::BigFishAnchorStatus::Inferred,
|
||
_ => module_big_fish::BigFishAnchorStatus::Missing,
|
||
},
|
||
}
|
||
}
|
||
|
||
fn map_big_fish_agent_message_response(
|
||
message: BigFishAgentMessageRecord,
|
||
) -> BigFishAgentMessageResponse {
|
||
BigFishAgentMessageResponse {
|
||
id: message.message_id,
|
||
role: message.role,
|
||
kind: message.kind,
|
||
text: message.text,
|
||
created_at: message.created_at,
|
||
}
|
||
}
|
||
|
||
fn map_big_fish_work_summary_response(
|
||
item: BigFishWorkSummaryRecord,
|
||
) -> BigFishWorkSummaryResponse {
|
||
BigFishWorkSummaryResponse {
|
||
work_id: item.work_id,
|
||
source_session_id: item.source_session_id,
|
||
owner_user_id: item.owner_user_id,
|
||
title: item.title,
|
||
subtitle: item.subtitle,
|
||
summary: item.summary,
|
||
cover_image_src: item.cover_image_src,
|
||
status: item.status,
|
||
updated_at: current_timestamp_micros_to_string(item.updated_at_micros),
|
||
publish_ready: item.publish_ready,
|
||
level_count: item.level_count,
|
||
level_main_image_ready_count: item.level_main_image_ready_count,
|
||
level_motion_ready_count: item.level_motion_ready_count,
|
||
background_ready: item.background_ready,
|
||
}
|
||
}
|
||
|
||
fn build_big_fish_welcome_text(seed_text: &str) -> String {
|
||
if seed_text.trim().is_empty() {
|
||
return "我会先帮你确定大鱼吃小鱼的核心锚点。可以从主题生态、成长阶梯或风险节奏开始。"
|
||
.to_string();
|
||
}
|
||
"我已经收到你的玩法起点，会先把它整理成锚点并准备结果页草稿。".to_string()
|
||
}
|
||
|
||
struct BigFishDashScopeSettings {
|
||
base_url: String,
|
||
api_key: String,
|
||
request_timeout_ms: u64,
|
||
}
|
||
|
||
struct BigFishGeneratedImage {
|
||
image_url: String,
|
||
task_id: String,
|
||
}
|
||
|
||
struct BigFishDownloadedImage {
|
||
mime_type: String,
|
||
extension: String,
|
||
bytes: Vec<u8>,
|
||
}
|
||
|
||
struct BigFishFormalAssetContext {
|
||
entity_id: String,
|
||
prompt: String,
|
||
negative_prompt: String,
|
||
size: String,
|
||
asset_object_kind: String,
|
||
binding_slot: String,
|
||
path_segments: Vec<String>,
|
||
apply_transparent_background_post_process: bool,
|
||
}
|
||
|
||
const BIG_FISH_TEXT_TO_IMAGE_MODEL: &str = "wan2.2-t2i-flash";
|
||
const BIG_FISH_ENTITY_KIND: &str = "big_fish_session";
|
||
|
||
async fn generate_big_fish_formal_asset(
|
||
state: &AppState,
|
||
owner_user_id: &str,
|
||
session_id: &str,
|
||
asset_kind: &str,
|
||
level: Option<u32>,
|
||
motion_key: Option<&str>,
|
||
generated_at_micros: i64,
|
||
) -> Result<String, AppError> {
|
||
let session = state
|
||
.spacetime_client()
|
||
.get_big_fish_session(session_id.to_string(), owner_user_id.to_string())
|
||
.await
|
||
.map_err(map_big_fish_client_error)?;
|
||
let draft = session.draft.as_ref().ok_or_else(|| {
|
||
AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
|
||
"provider": "big-fish",
|
||
"message": "玩法草稿尚未编译，不能生成正式图片。",
|
||
}))
|
||
})?;
|
||
let context = build_big_fish_formal_asset_context(
|
||
&session,
|
||
draft,
|
||
asset_kind,
|
||
level,
|
||
motion_key,
|
||
generated_at_micros,
|
||
)?;
|
||
let settings = require_big_fish_dashscope_settings(state)?;
|
||
let http_client = build_big_fish_dashscope_http_client(&settings)?;
|
||
let generated = create_big_fish_text_to_image_generation(
|
||
&http_client,
|
||
&settings,
|
||
context.prompt.as_str(),
|
||
context.negative_prompt.as_str(),
|
||
context.size.as_str(),
|
||
)
|
||
.await?;
|
||
let downloaded = download_big_fish_remote_image(
|
||
&http_client,
|
||
generated.image_url.as_str(),
|
||
"下载 Big Fish 正式图片失败",
|
||
context.apply_transparent_background_post_process,
|
||
)
|
||
.await?;
|
||
|
||
persist_big_fish_formal_asset(
|
||
state,
|
||
owner_user_id,
|
||
&context,
|
||
generated,
|
||
downloaded,
|
||
generated_at_micros,
|
||
)
|
||
.await
|
||
}
|
||
|
||
fn build_big_fish_formal_asset_context(
|
||
session: &BigFishSessionRecord,
|
||
draft: &BigFishGameDraftRecord,
|
||
asset_kind: &str,
|
||
level: Option<u32>,
|
||
motion_key: Option<&str>,
|
||
generated_at_micros: i64,
|
||
) -> Result<BigFishFormalAssetContext, AppError> {
|
||
let asset_id = format!("asset-{generated_at_micros}");
|
||
match asset_kind {
|
||
"level_main_image" => {
|
||
let level = find_big_fish_level_blueprint(draft, level)?;
|
||
let level_part = build_big_fish_level_part(Some(level.level));
|
||
Ok(BigFishFormalAssetContext {
|
||
entity_id: session.session_id.clone(),
|
||
prompt: build_big_fish_level_main_image_prompt(draft, level),
|
||
negative_prompt: BIG_FISH_TRANSPARENT_ASSET_NEGATIVE_PROMPT.to_string(),
|
||
size: "1024*1024".to_string(),
|
||
asset_object_kind: "big_fish_level_main_image".to_string(),
|
||
binding_slot: format!("level_main_image:{level_part}"),
|
||
path_segments: vec![
|
||
sanitize_big_fish_path_segment(session.session_id.as_str(), "session"),
|
||
"level-main-image".to_string(),
|
||
level_part,
|
||
asset_id,
|
||
],
|
||
apply_transparent_background_post_process: true,
|
||
})
|
||
}
|
||
"level_motion" => {
|
||
let level = find_big_fish_level_blueprint(draft, level)?;
|
||
let motion_key = motion_key
|
||
.map(str::trim)
|
||
.filter(|value| matches!(*value, "idle_float" | "move_swim"))
|
||
.ok_or_else(|| {
|
||
AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
|
||
"provider": "big-fish",
|
||
"message": "motionKey 必须是 idle_float 或 move_swim。",
|
||
}))
|
||
})?;
|
||
let level_part = build_big_fish_level_part(Some(level.level));
|
||
Ok(BigFishFormalAssetContext {
|
||
entity_id: session.session_id.clone(),
|
||
prompt: build_big_fish_level_motion_prompt(draft, level, motion_key),
|
||
negative_prompt: BIG_FISH_TRANSPARENT_ASSET_NEGATIVE_PROMPT.to_string(),
|
||
size: "1024*1024".to_string(),
|
||
asset_object_kind: "big_fish_level_motion".to_string(),
|
||
binding_slot: format!("level_motion:{level_part}:{motion_key}"),
|
||
path_segments: vec![
|
||
sanitize_big_fish_path_segment(session.session_id.as_str(), "session"),
|
||
"level-motion".to_string(),
|
||
level_part,
|
||
sanitize_big_fish_path_segment(motion_key, "motion"),
|
||
asset_id,
|
||
],
|
||
apply_transparent_background_post_process: true,
|
||
})
|
||
}
|
||
"stage_background" => Ok(BigFishFormalAssetContext {
|
||
entity_id: session.session_id.clone(),
|
||
prompt: build_big_fish_stage_background_prompt(draft),
|
||
negative_prompt: BIG_FISH_DEFAULT_NEGATIVE_PROMPT.to_string(),
|
||
size: "720*1280".to_string(),
|
||
asset_object_kind: "big_fish_stage_background".to_string(),
|
||
binding_slot: "stage_background".to_string(),
|
||
path_segments: vec![
|
||
sanitize_big_fish_path_segment(session.session_id.as_str(), "session"),
|
||
"stage-background".to_string(),
|
||
asset_id,
|
||
],
|
||
apply_transparent_background_post_process: false,
|
||
}),
|
||
_ => Err(
|
||
AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
|
||
"provider": "big-fish",
|
||
"message": format!("assetKind `{asset_kind}` 不支持正式图片生成。"),
|
||
})),
|
||
),
|
||
}
|
||
}
|
||
|
||
fn find_big_fish_level_blueprint(
|
||
draft: &BigFishGameDraftRecord,
|
||
level: Option<u32>,
|
||
) -> Result<&BigFishLevelBlueprintRecord, AppError> {
|
||
let level = level.ok_or_else(|| {
|
||
AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
|
||
"provider": "big-fish",
|
||
"message": "level 是等级资产生成的必填项。",
|
||
}))
|
||
})?;
|
||
draft
|
||
.levels
|
||
.iter()
|
||
.find(|blueprint| blueprint.level == level)
|
||
.ok_or_else(|| {
|
||
AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
|
||
"provider": "big-fish",
|
||
"message": format!("level `{level}` 不存在于当前 Big Fish 草稿。"),
|
||
}))
|
||
})
|
||
}
|
||
|
||
fn require_big_fish_dashscope_settings(
|
||
state: &AppState,
|
||
) -> Result<BigFishDashScopeSettings, AppError> {
|
||
let base_url = state.config.dashscope_base_url.trim().trim_end_matches('/');
|
||
if base_url.is_empty() {
|
||
return Err(
|
||
AppError::from_status(StatusCode::SERVICE_UNAVAILABLE).with_details(json!({
|
||
"provider": "dashscope",
|
||
"reason": "DASHSCOPE_BASE_URL 未配置",
|
||
})),
|
||
);
|
||
}
|
||
|
||
let api_key = state
|
||
.config
|
||
.dashscope_api_key
|
||
.as_deref()
|
||
.map(str::trim)
|
||
.filter(|value| !value.is_empty())
|
||
.ok_or_else(|| {
|
||
AppError::from_status(StatusCode::SERVICE_UNAVAILABLE).with_details(json!({
|
||
"provider": "dashscope",
|
||
"reason": "DASHSCOPE_API_KEY 未配置",
|
||
}))
|
||
})?;
|
||
|
||
Ok(BigFishDashScopeSettings {
|
||
base_url: base_url.to_string(),
|
||
api_key: api_key.to_string(),
|
||
request_timeout_ms: state.config.dashscope_image_request_timeout_ms.max(1),
|
||
})
|
||
}
|
||
|
||
fn build_big_fish_dashscope_http_client(
|
||
settings: &BigFishDashScopeSettings,
|
||
) -> Result<reqwest::Client, AppError> {
|
||
reqwest::Client::builder()
|
||
.timeout(Duration::from_millis(settings.request_timeout_ms))
|
||
.build()
|
||
.map_err(|error| {
|
||
AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR).with_details(json!({
|
||
"provider": "dashscope",
|
||
"message": format!("构造 DashScope HTTP 客户端失败：{error}"),
|
||
}))
|
||
})
|
||
}
|
||
|
||
async fn create_big_fish_text_to_image_generation(
|
||
http_client: &reqwest::Client,
|
||
settings: &BigFishDashScopeSettings,
|
||
prompt: &str,
|
||
negative_prompt: &str,
|
||
size: &str,
|
||
) -> Result<BigFishGeneratedImage, AppError> {
|
||
let mut parameters = Map::from_iter([
|
||
("n".to_string(), json!(1)),
|
||
("size".to_string(), Value::String(size.to_string())),
|
||
("prompt_extend".to_string(), Value::Bool(true)),
|
||
("watermark".to_string(), Value::Bool(false)),
|
||
]);
|
||
if !negative_prompt.trim().is_empty() {
|
||
parameters.insert(
|
||
"negative_prompt".to_string(),
|
||
Value::String(negative_prompt.trim().to_string()),
|
||
);
|
||
}
|
||
|
||
let response = http_client
|
||
.post(format!(
|
||
"{}/services/aigc/text2image/image-synthesis",
|
||
settings.base_url
|
||
))
|
||
.header(
|
||
reqwest::header::AUTHORIZATION,
|
||
format!("Bearer {}", settings.api_key),
|
||
)
|
||
.header(reqwest::header::CONTENT_TYPE, "application/json")
|
||
.header("X-DashScope-Async", "enable")
|
||
.json(&json!({
|
||
"model": BIG_FISH_TEXT_TO_IMAGE_MODEL,
|
||
"input": {
|
||
"prompt": prompt,
|
||
},
|
||
"parameters": parameters,
|
||
}))
|
||
.send()
|
||
.await
|
||
.map_err(|error| {
|
||
map_big_fish_dashscope_request_error(format!("创建 Big Fish 图片生成任务失败：{error}"))
|
||
})?;
|
||
let status = response.status();
|
||
let response_text = response.text().await.map_err(|error| {
|
||
map_big_fish_dashscope_request_error(format!("读取 Big Fish 图片生成响应失败：{error}"))
|
||
})?;
|
||
if !status.is_success() {
|
||
return Err(map_big_fish_dashscope_upstream_error(
|
||
response_text.as_str(),
|
||
"创建 Big Fish 图片生成任务失败",
|
||
));
|
||
}
|
||
let payload =
|
||
parse_big_fish_json_payload(response_text.as_str(), "解析 Big Fish 图片生成响应失败")?;
|
||
let task_id = extract_big_fish_task_id(&payload).ok_or_else(|| {
|
||
AppError::from_status(StatusCode::BAD_GATEWAY).with_details(json!({
|
||
"provider": "dashscope",
|
||
"message": "Big Fish 图片生成任务未返回 task_id",
|
||
}))
|
||
})?;
|
||
let deadline = Instant::now() + Duration::from_millis(settings.request_timeout_ms);
|
||
|
||
while Instant::now() < deadline {
|
||
let poll_response = http_client
|
||
.get(format!("{}/tasks/{}", settings.base_url, task_id))
|
||
.header(
|
||
reqwest::header::AUTHORIZATION,
|
||
format!("Bearer {}", settings.api_key),
|
||
)
|
||
.send()
|
||
.await
|
||
.map_err(|error| {
|
||
map_big_fish_dashscope_request_error(format!(
|
||
"查询 Big Fish 图片生成任务失败：{error}"
|
||
))
|
||
})?;
|
||
let poll_status = poll_response.status();
|
||
let poll_text = poll_response.text().await.map_err(|error| {
|
||
map_big_fish_dashscope_request_error(format!(
|
||
"读取 Big Fish 图片生成任务响应失败：{error}"
|
||
))
|
||
})?;
|
||
if !poll_status.is_success() {
|
||
return Err(map_big_fish_dashscope_upstream_error(
|
||
poll_text.as_str(),
|
||
"查询 Big Fish 图片生成任务失败",
|
||
));
|
||
}
|
||
let poll_payload =
|
||
parse_big_fish_json_payload(poll_text.as_str(), "解析 Big Fish 图片生成任务响应失败")?;
|
||
let task_status = find_first_big_fish_string_by_key(&poll_payload, "task_status")
|
||
.unwrap_or_default()
|
||
.trim()
|
||
.to_string();
|
||
if task_status == "SUCCEEDED" {
|
||
let image_url = extract_big_fish_image_urls(&poll_payload)
|
||
.into_iter()
|
||
.next()
|
||
.ok_or_else(|| {
|
||
AppError::from_status(StatusCode::BAD_GATEWAY).with_details(json!({
|
||
"provider": "dashscope",
|
||
"message": "Big Fish 图片生成成功但未返回图片地址",
|
||
}))
|
||
})?;
|
||
return Ok(BigFishGeneratedImage { image_url, task_id });
|
||
}
|
||
if matches!(task_status.as_str(), "FAILED" | "UNKNOWN") {
|
||
return Err(map_big_fish_dashscope_upstream_error(
|
||
poll_text.as_str(),
|
||
"Big Fish 图片生成任务失败",
|
||
));
|
||
}
|
||
|
||
sleep(Duration::from_secs(2)).await;
|
||
}
|
||
|
||
Err(
|
||
AppError::from_status(StatusCode::BAD_GATEWAY).with_details(json!({
|
||
"provider": "dashscope",
|
||
"message": "Big Fish 图片生成超时或未返回图片地址",
|
||
})),
|
||
)
|
||
}
|
||
|
||
async fn download_big_fish_remote_image(
|
||
http_client: &reqwest::Client,
|
||
image_url: &str,
|
||
fallback_message: &str,
|
||
apply_transparent_background_post_process: bool,
|
||
) -> Result<BigFishDownloadedImage, AppError> {
|
||
let response = http_client.get(image_url).send().await.map_err(|error| {
|
||
map_big_fish_dashscope_request_error(format!("{fallback_message}：{error}"))
|
||
})?;
|
||
let status = response.status();
|
||
let content_type = response
|
||
.headers()
|
||
.get(reqwest::header::CONTENT_TYPE)
|
||
.and_then(|value| value.to_str().ok())
|
||
.unwrap_or("image/jpeg")
|
||
.to_string();
|
||
let bytes = response.bytes().await.map_err(|error| {
|
||
map_big_fish_dashscope_request_error(format!("{fallback_message}：{error}"))
|
||
})?;
|
||
if !status.is_success() {
|
||
return Err(
|
||
AppError::from_status(StatusCode::BAD_GATEWAY).with_details(json!({
|
||
"provider": "dashscope",
|
||
"message": fallback_message,
|
||
"status": status.as_u16(),
|
||
})),
|
||
);
|
||
}
|
||
|
||
let mime_type = normalize_big_fish_downloaded_image_mime_type(content_type.as_str());
|
||
let mut normalized_bytes = bytes.to_vec();
|
||
let mut normalized_mime_type = mime_type;
|
||
let mut extension = big_fish_mime_to_extension(normalized_mime_type.as_str()).to_string();
|
||
|
||
// 中文注释：Big Fish 的等级主图与动作关键帧要和 RPG 角色主图保持同一后处理口径。
|
||
// 因此在上游已经输出 PNG 时，统一补一层透明背景 alpha 清理，避免只靠 prompt 约束导致残留底色。
|
||
if apply_transparent_background_post_process
|
||
&& normalized_mime_type == "image/png"
|
||
&& let Some(optimized) = try_apply_background_alpha_to_png(normalized_bytes.as_slice())
|
||
{
|
||
normalized_bytes = optimized;
|
||
normalized_mime_type = "image/png".to_string();
|
||
extension = "png".to_string();
|
||
}
|
||
|
||
Ok(BigFishDownloadedImage {
|
||
extension,
|
||
mime_type: normalized_mime_type,
|
||
bytes: normalized_bytes,
|
||
})
|
||
}
|
||
|
||
async fn persist_big_fish_formal_asset(
|
||
state: &AppState,
|
||
owner_user_id: &str,
|
||
context: &BigFishFormalAssetContext,
|
||
generated: BigFishGeneratedImage,
|
||
downloaded: BigFishDownloadedImage,
|
||
generated_at_micros: i64,
|
||
) -> Result<String, AppError> {
|
||
let oss_client = state.oss_client().ok_or_else(|| {
|
||
AppError::from_status(StatusCode::SERVICE_UNAVAILABLE).with_details(json!({
|
||
"provider": "aliyun-oss",
|
||
"reason": "OSS 未完成环境变量配置",
|
||
}))
|
||
})?;
|
||
let http_client = reqwest::Client::new();
|
||
let put_result = oss_client
|
||
.put_object(
|
||
&http_client,
|
||
OssPutObjectRequest {
|
||
prefix: LegacyAssetPrefix::BigFishAssets,
|
||
path_segments: context.path_segments.clone(),
|
||
file_name: format!("image.{}", downloaded.extension),
|
||
content_type: Some(downloaded.mime_type.clone()),
|
||
access: OssObjectAccess::Private,
|
||
metadata: build_big_fish_asset_metadata(
|
||
context.asset_object_kind.as_str(),
|
||
owner_user_id,
|
||
BIG_FISH_ENTITY_KIND,
|
||
context.entity_id.as_str(),
|
||
context.binding_slot.as_str(),
|
||
),
|
||
body: downloaded.bytes,
|
||
},
|
||
)
|
||
.await
|
||
.map_err(map_big_fish_asset_oss_error)?;
|
||
let head = oss_client
|
||
.head_object(
|
||
&http_client,
|
||
OssHeadObjectRequest {
|
||
object_key: put_result.object_key.clone(),
|
||
},
|
||
)
|
||
.await
|
||
.map_err(map_big_fish_asset_oss_error)?;
|
||
let asset_object = state
|
||
.spacetime_client()
|
||
.confirm_asset_object(
|
||
build_asset_object_upsert_input(
|
||
generate_asset_object_id(generated_at_micros),
|
||
head.bucket,
|
||
head.object_key,
|
||
AssetObjectAccessPolicy::Private,
|
||
head.content_type.or(Some(downloaded.mime_type)),
|
||
head.content_length,
|
||
head.etag,
|
||
context.asset_object_kind.clone(),
|
||
Some(generated.task_id),
|
||
Some(owner_user_id.to_string()),
|
||
None,
|
||
Some(context.entity_id.clone()),
|
||
generated_at_micros,
|
||
)
|
||
.map_err(map_big_fish_asset_object_prepare_error)?,
|
||
)
|
||
.await
|
||
.map_err(map_big_fish_asset_spacetime_error)?;
|
||
state
|
||
.spacetime_client()
|
||
.bind_asset_object_to_entity(
|
||
build_asset_entity_binding_input(
|
||
generate_asset_binding_id(generated_at_micros),
|
||
asset_object.asset_object_id,
|
||
BIG_FISH_ENTITY_KIND.to_string(),
|
||
context.entity_id.clone(),
|
||
context.binding_slot.clone(),
|
||
context.asset_object_kind.clone(),
|
||
Some(owner_user_id.to_string()),
|
||
None,
|
||
generated_at_micros,
|
||
)
|
||
.map_err(map_big_fish_asset_binding_prepare_error)?,
|
||
)
|
||
.await
|
||
.map_err(map_big_fish_asset_spacetime_error)?;
|
||
|
||
Ok(put_result.legacy_public_path)
|
||
}
|
||
|
||
fn build_big_fish_asset_metadata(
|
||
asset_kind: &str,
|
||
owner_user_id: &str,
|
||
entity_kind: &str,
|
||
entity_id: &str,
|
||
slot: &str,
|
||
) -> BTreeMap<String, String> {
|
||
BTreeMap::from([
|
||
("asset_kind".to_string(), asset_kind.to_string()),
|
||
("owner_user_id".to_string(), owner_user_id.to_string()),
|
||
("entity_kind".to_string(), entity_kind.to_string()),
|
||
("entity_id".to_string(), entity_id.to_string()),
|
||
("slot".to_string(), slot.to_string()),
|
||
])
|
||
}
|
||
|
||
fn parse_big_fish_json_payload(raw_text: &str, fallback_message: &str) -> Result<Value, AppError> {
|
||
serde_json::from_str::<Value>(raw_text).map_err(|error| {
|
||
AppError::from_status(StatusCode::BAD_GATEWAY).with_details(json!({
|
||
"provider": "dashscope",
|
||
"message": format!("{fallback_message}：{error}"),
|
||
}))
|
||
})
|
||
}
|
||
|
||
fn extract_big_fish_task_id(payload: &Value) -> Option<String> {
|
||
find_first_big_fish_string_by_key(payload, "task_id")
|
||
}
|
||
|
||
fn extract_big_fish_image_urls(payload: &Value) -> Vec<String> {
|
||
let mut urls = Vec::new();
|
||
collect_big_fish_strings_by_key(payload, "image", &mut urls);
|
||
collect_big_fish_strings_by_key(payload, "url", &mut urls);
|
||
let mut deduped = Vec::new();
|
||
for url in urls {
|
||
if !deduped.contains(&url) {
|
||
deduped.push(url);
|
||
}
|
||
}
|
||
deduped
|
||
}
|
||
|
||
fn find_first_big_fish_string_by_key(payload: &Value, target_key: &str) -> Option<String> {
|
||
let mut results = Vec::new();
|
||
collect_big_fish_strings_by_key(payload, target_key, &mut results);
|
||
results.into_iter().next()
|
||
}
|
||
|
||
fn collect_big_fish_strings_by_key(payload: &Value, target_key: &str, results: &mut Vec<String>) {
|
||
match payload {
|
||
Value::Array(entries) => {
|
||
for entry in entries {
|
||
collect_big_fish_strings_by_key(entry, target_key, results);
|
||
}
|
||
}
|
||
Value::Object(object) => {
|
||
for (key, value) in object {
|
||
if key == target_key
|
||
&& let Some(text) = value.as_str()
|
||
{
|
||
results.push(text.to_string());
|
||
}
|
||
collect_big_fish_strings_by_key(value, target_key, results);
|
||
}
|
||
}
|
||
_ => {}
|
||
}
|
||
}
|
||
|
||
fn normalize_big_fish_downloaded_image_mime_type(content_type: &str) -> String {
|
||
let mime_type = content_type
|
||
.split(';')
|
||
.next()
|
||
.map(str::trim)
|
||
.unwrap_or("image/jpeg");
|
||
match mime_type {
|
||
"image/png" | "image/webp" | "image/jpeg" | "image/jpg" | "image/gif" => {
|
||
mime_type.to_string()
|
||
}
|
||
_ => "image/jpeg".to_string(),
|
||
}
|
||
}
|
||
|
||
fn big_fish_mime_to_extension(mime_type: &str) -> &str {
|
||
match mime_type {
|
||
"image/png" => "png",
|
||
"image/webp" => "webp",
|
||
"image/gif" => "gif",
|
||
_ => "jpg",
|
||
}
|
||
}
|
||
|
||
fn map_big_fish_dashscope_request_error(message: String) -> AppError {
|
||
AppError::from_status(StatusCode::BAD_GATEWAY).with_details(json!({
|
||
"provider": "dashscope",
|
||
"message": message,
|
||
}))
|
||
}
|
||
|
||
fn map_big_fish_dashscope_upstream_error(raw_text: &str, fallback_message: &str) -> AppError {
|
||
AppError::from_status(StatusCode::BAD_GATEWAY).with_details(json!({
|
||
"provider": "dashscope",
|
||
"message": parse_big_fish_api_error_message(raw_text, fallback_message),
|
||
}))
|
||
}
|
||
|
||
fn parse_big_fish_api_error_message(raw_text: &str, fallback_message: &str) -> String {
|
||
let trimmed = raw_text.trim();
|
||
if trimmed.is_empty() {
|
||
return fallback_message.to_string();
|
||
}
|
||
if let Ok(payload) = serde_json::from_str::<Value>(trimmed)
|
||
&& let Some(message) = find_first_big_fish_string_by_key(&payload, "message")
|
||
.or_else(|| find_first_big_fish_string_by_key(&payload, "code"))
|
||
{
|
||
return message;
|
||
}
|
||
let excerpt = trimmed.chars().take(240).collect::<String>();
|
||
format!("{fallback_message}：{excerpt}")
|
||
}
|
||
|
||
fn map_big_fish_asset_object_prepare_error(error: AssetObjectFieldError) -> AppError {
|
||
AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
|
||
"provider": "asset-object",
|
||
"message": error.to_string(),
|
||
}))
|
||
}
|
||
|
||
fn map_big_fish_asset_binding_prepare_error(error: AssetObjectFieldError) -> AppError {
|
||
AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
|
||
"provider": "asset-entity-binding",
|
||
"message": error.to_string(),
|
||
}))
|
||
}
|
||
|
||
fn map_big_fish_asset_spacetime_error(error: SpacetimeClientError) -> AppError {
|
||
AppError::from_status(StatusCode::BAD_GATEWAY).with_details(json!({
|
||
"provider": "spacetimedb",
|
||
"message": error.to_string(),
|
||
}))
|
||
}
|
||
|
||
fn map_big_fish_asset_oss_error(error: platform_oss::OssError) -> AppError {
|
||
let status = match error {
|
||
platform_oss::OssError::InvalidConfig(_) | platform_oss::OssError::InvalidRequest(_) => {
|
||
StatusCode::BAD_REQUEST
|
||
}
|
||
platform_oss::OssError::ObjectNotFound(_) => StatusCode::NOT_FOUND,
|
||
platform_oss::OssError::Request(_)
|
||
| platform_oss::OssError::SerializePolicy(_)
|
||
| platform_oss::OssError::Sign(_) => StatusCode::BAD_GATEWAY,
|
||
};
|
||
AppError::from_status(status).with_details(json!({
|
||
"provider": "aliyun-oss",
|
||
"message": error.to_string(),
|
||
}))
|
||
}
|
||
|
||
fn build_big_fish_level_part(level: Option<u32>) -> String {
|
||
level
|
||
.map(|value| format!("level-{value}"))
|
||
.unwrap_or_else(|| "stage".to_string())
|
||
}
|
||
|
||
fn sanitize_big_fish_path_segment(value: &str, fallback: &str) -> String {
|
||
let sanitized = value
|
||
.trim()
|
||
.chars()
|
||
.map(|ch| {
|
||
if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' {
|
||
ch
|
||
} else {
|
||
'-'
|
||
}
|
||
})
|
||
.collect::<String>()
|
||
.trim_matches('-')
|
||
.to_string();
|
||
if sanitized.is_empty() {
|
||
fallback.to_string()
|
||
} else {
|
||
sanitized
|
||
}
|
||
}
|
||
|
||
fn ensure_non_empty(
|
||
request_context: &RequestContext,
|
||
value: &str,
|
||
field_name: &str,
|
||
) -> Result<(), Response> {
|
||
if value.trim().is_empty() {
|
||
return Err(big_fish_bad_request(
|
||
request_context,
|
||
format!("{field_name} is required").as_str(),
|
||
));
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
fn big_fish_bad_request(request_context: &RequestContext, message: &str) -> Response {
|
||
big_fish_error_response(
|
||
request_context,
|
||
AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
|
||
"provider": "big-fish",
|
||
"message": message,
|
||
})),
|
||
)
|
||
}
|
||
|
||
fn big_fish_sse_json_event_or_error(event_name: &str, payload: Value) -> Event {
|
||
match serde_json::to_string(&payload) {
|
||
Ok(payload_text) => Event::default().event(event_name).data(payload_text),
|
||
Err(_) => big_fish_sse_error_event_message("SSE payload 序列化失败".to_string()),
|
||
}
|
||
}
|
||
|
||
fn big_fish_sse_error_event_message(message: String) -> Event {
|
||
let payload = format!(
|
||
"{{\"message\":{}}}",
|
||
serde_json::to_string(&message)
|
||
.unwrap_or_else(|_| "\"SSE 错误事件序列化失败\"".to_string())
|
||
);
|
||
Event::default().event("error").data(payload)
|
||
}
|
||
|
||
fn map_big_fish_client_error(error: SpacetimeClientError) -> AppError {
|
||
let status = match &error {
|
||
SpacetimeClientError::Procedure(message)
|
||
if message.contains("big_fish_creation_session 不存在") =>
|
||
{
|
||
StatusCode::NOT_FOUND
|
||
}
|
||
SpacetimeClientError::Procedure(message)
|
||
if message.contains("不能为空")
|
||
|| message.contains("尚未编译")
|
||
|| message.contains("不允许")
|
||
|| message.contains("非法")
|
||
|| message.contains("缺少") =>
|
||
{
|
||
StatusCode::BAD_REQUEST
|
||
}
|
||
SpacetimeClientError::Runtime(_) => StatusCode::BAD_REQUEST,
|
||
SpacetimeClientError::Timeout => StatusCode::GATEWAY_TIMEOUT,
|
||
_ => StatusCode::BAD_GATEWAY,
|
||
};
|
||
|
||
let message = match &error {
|
||
SpacetimeClientError::Timeout => "SpacetimeDB 会话读取超时，请稍后重试。".to_string(),
|
||
SpacetimeClientError::ConnectDropped => {
|
||
"SpacetimeDB 会话连接已断开，请稍后重试。".to_string()
|
||
}
|
||
_ => error.to_string(),
|
||
};
|
||
|
||
AppError::from_status(status).with_details(json!({
|
||
"provider": "spacetimedb",
|
||
"message": message,
|
||
}))
|
||
}
|
||
|
||
fn should_soft_fallback_big_fish_gallery(error: &SpacetimeClientError) -> bool {
|
||
match error {
|
||
// 公开广场是首页可选数据，SpacetimeDB procedure 运行态短暂失败时不应阻断平台首屏。
|
||
SpacetimeClientError::Runtime(_) | SpacetimeClientError::ConnectDropped => true,
|
||
SpacetimeClientError::Procedure(message) => {
|
||
message.contains("list_big_fish_works")
|
||
|| message.contains("procedure")
|
||
|| message.contains("No such procedure")
|
||
}
|
||
_ => false,
|
||
}
|
||
}
|
||
|
||
fn big_fish_error_response(request_context: &RequestContext, error: AppError) -> Response {
|
||
error.into_response_with_context(Some(request_context))
|
||
}
|
||
|
||
fn current_utc_micros() -> i64 {
|
||
let duration = SystemTime::now()
|
||
.duration_since(UNIX_EPOCH)
|
||
.expect("system clock should be after unix epoch");
|
||
i64::try_from(duration.as_micros()).expect("current unix micros should fit in i64")
|
||
}
|
||
|
||
fn current_timestamp_micros_to_string(value: i64) -> String {
|
||
format_timestamp_micros(value)
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
#[test]
|
||
fn big_fish_gallery_soft_fallbacks_for_runtime_errors() {
|
||
assert!(should_soft_fallback_big_fish_gallery(
|
||
&SpacetimeClientError::Runtime("procedure runtime error".to_string())
|
||
));
|
||
assert!(should_soft_fallback_big_fish_gallery(
|
||
&SpacetimeClientError::ConnectDropped
|
||
));
|
||
assert!(should_soft_fallback_big_fish_gallery(
|
||
&SpacetimeClientError::Procedure("No such procedure: list_big_fish_works".to_string(),)
|
||
));
|
||
}
|
||
|
||
#[test]
|
||
fn big_fish_gallery_keeps_timeout_errors_visible() {
|
||
assert!(!should_soft_fallback_big_fish_gallery(
|
||
&SpacetimeClientError::Timeout
|
||
));
|
||
}
|
||
}
|