Asfmq
2c8d0b8f8b
- LAMOST 新增 DR12/13/14 及子版本（v0/v1.0/v1.1/v2.0）维度，Internal 发布标记需登录认证并前端灰显，release×subtype 交叉约束下沉至 capabilities 统一声明 - ObservationFetcher trait 扩展版本/认证/交叉约束能力声明，version 参数贯穿 client→service→API→Agent tool→前端全链路 - 安全：observation cache SQL 全参数绑定 + LIKE 转义、cone_cache_hash 加长度前缀防碰撞、DESI survey/program 白名单防穿越 - 异步化：persist/cached_files_total_size/maybe_persist_tool_result迁移到 tokio::fs；cancelled_runs 与 session_permission_checkers改用 DashMap；auth 读锁优先 + 60s 节流 - Gaia 去 native-tls 改禁用连接池规避 UnexpectedEof，reqwest 移除 native-tls feature - 重构：Source/ProductType from_str 集中解析、download 模块拆分为 try_download_pdf/html、AgentRuntime::init 抽取共享逻辑 - 部署：新增 deploy.sh 一键打包推送脚本、catch-panic 启用
273 lines
8.4 KiB
Rust
273 lines
8.4 KiB
Rust
// src/services/observation/dispatch.rs
|
||
//
|
||
// 统一编排入口 —— 把 (坐标/标识符) + (Source, ProductType) 翻译成对 fetcher 的调用
|
||
//
|
||
// 这是 observation 服务的最上层：api/observation.rs 和 agent tools 都通过本模块调用。
|
||
// 本模块只负责编排（选源、并发控制、失败聚合），不关心各源 HTTP 细节（那是 fetcher 的职责）。
|
||
|
||
use crate::api::AppState;
|
||
use crate::services::observation::fetcher::{pick_nearest, Candidate};
|
||
use crate::services::observation::registry::ObservationRegistry;
|
||
use crate::services::observation::types::{
|
||
DownloadFailure, FindStrategy, ObservationBatch, ObservationProduct, ObservationRequest,
|
||
ProductSpec,
|
||
};
|
||
use anyhow::Result;
|
||
use tracing::{info, warn};
|
||
|
||
/// 统一检索入口：按坐标 cone 检索候选源（**不下载**）
|
||
///
|
||
/// 与 `download_observation` 的 ByCoordinates 模式相比，本函数只执行 cone_search_cached
|
||
/// 这一步，把命中的候选源列表返回给调用方预览，由调用方决定是否下载、下载哪些。
|
||
///
|
||
/// 标识符模式无需"检索"——标识符本身就是确定源，调用方直接走 download 即可。
|
||
pub async fn search_observation(
|
||
state: &AppState,
|
||
registry: &ObservationRegistry,
|
||
source: crate::services::observation::types::Source,
|
||
product: &ProductSpec,
|
||
ra: f64,
|
||
dec: f64,
|
||
radius_deg: f64,
|
||
release: Option<&str>,
|
||
version: Option<&str>,
|
||
) -> Result<Vec<Candidate>> {
|
||
let subtype = product.subtype.as_deref();
|
||
let fetcher = registry.get(source, product.product, subtype)?;
|
||
info!(
|
||
"[Observation] search source={:?} product={:?} ra={:.4} dec={:.4} radius={:.4}°",
|
||
source, product.product, ra, dec, radius_deg
|
||
);
|
||
let matched = fetcher
|
||
.cone_search_cached(state, ra, dec, radius_deg, release, subtype, version)
|
||
.await?;
|
||
Ok(matched)
|
||
}
|
||
|
||
/// 统一下载入口：按坐标或按标识符下载观测数据
|
||
///
|
||
/// 这是观测数据的唯一上层入口。调用方通过 `ObservationRequest` 选择模式，force 控制是否忽略缓存。
|
||
pub async fn download_observation(
|
||
state: &AppState,
|
||
registry: &ObservationRegistry,
|
||
request: &ObservationRequest,
|
||
force: bool,
|
||
) -> Result<ObservationBatch> {
|
||
match request {
|
||
ObservationRequest::ByCoordinates {
|
||
source,
|
||
product,
|
||
ra,
|
||
dec,
|
||
radius_deg,
|
||
strategy,
|
||
release,
|
||
version,
|
||
} => {
|
||
download_by_coordinates(
|
||
state,
|
||
registry,
|
||
*source,
|
||
product,
|
||
*ra,
|
||
*dec,
|
||
*radius_deg,
|
||
*strategy,
|
||
release.as_deref(),
|
||
version.as_deref(),
|
||
force,
|
||
)
|
||
.await
|
||
}
|
||
ObservationRequest::ByIdentifier {
|
||
source,
|
||
product,
|
||
identifiers,
|
||
release,
|
||
version,
|
||
} => {
|
||
download_by_identifiers(
|
||
state,
|
||
registry,
|
||
*source,
|
||
product,
|
||
identifiers,
|
||
release.as_deref(),
|
||
version.as_deref(),
|
||
force,
|
||
)
|
||
.await
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 按坐标：cone 检索 → 选源 → 逐个下载
|
||
async fn download_by_coordinates(
|
||
state: &AppState,
|
||
registry: &ObservationRegistry,
|
||
source: crate::services::observation::types::Source,
|
||
product: &ProductSpec,
|
||
ra: f64,
|
||
dec: f64,
|
||
radius_deg: f64,
|
||
strategy: FindStrategy,
|
||
release: Option<&str>,
|
||
version: Option<&str>,
|
||
force: bool,
|
||
) -> Result<ObservationBatch> {
|
||
let subtype = product.subtype.as_deref();
|
||
let fetcher = registry.get(source, product.product, subtype)?;
|
||
info!(
|
||
"[Observation] by_coords source={:?} product={:?} ra={:.4} dec={:.4} radius={:.4}° strategy={:?}",
|
||
source, product.product, ra, dec, radius_deg, strategy
|
||
);
|
||
|
||
let matched = fetcher
|
||
.cone_search_cached(state, ra, dec, radius_deg, release, subtype, version)
|
||
.await?;
|
||
let matched_count = matched.len();
|
||
|
||
if matched_count == 0 {
|
||
return Ok(ObservationBatch {
|
||
source,
|
||
product: product.clone(),
|
||
ra: Some(ra),
|
||
dec: Some(dec),
|
||
radius_deg: Some(radius_deg),
|
||
matched_count: 0,
|
||
products: Vec::new(),
|
||
failures: Vec::new(),
|
||
});
|
||
}
|
||
|
||
let targets: Vec<usize> = match strategy {
|
||
FindStrategy::Nearest => vec![pick_nearest(&matched, ra, dec)],
|
||
FindStrategy::All => (0..matched.len()).collect(),
|
||
};
|
||
|
||
let mut products = Vec::with_capacity(targets.len());
|
||
let mut failures = Vec::new();
|
||
for idx in targets {
|
||
let candidate = &matched[idx];
|
||
match fetcher
|
||
.fetch(state, candidate, release, subtype, version, force)
|
||
.await
|
||
{
|
||
Ok(p) => products.push(p),
|
||
Err(e) => {
|
||
let label = candidate.label.clone();
|
||
warn!("[Observation] 下载失败 {}: {}", label, e);
|
||
failures.push(DownloadFailure {
|
||
source_label: label,
|
||
error: format!("{:#}", e),
|
||
});
|
||
}
|
||
}
|
||
}
|
||
|
||
Ok(ObservationBatch {
|
||
source,
|
||
product: product.clone(),
|
||
ra: Some(ra),
|
||
dec: Some(dec),
|
||
radius_deg: Some(radius_deg),
|
||
matched_count,
|
||
products,
|
||
failures,
|
||
})
|
||
}
|
||
|
||
/// 按标识符：解析各源标识字符串 → 逐个下载（跳过 cone 检索）
|
||
async fn download_by_identifiers(
|
||
state: &AppState,
|
||
registry: &ObservationRegistry,
|
||
source: crate::services::observation::types::Source,
|
||
product: &ProductSpec,
|
||
identifiers: &[String],
|
||
release: Option<&str>,
|
||
version: Option<&str>,
|
||
force: bool,
|
||
) -> Result<ObservationBatch> {
|
||
let subtype = product.subtype.as_deref();
|
||
let fetcher = registry.get(source, product.product, subtype)?;
|
||
if identifiers.is_empty() {
|
||
return Err(anyhow::anyhow!(
|
||
"标识符列表不能为空（source={:?} product={:?}）",
|
||
source,
|
||
product.product
|
||
));
|
||
}
|
||
info!(
|
||
"[Observation] by_id source={:?} product={:?} count={} release={:?}",
|
||
source,
|
||
product.product,
|
||
identifiers.len(),
|
||
release
|
||
);
|
||
|
||
let mut products = Vec::with_capacity(identifiers.len());
|
||
let mut failures = Vec::new();
|
||
for id in identifiers {
|
||
// 先把标识符解析为 Candidate（验证格式 + 提取字段）
|
||
let candidate = match fetcher
|
||
.resolve_identifier(id, release, subtype, version)
|
||
.await
|
||
{
|
||
Ok(c) => c,
|
||
Err(e) => {
|
||
warn!("[Observation] 标识符解析失败 {}: {}", id, e);
|
||
failures.push(DownloadFailure {
|
||
source_label: id.clone(),
|
||
error: format!("{:#}", e),
|
||
});
|
||
continue;
|
||
}
|
||
};
|
||
match fetcher
|
||
.fetch(state, &candidate, release, subtype, version, force)
|
||
.await
|
||
{
|
||
Ok(p) => products.push(p),
|
||
Err(e) => {
|
||
warn!("[Observation] 下载失败 {}: {}", id, e);
|
||
failures.push(DownloadFailure {
|
||
source_label: id.clone(),
|
||
error: format!("{:#}", e),
|
||
});
|
||
}
|
||
}
|
||
}
|
||
|
||
Ok(ObservationBatch {
|
||
source,
|
||
product: product.clone(),
|
||
ra: None,
|
||
dec: None,
|
||
radius_deg: None,
|
||
matched_count: products.len() + failures.len(),
|
||
products,
|
||
failures,
|
||
})
|
||
}
|
||
|
||
/// 便捷：从单个标识符构造请求并下载（供 CLI / 简单场景使用）
|
||
pub async fn download_one(
|
||
state: &AppState,
|
||
registry: &ObservationRegistry,
|
||
source: crate::services::observation::types::Source,
|
||
product: ProductSpec,
|
||
identifier: &str,
|
||
release: Option<&str>,
|
||
version: Option<&str>,
|
||
force: bool,
|
||
) -> Result<ObservationProduct> {
|
||
let subtype = product.subtype.as_deref();
|
||
let fetcher = registry.get(source, product.product, subtype)?;
|
||
let candidate = fetcher
|
||
.resolve_identifier(identifier, release, subtype, version)
|
||
.await?;
|
||
fetcher
|
||
.fetch(state, &candidate, release, subtype, version, force)
|
||
.await
|
||
}
|