lxy
795935bf6a
- 文件消息入站:20MB 上限落盘引用,模型经 read_file 按需读盘;下载带 3 次退避+120s 超时防线,复用媒体域 hash 去重/OSS 双写 - 群聊门控对齐 u-claw 模式:前移至类型分流前,@自己/@所有人响应,未艾特静默旁观(原门控后置致未支持类型误回提示卡) - 修复补充注入丢失:pending_injects 未消化计数+收敛出口续轮+桌面注入 IPC+前端本地气泡 - df-ai-core 测试构造器补 source 字段(溯源字段遗漏,全量门禁暴露)
350 lines
14 KiB
Rust
350 lines
14 KiB
Rust
//! 媒体存储域(#958 图片入站引用化 + 跨设备文件服务)
|
|
//!
|
|
//! 单一归属:媒体字节 → 存储引用(`ContentPart::Image.url`)的全部决策收在本模块——
|
|
//! hash 去重落盘(同内容同文件,转发/重发不重复存储)、可选 OSS 双写(跨设备可解析)、
|
|
//! 引用形态选择(oss `https://` / 本机 `file://`)。飞书协议(下载/上传回传)留 feishu_channel,
|
|
//! 本模块零协议依赖。
|
|
//!
|
|
//! 引用设计:不自造 scheme——`https://`(OSS)是 OpenAI 端点原生直传格式,anthropic 侧
|
|
//! 已有 url→base64 预拉;`file://` 由 df-ai::image_ref 发送前展开为 base64。展开按 url
|
|
//! 形态自然分流,无自定义协议。
|
|
//!
|
|
//! OSS 失败语义:上传失败不阻塞——本机快照已落盘,引用降级 `file://`(本机可解析,
|
|
//! 跨设备解析降级为「图片文件缺失」文本,诚实不炸请求)。
|
|
|
|
use df_ai::provider::ContentPart;
|
|
use serde::Deserialize;
|
|
|
|
/// OSS 配置(app_settings KV `media-oss-config` JSON;缺省=纯本地模式)。
|
|
/// public_base_url 为空时取 `https://{bucket}.{endpoint}`(bucket 私有时需自备 CDN 域)
|
|
#[derive(Debug, Clone, Deserialize)]
|
|
pub struct OssConfig {
|
|
pub bucket: String,
|
|
/// 如 oss-cn-hangzhou.aliyuncs.com
|
|
pub endpoint: String,
|
|
pub access_key_id: String,
|
|
pub access_key_secret: String,
|
|
#[serde(default)]
|
|
pub public_base_url: String,
|
|
}
|
|
|
|
/// app_settings 键:媒体 OSS 配置(KV JSON,设置面板后续批;当前可手配)
|
|
pub const KV_MEDIA_OSS_CONFIG: &str = "media-oss-config";
|
|
|
|
/// 读 OSS 配置(KV;未配置/解析失败 → None=纯本地模式)。媒体域唯一读取口,
|
|
/// 飞书入站与 spawn offload 共用
|
|
pub async fn read_oss_config(db: &df_storage::db::Database) -> Option<OssConfig> {
|
|
let repo = df_storage::crud::SettingsRepo::new(db);
|
|
let raw = repo.get(KV_MEDIA_OSS_CONFIG).await.ok().flatten()?;
|
|
serde_json::from_str(&raw).ok()
|
|
}
|
|
|
|
/// base64 图片片 → 落盘引用片(spawn 统一 offload,#958 引用化):
|
|
/// 桌面粘贴图/前端上传在此收口,消息只存引用,SQLite 不进大 base64 字符串
|
|
/// (历史重放不再全量携带)。落盘失败保原 base64 片(不丢图,仅失去引用化收益);
|
|
/// 引用片/文本片原样透传。media_type 缺失的片无法规范落盘,原样保留
|
|
pub async fn offload_image_parts(parts: Vec<ContentPart>, store: &MediaStore) -> Vec<ContentPart> {
|
|
use base64::Engine as _;
|
|
let mut out = Vec::with_capacity(parts.len());
|
|
for p in parts {
|
|
out.push(match p {
|
|
ContentPart::Image { url, base64: Some(b64), media_type, alt } => {
|
|
let decoded = base64::engine::general_purpose::STANDARD
|
|
.decode(b64.as_bytes())
|
|
.ok()
|
|
.filter(|b| !b.is_empty());
|
|
match (decoded, media_type.clone()) {
|
|
(Some(bytes), Some(mt)) => match store.save(&bytes, &mt, None).await {
|
|
Ok(url) => ContentPart::Image {
|
|
url: Some(url),
|
|
base64: None,
|
|
media_type,
|
|
alt,
|
|
},
|
|
Err(e) => {
|
|
tracing::warn!(err = %e, "图片落盘失败,保留 base64 片");
|
|
ContentPart::Image { url, base64: Some(b64), media_type, alt }
|
|
}
|
|
},
|
|
// 解码失败/空图/无 media_type:原样透传,provider 层按既有语义处置
|
|
_ => ContentPart::Image { url, base64: Some(b64), media_type, alt },
|
|
}
|
|
}
|
|
other => other,
|
|
});
|
|
}
|
|
out
|
|
}
|
|
|
|
/// 媒体存储。`oss` 为 None 时纯本地模式(引用全部 file://)
|
|
pub struct MediaStore {
|
|
data_dir: std::path::PathBuf,
|
|
oss: Option<OssConfig>,
|
|
}
|
|
|
|
impl MediaStore {
|
|
pub fn new(data_dir: std::path::PathBuf, oss: Option<OssConfig>) -> Self {
|
|
Self { data_dir, oss }
|
|
}
|
|
|
|
/// 落盘并返回引用 url。本地快照必落(hash 去重,同内容幂等);OSS 配置在则上传,
|
|
/// 成功引用 `https://`(跨设备可解析),失败降级 `file://`(不阻塞,本机零损)。
|
|
///
|
|
/// `original_name`:原始文件名(文件消息带,图片消息通常 None)——缀在 hash 后保
|
|
/// 目录可读,清洗路径穿越字符防越界;None/空则纯 hash 名
|
|
pub async fn save(
|
|
&self,
|
|
bytes: &[u8],
|
|
media_type: &str,
|
|
original_name: Option<&str>,
|
|
) -> Result<String, String> {
|
|
self.save_with_path(bytes, media_type, original_name)
|
|
.await
|
|
.map(|(url, _)| url)
|
|
}
|
|
|
|
/// 落盘并返回 (引用 url, 本机绝对路径)。文件消息入站用:模型 read_file 只认本机
|
|
/// 路径,OSS `https://` 引用对本机读取无意义,须另取落盘路径;图片入站只要引用,
|
|
/// 走 save 薄包装。落盘/双写/降级语义与 save 完全一致
|
|
pub async fn save_with_path(
|
|
&self,
|
|
bytes: &[u8],
|
|
media_type: &str,
|
|
original_name: Option<&str>,
|
|
) -> Result<(String, std::path::PathBuf), String> {
|
|
let key = content_key(bytes, media_type, original_name);
|
|
let dir = self.data_dir.join("media");
|
|
tokio::fs::create_dir_all(&dir)
|
|
.await
|
|
.map_err(|e| format!("创建媒体目录失败: {e}"))?;
|
|
let path = dir.join(&key);
|
|
// 存在即命中(hash 去重):跳过重写,读侧幂等
|
|
if !tokio::fs::try_exists(&path).await.unwrap_or(false) {
|
|
if let Err(e) = tokio::fs::write(&path, bytes).await {
|
|
// 半写残件不留盘(对齐 u-claw fetchResource 防线)
|
|
tokio::fs::remove_file(&path).await.ok();
|
|
return Err(format!("媒体写入失败: {e}"));
|
|
}
|
|
}
|
|
let url = match &self.oss {
|
|
Some(cfg) => match upload_to_oss(cfg, &key, bytes, media_type).await {
|
|
Ok(()) => public_url(cfg, &key),
|
|
Err(e) => {
|
|
tracing::warn!(key = %key, err = %e, "OSS 上传失败,引用降级本机 file://");
|
|
local_ref(&path)
|
|
}
|
|
},
|
|
None => local_ref(&path),
|
|
};
|
|
Ok((url, path))
|
|
}
|
|
}
|
|
|
|
/// 内容 sha256 全量 hex。落盘键与展示名回退共用同一 hash 口径(飞书文件入站无原始名
|
|
/// 时展示名取 `file_<hash 前 8>`,与落盘 hash 命名同源可对账)
|
|
pub fn content_hash(bytes: &[u8]) -> String {
|
|
use sha2::Digest as _;
|
|
hex::encode(sha2::Sha256::digest(bytes))
|
|
}
|
|
|
|
/// 内容寻址文件名:`<sha256前16hex>[_<清洗原始名>]<ext>`(对齐 u-claw safeAttachName
|
|
/// 模式)——hash 前缀保去重(同内容必同文件)且排序聚簇,原始名缀后保目录可读。
|
|
/// 平铺不分会话目录:会话归属经 DB 消息引用反查,文件系统不编码会话(否则同图跨会话
|
|
/// 重复存储,去重失效);个人桌面应用媒体量级百~千,hash 前缀平铺可管理,分桶待量级上来再说
|
|
fn content_key(bytes: &[u8], media_type: &str, original_name: Option<&str>) -> String {
|
|
let digest = content_hash(bytes)[..16].to_string();
|
|
// 扩展名:原始名自带 ext 优先(保真),否则按 media_type 规范化
|
|
let name = original_name.map(sanitize_name).filter(|s| !s.is_empty());
|
|
let ext = name
|
|
.as_deref()
|
|
.and_then(|n| {
|
|
std::path::Path::new(n)
|
|
.extension()
|
|
.map(|e| format!(".{}", e.to_string_lossy()))
|
|
})
|
|
.unwrap_or_else(|| ext_of(media_type).to_string());
|
|
match name {
|
|
Some(n) => format!("{digest}_{n}"),
|
|
None => format!("{digest}{ext}"),
|
|
}
|
|
}
|
|
|
|
/// 原始文件名清洗(防路径穿越/过长):/ \ .. → _,截 60 字符
|
|
fn sanitize_name(name: &str) -> String {
|
|
let cleaned: String = name
|
|
.replace('/', "_")
|
|
.replace('\\', "_")
|
|
.replace("..", "_")
|
|
.chars()
|
|
.take(60)
|
|
.collect();
|
|
cleaned.trim().to_string()
|
|
}
|
|
|
|
fn ext_of(media_type: &str) -> &'static str {
|
|
match media_type {
|
|
"image/png" => ".png",
|
|
"image/jpeg" => ".jpg",
|
|
"image/gif" => ".gif",
|
|
"image/webp" => ".webp",
|
|
_ => ".bin",
|
|
}
|
|
}
|
|
|
|
/// 本机引用(file:// + 原生绝对路径;df-ai::image_ref 展开端同前缀约定)
|
|
fn local_ref(path: &std::path::Path) -> String {
|
|
format!("file://{}", path.display())
|
|
}
|
|
|
|
/// OSS 公网引用:自定义 CDN 域优先,缺省桶域名
|
|
fn public_url(cfg: &OssConfig, key: &str) -> String {
|
|
if cfg.public_base_url.is_empty() {
|
|
format!("https://{}.{}/{}", cfg.bucket, cfg.endpoint, key)
|
|
} else {
|
|
format!("{}/{}", cfg.public_base_url.trim_end_matches('/'), key)
|
|
}
|
|
}
|
|
|
|
/// 阿里云 OSS PUT 上传(Header 签名版,无需 SDK):
|
|
/// Signature = base64(hmac-sha1(SK, "PUT\n\n{ctype}\n{date}\n/{bucket}/{key}"))
|
|
async fn upload_to_oss(
|
|
cfg: &OssConfig,
|
|
key: &str,
|
|
bytes: &[u8],
|
|
content_type: &str,
|
|
) -> Result<(), String> {
|
|
use hmac::Mac as _;
|
|
let url = format!("https://{}.{}/{}", cfg.bucket, cfg.endpoint, key);
|
|
let date = httpdate::fmt_http_date(std::time::SystemTime::now());
|
|
let resource = format!("/{}/{}", cfg.bucket, key);
|
|
let string_to_sign = format!("PUT\n\n{content_type}\n{date}\n{resource}");
|
|
let mut mac = hmac::Hmac::<sha1::Sha1>::new_from_slice(cfg.access_key_secret.as_bytes())
|
|
.map_err(|e| format!("OSS 签名初始化失败: {e}"))?;
|
|
mac.update(string_to_sign.as_bytes());
|
|
let sig = base64_sign(&mac.finalize().into_bytes());
|
|
let client = reqwest::Client::new();
|
|
let resp = client
|
|
.put(&url)
|
|
.header("Date", &date)
|
|
.header("Content-Type", content_type)
|
|
.header(
|
|
"Authorization",
|
|
format!("OSS {}:{sig}", cfg.access_key_id),
|
|
)
|
|
.body(bytes.to_vec())
|
|
.timeout(std::time::Duration::from_secs(60))
|
|
.send()
|
|
.await
|
|
.map_err(|e| format!("OSS 请求失败: {e}"))?;
|
|
if resp.status().is_success() {
|
|
Ok(())
|
|
} else {
|
|
let status = resp.status();
|
|
let body = resp.text().await.unwrap_or_default();
|
|
Err(format!("OSS 上传失败 {status}: {}", truncate_body(&body)))
|
|
}
|
|
}
|
|
|
|
/// hmac 输出 → 标准 base64(签名编码)
|
|
fn base64_sign(digest: &[u8]) -> String {
|
|
use base64::Engine as _;
|
|
base64::engine::general_purpose::STANDARD.encode(digest)
|
|
}
|
|
|
|
/// 错误体截断(防 XML 错误页刷日志)
|
|
fn truncate_body(s: &str) -> &str {
|
|
match s.char_indices().nth(200) {
|
|
Some((i, _)) => &s[..i],
|
|
None => s,
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[tokio::test]
|
|
async fn save_dedupes_by_content_and_refs_local() {
|
|
let dir = std::env::temp_dir().join(format!("df-media-test-{}", df_types::types::new_id()));
|
|
let store = MediaStore::new(dir.clone(), None);
|
|
let url1 = store.save(b"\x89PNG same", "image/png", None).await.unwrap();
|
|
let url2 = store.save(b"\x89PNG same", "image/png", None).await.unwrap();
|
|
let url3 = store.save(b"\x89PNG other", "image/png", None).await.unwrap();
|
|
// 同内容同引用(hash 去重),不同内容不同引用
|
|
assert_eq!(url1, url2);
|
|
assert_ne!(url1, url3);
|
|
assert!(url1.starts_with("file://"), "纯本地模式引用 file://, 实际 {url1}");
|
|
let path = std::path::Path::new(&url1["file://".len()..]);
|
|
assert_eq!(tokio::fs::read(path).await.unwrap(), b"\x89PNG same");
|
|
tokio::fs::remove_dir_all(&dir).await.ok();
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn original_name_kept_readable_and_sanitized() {
|
|
let dir = std::env::temp_dir().join(format!("df-media-test-{}", df_types::types::new_id()));
|
|
let store = MediaStore::new(dir.clone(), None);
|
|
let url = store
|
|
.save(b"file-bytes", "application/octet-stream", Some("../报表/最终 版本.xlsx"))
|
|
.await
|
|
.unwrap();
|
|
let name = &url["file://".len()..];
|
|
let fname = std::path::Path::new(name).file_name().unwrap().to_string_lossy();
|
|
assert!(fname.contains('_'), "hash 与原始名以 _ 分隔: {fname}");
|
|
assert!(!fname.contains("..") && !fname.contains('/'), "路径穿越字符已清洗: {fname}");
|
|
assert!(fname.ends_with(".xlsx"), "原始名自带 ext 优先: {fname}");
|
|
// 同内容同名 → 同文件(去重不因名字破坏)
|
|
let url2 = store
|
|
.save(b"file-bytes", "application/octet-stream", Some("../报表/最终 版本.xlsx"))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(url, url2);
|
|
tokio::fs::remove_dir_all(&dir).await.ok();
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn save_with_path_returns_readable_local_path() {
|
|
// 文件消息入站契约:引用 url 之外另给本机路径(模型 read_file 只认本机路径),
|
|
// 通用文件落盘字节无损可读
|
|
let dir = std::env::temp_dir().join(format!("df-media-test-{}", df_types::types::new_id()));
|
|
let store = MediaStore::new(dir.clone(), None);
|
|
let (url, path) = store
|
|
.save_with_path(b"doc-bytes", "application/octet-stream", Some("合同.pdf"))
|
|
.await
|
|
.unwrap();
|
|
assert!(url.starts_with("file://"));
|
|
assert_eq!(
|
|
path.file_name().unwrap().to_string_lossy(),
|
|
format!("{}_合同.pdf", &content_hash(b"doc-bytes")[..16])
|
|
);
|
|
assert_eq!(tokio::fs::read(&path).await.unwrap(), b"doc-bytes");
|
|
tokio::fs::remove_dir_all(&dir).await.ok();
|
|
}
|
|
|
|
#[test]
|
|
fn content_key_stable_and_ext_aware() {
|
|
let k1 = content_key(b"abc", "image/png", None);
|
|
let k2 = content_key(b"abc", "image/png", None);
|
|
let k3 = content_key(b"abc", "image/jpeg", None);
|
|
assert_eq!(k1, k2);
|
|
assert_ne!(k1, k3, "同字节不同 media_type 应不同文件(png/jpg 不混)");
|
|
assert!(k1.ends_with(".png") && k1.len() == 16 + 4);
|
|
}
|
|
|
|
#[test]
|
|
fn public_url_prefers_cdn_domain() {
|
|
let cfg = OssConfig {
|
|
bucket: "b".into(),
|
|
endpoint: "oss-cn-hangzhou.aliyuncs.com".into(),
|
|
access_key_id: "ak".into(),
|
|
access_key_secret: "sk".into(),
|
|
public_base_url: "https://cdn.example.com/".into(),
|
|
};
|
|
assert_eq!(public_url(&cfg, "k.png"), "https://cdn.example.com/k.png");
|
|
let cfg2 = OssConfig { public_base_url: String::new(), ..cfg };
|
|
assert_eq!(
|
|
public_url(&cfg2, "k.png"),
|
|
"https://b.oss-cn-hangzhou.aliyuncs.com/k.png"
|
|
);
|
|
}
|
|
}
|