tdm_server_rust/service/

editor_import_service.rs

//! 图源导入服务 (Editor Provider Import)
//!
//! 把话数的图源（`provider_file_oss_id` 指向的单图或压缩包）转换为编辑器页面：
//!
//! 1. 通过现有 OSS 通道下载图源字节
//! 2. 解压/识别为按页序排序的图片序列
//! 3. 逐张上传到图片 OSS（`tencent.image_bucket` / `image_cdn_domain`）
//! 4. 写入 `editor_page` 记录并标记 `uploaded=true`
//!
//! 导入后编辑器只消费图片 OSS 地址，运行期不再读取压缩包。
//!
//! 幂等：已存在同话数页面时直接返回；`force=true` 时删除旧页面、旧标点后重建。

use crate::app::AppState;
use crate::cache::on_episode_mutated;
use crate::entity::editor::{
    EditorPageInfo, EditorUploadCredential, EditorUploadCredentialsResponse, EditorUploadFile,
    ImportProviderResult, RegisterEditorPagesRequest, SourceArchiveUploadCredential,
};
use crate::error::{ApiResult, AppError};
use crate::repository::editor_page_repo::{EditorPageRepository, NewEditorPage};
use crate::repository::editor_unit_repo::EditorUnitRepository;
use crate::repository::episode_repo::EpisodeRepository;
use crate::repository::oss_repo::OssRepository;
use crate::service::editor_source_archive_service::{archive_object_key, compute_archive_hash};
use crate::service::oss_service::OssService;
use crate::utils::archive_extract::{self, ExtractedImage};
use reqwest::header::CONTENT_TYPE;
use sea_orm::{ConnectionTrait, TransactionTrait};
use std::collections::HashSet;
use uuid::Uuid;

/// 单次图源导入允许的最大页面数，避免异常请求耗尽凭证、内存和数据库资源。
const MAX_EDITOR_PAGE_COUNT: usize = 500;
/// 编辑器记录允许的最大图片边长。
const MAX_EDITOR_IMAGE_DIMENSION: i32 = 100_000;

/// 图源导入服务
pub struct EditorImportService;

impl EditorImportService {
    /// 从话数图源导入并生成编辑器页面。
    ///
    /// ## 参数
    /// - `state`: 应用状态
    /// - `episode_id`: 话数 ID
    /// - `force`: 是否强制重建（删除旧页面与旧标点后重新导入）
    ///
    /// ## 返回
    /// - [`ImportProviderResult`]：是否新建、页面总数与页面列表
    ///
    /// ## 错误
    /// - 话数不存在或无图源
    /// - 图源下载、解压、上传失败
    #[tracing::instrument(skip_all, level = "debug")]
    pub async fn import_provider_pages(
        state: &AppState,
        episode_id: i32,
        force: bool,
    ) -> ApiResult<ImportProviderResult> {
        let episode_repo = EpisodeRepository::new(state.db.clone());
        // 校验话数存在并取得图源 OSS 记录
        let manga_id = episode_repo
            .get_manga_id_by_episode_id(episode_id)
            .await?
            .ok_or_else(|| AppError::business(format!("漫画单话 {episode_id} 不存在喵")))?;
        let provider_oss_id = episode_repo
            .get_provider_oss_id(episode_id)
            .await?
            .ok_or_else(|| AppError::business("当前话数尚未上传图源喵"))?;

        let page_repo = EditorPageRepository::new(state.db.clone());
        let existing_pages = page_repo.list_by_episode(episode_id).await?;

        // 幂等：已有页面且非强制，直接返回
        if !existing_pages.is_empty() && !force {
            return Ok(ImportProviderResult {
                episode_id,
                imported: false,
                page_count: existing_pages.len(),
                pages: existing_pages
                    .into_iter()
                    .map(EditorPageInfo::from)
                    .collect(),
            });
        }

        if force && !existing_pages.is_empty() {
            let db = crate::db::from_sqlx_pool(state.db.clone());
            ensure_rebuild_allowed_with(&db, episode_id, &existing_pages).await?;
        }
        let expected_page_ids = sorted_page_ids(&existing_pages);

        // 下载图源（复用现有 provider 下载通道）
        let (filename, bytes) =
            OssService::download_file_proxy(state, episode_id, "provider".to_string()).await?;

        // 解压/识别为按页序排序的图片
        let images = archive_extract::extract_images(&filename, &bytes)?;

        // 逐张上传到图片 OSS
        let oss_repo = OssRepository::new(state.db.clone(), (*state.config).clone());
        let bucket = oss_repo.image_bucket_name();
        let max_bytes = state.config.tencent.max_file_size * 1024 * 1024;
        let mut new_pages = Vec::with_capacity(images.len());
        for (idx, img) in images.iter().enumerate() {
            let object_key = build_image_object_key(episode_id, idx, &img.name);
            let put_url = oss_repo
                .presigned_image_put_url(&object_key, max_bytes)
                .await?;
            upload_image(state, &put_url, img).await?;
            let (width, height) = image_dimensions(&img.data);
            new_pages.push(NewEditorPage {
                page_index: idx as i32,
                source_oss_id: Some(provider_oss_id),
                source_file_name: Some(img.name.clone()),
                image_object_key: object_key.clone(),
                image_bucket: Some(bucket.clone()),
                image_url: oss_repo.image_display_url(&object_key),
                image_width: width,
                image_height: height,
            });
        }

        let (inserted, imported) =
            replace_pages_atomically(state, episode_id, force, &expected_page_ids, new_pages)
                .await?;
        tracing::info!(
            "图源导入完成 episode_id={episode_id} manga_id={manga_id} pages={}",
            inserted.len()
        );
        on_episode_mutated(state, manga_id, &[]).await;
        Ok(ImportProviderResult {
            episode_id,
            imported,
            page_count: inserted.len(),
            pages: inserted.into_iter().map(EditorPageInfo::from).collect(),
        })
    }

    /// 为前端图源解压直传准备批量上传凭证。
    ///
    /// 前端把图源 zip 解压（或散图识别）后，调用本接口一次拿到全部页面的图片 OSS
    /// 预签名 PUT URL（单次 STS），随后逐张直传图片字节，最后调用
    /// [`register_pages`](Self::register_pages) 落库。
    ///
    /// ## 参数
    /// - `episode_id`: 话数 ID
    /// - `files`: 待上传文件（页序 + 原始文件名，用于推断对象键后缀）
    #[tracing::instrument(skip_all, level = "debug")]
    pub async fn prepare_upload_credentials(
        state: &AppState,
        episode_id: i32,
        files: Vec<EditorUploadFile>,
    ) -> ApiResult<EditorUploadCredentialsResponse> {
        validate_upload_files(&files)?;
        // 校验话数存在
        EpisodeRepository::new(state.db.clone())
            .get_manga_id_by_episode_id(episode_id)
            .await?
            .ok_or_else(|| AppError::business(format!("漫画单话 {episode_id} 不存在喵")))?;
        let expected_page_ids = sorted_page_ids(
            &EditorPageRepository::new(state.db.clone())
                .list_by_episode(episode_id)
                .await?,
        );

        let oss_repo = OssRepository::new(state.db.clone(), (*state.config).clone());
        let bucket = oss_repo.image_bucket_name();
        let prefix = format!("editor/episode_{episode_id}/");
        let object_keys: Vec<String> = files
            .iter()
            .map(|f| build_image_object_key(episode_id, f.index as usize, &f.filename))
            .collect();
        // 编辑器页面图沿用图源导入上限，避免大于通用图片上限的页面被 STS 策略拒绝。
        let max_bytes = state
            .config
            .tencent
            .max_file_size
            .max(state.config.tencent.image_max_file_size)
            * 1024
            * 1024;
        let urls = oss_repo
            .presigned_image_put_urls(&prefix, &object_keys, max_bytes)
            .await?;

        let archive_hash = compute_archive_hash(
            episode_id,
            object_keys
                .iter()
                .enumerate()
                .map(|(idx, key)| (idx as i32, key.as_str())),
        );
        let archive_key = archive_object_key(episode_id, archive_hash);
        let archive_max = max_bytes * files.len().max(1) as u64;
        let archive_put_url = oss_repo
            .presigned_image_put_url(&archive_key, archive_max)
            .await?;

        let file_creds = files
            .into_iter()
            .zip(object_keys)
            .zip(urls)
            .map(|((f, object_key), presigned_url)| EditorUploadCredential {
                index: f.index,
                image_url: oss_repo.image_display_url(&object_key),
                object_key,
                presigned_url,
                bucket: bucket.clone(),
            })
            .collect();
        Ok(EditorUploadCredentialsResponse {
            files: file_creds,
            source_archive: SourceArchiveUploadCredential {
                object_key: archive_key.clone(),
                presigned_url: archive_put_url,
                download_url: oss_repo.image_display_url(&archive_key),
            },
            expected_page_ids,
        })
    }

    /// 注册前端已上传到图片 OSS 的页面。
    ///
    /// 与服务端 `import_provider_pages` 等价的落库逻辑，但图片由前端直传，
    /// 不再依赖 `provider_file_oss_id` 压缩包。
    ///
    /// 幂等：已有页面且非 `force` 时直接返回旧页面；强制重建会校验页面快照及下游工作，
    /// 并在同一事务内替换旧页面与标点。
    #[tracing::instrument(skip_all, level = "debug")]
    pub async fn register_pages(
        state: &AppState,
        episode_id: i32,
        req: RegisterEditorPagesRequest,
    ) -> ApiResult<ImportProviderResult> {
        let manga_id = EpisodeRepository::new(state.db.clone())
            .get_manga_id_by_episode_id(episode_id)
            .await?
            .ok_or_else(|| AppError::business(format!("漫画单话 {episode_id} 不存在喵")))?;

        let oss_repo = OssRepository::new(state.db.clone(), (*state.config).clone());
        let bucket = oss_repo.image_bucket_name();

        // 校验并按 page_index 排序，禁止伪造其他话数的对象键。
        let pages = validate_registered_pages(episode_id, req.pages)?;
        let new_pages: Vec<NewEditorPage> = pages
            .into_iter()
            .map(|p| NewEditorPage {
                page_index: p.page_index,
                source_oss_id: None,
                source_file_name: p.source_file_name,
                image_url: oss_repo.image_display_url(&p.object_key),
                image_object_key: p.object_key,
                image_bucket: Some(bucket.clone()),
                image_width: p.image_width,
                image_height: p.image_height,
            })
            .collect();

        let (inserted, imported) = replace_pages_atomically(
            state,
            episode_id,
            req.force,
            &req.expected_page_ids,
            new_pages,
        )
        .await?;
        tracing::info!(
            "前端直传页面注册完成 episode_id={episode_id} pages={}",
            inserted.len()
        );
        on_episode_mutated(state, manga_id, &[]).await;
        Ok(ImportProviderResult {
            episode_id,
            imported,
            page_count: inserted.len(),
            pages: inserted.into_iter().map(EditorPageInfo::from).collect(),
        })
    }
}

/// 在短事务内原子切换页面集，失败时完整保留旧页面和标点。
async fn replace_pages_atomically(
    state: &AppState,
    episode_id: i32,
    force: bool,
    expected_page_ids: &[i64],
    new_pages: Vec<NewEditorPage>,
) -> ApiResult<(Vec<crate::sea_entity::editor_page::Model>, bool)> {
    let db = crate::db::from_sqlx_pool(state.db.clone());
    let txn = db.begin().await?;
    let episode = EpisodeRepository::lock_by_id_with(&txn, episode_id)
        .await?
        .ok_or_else(|| AppError::business(format!("漫画单话 {episode_id} 不存在喵")))?;
    let existing_pages = EditorPageRepository::list_by_episode_with(&txn, episode_id).await?;

    // 网络超时后的同内容重试直接视为成功，避免再次删除已提交页面。
    if page_sets_equal(&existing_pages, &new_pages) {
        txn.commit().await?;
        return Ok((existing_pages, false));
    }

    if !existing_pages.is_empty() && !force {
        txn.commit().await?;
        return Ok((existing_pages, false));
    }

    if force && sorted_page_ids(&existing_pages) != sorted_ids(expected_page_ids) {
        return Err(AppError::business(
            "图源页面已被其他人更新，请刷新后重新上传喵",
        ));
    }

    if force && !existing_pages.is_empty() {
        ensure_rebuild_allowed_with(&txn, episode_id, &existing_pages).await?;
        if episode.translator_file.is_some()
            || episode.proofreader_file.is_some()
            || episode.timer_file.is_some()
            || episode.translator_file_oss_id.is_some()
            || episode.proofreader_file_oss_id.is_some()
            || episode.letterer_file_oss_id.is_some()
            || episode.timer_file_oss_id.is_some()
            || episode.publish_link.is_some()
        {
            return Err(AppError::business(
                "该话已有下游稿件或发布记录，禁止直接重建图源喵",
            ));
        }
        EditorUnitRepository::delete_by_episode_with(&txn, episode_id).await?;
        EditorPageRepository::delete_by_episode_with(&txn, episode_id).await?;
    }

    let inserted = EditorPageRepository::insert_pages_with(&txn, episode_id, new_pages).await?;
    txn.commit().await?;
    Ok((inserted, true))
}

/// 检查当前页面是否已有需要保护的在线编辑或岗位交稿。
async fn ensure_rebuild_allowed_with<C: ConnectionTrait>(
    db: &C,
    episode_id: i32,
    existing_pages: &[crate::sea_entity::editor_page::Model],
) -> ApiResult<()> {
    if existing_pages.iter().any(|page| page.translation_completed)
        || EditorUnitRepository::count_by_episode_with(db, episode_id).await? > 0
        || EpisodeRepository::has_downstream_submission_with(db, episode_id).await?
    {
        return Err(AppError::business(
            "该话已有翻译或校对内容，禁止直接重建图源喵",
        ));
    }
    Ok(())
}

/// 校验上传凭证请求，限制页面数量并要求页序连续。
fn validate_upload_files(files: &[EditorUploadFile]) -> ApiResult<()> {
    if files.is_empty() {
        return Err(AppError::business("没有待上传的页面喵"));
    }
    if files.len() > MAX_EDITOR_PAGE_COUNT {
        return Err(AppError::business(format!(
            "单话图源不能超过 {MAX_EDITOR_PAGE_COUNT} 页喵"
        )));
    }
    let mut indices: Vec<i32> = files.iter().map(|file| file.index).collect();
    indices.sort_unstable();
    if indices
        .iter()
        .enumerate()
        .any(|(expected, actual)| *actual != expected as i32)
    {
        return Err(AppError::business("图源页序必须从 0 开始且连续喵"));
    }
    if files
        .iter()
        .any(|file| file.filename.trim().is_empty() || file.filename.len() > 255)
    {
        return Err(AppError::business("图源文件名为空或过长喵"));
    }
    Ok(())
}

/// 校验并排序前端已上传的页面清单。
fn validate_registered_pages(
    episode_id: i32,
    mut pages: Vec<crate::entity::editor::RegisterEditorPage>,
) -> ApiResult<Vec<crate::entity::editor::RegisterEditorPage>> {
    if pages.is_empty() {
        return Err(AppError::business("没有可注册的页面喵"));
    }
    if pages.len() > MAX_EDITOR_PAGE_COUNT {
        return Err(AppError::business(format!(
            "单话图源不能超过 {MAX_EDITOR_PAGE_COUNT} 页喵"
        )));
    }
    pages.sort_by_key(|page| page.page_index);
    let expected_prefix = format!("editor/episode_{episode_id}/");
    let mut object_keys = HashSet::with_capacity(pages.len());
    for (index, page) in pages.iter().enumerate() {
        if page.page_index != index as i32 {
            return Err(AppError::business("图源页序必须从 0 开始且连续喵"));
        }
        if !page.object_key.starts_with(&expected_prefix)
            || !object_keys.insert(page.object_key.as_str())
        {
            return Err(AppError::business("图源对象键无效或重复喵"));
        }
        if page
            .source_file_name
            .as_ref()
            .is_some_and(|name| name.len() > 255)
        {
            return Err(AppError::business("图源文件名过长喵"));
        }
        if [page.image_width, page.image_height]
            .into_iter()
            .flatten()
            .any(|dimension| !(1..=MAX_EDITOR_IMAGE_DIMENSION).contains(&dimension))
        {
            return Err(AppError::business("图源图片尺寸无效喵"));
        }
    }
    Ok(pages)
}

/// 判断当前页面集是否与本次注册内容完全相同。
fn page_sets_equal(
    existing: &[crate::sea_entity::editor_page::Model],
    incoming: &[NewEditorPage],
) -> bool {
    existing.len() == incoming.len()
        && existing.iter().zip(incoming).all(|(current, next)| {
            current.page_index == next.page_index
                && current.image_object_key == next.image_object_key
        })
}

/// 对页面模型 ID 排序，生成并发更新快照。
fn sorted_page_ids(pages: &[crate::sea_entity::editor_page::Model]) -> Vec<i64> {
    sorted_ids(&pages.iter().map(|page| page.id).collect::<Vec<_>>())
}

/// 返回排序后的 ID 副本。
fn sorted_ids(ids: &[i64]) -> Vec<i64> {
    let mut sorted = ids.to_vec();
    sorted.sort_unstable();
    sorted
}

/// 生成页面图片在图片 OSS 中的对象键
fn build_image_object_key(episode_id: i32, index: usize, source_name: &str) -> String {
    let ext = source_name
        .rsplit('.')
        .next()
        .filter(|e| e.len() <= 5 && !e.contains('/'))
        .map(|e| e.to_ascii_lowercase())
        .unwrap_or_else(|| "jpg".to_string());
    format!(
        "editor/episode_{episode_id}/{:04}_{}.{ext}",
        index,
        Uuid::new_v4()
    )
}

/// 通过预签名 PUT URL 上传单张图片
async fn upload_image(state: &AppState, put_url: &str, img: &ExtractedImage) -> ApiResult<()> {
    let content_type = mime_guess::from_path(&img.name)
        .first_or_octet_stream()
        .to_string();
    let resp = state
        .http_client
        .put(put_url)
        .header(CONTENT_TYPE, content_type)
        .body(img.data.clone())
        .send()
        .await
        .map_err(|e| AppError::Oss {
            code: None,
            msg: format!("上传图片到 OSS 失败：{e}"),
        })?;
    if !resp.status().is_success() {
        return Err(AppError::Oss {
            code: None,
            msg: format!("上传图片到 OSS 返回非 2xx：HTTP {}", resp.status()),
        });
    }
    Ok(())
}

/// 解析图片像素宽高（失败返回 None，不阻断导入）
fn image_dimensions(bytes: &[u8]) -> (Option<i32>, Option<i32>) {
    match imagesize::blob_size(bytes) {
        Ok(size) => (Some(size.width as i32), Some(size.height as i32)),
        Err(_) => (None, None),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::entity::editor::RegisterEditorPage;

    fn upload_file(index: i32, filename: &str) -> EditorUploadFile {
        EditorUploadFile {
            index,
            filename: filename.to_string(),
        }
    }

    fn registered_page(episode_id: i32, page_index: i32, suffix: &str) -> RegisterEditorPage {
        RegisterEditorPage {
            page_index,
            object_key: format!("editor/episode_{episode_id}/{suffix}.jpg"),
            source_file_name: Some(format!("{suffix}.jpg")),
            image_width: Some(1200),
            image_height: Some(1800),
        }
    }

    #[test]
    fn upload_files_require_contiguous_unique_indices() {
        assert!(
            validate_upload_files(&[upload_file(0, "001.jpg"), upload_file(1, "002.jpg"),]).is_ok()
        );
        assert!(
            validate_upload_files(&[upload_file(0, "001.jpg"), upload_file(0, "002.jpg"),])
                .is_err()
        );
        assert!(validate_upload_files(&[upload_file(1, "002.jpg")]).is_err());
    }

    #[test]
    fn registered_pages_are_sorted_and_scoped_to_episode() {
        let pages = validate_registered_pages(
            42,
            vec![
                registered_page(42, 1, "0001_b"),
                registered_page(42, 0, "0000_a"),
            ],
        )
        .expect("valid pages");
        assert_eq!(
            pages.iter().map(|page| page.page_index).collect::<Vec<_>>(),
            vec![0, 1]
        );

        assert!(validate_registered_pages(42, vec![registered_page(7, 0, "0000_a")]).is_err());
    }

    #[test]
    fn registered_pages_reject_duplicate_object_keys_and_invalid_dimensions() {
        let duplicate = registered_page(42, 0, "same");
        let mut duplicate_second = duplicate.clone();
        duplicate_second.page_index = 1;
        assert!(validate_registered_pages(42, vec![duplicate, duplicate_second]).is_err());

        let mut invalid_dimension = registered_page(42, 0, "0000_a");
        invalid_dimension.image_width = Some(0);
        assert!(validate_registered_pages(42, vec![invalid_dimension]).is_err());
    }

    #[test]
    fn sorted_ids_returns_snapshot_in_stable_order() {
        assert_eq!(sorted_ids(&[9, 2, 5]), vec![2, 5, 9]);
    }
}