新增: 媒体引用化存储与消息溯源
- media_store 模块: sha256 去重落盘 + 阿里云 OSS 双写(签名手写), OSS 失败降级本机引用 - 飞书图片入站: image 消息下载装配 vision 片, 3 次退避+120s 超时+残件清理防线 - 出站回传: 报告 [交付] 路径行自动上传飞书原生附件, 凭据/密钥名拦截兜底 - 溯源 V52: ai_messages.source 列, 飞书 msg_id/chat_id/发送者随消息落库 - spawn offload: 桌面粘贴图 base64 统一转落盘引用, SQLite 不进大字符串 - 引用统一 url 字段(https/file://)不自造 scheme, image_ref 发送前按需展开
This commit is contained in:
@@ -0,0 +1,309 @@
|
||||
//! 媒体存储域(#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> {
|
||||
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}"));
|
||||
}
|
||||
}
|
||||
match &self.oss {
|
||||
Some(cfg) => match upload_to_oss(cfg, &key, bytes, media_type).await {
|
||||
Ok(()) => Ok(public_url(cfg, &key)),
|
||||
Err(e) => {
|
||||
tracing::warn!(key = %key, err = %e, "OSS 上传失败,引用降级本机 file://");
|
||||
Ok(local_ref(&path))
|
||||
}
|
||||
},
|
||||
None => Ok(local_ref(&path)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 内容寻址文件名:`<sha256前16hex>[_<清洗原始名>]<ext>`(对齐 u-claw safeAttachName
|
||||
/// 模式)——hash 前缀保去重(同内容必同文件)且排序聚簇,原始名缀后保目录可读。
|
||||
/// 平铺不分会话目录:会话归属经 DB 消息引用反查,文件系统不编码会话(否则同图跨会话
|
||||
/// 重复存储,去重失效);个人桌面应用媒体量级百~千,hash 前缀平铺可管理,分桶待量级上来再说
|
||||
fn content_key(bytes: &[u8], media_type: &str, original_name: Option<&str>) -> String {
|
||||
use sha2::Digest as _;
|
||||
let digest = hex::encode(&sha2::Sha256::digest(bytes)[..8]);
|
||||
// 扩展名:原始名自带 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();
|
||||
}
|
||||
|
||||
#[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"
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user