nyaKazuha/BlueArchiveToolkit
1
0
Fork 0
mirror of https://github.com/Yuyi-Oak/BlueArchiveToolkit.git synced 2026-09-18 06:34:54 +08:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

fix: 完成下载并发与翻译交接链路

Browse Source
This commit is contained in:
nyaKazuha
2026-08-02 22:34:23 +08:00
parent d533c88108
commit 8b64cc94f3
19 changed files with 2923 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
infrastructure/src/bin/bat_official_sync.rs
+426 -22
View File
@@ -3,6 +3,8 @@ use bat_assetbundle::UnitySerializedReplacementValue;
use bat_core::domain::{Resource, ResourceType};
use bat_core::repositories::resource_repository::{ResourceQuery, ResourceRepository};
use bat_core::{ApiError, ErrorCode};
#[cfg(test)]
use bat_infrastructure::DEFAULT_DOWNLOAD_CONCURRENCY;
use bat_infrastructure::{
apply_patch_file, apply_unityfs_field_patch_file, apply_unityfs_string_field_patch_file,
apply_unityfs_text_asset_patch_file, changed_endpoint_urls, diff_extended_snapshot,
@@ -16,11 +18,12 @@ use bat_infrastructure::{
OfficialUpdateConfig, OfficialUpdateProgress, OfficialUpdateReport, OfficialUpdateService,
OfficialUpdateSnapshot, OfficialUpdateStatus, OfficialVerificationSummary,
OfficialVersionRecord, OfficialVersionState, PatchApplyKind, PatchApplyParams,
PatchApplyReport, ReleaseFlowStatusCode, SqliteResourceRepository, UnityFsFieldPatchParams,
PatchApplyReport, ReleaseFlowStatusCode, SqliteResourceRepository,
SqliteTranslationTaskRepository, TranslationTaskStatus, UnityFsFieldPatchParams,
UnityFsPatchReport, UnityFsStringFieldPatchParams, UnityFsTextAssetPatchParams,
LOCALIZED_CURRENT_LINK, LOCALIZED_PATCH_MANIFEST_FILE, LOCALIZED_VERSIONS_DIR,
LOCALIZED_VERSION_STATE_FILE, OFFICIAL_PARSE_CACHE_FILE, OFFICIAL_TEXTUNIT_INDEX_FILE,
PRIVATE_FILE_MODE,
LOCALIZED_VERSION_STATE_FILE, MAX_DOWNLOAD_CONCURRENCY, MIN_DOWNLOAD_CONCURRENCY,
OFFICIAL_PARSE_CACHE_FILE, OFFICIAL_TEXTUNIT_INDEX_FILE, PRIVATE_FILE_MODE,
};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
@@ -175,6 +178,7 @@ fn run() -> anyhow::Result<i32> {
| CliCommand::ParseTextUnits
| CliCommand::ParseErrors
| CliCommand::TranslationTasks
| CliCommand::TranslationHandoff
| CliCommand::LocalizedStatus
| CliCommand::ResourceIndex => {
run_readonly_query_command(&options)?;
@@ -243,12 +247,14 @@ struct CliOptions {
query_bundle_path: Option<String>,
query_archive_entry: Option<String>,
query_task_status: Option<String>,
query_worker_status: Option<String>,
query_parse_status: Option<String>,
query_path_id: Option<i64>,
query_class_id: Option<i32>,
query_field_path: Option<String>,
query_format: Option<String>,
query_has_reason: Option<bool>,
query_has_failure_reason: Option<bool>,
query_option_explicit: bool,
patch_kind: Option<PatchApplyKind>,
patch_source_path: Option<PathBuf>,
@@ -302,12 +308,14 @@ impl Default for CliOptions {
query_bundle_path: None,
query_archive_entry: None,
query_task_status: None,
query_worker_status: None,
query_parse_status: None,
query_path_id: None,
query_class_id: None,
query_field_path: None,
query_format: None,
query_has_reason: None,
query_has_failure_reason: None,
query_option_explicit: false,
patch_kind: None,
patch_source_path: None,
@@ -349,6 +357,7 @@ enum CliCommand {
ParseTextUnits,
ParseErrors,
TranslationTasks,
TranslationHandoff,
LocalizedStatus,
ResourceIndex,
PatchApply,
@@ -800,6 +809,8 @@ const RPC_METHOD_PARSE_STATUS: &str = "parse.status";
const RPC_METHOD_PARSE_TEXT_UNITS: &str = "parse.text_units";
const RPC_METHOD_PARSE_ERRORS: &str = "parse.errors";
const RPC_METHOD_TRANSLATION_TASKS: &str = "translation.tasks";
const RPC_METHOD_TRANSLATION_HANDOFF: &str = "translation.handoff";
const RPC_METHOD_TRANSLATION_TASK_UPDATE: &str = "translation.task.update";
const RPC_METHOD_LOCALIZED_STATUS: &str = "localized.status";
const RPC_METHOD_CATALOG_STATUS: &str = "catalog.status";
const RPC_METHOD_CATALOG_VERSIONS: &str = "catalog.versions";
@@ -2278,6 +2289,16 @@ fn dispatch_rpc_method(
build_translation_tasks_report(state_dir, query, offset, limit),
)
}
RPC_METHOD_TRANSLATION_HANDOFF => rpc_envelope_from_result(
request_id,
"translation.handoff",
build_translation_handoff_report(state_dir),
),
RPC_METHOD_TRANSLATION_TASK_UPDATE => rpc_envelope_from_result(
request_id,
"translation.task.update",
update_translation_task_status_report(state_dir, request.params.as_ref()),
),
RPC_METHOD_LOCALIZED_STATUS => rpc_envelope_from_result(
request_id,
"localized.status",
@@ -2873,17 +2894,46 @@ fn query_resource_repository(
})
}
fn query_translation_task_repository(
repository_path: &Path,
query: &OfficialTextUnitTaskQuery,
offset: usize,
limit: usize,
) -> anyhow::Result<(u64, Vec<bat_infrastructure::PersistedTranslationTask>)> {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
runtime.block_on(async {
let repository = SqliteTranslationTaskRepository::open(repository_path)
.await
.map_err(|error| anyhow::anyhow!("{error}"))?;
let total_entries = repository
.count(query)
.await
.map_err(|error| anyhow::anyhow!("{error}"))?;
let entries = repository
.list(query)
.await
.map_err(|error| anyhow::anyhow!("{error}"))?
.into_iter()
.skip(offset)
.take(limit)
.collect();
Ok((total_entries, entries))
})
}
fn repository_file_exists_no_symlink(path: &Path) -> anyhow::Result<bool> {
sqlite_file_exists_no_symlink(path, "资源索引数据库")
}
fn sqlite_file_exists_no_symlink(path: &Path, label: &str) -> anyhow::Result<bool> {
match fs::symlink_metadata(path) {
Ok(metadata) if metadata.file_type().is_symlink() => Err(anyhow::anyhow!(
"资源索引数据库不能是 symlink：{}",
path.display()
)),
Ok(metadata) if metadata.file_type().is_symlink() => {
Err(anyhow::anyhow!("{label}不能是 symlink：{}", path.display()))
}
Ok(metadata) if metadata.is_file() => Ok(true),
Ok(_) => Err(anyhow::anyhow!(
"资源索引数据库不是普通文件：{}",
path.display()
)),
Ok(_) => Err(anyhow::anyhow!("{label}不是普通文件：{}", path.display())),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
Err(error) => Err(error.into()),
}
@@ -3089,6 +3139,8 @@ fn build_translation_tasks_report(
let task_queue_path = record
.resource_root
.join(bat_infrastructure::OFFICIAL_TEXTUNIT_TASK_QUEUE_FILE);
let task_repository_path =
SqliteTranslationTaskRepository::repository_path(&record.resource_root);
let Some(queue) = bat_infrastructure::read_textunit_task_queue_at(&record.resource_root)
.map_err(anyhow::Error::msg)?
else {
@@ -3097,21 +3149,56 @@ fn build_translation_tasks_report(
"current_version_id": record.id,
"resource_root": record.resource_root,
"textunit_task_queue_path": task_queue_path,
"task_repository_path": task_repository_path,
"task_repository_available": false,
}));
};
let matches = bat_infrastructure::query_textunit_tasks(&queue, &query);
let total_entries = matches.len();
let entries = matches
.into_iter()
.skip(offset)
.take(limit)
.cloned()
.collect::<Vec<_>>();
let task_repository_available =
sqlite_file_exists_no_symlink(&task_repository_path, "翻译任务状态数据库")?;
let (total_entries, entries) = if task_repository_available {
query_translation_task_repository(&task_repository_path, &query, offset, limit)?
} else {
let mut queue_query = query.clone();
queue_query.task_status = None;
queue_query.has_failure_reason = None;
let matches = bat_infrastructure::query_textunit_tasks(&queue, &queue_query);
let persisted = matches
.into_iter()
.cloned()
.map(|task| {
bat_infrastructure::PersistedTranslationTask::from_queued_task(
task,
queue.generated_unix_seconds,
)
})
.filter(|task| {
query
.task_status
.as_ref()
.is_none_or(|status| task.task_status.as_str() == status)
})
.filter(|task| {
query
.has_failure_reason
.is_none_or(|has_reason| task.failure_reason.is_some() == has_reason)
})
.collect::<Vec<_>>();
let total_entries = persisted.len();
let entries = persisted
.into_iter()
.skip(offset)
.take(limit)
.collect::<Vec<_>>();
(total_entries as u64, entries)
};
Ok(serde_json::json!({
"available": true,
"current_version_id": record.id,
"resource_root": record.resource_root,
"textunit_task_queue_path": task_queue_path,
"task_repository_path": task_repository_path,
"task_repository_available": task_repository_available,
"task_repository_schema_version": bat_infrastructure::TRANSLATION_TASK_SCHEMA_VERSION,
"summary": queue.summary,
"total_entries": total_entries,
"offset": offset,
@@ -3121,6 +3208,124 @@ fn build_translation_tasks_report(
}))
}
/// `translation.handoff`：当前已发布版本的动态翻译交接视图。
fn build_translation_handoff_report(state_dir: &Path) -> anyhow::Result<serde_json::Value> {
let (_, version_state) = read_daemon_resource_state(state_dir)?;
let current = version_state
.as_ref()
.and_then(|state| state.current_completed_version.as_ref());
let Some(record) = current else {
return Ok(serde_json::json!({ "available": false }));
};
let resource_root = &record.resource_root;
let task_queue_path = resource_root.join(bat_infrastructure::OFFICIAL_TEXTUNIT_TASK_QUEUE_FILE);
let handoff_path = resource_root.join(bat_infrastructure::TRANSLATION_HANDOFF_FILE);
let repository_path = SqliteTranslationTaskRepository::repository_path(resource_root);
let Some(queue) = bat_infrastructure::read_textunit_task_queue_at(resource_root)
.map_err(anyhow::Error::msg)?
else {
return Ok(serde_json::json!({
"available": false,
"current_version_id": record.id,
"resource_root": resource_root,
"textunit_task_queue_path": task_queue_path,
"translation_handoff_path": handoff_path,
"task_repository_path": repository_path,
}));
};
let task_repository_available =
sqlite_file_exists_no_symlink(&repository_path, "翻译任务状态数据库")?;
let tasks = if task_repository_available {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
runtime.block_on(async {
let repository = SqliteTranslationTaskRepository::open(&repository_path)
.await
.map_err(|error| anyhow::anyhow!("{error}"))?;
repository
.list(&OfficialTextUnitTaskQuery::default())
.await
.map_err(|error| anyhow::anyhow!("{error}"))
})?
} else {
queue
.tasks
.iter()
.cloned()
.map(|task| {
bat_infrastructure::PersistedTranslationTask::from_queued_task(
task,
queue.generated_unix_seconds,
)
})
.collect::<Vec<_>>()
};
let handoff = bat_infrastructure::build_translation_handoff(&queue, &tasks);
let handoff_file_available = sqlite_file_exists_no_symlink(&handoff_path, "翻译 handoff")?;
Ok(serde_json::json!({
"available": true,
"current_version_id": record.id,
"resource_root": resource_root,
"textunit_task_queue_path": task_queue_path,
"translation_handoff_path": handoff_path,
"translation_handoff_file_available": handoff_file_available,
"task_repository_path": repository_path,
"task_repository_available": task_repository_available,
"task_repository_schema_version": bat_infrastructure::TRANSLATION_TASK_SCHEMA_VERSION,
"handoff_schema_version": bat_infrastructure::TRANSLATION_HANDOFF_SCHEMA_VERSION,
"handoff": handoff,
}))
}
/// `translation.task.update`：写入 provider worker 的可回查状态。
fn update_translation_task_status_report(
state_dir: &Path,
params: Option<&serde_json::Value>,
) -> anyhow::Result<serde_json::Value> {
let task_id = rpc_string_param(params, "task_id")
.ok_or_else(|| anyhow::anyhow!("translation.task.update 缺少 task_id"))?;
let status_label = rpc_string_param(params, "status")
.ok_or_else(|| anyhow::anyhow!("translation.task.update 缺少 status"))?;
let status = TranslationTaskStatus::parse(status_label)
.ok_or_else(|| anyhow::anyhow!("不支持的翻译任务 worker 状态：{status_label}"))?;
let failure_reason = rpc_string_param(params, "failure_reason")
.or_else(|| rpc_string_param(params, "reason"))
.map(str::to_string);
let provider_run_id = rpc_string_param(params, "provider_run_id").map(str::to_string);
let (_, version_state) = read_daemon_resource_state(state_dir)?;
let current = version_state
.as_ref()
.and_then(|state| state.current_completed_version.as_ref())
.ok_or_else(|| anyhow::anyhow!("没有可更新翻译任务的当前官方 release"))?;
let repository_path = SqliteTranslationTaskRepository::repository_path(&current.resource_root);
if !sqlite_file_exists_no_symlink(&repository_path, "翻译任务状态数据库")? {
return Err(anyhow::anyhow!(
"翻译任务状态数据库不存在：{}",
repository_path.display()
));
}
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
let task = runtime.block_on(async {
let repository = SqliteTranslationTaskRepository::open(&repository_path)
.await
.map_err(|error| anyhow::anyhow!("{error}"))?;
repository
.update_status(task_id, status, failure_reason, provider_run_id)
.await
.map_err(|error| anyhow::anyhow!("{error}"))
})?;
Ok(serde_json::json!({
"available": true,
"current_version_id": current.id,
"task_repository_path": repository_path,
"entry": task,
}))
}
fn textunit_query_json(query: &OfficialTextUnitQuery) -> serde_json::Value {
serde_json::json!({
"destination": query.destination.clone(),
@@ -3141,9 +3346,11 @@ fn translation_task_query_json(query: &OfficialTextUnitTaskQuery) -> serde_json:
"path_pattern": query.path_pattern.clone(),
"archive_entry": query.archive_entry.clone(),
"status": query.status.clone(),
"task_status": query.task_status.clone(),
"parse_status": query.parse_status.clone(),
"text_unit_format": query.text_unit_format.clone(),
"has_reason": query.has_reason,
"has_failure_reason": query.has_failure_reason,
})
}
@@ -3398,11 +3605,13 @@ fn rpc_translation_task_query_params(
status: rpc_string_param(params, "status")
.or_else(|| rpc_string_param(params, "task_status"))
.map(str::to_string),
task_status: rpc_string_param(params, "worker_status").map(str::to_string),
parse_status: rpc_string_param(params, "parse_status").map(str::to_string),
text_unit_format: rpc_string_param(params, "text_unit_format")
.or_else(|| rpc_string_param(params, "format"))
.map(str::to_string),
has_reason: rpc_bool_param(params, "has_reason"),
has_failure_reason: rpc_bool_param(params, "has_failure_reason"),
};
Ok((query, offset, limit))
}
@@ -4357,6 +4566,7 @@ fn readonly_query_rpc_method(command: CliCommand) -> Option<&'static str> {
CliCommand::ParseTextUnits => Some(RPC_METHOD_PARSE_TEXT_UNITS),
CliCommand::ParseErrors => Some(RPC_METHOD_PARSE_ERRORS),
CliCommand::TranslationTasks => Some(RPC_METHOD_TRANSLATION_TASKS),
CliCommand::TranslationHandoff => Some(RPC_METHOD_TRANSLATION_HANDOFF),
CliCommand::LocalizedStatus => Some(RPC_METHOD_LOCALIZED_STATUS),
CliCommand::ResourceIndex => Some(RPC_METHOD_RESOURCE_INDEX),
_ => None,
@@ -4471,6 +4681,9 @@ fn readonly_query_rpc_params(options: &CliOptions) -> Option<serde_json::Value>
if let Some(status) = options.query_task_status.as_ref() {
params.insert("status".to_string(), serde_json::json!(status));
}
if let Some(status) = options.query_worker_status.as_ref() {
params.insert("worker_status".to_string(), serde_json::json!(status));
}
if let Some(parse_status) = options.query_parse_status.as_ref() {
params.insert("parse_status".to_string(), serde_json::json!(parse_status));
}
@@ -4480,6 +4693,12 @@ fn readonly_query_rpc_params(options: &CliOptions) -> Option<serde_json::Value>
if let Some(has_reason) = options.query_has_reason {
params.insert("has_reason".to_string(), serde_json::json!(has_reason));
}
if let Some(has_failure_reason) = options.query_has_failure_reason {
params.insert(
"has_failure_reason".to_string(),
serde_json::json!(has_failure_reason),
);
}
}
_ => {}
}
@@ -4515,6 +4734,7 @@ fn build_readonly_query_report(
options.query_offset,
options.query_limit,
),
RPC_METHOD_TRANSLATION_HANDOFF => build_translation_handoff_report(&options.state_dir),
RPC_METHOD_LOCALIZED_STATUS => {
build_localized_status_report(&options.state_dir, &options.config)
}
@@ -4539,13 +4759,15 @@ fn validate_readonly_query_options(options: &CliOptions) -> anyhow::Result<()> {
|| options.query_field_path.is_some();
let has_translation_task_filter = options.query_task_id.is_some()
|| options.query_task_status.is_some()
|| options.query_has_reason.is_some();
|| options.query_worker_status.is_some()
|| options.query_has_reason.is_some()
|| options.query_has_failure_reason.is_some();
match options.command {
CliCommand::ResourceIndex => {
if has_parse_object_filter || has_translation_task_filter {
return Err(anyhow::anyhow!(
"--path-id/--class-id/--field-path 只适用于 parse-text-units 或 parse-errors；--task-id/--task-status/--has-reason 只适用于 translation-tasks"
"--path-id/--class-id/--field-path 只适用于 parse-text-units 或 parse-errors；--task-id/--task-status/--worker-status/--has-reason/--has-failure-reason 只适用于 translation-tasks"
));
}
}
@@ -4556,7 +4778,7 @@ fn validate_readonly_query_options(options: &CliOptions) -> anyhow::Result<()> {
|| options.query_parse_status.is_some()
{
return Err(anyhow::anyhow!(
"--resource-type/--hash/--release-id/--platform/--bundle-path/--parse-status 只适用于 resource-index 或 translation-tasks；--task-id/--task-status/--has-reason 只适用于 translation-tasks"
"--resource-type/--hash/--release-id/--platform/--bundle-path/--parse-status 只适用于 resource-index 或 translation-tasks；--task-id/--task-status/--worker-status/--has-reason/--has-failure-reason 只适用于 translation-tasks"
));
}
}
@@ -4567,6 +4789,11 @@ fn validate_readonly_query_options(options: &CliOptions) -> anyhow::Result<()> {
));
}
}
CliCommand::TranslationHandoff if options.query_option_explicit => {
return Err(anyhow::anyhow!(
"translation-handoff 不接受查询过滤参数；请使用 translation-tasks 查询单项任务"
));
}
CliCommand::ParseStatus | CliCommand::LocalizedStatus if options.query_option_explicit => {
return Err(anyhow::anyhow!(
"查询过滤参数只适用于 resource-index、parse-text-units、parse-errors 或 translation-tasks"
@@ -4788,6 +5015,8 @@ fn translation_task_query_from_options(options: &CliOptions) -> OfficialTextUnit
parse_status: options.query_parse_status.clone(),
text_unit_format: options.query_format.clone(),
has_reason: options.query_has_reason,
has_failure_reason: options.query_has_failure_reason,
task_status: options.query_worker_status.clone(),
}
}
@@ -6446,6 +6675,8 @@ fn daemon_child_args(options: &CliOptions) -> Vec<String> {
}
args.push("--curl".to_string());
args.push(config.curl_command.to_string_lossy().to_string());
args.push("--download-concurrency".to_string());
args.push(config.download_concurrency.to_string());
match config.curl_proxy.mode() {
CurlProxyMode::Auto if options.proxy_option_explicit => {
args.push("--proxy".to_string());
@@ -6960,6 +7191,8 @@ BAT_AUTO_DISCOVER=1
# 逗号分隔：windows,android
#BAT_PLATFORMS=windows,android
#BAT_CURL=curl
# 最大并发下载数（默认 8；范围 1..=256）
#BAT_DOWNLOAD_CONCURRENCY=8
#BAT_UNZIP=unzip
# ---- 输出 ----
@@ -7138,6 +7371,10 @@ fn apply_bat_env_overrides(
if let Some(v) = value("BAT_CURL") {
options.config.curl_command = PathBuf::from(v);
}
if let Some(v) = value("BAT_DOWNLOAD_CONCURRENCY") {
options.config.download_concurrency =
parse_download_concurrency(&v, "环境变量 BAT_DOWNLOAD_CONCURRENCY")?;
}
if let Some(v) = value("BAT_UNZIP") {
options.config.unzip_command = PathBuf::from(v);
}
@@ -7245,6 +7482,10 @@ fn parse_args_with_env(
ensure_command_not_set(options.command, "translation-tasks")?;
options.command = CliCommand::TranslationTasks;
}
"translation-handoff" => {
ensure_command_not_set(options.command, "translation-handoff")?;
options.command = CliCommand::TranslationHandoff;
}
"localized-status" => {
ensure_command_not_set(options.command, "localized-status")?;
options.command = CliCommand::LocalizedStatus;
@@ -7359,6 +7600,11 @@ fn parse_args_with_env(
"--curl" => {
options.config.curl_command = PathBuf::from(next_option_value(&mut args, &flag)?);
}
"--download-concurrency" => {
options.config.download_concurrency =
parse_download_concurrency(&next_option_value(&mut args, &flag)?, &flag)?;
options.sync_option_explicit = true;
}
"--proxy" => {
options.config.curl_proxy =
parse_proxy_config(&next_option_value(&mut args, &flag)?)?;
@@ -7543,6 +7789,10 @@ fn parse_args_with_env(
options.query_task_status = Some(next_option_value(&mut args, &flag)?);
options.query_option_explicit = true;
}
"--worker-status" => {
options.query_worker_status = Some(next_option_value(&mut args, &flag)?);
options.query_option_explicit = true;
}
"--parse-status" => {
options.query_parse_status = Some(next_option_value(&mut args, &flag)?);
options.query_option_explicit = true;
@@ -7587,6 +7837,14 @@ fn parse_args_with_env(
options.query_has_reason = Some(false);
options.query_option_explicit = true;
}
"--has-failure-reason" => {
options.query_has_failure_reason = Some(true);
options.query_option_explicit = true;
}
"--no-failure-reason" => {
options.query_has_failure_reason = Some(false);
options.query_option_explicit = true;
}
"--patch-kind" => {
options.patch_kind = Some(parse_patch_apply_kind(&next_option_value(
&mut args, &flag,
@@ -7704,6 +7962,7 @@ fn parse_args_with_env(
| CliCommand::ParseTextUnits
| CliCommand::ParseErrors
| CliCommand::TranslationTasks
| CliCommand::TranslationHandoff
| CliCommand::LocalizedStatus
| CliCommand::ResourceIndex => {
validate_readonly_query_options(&options)?;
@@ -7833,6 +8092,18 @@ fn tools_are_non_default(config: &OfficialUpdateConfig, baseline: &OfficialUpdat
|| config.unzip_command != baseline.unzip_command
}
fn parse_download_concurrency(value: &str, source: &str) -> anyhow::Result<usize> {
let parsed = value
.parse::<usize>()
.map_err(|error| anyhow::anyhow!("{source} 无效：{error}"))?;
if !(MIN_DOWNLOAD_CONCURRENCY..=MAX_DOWNLOAD_CONCURRENCY).contains(&parsed) {
return Err(anyhow::anyhow!(
"{source} 必须在 {MIN_DOWNLOAD_CONCURRENCY}..={MAX_DOWNLOAD_CONCURRENCY} 范围内"
));
}
Ok(parsed)
}
fn next_option_value(
args: &mut impl Iterator<Item = String>,
flag: &str,
@@ -7896,6 +8167,7 @@ fn print_usage(binary: &str) {
eprintln!(" parse-text-units Query current official TextUnit detail index");
eprintln!(" parse-errors Query current official parse/extraction diagnostics");
eprintln!(" translation-tasks Query current offline TextUnit translation task status");
eprintln!(" translation-handoff Query current translation job/unit/provider handoff");
eprintln!(" localized-status Show localized release status for current official release");
eprintln!(" resource-index Query CAS + ResourceRepository index");
eprintln!(" patch-apply Apply a Binary/JSON/Text patch file");
@@ -7946,6 +8218,9 @@ fn print_usage(binary: &str) {
eprintln!(" --import-resource-db <PATH> SQLite ResourceRepository path");
eprintln!(" --snapshot <PATH> Override snapshot path (default: <output>/current/official-sync-snapshot.json)");
eprintln!(" --curl <PATH> curl executable (default: curl)");
eprintln!(
" --download-concurrency <N> Bounded parallel downloads (default: 8, range 1..=256)"
);
eprintln!(" --proxy <URL|auto|none> curl proxy override (default: auto from env)");
eprintln!(" --no-proxy Force direct curl connections");
eprintln!(" --unzip <PATH> unzip executable (default: unzip)");
@@ -7970,6 +8245,9 @@ fn print_usage(binary: &str) {
eprintln!(" --bundle-path <PATH> Filter resource-index by metadata bundle path");
eprintln!(" --archive-entry <PATH> Filter resource-index, parse detail, or translation-tasks by ZIP/archive entry");
eprintln!(" --task-status <STATUS> Filter translation-tasks by task status");
eprintln!(
" --worker-status <STATUS> Filter translation-tasks by provider worker status"
);
eprintln!(" --parse-status <STATUS> Filter resource-index or translation-tasks by parse status");
eprintln!(" --path-id <ID> Filter parse detail by Unity object path ID");
eprintln!(" --class-id <ID> Filter parse detail by Unity class ID");
@@ -7978,6 +8256,9 @@ fn print_usage(binary: &str) {
eprintln!(
" --has-reason | --no-reason Filter translation-tasks by diagnostic reason presence"
);
eprintln!(
" --has-failure-reason | --no-failure-reason Filter translation-tasks by provider failure reason"
);
eprintln!();
eprintln!("Write patch:");
eprintln!(" --patch-kind <binary|json|text> Patch type for patch-apply");
@@ -8098,6 +8379,10 @@ mod tests {
#[test]
fn env_defaults_apply_and_cli_overrides() {
assert_eq!(
CliOptions::default().config.download_concurrency,
DEFAULT_DOWNLOAD_CONCURRENCY
);
let options = parse_with_env(
&["bat"],
&[
@@ -8109,6 +8394,7 @@ mod tests {
("BAT_AUTO_DISCOVER", "1"),
("BAT_STATE_DIR", "/srv/state"),
("BAT_INTERVAL_SECONDS", "120"),
("BAT_DOWNLOAD_CONCURRENCY", "3"),
],
)
.unwrap();
@@ -8127,6 +8413,7 @@ mod tests {
Some(PathBuf::from("/srv/bat/resources.sqlite"))
);
assert!(options.config.auto_discover);
assert_eq!(options.config.download_concurrency, 3);
assert_eq!(options.state_dir, PathBuf::from("/srv/state"));
assert_eq!(options.interval, Duration::from_secs(120));
@@ -8206,6 +8493,20 @@ mod tests {
assert!(parse_with_env(&["bat"], &[("BAT_WATCH", "maybe")]).is_err());
assert!(parse_with_env(&["bat"], &[("BAT_INTERVAL_SECONDS", "abc")]).is_err());
assert!(parse_with_env(&["bat"], &[("BAT_PROXY", "ftp://x")]).is_err());
assert!(parse_with_env(&["bat"], &[("BAT_DOWNLOAD_CONCURRENCY", "0")]).is_err());
assert!(parse_with_env(&["bat"], &[("BAT_DOWNLOAD_CONCURRENCY", "257")]).is_err());
assert!(parse(&["bat", "--download-concurrency", "0"]).is_err());
assert!(parse(&["bat", "--download-concurrency", "257"]).is_err());
}
#[test]
fn cli_download_concurrency_is_preserved_for_daemon_child() {
let options = parse(&["bat", "--download-concurrency", "4"]).unwrap();
assert_eq!(options.config.download_concurrency, 4);
let args = daemon_child_args(&options);
assert!(args
.windows(2)
.any(|pair| pair == ["--download-concurrency", "4"]));
}
#[test]
@@ -11242,6 +11543,34 @@ mod tests {
bat_infrastructure::write_textunit_task_queue_at(current_dir, &queue).unwrap();
}
fn write_translation_task_repository_fixture(current_dir: &Path) {
let queue = bat_infrastructure::read_textunit_task_queue_at(current_dir)
.unwrap()
.unwrap();
let repository_path =
bat_infrastructure::SqliteTranslationTaskRepository::repository_path(current_dir);
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
runtime.block_on(async {
let repository =
bat_infrastructure::SqliteTranslationTaskRepository::new(&repository_path)
.await
.unwrap();
repository.sync_queue(&queue).await.unwrap();
repository
.update_status(
"textunit/v-current/Bundle/a.bundle",
bat_infrastructure::TranslationTaskStatus::Failed,
Some("Crowdin provider rejected the payload".to_string()),
Some("crowdin-run-1".to_string()),
)
.await
.unwrap();
});
}
fn write_textunit_index_fixture(current_dir: &Path) {
let index = bat_infrastructure::OfficialTextUnitIndex {
version: bat_infrastructure::OFFICIAL_TEXTUNIT_INDEX_VERSION,
@@ -11326,6 +11655,7 @@ mod tests {
let output_root = temp.path().join("output");
let current_dir = write_catalog_fixture(&state_dir, &output_root, "bundle-b2", None);
write_textunit_task_queue_fixture(&current_dir);
write_translation_task_repository_fixture(&current_dir);
let envelope = dispatch_rpc_method(
&rpc_request(
@@ -11392,6 +11722,80 @@ mod tests {
value["data"]["entries"][0]["reason"],
serde_json::Value::Null
);
let envelope = dispatch_rpc_method(
&rpc_request(
"translation.tasks",
Some(serde_json::json!({
"worker_status": "failed",
"has_failure_reason": true,
"offset": 0,
"limit": 10
})),
),
&state_dir,
&new_daemon_control(),
&test_task_context(),
"req-translation-3".to_string(),
);
let value = serde_json::to_value(&envelope).unwrap();
assert_eq!(value["ok"], true);
assert_eq!(value["data"]["task_repository_available"], true);
assert_eq!(value["data"]["total_entries"], 1);
assert_eq!(value["data"]["entries"][0]["task_status"], "failed");
assert_eq!(
value["data"]["entries"][0]["failure_reason"],
"Crowdin provider rejected the payload"
);
assert_eq!(value["data"]["entries"][0]["attempt_count"], 0);
let envelope = dispatch_rpc_method(
&rpc_request(
"translation.task.update",
Some(serde_json::json!({
"task_id": "textunit/v-current/Bundle/a.bundle",
"status": "completed",
"provider_run_id": "crowdin-run-2"
})),
),
&state_dir,
&new_daemon_control(),
&test_task_context(),
"req-translation-4".to_string(),
);
let value = serde_json::to_value(&envelope).unwrap();
assert_eq!(value["ok"], true);
assert_eq!(value["data"]["entry"]["task_status"], "completed");
assert_eq!(
value["data"]["entry"]["failure_reason"],
serde_json::Value::Null
);
assert_eq!(value["data"]["entry"]["provider_run_id"], "crowdin-run-2");
assert!(value["data"]["entry"]["completed_unix_seconds"].is_number());
let envelope = dispatch_rpc_method(
&rpc_request("translation.handoff", None),
&state_dir,
&new_daemon_control(),
&test_task_context(),
"req-translation-handoff".to_string(),
);
let value = serde_json::to_value(&envelope).unwrap();
assert_eq!(value["ok"], true);
assert_eq!(value["data"]["available"], true);
assert_eq!(
value["data"]["handoff_schema_version"],
bat_infrastructure::TRANSLATION_HANDOFF_SCHEMA_VERSION
);
assert_eq!(
value["data"]["handoff"]["units"].as_array().unwrap().len(),
2
);
assert_eq!(value["data"]["handoff"]["units"][0]["status"], "translated");
assert_eq!(
value["data"]["handoff"]["provider_runs"][0]["provider_run_id"],
"crowdin-run-2"
);
}
#[test]
infrastructure/src/downloader.rs
+260
View File
@@ -0,0 +1,260 @@
//! Downloader backend and bounded scheduling contracts.
//!
//! The scheduler is deliberately independent from curl, manifests, and
//! official URL rules. Those concerns belong to a backend and the caller,
//! which keeps retry, proxy, and verification policy composable.
use std::sync::{mpsc, Arc, Mutex};
/// Lowest supported download concurrency.
pub const MIN_DOWNLOAD_CONCURRENCY: usize = 1;
/// Highest supported download concurrency.
pub const MAX_DOWNLOAD_CONCURRENCY: usize = 256;
/// Default official download concurrency.
pub const DEFAULT_DOWNLOAD_CONCURRENCY: usize = 8;
/// A backend that executes one already-planned download task.
pub trait DownloaderBackend<T>: Send + Sync {
/// Successful result returned for one task.
type Output: Send;
/// Failure returned for one task.
type Error: Send;
/// Executes one task. The scheduler owns ordering and concurrency only.
fn download(&self, task: T) -> Result<Self::Output, Self::Error>;
}
/// A bounded worker scheduler.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DownloadScheduler {
max_concurrency: usize,
}
impl DownloadScheduler {
/// Creates a scheduler with the supported bounded range.
///
/// The official CLI validates input and reports out-of-range values.
/// This lower-level constructor remains total for library callers and
/// clamps values to the same safety bounds.
pub fn new(max_concurrency: usize) -> Self {
Self {
max_concurrency: max_concurrency
.clamp(MIN_DOWNLOAD_CONCURRENCY, MAX_DOWNLOAD_CONCURRENCY),
}
}
/// Returns the configured upper bound.
pub fn max_concurrency(self) -> usize {
self.max_concurrency
}
/// Executes tasks with a bounded number of workers.
///
/// Results are returned in input order even when workers finish out of
/// order. A failed task does not cause additional tasks to be scheduled
/// after it, because already-started bounded work must be joined cleanly;
/// callers decide whether a failed result invalidates the whole release.
pub fn execute<T, B>(self, backend: &B, tasks: Vec<T>) -> Vec<Result<B::Output, B::Error>>
where
T: Send + 'static,
B: DownloaderBackend<T>,
{
self.execute_with_observer(
backend,
tasks,
|_, _| Ok::<(), std::convert::Infallible>(()),
)
.expect("infallible download observer cannot fail")
}
/// Executes tasks and observes each result as soon as a worker returns it.
///
/// The observer runs on the coordinator thread, while worker threads
/// immediately take another pending task after sending their result. An
/// observer error stops further observation but still drains and joins all
/// workers before returning, so no background transfer is left detached.
pub fn execute_with_observer<T, B, F, E>(
self,
backend: &B,
tasks: Vec<T>,
mut observer: F,
) -> Result<Vec<Result<B::Output, B::Error>>, E>
where
T: Send + 'static,
B: DownloaderBackend<T>,
F: FnMut(usize, &Result<B::Output, B::Error>) -> Result<(), E>,
{
if tasks.is_empty() {
return Ok(Vec::new());
}
if self.max_concurrency == 1 {
let mut results = Vec::with_capacity(tasks.len());
for (index, task) in tasks.into_iter().enumerate() {
let result = backend.download(task);
observer(index, &result)?;
results.push(result);
}
return Ok(results);
}
let total = tasks.len();
let worker_count = self.max_concurrency.min(total);
let pending = Arc::new(Mutex::new(tasks.into_iter().enumerate()));
let (result_sender, result_receiver) = mpsc::channel();
std::thread::scope(|scope| {
for _ in 0..worker_count {
let pending = Arc::clone(&pending);
let result_sender = result_sender.clone();
scope.spawn(move || loop {
let task = pending
.lock()
.expect("download scheduler task queue poisoned")
.next();
let Some((index, task)) = task else {
break;
};
let result = backend.download(task);
if result_sender.send((index, result)).is_err() {
break;
}
});
}
drop(result_sender);
let mut results = std::iter::repeat_with(|| None)
.take(total)
.collect::<Vec<_>>();
let mut observer_error = None;
for (index, result) in result_receiver {
if observer_error.is_none() {
if let Err(error) = observer(index, &result) {
observer_error = Some(error);
}
}
results[index] = Some(result);
}
let results = results
.into_iter()
.map(|result| result.expect("download scheduler lost a task result"))
.collect();
match observer_error {
Some(error) => Err(error),
None => Ok(results),
}
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread;
use std::time::Duration;
struct TestBackend {
active: AtomicUsize,
max_active: AtomicUsize,
}
impl DownloaderBackend<usize> for TestBackend {
type Output = usize;
type Error = String;
fn download(&self, task: usize) -> Result<Self::Output, Self::Error> {
let active = self.active.fetch_add(1, Ordering::SeqCst) + 1;
self.max_active.fetch_max(active, Ordering::SeqCst);
thread::sleep(Duration::from_millis(2));
self.active.fetch_sub(1, Ordering::SeqCst);
Ok(task * 2)
}
}
#[test]
fn scheduler_preserves_result_order_and_respects_bound() {
let backend = TestBackend {
active: AtomicUsize::new(0),
max_active: AtomicUsize::new(0),
};
let results = DownloadScheduler::new(2).execute(&backend, (0..8).collect());
assert_eq!(
results.into_iter().map(Result::unwrap).collect::<Vec<_>>(),
(0..8).map(|value| value * 2).collect::<Vec<_>>()
);
assert!(backend.max_active.load(Ordering::SeqCst) <= 2);
assert!(backend.max_active.load(Ordering::SeqCst) >= 2);
}
#[test]
fn zero_concurrency_is_conservative() {
assert_eq!(
DownloadScheduler::new(0).max_concurrency(),
MIN_DOWNLOAD_CONCURRENCY
);
}
#[test]
fn scheduler_caps_untrusted_upper_bound() {
assert_eq!(
DownloadScheduler::new(usize::MAX).max_concurrency(),
MAX_DOWNLOAD_CONCURRENCY
);
}
#[test]
fn observer_receives_completion_without_a_global_barrier() {
struct UnevenBackend {
active: AtomicUsize,
task_two_started_while_task_zero_active: AtomicUsize,
}
impl DownloaderBackend<usize> for UnevenBackend {
type Output = usize;
type Error = String;
fn download(&self, task: usize) -> Result<Self::Output, Self::Error> {
if task == 0 {
self.active.fetch_add(1, Ordering::SeqCst);
thread::sleep(Duration::from_millis(20));
self.active.fetch_sub(1, Ordering::SeqCst);
} else {
if task == 2 && self.active.load(Ordering::SeqCst) > 0 {
self.task_two_started_while_task_zero_active
.fetch_add(1, Ordering::SeqCst);
}
thread::sleep(Duration::from_millis(if task == 1 { 1 } else { 5 }));
}
Ok(task)
}
}
let backend = UnevenBackend {
active: AtomicUsize::new(0),
task_two_started_while_task_zero_active: AtomicUsize::new(0),
};
let mut completed = Vec::new();
let results = DownloadScheduler::new(2)
.execute_with_observer(&backend, vec![0, 1, 2], |index, _| {
completed.push(index);
Ok::<(), ()>(())
})
.unwrap();
assert_eq!(
results.into_iter().map(Result::unwrap).collect::<Vec<_>>(),
vec![0, 1, 2]
);
assert_eq!(completed.len(), 3);
assert!(completed[0] == 1, "短任务应在长任务之前回传：{completed:?}");
assert_eq!(
backend
.task_two_started_while_task_zero_active
.load(Ordering::SeqCst),
1,
"worker 完成 task 1 后应立即领取 task 2"
);
}
}
infrastructure/src/lib.rs
+14
View File
@@ -12,6 +12,7 @@
pub mod cas;
mod curl_transfer;
pub mod downloader;
pub mod import;
pub mod localized_patch;
pub mod official_changes;
@@ -28,12 +29,17 @@ pub mod patch_ops;
pub mod path_security;
pub mod release_flow;
pub mod resources;
pub mod translation_tasks;
mod zip_validation;
pub use cas::FileSystemCasRepository;
pub use curl_transfer::{
redact_proxy_url, resolve_curl_proxy, CurlProxyConfig, CurlProxyMode, ResolvedCurlProxy,
};
pub use downloader::{
DownloadScheduler, DownloaderBackend, DEFAULT_DOWNLOAD_CONCURRENCY, MAX_DOWNLOAD_CONCURRENCY,
MIN_DOWNLOAD_CONCURRENCY,
};
pub use import::{
BundleSource, ImportedResource, ResourceImportCategory, ResourceImportReport,
ResourceImportService,
@@ -124,6 +130,14 @@ pub use path_security::{
};
pub use release_flow::ReleaseFlowStatusCode;
pub use resources::{InMemoryResourceRepository, SqliteResourceRepository};
pub use translation_tasks::{
build_translation_handoff, read_translation_handoff_at, sync_translation_task_repository_at,
write_translation_handoff_at, PersistedTranslationTask, ProviderRun, ProviderRunStatus,
SqliteTranslationTaskRepository, TranslationHandoff, TranslationJob, TranslationJobStatus,
TranslationTaskStatus, TranslationTaskSyncReport, TranslationUnit, TranslationUnitStatus,
TRANSLATION_HANDOFF_FILE, TRANSLATION_HANDOFF_SCHEMA_VERSION, TRANSLATION_TASK_REPOSITORY_FILE,
TRANSLATION_TASK_SCHEMA_VERSION,
};
/// Infrastructure 版本号
pub const VERSION: &str = env!("CARGO_PKG_VERSION");
infrastructure/src/official_download.rs
+317 -53
View File
@@ -1,6 +1,10 @@
//! Official JP resource download execution.
use crate::curl_transfer::{run_curl_with_retry_with_proxy, CurlProxyConfig, CurlRetryError};
use crate::downloader::{
DownloadScheduler, DownloaderBackend, DEFAULT_DOWNLOAD_CONCURRENCY, MAX_DOWNLOAD_CONCURRENCY,
MIN_DOWNLOAD_CONCURRENCY,
};
use crate::official_pull::OfficialResourcePullPlan;
use crate::path_security::{
ensure_path_within_root, ensure_safe_directory_path, ensure_safe_file_target,
@@ -9,7 +13,10 @@ use crate::path_security::{
use crate::zip_validation::{
path_has_zip_extension, url_or_path_has_zip_extension, validate_zip_structure,
};
use bat_adapters::official::yostar_jp::{is_official_yostar_jp_url, YostarJpResourceEndpointKind};
use bat_adapters::official::yostar_jp::YostarJpResourceEndpointKind;
use bat_adapters::official::{
destination_under_root, DownloadUrlMapper, OfficialResourceBackend, YostarJpBackend,
};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, HashSet};
use std::fs::{self, File};
@@ -539,9 +546,11 @@ impl OfficialLocalResourceState {
#[derive(Debug, Clone)]
pub struct OfficialResourcePullService {
output_root: PathBuf,
backend: YostarJpBackend,
curl_command: PathBuf,
curl_proxy: CurlProxyConfig,
retry_attempts: usize,
max_concurrency: usize,
}
impl OfficialResourcePullService {
@@ -557,9 +566,11 @@ impl OfficialResourcePullService {
) -> Self {
Self {
output_root: output_root.into(),
backend: YostarJpBackend,
curl_command: curl_command.into(),
curl_proxy: CurlProxyConfig::default(),
retry_attempts: DEFAULT_RETRY_ATTEMPTS,
max_concurrency: DEFAULT_DOWNLOAD_CONCURRENCY,
}
}
@@ -575,6 +586,22 @@ impl OfficialResourcePullService {
self
}
/// Sets the bounded number of concurrent downloads.
///
/// The default is [`DEFAULT_DOWNLOAD_CONCURRENCY`]. Values are kept
/// within the supported `1..=256` range; the public update configuration
/// validates input before constructing this service.
pub fn with_max_concurrency(mut self, max_concurrency: usize) -> Self {
self.max_concurrency =
max_concurrency.clamp(MIN_DOWNLOAD_CONCURRENCY, MAX_DOWNLOAD_CONCURRENCY);
self
}
/// Returns the configured download concurrency.
pub fn max_concurrency(&self) -> usize {
self.max_concurrency
}
/// Returns the output root used for downloaded files.
pub fn output_root(&self) -> &Path {
&self.output_root
@@ -666,7 +693,7 @@ impl OfficialResourcePullService {
}
let mut planned: Vec<PlannedDownload> = Vec::with_capacity(total);
for url in urls {
if !is_official_yostar_jp_url(&url) {
if !self.backend.is_official_url(&url) {
return Err(DownloadError::new(
bat_core::ErrorCode::NON_OFFICIAL_URL,
format!("拒绝下载非官方 URL：{url}"),
@@ -695,6 +722,17 @@ impl OfficialResourcePullService {
});
}
if self.max_concurrency > 1 {
return self.pull_planned_concurrently(
planned,
manifest,
official_hash_pairs,
total,
&mut progress,
&mut should_cancel,
);
}
// Phase B：按 plan 顺序处理每个 URL。下载或复用完成并写入 manifest 后，
// 立即尝试校验已经到齐的官方 `.bytes/.hash` pair，避免把文件级问题延后到整轮末尾。
let mut completed_count = 0usize;
@@ -824,6 +862,183 @@ impl OfficialResourcePullService {
})
}
fn pull_planned_concurrently(
&self,
planned: Vec<PlannedDownload>,
mut manifest: OfficialDownloadManifest,
official_hash_pairs: Vec<OfficialSeedHashPair>,
total: usize,
progress: &mut impl FnMut(OfficialResourcePullProgress),
should_cancel: &mut impl FnMut() -> bool,
) -> Result<OfficialResourcePullReport, DownloadError> {
if should_cancel() {
return Err("官方资源拉取已被停止请求中断".to_string().into());
}
for item in &planned {
progress(OfficialResourcePullProgress::started(
0,
total,
item.url.clone(),
));
}
let backend = CurlDownloadBackend { service: self };
let mut completed_count = 0usize;
let mut items_by_plan_index: Vec<Option<OfficialResourcePullItem>> =
(0..total).map(|_| None).collect();
let mut verified_hashes = Vec::new();
let mut verified_hash_urls = HashSet::<String>::new();
let mut processed_urls = HashSet::<String>::new();
DownloadScheduler::new(self.max_concurrency).execute_with_observer(
&backend,
planned.clone(),
|plan_index, result| {
let item = &planned[plan_index];
if should_cancel() {
return Err("官方资源拉取已被停止请求中断".to_string().into());
}
let needs_manifest = item.existing.is_none();
let verification = match result {
Ok(pull_result) => {
let verification_result = self
.clear_quarantine_entry(&item.url)
.and_then(|_| {
if needs_manifest {
self.record_download_manifest_entry(
&mut manifest,
&item.url,
&item.destination,
)
} else {
Ok(pull_result.verification.clone())
}
})
.and_then(|verification| {
if needs_manifest {
self.write_download_manifest(&manifest)
.map(|_| verification)
} else {
Ok(verification)
}
});
match verification_result {
Ok(verification) => verification,
Err(error) => {
let error = PullOneError::plain(format!(
"记录下载 manifest 失败：{error}"
));
self.record_quarantine_entry(
&item.url,
&item.destination,
&error,
)?;
progress(OfficialResourcePullProgress::failed(
completed_count,
total,
item.url.clone(),
&error,
));
return Err(DownloadError::new(
error.error_code(),
format!(
"官方资源下载失败：URL 已进入 quarantine，中止本轮同步、不发布不完整资源；url={} quarantine={}；{}",
item.url,
self.download_quarantine_path().display(),
error.message
),
));
}
}
}
Err(error) => {
self.record_quarantine_entry(&item.url, &item.destination, error)?;
progress(OfficialResourcePullProgress::failed(
completed_count,
total,
item.url.clone(),
error,
));
return Err(DownloadError::new(
error.error_code(),
format!(
"官方资源下载失败：URL 已进入 quarantine，中止本轮同步、不发布不完整资源；url={} quarantine={}；{}",
item.url,
self.download_quarantine_path().display(),
error.message
),
));
}
};
completed_count += 1;
progress(OfficialResourcePullProgress::finished(
completed_count,
total,
item.url.clone(),
result
.as_ref()
.expect("successful result handled above")
.status,
result
.as_ref()
.expect("successful result handled above")
.bytes,
result
.as_ref()
.expect("successful result handled above")
.transferred_bytes,
verification,
));
processed_urls.insert(item.url.clone());
let newly_verified_hashes = self.verify_ready_official_hashes(
&official_hash_pairs,
&processed_urls,
&mut verified_hash_urls,
&mut verified_hashes,
&mut manifest,
)?;
for verification in newly_verified_hashes {
progress(OfficialResourcePullProgress::verification(
completed_count,
total,
verification.data_url.clone(),
verification,
));
}
let pull_result = result
.as_ref()
.expect("successful result handled above");
items_by_plan_index[plan_index] = Some(OfficialResourcePullItem {
url: item.url.clone(),
destination: item.destination.clone(),
bytes: pull_result.bytes,
transferred_bytes: pull_result.transferred_bytes,
status: pull_result.status,
});
Ok(())
},
)?;
self.verify_all_official_hashes_are_complete(&official_hash_pairs, &verified_hash_urls)?;
let items = items_by_plan_index
.into_iter()
.enumerate()
.map(|(index, item)| {
item.ok_or_else(|| {
DownloadError::from(format!(
"并发下载结果缺少 plan index={index}，拒绝发布不完整资源"
))
})
})
.collect::<Result<Vec<_>, _>>()?;
Ok(OfficialResourcePullReport {
items,
verified_hashes,
})
}
/// Audits every URL in a pull plan against the local download manifest.
///
/// This performs no network I/O. It checks that each URL has a manifest
@@ -914,7 +1129,7 @@ impl OfficialResourcePullService {
/// `TableCatalog.bytes`, `BundlePackingInfo.bytes`, and
/// `MediaCatalog.bytes`.
pub fn fetch_bytes(&self, url: &str) -> Result<Vec<u8>, DownloadError> {
if !is_official_yostar_jp_url(url) {
if !self.backend.is_official_url(url) {
return Err(DownloadError::new(
bat_core::ErrorCode::NON_OFFICIAL_URL,
format!("拒绝拉取非官方 URL：{url}"),
@@ -1581,36 +1796,12 @@ impl OfficialResourcePullService {
fn destination_for_url(&self, url: &str) -> Result<PathBuf, String> {
self.ensure_output_root_safe()?;
if !is_official_yostar_jp_url(url) {
if !self.backend.is_official_url(url) {
return Err(format!("URL 不是官方 JP host：{url}"));
}
let rest = url
.strip_prefix("https://")
.ok_or_else(|| format!("官方 URL 必须使用 https：{url}"))?;
let (host, path) = rest
.split_once('/')
.ok_or_else(|| format!("官方 URL 缺少路径：{url}"))?;
let mut relative_destination = PathBuf::from(sanitize_segment(host));
for segment in path.split('/') {
if segment.is_empty() {
continue;
}
if segment == "." || segment == ".." {
return Err(format!("官方 URL 包含不安全路径片段：{url}"));
}
// 官方资源 URL 不携带 query/fragment。若出现则直接拒绝，而非静默剥除——
// 否则仅 query 不同的两个 URL 会映射到同一目标文件而相互覆盖，
// 并导致每轮 hash 复用校验失配、反复重下。
if segment.contains('?') || segment.contains('#') {
return Err(format!("官方资源 URL 不允许包含 query 或 fragment：{url}"));
}
relative_destination.push(sanitize_segment(segment));
}
let destination = self.output_root.join(relative_destination);
let relative_destination = self.backend.relative_destination(url)?;
let destination = destination_under_root(&self.output_root, &relative_destination)?;
ensure_path_within_root(&self.output_root, &destination)?;
Ok(destination)
}
@@ -1834,6 +2025,7 @@ fn default_download_quarantine_version() -> u32 {
/// Phase A 产出的单个下载计划项：URL、目标路径，以及若命中本地 manifest
/// 校验则带上「已验证可跳过」的结果（`existing`）。
#[derive(Debug, Clone)]
struct PlannedDownload {
url: String,
destination: PathBuf,
@@ -1854,6 +2046,21 @@ struct PullOneError {
retry_error: Option<CurlRetryError>,
}
struct CurlDownloadBackend<'a> {
service: &'a OfficialResourcePullService,
}
impl DownloaderBackend<PlannedDownload> for CurlDownloadBackend<'_> {
type Output = PullOneResult;
type Error = PullOneError;
fn download(&self, task: PlannedDownload) -> Result<Self::Output, Self::Error> {
task.existing
.map(Ok)
.unwrap_or_else(|| self.service.pull_one(&task.url, &task.destination))
}
}
impl PullOneError {
fn plain(message: String) -> Self {
Self {
@@ -2088,20 +2295,6 @@ fn read_u32_le(bytes: &[u8], offset: usize) -> u32 {
])
}
fn sanitize_segment(segment: &str) -> String {
segment
.chars()
.map(|ch| {
if ch.is_control() || matches!(ch, '/' | '\\' | ':' | '*' | '?' | '"' | '<' | '>' | '|')
{
'_'
} else {
ch
}
})
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
@@ -3263,21 +3456,26 @@ exit 22
.official_hash
.as_ref()
.expect("verification event must carry official hash detail");
let hash_finished_index = events
.iter()
.position(|event| {
event.kind == OfficialResourcePullProgressKind::Finished
&& event.url == hash.hash_url
let pair_finished_indices = [hash.data_url.as_str(), hash.hash_url.as_str()]
.into_iter()
.map(|url| {
events
.iter()
.position(|event| {
event.kind == OfficialResourcePullProgressKind::Finished
&& event.url == url
})
.expect("hash pair member must finish before verification")
})
.expect("hash sidecar must finish before verification");
.collect::<Vec<_>>();
let verification_index = events
.iter()
.position(|event| std::ptr::eq(event, *verification_event))
.expect("verification event must be present in event stream");
assert_eq!(
verification_index,
hash_finished_index + 1,
"official hash verification must run immediately after sidecar is complete"
pair_finished_indices.into_iter().max().unwrap() + 1,
"official hash verification must run immediately after the hash pair is complete"
);
}
@@ -3320,7 +3518,8 @@ exit 22
write_fake_curl(&curl_path);
// 顺序下载：每个 URL 恰好一次 started + 一次 finished，全部文件落盘。
let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path)
.with_max_concurrency(1);
let plan = build_official_pull_plan_for_platforms(
discovery_plan(),
inventory(),
@@ -3354,6 +3553,71 @@ exit 22
assert_eq!(manifest.entries.len(), all_urls.len());
}
#[test]
fn downloads_run_with_bounded_concurrency_and_keep_report_order() {
let out_dir = TempDir::new().unwrap();
let bin_dir = TempDir::new().unwrap();
let curl_path = bin_dir.path().join("curl");
write_fake_curl(&curl_path);
let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path)
.with_max_concurrency(3);
let plan = build_official_pull_plan_for_platforms(
discovery_plan(),
inventory(),
&[PatchPlatform::Windows],
);
let all_urls = plan.all_urls().unwrap();
let mut events = Vec::new();
let report = service
.pull_with_progress(&plan, |event| events.push(event))
.unwrap();
assert_eq!(service.max_concurrency(), 3);
assert_eq!(report.items.len(), all_urls.len());
assert_eq!(
report
.items
.iter()
.map(|item| &item.url)
.collect::<Vec<_>>(),
all_urls.iter().collect::<Vec<_>>()
);
for url in &all_urls {
assert_eq!(
events
.iter()
.filter(|event| {
event.url == *url && event.kind == OfficialResourcePullProgressKind::Started
})
.count(),
1,
"url {url} 的 started 次数"
);
assert_eq!(
events
.iter()
.filter(|event| {
event.url == *url
&& event.kind == OfficialResourcePullProgressKind::Finished
})
.count(),
1,
"url {url} 的 finished 次数"
);
}
let finished_indices = events
.iter()
.filter(|event| event.kind == OfficialResourcePullProgressKind::Finished)
.map(|event| event.index)
.collect::<Vec<_>>();
assert_eq!(finished_indices, (1..=all_urls.len()).collect::<Vec<_>>());
assert_eq!(
service.read_download_manifest().unwrap().entries.len(),
all_urls.len()
);
}
#[test]
fn retries_transient_download_failures() {
let out_dir = TempDir::new().unwrap();
infrastructure/src/official_textunit_queue.rs
+19 -1
View File
@@ -16,6 +16,7 @@ use crate::path_security::{
ensure_path_within_root, ensure_safe_file_target, read_file_no_symlink, write_file_atomic,
STATE_FILE_MODE,
};
use crate::sync_translation_task_repository_at;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
@@ -121,6 +122,10 @@ pub struct OfficialTextUnitTaskQuery {
pub text_unit_format: Option<String>,
/// Filter tasks by whether a diagnostic reason is present.
pub has_reason: Option<bool>,
/// Filter tasks by whether a provider failure reason is present.
pub has_failure_reason: Option<bool>,
/// Filter by mutable provider-worker status.
pub task_status: Option<String>,
}
/// Aggregate counters for an incremental TextUnit task queue.
@@ -355,6 +360,16 @@ pub fn write_official_textunit_queues(
write_textunit_task_queue_at(resource_root, &task_queue)?;
let task_queue_path = resource_root.join(OFFICIAL_TEXTUNIT_TASK_QUEUE_FILE);
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|error| format!("构建官方 TextUnit 任务状态同步运行时失败：{error}"))?;
runtime
.block_on(sync_translation_task_repository_at(
resource_root,
&task_queue,
))
.map_err(|error| format!("同步官方 TextUnit 任务状态到 SQLite 失败：{error}"))?;
let crowdin_queue =
CrowdinTextUnitQueue::from_textunit_task_queue(task_queue_path.clone(), &task_queue);
write_crowdin_textunit_queue_at(resource_root, &crowdin_queue)?;
@@ -492,7 +507,10 @@ fn parse_entries_by_destination(
by_destination
}
fn textunit_task_matches(task: &OfficialTextUnitTask, query: &OfficialTextUnitTaskQuery) -> bool {
pub(crate) fn textunit_task_matches(
task: &OfficialTextUnitTask,
query: &OfficialTextUnitTaskQuery,
) -> bool {
if query
.task_id
.as_ref()
infrastructure/src/official_update.rs
+101 -17
View File
@@ -29,6 +29,9 @@ use crate::path_security::{
read_file_no_symlink, validate_output_root, write_file_atomic, STATE_FILE_MODE,
};
use crate::release_flow::ReleaseFlowStatusCode;
use crate::translation_tasks::{
build_translation_handoff, sync_translation_task_repository_at, write_translation_handoff_at,
};
use crate::{
build_official_pull_plan_for_platform_inventory, build_official_sync_plan,
changed_endpoint_urls, default_official_platforms, DownloadError,
@@ -42,15 +45,17 @@ use crate::{
read_parse_cache_at, OfficialParseCacheService, OfficialParseConfig, OfficialParseSummary,
};
use crate::{FileSystemCasRepository, SqliteResourceRepository};
use crate::{DEFAULT_DOWNLOAD_CONCURRENCY, MAX_DOWNLOAD_CONCURRENCY, MIN_DOWNLOAD_CONCURRENCY};
use bat_adapters::official::game_main_config::YostarJpGameMainConfig;
use bat_adapters::official::inventory::{
YostarJpPlatformCatalogInventory, YostarJpPlatformDownloadInventory,
};
use bat_adapters::official::launcher::YostarJpLauncherManifestFile;
use bat_adapters::official::yostar_jp::{
server_info_url, PatchPlatform, YostarJpResourceDiscoveryPlan, YostarJpResourceEndpoint,
PatchPlatform, YostarJpResourceDiscoveryPlan, YostarJpResourceEndpoint,
YostarJpResourceEndpointKind, YostarJpServerInfo, YostarJpSyncSnapshot,
};
use bat_adapters::official::{
OfficialResourceBackend, PlatformCatalogInput, YostarJpBackend,
YostarJpPlatformDownloadInventory,
};
use bat_core::ErrorCode;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
@@ -115,6 +120,10 @@ pub struct OfficialUpdateConfig {
pub curl_command: PathBuf,
/// Proxy selection used by all official `curl` transfers.
pub curl_proxy: CurlProxyConfig,
/// Maximum number of resource downloads executed concurrently.
///
/// `8` is the default; values are accepted only in `1..=256`.
pub download_concurrency: usize,
/// Unzip command used when a metadata change requires GameMainConfig parsing.
pub unzip_command: PathBuf,
/// Dry run reports decisions and optional plan URLs without writing sync state.
@@ -151,6 +160,7 @@ impl Default for OfficialUpdateConfig {
snapshot_path: None,
curl_command: PathBuf::from("curl"),
curl_proxy: CurlProxyConfig::default(),
download_concurrency: DEFAULT_DOWNLOAD_CONCURRENCY,
unzip_command: PathBuf::from("unzip"),
dry_run: false,
plan: false,
@@ -844,6 +854,8 @@ pub struct OfficialUpdateReport {
pub crowdin_textunit_queue_path: Option<PathBuf>,
/// Incremental TextUnit task queue summary.
pub textunit_task_summary: Option<OfficialTextUnitTaskSummary>,
/// Versioned translation worker handoff view.
pub translation_handoff_path: Option<PathBuf>,
/// Snapshot path written after success.
pub snapshot_written: Option<PathBuf>,
/// Launcher bootstrap artifact written for this run.
@@ -1221,7 +1233,8 @@ impl OfficialUpdateService {
&active_resource_root,
&config.curl_command,
)
.with_proxy_config(config.curl_proxy.clone());
.with_proxy_config(config.curl_proxy.clone())
.with_max_concurrency(config.download_concurrency);
let snapshot_path = snapshot_path_for(config, &active_resource_root);
let bootstrap_cache_path = config.bootstrap_cache_path();
@@ -1668,6 +1681,7 @@ impl OfficialUpdateService {
textunit_task_queue_path: None,
crowdin_textunit_queue_path: None,
textunit_task_summary: None,
translation_handoff_path: None,
snapshot_written: None,
launcher_bootstrap_artifact_path: None,
repository_import_enabled: config.import_repository,
@@ -1846,7 +1860,8 @@ impl OfficialUpdateService {
&publish_plan.staging_path,
&config.curl_command,
)
.with_proxy_config(config.curl_proxy.clone());
.with_proxy_config(config.curl_proxy.clone())
.with_max_concurrency(config.download_concurrency);
let pruned_stale_resource_count = staging_fetcher
.prune_stale_manifest_entries(&pull_plan)
.map_err(anyhow::Error::msg)?;
@@ -2201,6 +2216,7 @@ fn run_post_sync_textunit_queue_if_needed(
&& is_crowdin_textunit_queue_current(resource_root, &queue)
.map_err(anyhow::Error::msg)?
{
sync_translation_task_repository(resource_root, &queue)?;
progress(OfficialUpdateProgress::new(
"textunit",
format!(
@@ -2212,7 +2228,8 @@ fn run_post_sync_textunit_queue_if_needed(
));
report.textunit_task_queue_path = Some(task_queue_path);
report.crowdin_textunit_queue_path = Some(crowdin_queue_path);
report.textunit_task_summary = Some(queue.summary);
report.textunit_task_summary = Some(queue.summary.clone());
write_translation_handoff_for_release(resource_root, &queue, report, progress)?;
return Ok(());
}
progress(OfficialUpdateProgress::new(
@@ -2227,6 +2244,19 @@ fn run_post_sync_textunit_queue_if_needed(
run_post_sync_textunit_queue(resource_root, report, progress, should_cancel)
}
fn sync_translation_task_repository(
resource_root: &Path,
queue: &crate::official_textunit_queue::OfficialTextUnitTaskQueue,
) -> anyhow::Result<()> {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
runtime
.block_on(sync_translation_task_repository_at(resource_root, queue))
.map_err(|error| anyhow::anyhow!("{error}"))?;
Ok(())
}
fn run_post_sync_textunit_queue(
resource_root: &Path,
report: &mut OfficialUpdateReport,
@@ -2257,6 +2287,11 @@ fn run_post_sync_textunit_queue(
),
));
apply_textunit_queue_report(report, queue_report);
if let Some(queue) =
read_textunit_task_queue_at(resource_root).map_err(anyhow::Error::msg)?
{
write_translation_handoff_for_release(resource_root, &queue, report, progress)?;
}
}
Err(error) => {
progress(OfficialUpdateProgress::new(
@@ -2277,6 +2312,48 @@ fn apply_textunit_queue_report(
report.textunit_task_summary = Some(queue_report.summary);
}
fn write_translation_handoff_for_release(
resource_root: &Path,
queue: &crate::official_textunit_queue::OfficialTextUnitTaskQueue,
report: &mut OfficialUpdateReport,
progress: &mut dyn FnMut(OfficialUpdateProgress),
) -> anyhow::Result<()> {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
let tasks = runtime.block_on(async {
let repository = crate::translation_tasks::SqliteTranslationTaskRepository::new(
crate::translation_tasks::SqliteTranslationTaskRepository::repository_path(
resource_root,
),
)
.await
.map_err(|error| anyhow::anyhow!("{error}"))?;
repository
.sync_queue(queue)
.await
.map_err(|error| anyhow::anyhow!("{error}"))?;
repository
.list(&crate::official_textunit_queue::OfficialTextUnitTaskQuery::default())
.await
.map_err(|error| anyhow::anyhow!("{error}"))
})?;
let handoff = build_translation_handoff(queue, &tasks);
write_translation_handoff_at(resource_root, &handoff).map_err(anyhow::Error::msg)?;
let path = resource_root.join(crate::translation_tasks::TRANSLATION_HANDOFF_FILE);
report.translation_handoff_path = Some(path.clone());
progress(OfficialUpdateProgress::new(
"textunit",
format!(
"翻译 handoff 已发布：任务={} provider_run={} {}",
handoff.units.len(),
handoff.provider_runs.len(),
path.display()
),
));
Ok(())
}
fn run_post_sync_repository_import(
config: &OfficialUpdateConfig,
resource_root: &Path,
@@ -2370,8 +2447,9 @@ fn build_pull_plan(
should_cancel: &mut dyn FnMut() -> bool,
) -> anyhow::Result<OfficialResourcePullPlan> {
check_shutdown_requested(should_cancel)?;
let discovery = server_info
.discovery_plan(connection_group, app_version, platforms)
let backend = YostarJpBackend;
let discovery = backend
.discovery_plan(server_info, connection_group, app_version, platforms)
.map_err(anyhow::Error::msg)?;
progress(OfficialUpdateProgress::new(
"plan",
@@ -2402,7 +2480,9 @@ fn load_server_info(
match source {
OfficialServerInfoSource::LocalPath(path) => Ok(fs::read(path)?),
OfficialServerInfoSource::OfficialFile(file_name) => {
let url = server_info_url(file_name).map_err(anyhow::Error::msg)?;
let url = YostarJpBackend
.server_info_url(file_name)
.map_err(anyhow::Error::msg)?;
fetcher.fetch_bytes(&url).map_err(anyhow::Error::new)
}
OfficialServerInfoSource::OfficialUrl(url) => {
@@ -2489,6 +2569,7 @@ fn waiting_for_official_resources_report(
textunit_task_queue_path: None,
crowdin_textunit_queue_path: None,
textunit_task_summary: None,
translation_handoff_path: None,
snapshot_written: None,
launcher_bootstrap_artifact_path: None,
repository_import_enabled: config.import_repository,
@@ -2894,6 +2975,12 @@ pub fn diff_extended_snapshot(
}
fn validate_update_paths(config: &OfficialUpdateConfig) -> Result<(), String> {
if !(MIN_DOWNLOAD_CONCURRENCY..=MAX_DOWNLOAD_CONCURRENCY).contains(&config.download_concurrency)
{
return Err(format!(
"下载并发数必须在 {MIN_DOWNLOAD_CONCURRENCY}..={MAX_DOWNLOAD_CONCURRENCY} 范围内"
));
}
validate_output_root(&config.output_root)?;
validate_output_root(&config.localized_output_root)?;
validate_separate_output_roots(&config.output_root, &config.localized_output_root)?;
@@ -4047,17 +4134,14 @@ fn build_inventory_from_seed_catalogs(
anyhow::anyhow!("官方发现结果缺少 {} MediaCatalog.bytes", platform.as_str())
})?;
platform_catalogs.push(YostarJpPlatformCatalogInventory::from_catalog_bytes(
*platform,
platform_catalogs.push(PlatformCatalogInput {
platform: *platform,
bundle_packing_info,
media_catalog,
));
});
}
Ok(YostarJpPlatformDownloadInventory::from_catalog_bytes(
table_catalog,
platform_catalogs,
))
Ok(YostarJpBackend.inventory(table_catalog, &platform_catalogs))
}
#[derive(Debug, Default)]
infrastructure/src/translation_tasks.rs
+1288
View File
File diff suppressed because it is too large Load Diff