feat: 手动上传绕防爬、下载错误诊断与健康检查工具;模块化重构 API 与批量同步

后端:
  - 将 handlers.rs (1338行) 拆分为 helpers/papers/notes/sync 四模块
  - 将 batch_sync.rs 拆分为 batch/{mod,meta,asset} 三模块
  - 新增 POST /api/upload 多部件文件上传接口
  - 新增 POST /api/no_resource 标记文献"无全文资源"
  - 新增 GET/POST /api/active_bibcode 追踪活跃文献
  - StandardPaper 结构体扩展 pdf_error / html_error 错误诊断字段
  - download.rs 记录下载失败详情至数据库
  - 新增 health_check 二进制工具,支持只读扫描与 --fix 自动修复
  - 移除 scratch/ 目录、recovered_handlers.rs 及调试日志

  前端:
  - 新建 CustomSelect 可复用组件,替换全部原生 select
  - LibraryPanel:同步按钮反馈动画、下载失败/无资源状态筛选与计数、
    文献类型筛选、状态优先排序、搜索一键清空
  - 详情弹窗:错误诊断展示、手动 PDF/HTML 上传区、无资源标记/恢复
  - SearchPanel:扩展文献类型徽章、下载失败状态提示
  - SyncPanel:同步启动乐观 UI 更新、日志容器内自动滚动
  - Tab 状态 localStorage 持久化、弹窗 z-index 修复
This commit is contained in:
fmq
2026-06-11 22:56:36 +08:00
parent cd6af4f995
commit 8cc2b74abc
43 changed files with 4512 additions and 3879 deletions
-1338
View File
File diff suppressed because it is too large Load Diff
+373
View File
@@ -0,0 +1,373 @@
// src/api/helpers.rs
use sqlx::{SqlitePool, Row};
use tracing::info;
use crate::clients::ads::AdsPaperDoc;
use crate::clients::arxiv::ArxivPaper;
use super::StandardPaper;
pub fn convert_ads_doc_to_standard(doc: &AdsPaperDoc) -> StandardPaper {
let title = doc.title.as_ref()
.and_then(|v: &Vec<String>| v.first())
.cloned()
.unwrap_or_else(|| doc.bibcode.clone());
let authors = doc.author.clone().unwrap_or_default();
let keywords = doc.keyword.clone().unwrap_or_default();
let doi = doc.doi.as_ref()
.and_then(|v: &Vec<String>| v.first())
.cloned()
.unwrap_or_default();
let mut arxiv_id = String::new();
if let Some(identifiers) = &doc.identifier {
for id in identifiers {
if id.starts_with("arXiv:") {
arxiv_id = id.replace("arXiv:", "").trim().to_string();
break;
}
}
}
if arxiv_id.is_empty() {
if doc.bibcode.starts_with("arXiv") {
arxiv_id = doc.bibcode.replace("arXiv", "").trim().to_string();
}
}
StandardPaper {
bibcode: doc.bibcode.clone(),
title,
authors,
year: doc.year.clone().unwrap_or_default(),
pub_journal: doc.pub_journal.clone().unwrap_or_default(),
keywords,
abstract_text: doc.abstract_text.clone().unwrap_or_default(),
doi,
arxiv_id,
citation_count: doc.citation_count.unwrap_or(0),
reference_count: doc.reference_count.unwrap_or(0),
is_downloaded: false,
has_markdown: false,
has_translation: false,
doctype: doc.doctype.clone().unwrap_or_else(|| "article".to_string()),
pdf_error: None,
html_error: None,
}
}
pub fn convert_arxiv_to_standard(doc: &ArxivPaper) -> StandardPaper {
StandardPaper {
bibcode: doc.id.clone(),
title: doc.title.clone(),
authors: doc.authors.clone(),
year: doc.year.clone(),
pub_journal: "arXiv Preprint".to_string(),
keywords: Vec::new(),
abstract_text: doc.abstract_text.clone(),
doi: doc.doi.clone().unwrap_or_default(),
arxiv_id: doc.id.clone(),
citation_count: 0,
reference_count: 0,
is_downloaded: false,
has_markdown: false,
has_translation: false,
doctype: "eprint".to_string(),
pdf_error: None,
html_error: None,
}
}
pub async fn save_paper_to_db(db: &SqlitePool, p: &StandardPaper) -> anyhow::Result<()> {
let authors_json = serde_json::to_string(&p.authors)?;
let keywords_json = serde_json::to_string(&p.keywords)?;
// 1. 如果存在 arxiv_id,检查是否有已存在的相同 arxiv_id 记录以防 duplicate
if !p.arxiv_id.is_empty() {
let existing_opt: Option<(String, Option<String>, Option<String>, Option<String>, Option<String>)> = sqlx::query_as(
"SELECT bibcode, pdf_path, html_path, markdown_path, translation_path FROM papers WHERE arxiv_id = ?"
)
.bind(&p.arxiv_id)
.fetch_optional(db)
.await?;
if let Some((existing_bibcode, _pdf, _html, _md, _tr)) = existing_opt {
if existing_bibcode != p.bibcode {
// 发现不同 bibcode 标识的同一篇文献记录,需要进行合并
// 如果已存在的记录使用的是临时 arXiv ID 作为 bibcode,且新记录使用的是正式 ADS bibcode,我们升级 bibcode 主键
let is_existing_temp = existing_bibcode == p.arxiv_id;
let is_new_formal = p.bibcode != p.arxiv_id;
if is_existing_temp && is_new_formal {
info!("发现相同 arXiv ID 的文献,将临时主键 {} 升级为正式 ADS Bibcode: {}", existing_bibcode, p.bibcode);
sqlx::query(
"UPDATE papers SET bibcode = ?, title = ?, authors = ?, year = ?, pub = ?, keywords = ?, abstract = ?, doi = ?, citation_count = ?, reference_count = ?, doctype = ? WHERE bibcode = ?"
)
.bind(&p.bibcode)
.bind(&p.title)
.bind(&authors_json)
.bind(&p.year)
.bind(&p.pub_journal)
.bind(&keywords_json)
.bind(&p.abstract_text)
.bind(&p.doi)
.bind(p.citation_count)
.bind(p.reference_count)
.bind(&p.doctype)
.bind(&existing_bibcode)
.execute(db)
.await?;
return Ok(());
} else {
// 如果已存在的是正式 ADS bibcode,而新插入的是临时 arXiv ID,直接忽略或更新元数据而不更改主键
info!("发现相同 arXiv ID 的文献 {} 已存在正式记录,忽略临时 arXiv 插入", existing_bibcode);
return Ok(());
}
}
}
}
// 2. 正常插入/冲突更新
sqlx::query(
"INSERT INTO papers (bibcode, title, authors, year, pub, keywords, abstract, doi, arxiv_id, citation_count, reference_count, doctype) \
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
ON CONFLICT(bibcode) DO UPDATE SET \
title=excluded.title, \
authors=excluded.authors, \
pub=excluded.pub, \
keywords=excluded.keywords, \
abstract=excluded.abstract, \
doi=excluded.doi, \
arxiv_id=excluded.arxiv_id, \
citation_count=excluded.citation_count, \
reference_count=excluded.reference_count, \
doctype=excluded.doctype"
)
.bind(&p.bibcode)
.bind(&p.title)
.bind(authors_json)
.bind(&p.year)
.bind(&p.pub_journal)
.bind(keywords_json)
.bind(&p.abstract_text)
.bind(&p.doi)
.bind(&p.arxiv_id)
.bind(p.citation_count)
.bind(p.reference_count)
.bind(&p.doctype)
.execute(db)
.await?;
Ok(())
}
pub async fn get_paper_from_db(db: &SqlitePool, library_dir: &std::path::Path, bibcode: &str) -> anyhow::Result<StandardPaper> {
let r = sqlx::query("SELECT bibcode, title, authors, year, pub, keywords, abstract, doi, arxiv_id, citation_count, reference_count, pdf_path, html_path, markdown_path, translation_path, doctype FROM papers WHERE bibcode = ?")
.bind(bibcode)
.fetch_one(db)
.await?;
let pdf_path: Option<String> = r.get(11);
let html_path: Option<String> = r.get(12);
let markdown_path: Option<String> = r.get(13);
let translation_path: Option<String> = r.get(14);
let doctype_val: Option<String> = r.get(15);
let authors_str: Option<String> = r.get(2);
let authors: Vec<String> = authors_str.and_then(|s| serde_json::from_str(&s).ok()).unwrap_or_default();
let keywords_str: Option<String> = r.get(5);
let keywords: Vec<String> = keywords_str.and_then(|s| serde_json::from_str(&s).ok()).unwrap_or_default();
let is_pdf_exist = pdf_path.as_ref().map(|p| library_dir.join(p).exists()).unwrap_or(false);
let is_html_exist = html_path.as_ref().map(|p| library_dir.join(p).exists()).unwrap_or(false);
let is_md_exist = markdown_path.as_ref().map(|p| library_dir.join(p).exists()).unwrap_or(false);
let is_tr_exist = translation_path.as_ref().map(|p| library_dir.join(p).exists()).unwrap_or(false);
let pdf_error = pdf_path.as_ref()
.filter(|p| p.starts_with("error:"))
.map(|p| p["error:".len()..].trim().to_string());
let html_error = html_path.as_ref()
.filter(|p| p.starts_with("error:"))
.map(|p| p["error:".len()..].trim().to_string());
Ok(StandardPaper {
bibcode: r.get(0),
title: r.get(1),
authors,
year: r.get(3),
pub_journal: r.get(4),
keywords,
abstract_text: r.get(6),
doi: r.get(7),
arxiv_id: r.get(8),
citation_count: r.get(9),
reference_count: r.get(10),
is_downloaded: is_pdf_exist || is_html_exist,
has_markdown: is_md_exist,
has_translation: is_tr_exist,
doctype: doctype_val.unwrap_or_else(|| "article".to_string()),
pdf_error,
html_error,
})
}
pub async fn check_paper_paths_in_db(
db: &SqlitePool,
library_dir: &std::path::Path,
bibcode: &str
) -> anyhow::Result<Option<(Option<String>, Option<String>, Option<String>, Option<String>)>> {
let r_opt = sqlx::query("SELECT pdf_path, html_path, markdown_path, translation_path FROM papers WHERE bibcode = ?")
.bind(bibcode)
.fetch_optional(db)
.await?;
if let Some(r) = r_opt {
let pdf: Option<String> = r.get(0);
let html: Option<String> = r.get(1);
let md: Option<String> = r.get(2);
let tr: Option<String> = r.get(3);
let pdf_res = pdf.filter(|p| library_dir.join(p).exists());
let html_res = html.filter(|p| library_dir.join(p).exists());
let md_res = md.filter(|p| library_dir.join(p).exists());
let tr_res = tr.filter(|p| library_dir.join(p).exists());
Ok(Some((pdf_res, html_res, md_res, tr_res)))
} else {
Ok(None)
}
}
#[cfg(test)]
mod tests {
use super::*;
use sqlx::sqlite::SqlitePoolOptions;
#[test]
fn test_convert_ads_doc_to_standard() {
let doc = AdsPaperDoc {
bibcode: "2026A&A...123..456X".to_string(),
title: Some(vec!["A Test Title".to_string()]),
author: Some(vec!["Author A".to_string(), "Author B".to_string()]),
year: Some("2026".to_string()),
pub_journal: Some("Astronomy & Astrophysics".to_string()),
keyword: Some(vec!["Keyword 1".to_string()]),
abstract_text: Some("This is abstract".to_string()),
doi: Some(vec!["10.1000/test.doi".to_string()]),
citation_count: Some(5),
reference_count: Some(10),
reference: None,
citation: None,
identifier: None,
doctype: Some("article".to_string()),
};
let paper = convert_ads_doc_to_standard(&doc);
assert_eq!(paper.bibcode, "2026A&A...123..456X");
assert_eq!(paper.title, "A Test Title");
assert_eq!(paper.authors, vec!["Author A", "Author B"]);
assert_eq!(paper.year, "2026");
assert_eq!(paper.pub_journal, "Astronomy & Astrophysics");
assert_eq!(paper.keywords, vec!["Keyword 1"]);
assert_eq!(paper.abstract_text, "This is abstract");
assert_eq!(paper.doi, "10.1000/test.doi");
assert_eq!(paper.arxiv_id, "");
assert_eq!(paper.citation_count, 5);
assert_eq!(paper.reference_count, 10);
}
#[test]
fn test_convert_ads_doc_to_standard_with_arxiv_identifier() {
let doc = AdsPaperDoc {
bibcode: "2026MNRAS.530.1234A".to_string(),
title: Some(vec!["Another Test Title".to_string()]),
author: Some(vec!["Author A".to_string()]),
year: Some("2026".to_string()),
pub_journal: Some("MNRAS".to_string()),
keyword: None,
abstract_text: None,
doi: None,
citation_count: None,
reference_count: None,
reference: None,
citation: None,
identifier: Some(vec!["2026MNRAS.530.1234A".to_string(), "arXiv:2606.12345".to_string()]),
doctype: Some("article".to_string()),
};
let paper = convert_ads_doc_to_standard(&doc);
assert_eq!(paper.bibcode, "2026MNRAS.530.1234A");
assert_eq!(paper.arxiv_id, "2606.12345");
}
#[test]
fn test_convert_arxiv_to_standard() {
let doc = ArxivPaper {
id: "2606.12345".to_string(),
title: "Arxiv Title".to_string(),
authors: vec!["Author C".to_string()],
year: "2026".to_string(),
abstract_text: "Arxiv abstract".to_string(),
doi: Some("10.1000/arxiv.doi".to_string()),
pdf_url: "https://arxiv.org/pdf/2606.12345.pdf".to_string(),
};
let paper = convert_arxiv_to_standard(&doc);
assert_eq!(paper.bibcode, "2606.12345");
assert_eq!(paper.title, "Arxiv Title");
assert_eq!(paper.authors, vec!["Author C"]);
assert_eq!(paper.year, "2026");
assert_eq!(paper.pub_journal, "arXiv Preprint");
assert_eq!(paper.doi, "10.1000/arxiv.doi");
assert_eq!(paper.arxiv_id, "2606.12345");
}
#[tokio::test]
async fn test_db_operations() -> anyhow::Result<()> {
let pool = SqlitePoolOptions::new()
.max_connections(1)
.connect("sqlite::memory:")
.await?;
// 运行迁移
sqlx::migrate!("./migrations")
.run(&pool)
.await?;
let paper = StandardPaper {
bibcode: "2026A&A...123..456X".to_string(),
title: "A Test Title".to_string(),
authors: vec!["Author A".to_string()],
year: "2026".to_string(),
pub_journal: "Astronomy & Astrophysics".to_string(),
keywords: vec!["Keyword 1".to_string()],
abstract_text: "This is abstract".to_string(),
doi: "10.1000/test.doi".to_string(),
arxiv_id: "".to_string(),
citation_count: 5,
reference_count: 10,
is_downloaded: false,
has_markdown: false,
has_translation: false,
doctype: "article".to_string(),
};
// 保存
save_paper_to_db(&pool, &paper).await?;
// 读取
let retrieved = get_paper_from_db(&pool, std::path::Path::new(""), "2026A&A...123..456X").await?;
assert_eq!(retrieved.title, paper.title);
assert_eq!(retrieved.authors, paper.authors);
assert_eq!(retrieved.keywords, paper.keywords);
// 检查路径状态(初始为 None)
let paths = check_paper_paths_in_db(&pool, std::path::Path::new(""), "2026A&A...123..456X").await?;
assert!(paths.is_some());
let (pdf, html, md, tr) = paths.unwrap();
assert!(pdf.is_none());
assert!(html.is_none());
assert!(md.is_none());
assert!(tr.is_none());
Ok(())
}
}
+78 -1
View File
@@ -1 +1,78 @@
pub mod handlers;
// src/api/mod.rs
use std::sync::Arc;
use serde::{Deserialize, Serialize};
use sqlx::SqlitePool;
use crate::Config;
use crate::services::translation::Dictionary;
use crate::clients::qiniu::QiniuClient;
use crate::clients::ads::AdsClient;
use crate::clients::arxiv::ArxivClient;
use crate::services::download::Downloader;
// 全局共享的 Axum 应用上下文状态
pub struct AppState {
pub config: Config,
pub db: SqlitePool,
pub dict: Dictionary,
pub qiniu: QiniuClient,
pub ads: AdsClient,
pub arxiv: ArxivClient,
pub downloader: Downloader,
pub harvest_status: Arc<tokio::sync::Mutex<crate::services::batch_sync::MetaSyncStatus>>,
pub process_status: Arc<tokio::sync::Mutex<crate::services::batch_sync::AssetSyncStatus>>,
pub active_bibcode: Arc<tokio::sync::Mutex<Option<String>>>,
}
// 统一标准化的文献格式,用于向前端传输
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct StandardPaper {
pub bibcode: String,
pub title: String,
pub authors: Vec<String>,
pub year: String,
pub pub_journal: String,
pub keywords: Vec<String>,
pub abstract_text: String,
pub doi: String,
pub arxiv_id: String,
pub citation_count: i32,
pub reference_count: i32,
pub is_downloaded: bool,
pub has_markdown: bool,
pub has_translation: bool,
pub doctype: String,
pub pdf_error: Option<String>,
pub html_error: Option<String>,
}
pub mod helpers;
pub mod papers;
pub mod notes;
pub mod sync;
// 提供兼容的 handlers 命名空间,避免修改 main.rs / batch_sync.rs 里的导入
pub mod handlers {
pub use super::helpers::{
convert_ads_doc_to_standard, convert_arxiv_to_standard, save_paper_to_db,
get_paper_from_db, check_paper_paths_in_db,
};
pub use super::papers::{
search_papers, download_paper, parse_paper, translate_paper,
get_citation_network, get_paper_detail, get_library, export_citations,
upload_paper_file, mark_no_resource, get_active_bibcode, set_active_bibcode,
SearchParams, DownloadRequest, ParseRequest, ParseResponse,
TranslateRequest, TranslateResponse, CitationsResponse, PaperDetailResponse,
ExportRequest, ExportResponse, MarkNoResourceRequest,
};
pub use super::notes::{
create_note, get_notes, delete_note,
NoteRecord, CreateNoteRequest, DeleteNoteParams, GetNotesParams,
};
pub use super::sync::{
run_meta_sync, get_meta_sync_count, get_meta_sync_status,
run_asset_sync, stop_asset_sync, get_sync_queries, delete_sync_query,
get_asset_sync_status, MetaSyncRunRequest, MetaSyncCountRequest,
MetaSyncCountResponse, AssetSyncRunRequest, SavedSyncQuery,
};
pub use super::{AppState, StandardPaper};
}
+113
View File
@@ -0,0 +1,113 @@
// src/api/notes.rs
use axum::{
extract::{Query, State},
http::StatusCode,
Json,
};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use sqlx::Row;
use super::AppState;
#[derive(Debug, Serialize, Deserialize)]
pub struct NoteRecord {
pub id: i64,
pub bibcode: String,
pub paragraph_index: i64,
pub note_text: String,
pub highlight_color: String,
pub selected_text: String,
pub created_at: String,
}
#[derive(Deserialize)]
pub struct CreateNoteRequest {
pub bibcode: String,
pub paragraph_index: i64,
pub note_text: Option<String>,
pub highlight_color: Option<String>,
pub selected_text: Option<String>,
}
#[derive(Deserialize)]
pub struct DeleteNoteParams {
pub id: i64,
}
#[derive(Deserialize)]
pub struct GetNotesParams {
pub bibcode: String,
}
// 创建笔记
pub async fn create_note(
State(state): State<Arc<AppState>>,
Json(req): Json<CreateNoteRequest>,
) -> Result<Json<NoteRecord>, (StatusCode, String)> {
let note_text = req.note_text.unwrap_or_default();
let highlight_color = req.highlight_color.unwrap_or_else(|| "yellow".to_string());
let selected_text = req.selected_text.unwrap_or_default();
let row = sqlx::query(
"INSERT INTO notes (bibcode, paragraph_index, note_text, highlight_color, selected_text) VALUES (?, ?, ?, ?, ?) RETURNING id, bibcode, paragraph_index, note_text, highlight_color, selected_text, created_at"
)
.bind(&req.bibcode)
.bind(req.paragraph_index)
.bind(&note_text)
.bind(&highlight_color)
.bind(&selected_text)
.fetch_one(&state.db)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("保存笔记失败: {}", e)))?;
Ok(Json(NoteRecord {
id: row.get(0),
bibcode: row.get(1),
paragraph_index: row.get(2),
note_text: row.get(3),
highlight_color: row.get(4),
selected_text: row.get(5),
created_at: row.get(6),
}))
}
// 查询某篇文献的全部笔记
pub async fn get_notes(
State(state): State<Arc<AppState>>,
Query(params): Query<GetNotesParams>,
) -> Result<Json<Vec<NoteRecord>>, (StatusCode, String)> {
let rows = sqlx::query(
"SELECT id, bibcode, paragraph_index, note_text, highlight_color, selected_text, created_at FROM notes WHERE bibcode = ? ORDER BY paragraph_index, created_at"
)
.bind(&params.bibcode)
.fetch_all(&state.db)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("查询笔记失败: {}", e)))?;
let notes: Vec<NoteRecord> = rows.iter().map(|r| NoteRecord {
id: r.get(0),
bibcode: r.get(1),
paragraph_index: r.get(2),
note_text: r.get(3),
highlight_color: r.get(4),
selected_text: r.get(5),
created_at: r.get(6),
}).collect();
Ok(Json(notes))
}
// 删除指定 id 的笔记
pub async fn delete_note(
State(state): State<Arc<AppState>>,
Query(params): Query<DeleteNoteParams>,
) -> Result<StatusCode, (StatusCode, String)> {
sqlx::query("DELETE FROM notes WHERE id = ?")
.bind(params.id)
.execute(&state.db)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("删除笔记失败: {}", e)))?;
Ok(StatusCode::NO_CONTENT)
}
+855
View File
@@ -0,0 +1,855 @@
// src/api/papers.rs
use axum::{
extract::{Query, State},
http::StatusCode,
Json,
};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use std::fs;
use tracing::{info, warn, error};
use sqlx::Row;
use super::{AppState, StandardPaper};
use super::helpers::{
convert_ads_doc_to_standard, convert_arxiv_to_standard, save_paper_to_db,
get_paper_from_db, check_paper_paths_in_db,
};
// 检索请求参数
#[derive(Debug, Deserialize)]
pub struct SearchParams {
pub q: String,
pub source: Option<String>, // "all" | "ads" | "arxiv"
pub rows: Option<i32>,
pub start: Option<i32>, // 分页起始偏移量
pub sort: Option<String>, // 排序字段
}
// ── GET /api/search ──
// 统一检索接口,合并去重 ADS 和 arXiv 数据
pub async fn search_papers(
State(state): State<Arc<AppState>>,
Query(params): Query<SearchParams>,
) -> Result<Json<Vec<StandardPaper>>, (StatusCode, String)> {
let source = params.source.unwrap_or_else(|| "all".to_string());
let rows = params.rows.unwrap_or(10);
let start = params.start.unwrap_or(0);
let sort = params.sort.as_deref().unwrap_or("relevance");
let mut results = Vec::new();
// 1. 检索 NASA ADS
if source == "all" || source == "ads" {
if !state.config.ads_api_key.is_empty() {
match state.ads.search(&params.q, start, rows, sort).await {
Ok(docs) => {
for doc in docs {
let paper = convert_ads_doc_to_standard(&doc);
// 入库 SQLite
if let Err(e) = save_paper_to_db(&state.db, &paper).await {
warn!("保存 ADS 文献至数据库失败: {}", e);
}
// 保存引用/参考文献关联拓扑
if let Some(refs) = doc.reference {
for ref_bib in refs {
let _ = sqlx::query("INSERT OR IGNORE INTO citations_references (source_bibcode, target_bibcode) VALUES (?, ?)")
.bind(&paper.bibcode)
.bind(&ref_bib)
.execute(&state.db)
.await;
}
}
if let Some(cits) = doc.citation {
for cit_bib in cits {
let _ = sqlx::query("INSERT OR IGNORE INTO citations_references (source_bibcode, target_bibcode) VALUES (?, ?)")
.bind(&cit_bib)
.bind(&paper.bibcode)
.execute(&state.db)
.await;
}
}
results.push(paper);
}
}
Err(e) => {
error!("ADS 检索执行失败: {}", e);
}
}
} else {
warn!("ADS_API_KEY 未配置,跳过 ADS 检索。");
}
}
// 2. 检索 arXiv
if source == "all" || source == "arxiv" {
match state.arxiv.search(&params.q, start, rows, sort).await {
Ok(papers) => {
for p in papers {
let paper = convert_arxiv_to_standard(&p);
// 入库 SQLite (使用 arXiv ID 暂作主键以作记录)
if let Err(e) = save_paper_to_db(&state.db, &paper).await {
warn!("保存 arXiv 文献至数据库失败: {}", e);
}
results.push(paper);
}
}
Err(e) => {
error!("arXiv 检索执行失败: {}", e);
}
}
}
// 对两端获取的数据进行去重合并,增加对相同 arxiv_id 的判断
let mut unique_results: Vec<StandardPaper> = Vec::new();
for r in results {
if !unique_results.iter().any(|u| u.bibcode == r.bibcode || (!u.doi.is_empty() && u.doi == r.doi) || (!u.arxiv_id.is_empty() && u.arxiv_id == r.arxiv_id)) {
let mut final_paper = r.clone();
// 如果本地数据库存在该文献,直接从数据库读取标准元数据(包括 is_downloaded, has_markdown, pdf_error, html_error 等)
if let Ok(db_paper) = get_paper_from_db(&state.db, &state.config.library_dir, &r.bibcode).await {
final_paper = db_paper;
}
unique_results.push(final_paper);
}
}
Ok(Json(unique_results))
}
// ── POST /api/download ──
#[derive(Deserialize)]
pub struct DownloadRequest {
pub bibcode: String,
pub force: Option<bool>, // 强制重新下载,即使已存在本地文件
}
// 一键双格式并行下载文献 (PDF + HTML),支持 force=true 强制重新下载
pub async fn download_paper(
State(state): State<Arc<AppState>>,
Json(req): Json<DownloadRequest>,
) -> Result<Json<StandardPaper>, (StatusCode, String)> {
let force = req.force.unwrap_or(false);
info!("接收到文献下载指令,标识符: {}, 强制重下: {}", req.bibcode, force);
let paper = get_paper_from_db(&state.db, &state.config.library_dir, &req.bibcode)
.await
.map_err(|e| (StatusCode::NOT_FOUND, format!("未找到该文献记录: {}", e)))?;
// force=true 时清除旧路径记录,强制重新下载
if force {
sqlx::query("UPDATE papers SET pdf_path = NULL, html_path = NULL WHERE bibcode = ?")
.bind(&req.bibcode)
.execute(&state.db)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("重置下载状态失败: {}", e)))?;
}
// 下载策略:
// 1. 如有 arXiv ID,优先走 arXiv 直连(绕过出版商防护墙,成功率高)
// 2. 否则走 ADS 网关多级回退(PUB_PDF → EPRINT_PDF → CrossRef
// 3. 若 ADS 路径 PDF/HTML 均失败但有 arXiv ID,再尝试 arXiv 作为兜底
let (pdf_res, html_res) = if !paper.arxiv_id.is_empty() {
info!("[下载] 优先使用 arXiv 通道: {}", paper.arxiv_id);
state.downloader.download_arxiv_direct(&paper.arxiv_id, &state.config.library_dir).await
} else {
let doi_opt = if !paper.doi.is_empty() { Some(paper.doi.as_str()) } else { None };
state.downloader.download_paper(&req.bibcode, doi_opt, &state.config.library_dir).await
};
if pdf_res.is_err() && html_res.is_err() {
let pdf_err = pdf_res.as_ref().err().unwrap();
let html_err = html_res.as_ref().err().unwrap();
error!("文献 {} PDF 和 HTML 均下载失败,无可用物理文件格式", req.bibcode);
let pdf_db_err = format!("error: {}", pdf_err);
let html_db_err = format!("error: {}", html_err);
let _ = sqlx::query("UPDATE papers SET pdf_path = ?, html_path = ? WHERE bibcode = ?")
.bind(&pdf_db_err)
.bind(&html_db_err)
.bind(&req.bibcode)
.execute(&state.db)
.await;
return Err((StatusCode::INTERNAL_SERVER_ERROR, format!("下载失败。PDF: {}, HTML: {}", pdf_err, html_err)));
}
let pdf_rel = match pdf_res {
Ok(p) => Some(p.strip_prefix(&state.config.library_dir).unwrap_or(&p).to_string_lossy().to_string()),
Err(e) => Some(format!("error: {}", e)),
};
let html_rel = match html_res {
Ok(p) => Some(p.strip_prefix(&state.config.library_dir).unwrap_or(&p).to_string_lossy().to_string()),
Err(e) => Some(format!("error: {}", e)),
};
// 回写存储路径至数据库
sqlx::query("UPDATE papers SET pdf_path = ?, html_path = ? WHERE bibcode = ?")
.bind(&pdf_rel)
.bind(&html_rel)
.bind(&req.bibcode)
.execute(&state.db)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("更新数据库失败: {}", e)))?;
let mut updated_paper = paper;
updated_paper.is_downloaded = true;
Ok(Json(updated_paper))
}
// ── POST /api/parse ──
#[derive(Deserialize)]
pub struct ParseRequest {
pub bibcode: String,
pub force: Option<bool>,
}
#[derive(Serialize)]
pub struct ParseResponse {
pub markdown: String,
}
// 将 HTML / PDF 转换为标准英文 Markdown 文本(HTML 优先,PDF 调用 MinerU 远程 API 且上传七牛云)
pub async fn parse_paper(
State(state): State<Arc<AppState>>,
Json(req): Json<ParseRequest>,
) -> Result<Json<ParseResponse>, (StatusCode, String)> {
info!("接收到文献结构化解析指令: {} (强制重新解析: {:?})", req.bibcode, req.force);
let (pdf_opt, html_opt, md_opt, _) = check_paper_paths_in_db(&state.db, &state.config.library_dir, &req.bibcode)
.await
.map_err(|e| (StatusCode::NOT_FOUND, format!("获取文献路径失败: {}", e)))?
.ok_or((StatusCode::NOT_FOUND, "该文献未注册在数据库中".to_string()))?;
let force = req.force.unwrap_or(false);
// 如果先前已经解析成功过且非强制重新解析,直读 Markdown 文件返回
if !force {
if let Some(md_rel) = md_opt {
let md_abs = state.config.library_dir.join(&md_rel);
if md_abs.exists() {
if let Ok(content) = fs::read_to_string(&md_abs) {
return Ok(Json(ParseResponse { markdown: content }));
}
}
}
}
let mut parsed_markdown = String::new();
let mut relative_md_path = String::new();
// 查询该文献的元数据以生成 Markdown YAML 头部信息
let paper = get_paper_from_db(&state.db, &state.config.library_dir, &req.bibcode)
.await
.map_err(|e| (StatusCode::NOT_FOUND, format!("未找到该文献元数据记录: {}", e)))?;
// 策略 1HTML 优先解析
if let Some(html_rel) = html_opt {
let html_abs = state.config.library_dir.join(&html_rel);
if html_abs.exists() {
match crate::services::parser::html_to_markdown(&html_abs) {
Ok(md) => {
let front_matter = format!(
"---\ntitle: {}\nauthor: [{}]\npublisher: {}\nsource: \"https://ui.adsabs.harvard.edu/abs/{}/abstract\"\ndate: \"{}\"\ntags: \"{}\"\n---\n\n",
serde_json::to_string(&paper.title).unwrap_or_else(|_| format!("\"{}\"", paper.title)),
paper.authors.iter().map(|a| format!("\"{}\"", a)).collect::<Vec<_>>().join(", "),
serde_json::to_string(&paper.pub_journal).unwrap_or_else(|_| format!("\"{}\"", paper.pub_journal)),
paper.bibcode,
paper.year,
paper.keywords.join(",")
);
parsed_markdown = format!("{}{}", front_matter, md);
let md_filename = format!("{}_en.md", req.bibcode);
let md_dest = state.config.library_dir.join("Markdown").join(&md_filename);
fs::create_dir_all(md_dest.parent().unwrap()).unwrap_or_default();
if fs::write(&md_dest, &parsed_markdown).is_ok() {
relative_md_path = format!("Markdown/{}", md_filename);
}
}
Err(e) => {
warn!("HTML 转换为 Markdown 失败 {}: {}。将自动降级使用 PDF 结构化解析。", req.bibcode, e);
}
}
}
}
// 策略 2:回退至 PDF 远程 MinerU 解析
if parsed_markdown.is_empty() {
if let Some(pdf_rel) = pdf_opt {
let pdf_abs = state.config.library_dir.join(&pdf_rel);
if pdf_abs.exists() {
match crate::services::parser::parse_pdf_via_mineru(&pdf_abs, &state.qiniu, &state.config).await {
Ok(md) => {
let front_matter = format!(
"---\ntitle: {}\nauthor: [{}]\npublisher: {}\nsource: \"https://ui.adsabs.harvard.edu/abs/{}/abstract\"\ndate: \"{}\"\ntags: \"{}\"\n---\n\n",
serde_json::to_string(&paper.title).unwrap_or_else(|_| format!("\"{}\"", paper.title)),
paper.authors.iter().map(|a| format!("\"{}\"", a)).collect::<Vec<_>>().join(", "),
serde_json::to_string(&paper.pub_journal).unwrap_or_else(|_| format!("\"{}\"", paper.pub_journal)),
paper.bibcode,
paper.year,
paper.keywords.join(",")
);
parsed_markdown = format!("{}{}", front_matter, md);
let md_filename = format!("{}_en.md", req.bibcode);
let md_dest = state.config.library_dir.join("Markdown").join(&md_filename);
fs::create_dir_all(md_dest.parent().unwrap()).unwrap_or_default();
if fs::write(&md_dest, &parsed_markdown).is_ok() {
relative_md_path = format!("Markdown/{}", md_filename);
}
}
Err(e) => {
error!("PDF layout 远程 MinerU 解析失败: {}", e);
return Err((StatusCode::INTERNAL_SERVER_ERROR, format!("PDF 结构解析失败: {}", e)));
}
}
} else {
error!("文献 {} 解析失败:本地 PDF 文件 {:?} 丢失", req.bibcode, pdf_abs);
return Err((StatusCode::NOT_FOUND, "本地 PDF 文件未找到".to_string()));
}
} else {
error!("文献 {} 解析失败:请先下载该文献的 HTML 或 PDF 文件", req.bibcode);
return Err((StatusCode::BAD_REQUEST, "请先下载该文献的 HTML 或 PDF 文件".to_string()));
}
}
// 更新本地解析路径至 SQLite 数据库
if !relative_md_path.is_empty() {
let _ = sqlx::query("UPDATE papers SET markdown_path = ? WHERE bibcode = ?")
.bind(&relative_md_path)
.bind(&req.bibcode)
.execute(&state.db)
.await;
}
Ok(Json(ParseResponse { markdown: parsed_markdown }))
}
// ── POST /api/translate ──
#[derive(Deserialize)]
pub struct TranslateRequest {
pub bibcode: String,
pub force: Option<bool>,
}
#[derive(Serialize)]
pub struct TranslateResponse {
pub translation: String,
}
// 文献中英双栏对比翻译接口(包含词表注入与本地物理缓存)
pub async fn translate_paper(
State(state): State<Arc<AppState>>,
Json(req): Json<TranslateRequest>,
) -> Result<Json<TranslateResponse>, (StatusCode, String)> {
let force = req.force.unwrap_or(false);
info!("接收到对比翻译请求: 文献={}, 强制重译={}", req.bibcode, force);
let (_, _, md_opt, tr_opt) = check_paper_paths_in_db(&state.db, &state.config.library_dir, &req.bibcode)
.await
.map_err(|e| (StatusCode::NOT_FOUND, format!("查询文献路径失败: {}", e)))?
.ok_or((StatusCode::NOT_FOUND, "该文献未注册在数据库中".to_string()))?;
// 若本地已存在翻译物理文件且未指明强制重译,直读本地缓存返回
if !force {
if let Some(tr_rel) = tr_opt {
let tr_abs = state.config.library_dir.join(&tr_rel);
if tr_abs.exists() {
if let Ok(content) = fs::read_to_string(&tr_abs) {
return Ok(Json(TranslateResponse { translation: content }));
}
}
}
}
// 检查英文解析文件是否存在
let md_rel = match md_opt {
Some(rel) => rel,
None => {
error!("文献 {} 翻译失败:文献未完成解析,缺少英文 Markdown 路径", req.bibcode);
return Err((StatusCode::BAD_REQUEST, "文献必须先完成解析方可翻译".to_string()));
}
};
let md_abs = state.config.library_dir.join(&md_rel);
if !md_abs.exists() {
error!("文献 {} 翻译失败:解析的英文 Markdown 文件 {:?} 不存在", req.bibcode, md_abs);
return Err((StatusCode::BAD_REQUEST, "解析 Markdown 文件丢失".to_string()));
}
let english_markdown = fs::read_to_string(&md_abs)
.map_err(|e| {
error!("文献 {} 翻译失败:读取解析内容失败: {}", req.bibcode, e);
(StatusCode::INTERNAL_SERVER_ERROR, format!("读取解析内容失败: {}", e))
})?;
// 调用 LLM 翻译服务并注入对照词表
let translated_markdown = crate::services::translation::translate_markdown(&english_markdown, &state.dict, &state.config)
.await
.map_err(|e| {
error!("文献 {} 翻译失败:调用 LLM 翻译发生错误: {}", req.bibcode, e);
(StatusCode::INTERNAL_SERVER_ERROR, format!("调用 LLM 翻译失败: {}", e))
})?;
// 翻译结果物理写入本地
let tr_filename = format!("{}_zh.md", req.bibcode);
let tr_dest = state.config.library_dir.join("Translation").join(&tr_filename);
fs::create_dir_all(tr_dest.parent().unwrap()).unwrap_or_default();
fs::write(&tr_dest, &translated_markdown)
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("写入翻译文件失败: {}", e)))?;
let relative_tr_path = format!("Translation/{}", tr_filename);
// 缓存路径更新入库
sqlx::query("UPDATE papers SET translation_path = ? WHERE bibcode = ?")
.bind(&relative_tr_path)
.bind(&req.bibcode)
.execute(&state.db)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("更新数据库翻译状态失败: {}", e)))?;
Ok(Json(TranslateResponse { translation: translated_markdown }))
}
#[derive(Debug, Serialize)]
pub struct CitationsResponse {
pub bibcode: String,
pub title: String,
pub citation_count: i32,
pub reference_count: i32,
pub references: Vec<String>, // 该文献参考文献 bibcode 数组
pub citations: Vec<String>, // 引用该文献的 bibcode 数组
pub citation_counts: std::collections::HashMap<String, i32>, // 相关文献与被引数映射
}
// 从 SQLite 查询引用关联,生成引用星系关系树
pub async fn get_citation_network(
State(state): State<Arc<AppState>>,
Query(params): Query<DownloadRequest>,
) -> Result<Json<CitationsResponse>, (StatusCode, String)> {
let paper = match get_paper_from_db(&state.db, &state.config.library_dir, &params.bibcode).await {
Ok(p) => p,
Err(_) => {
// 如果本地数据库查不到,尝试从 ADS 在线 API 动态获取
if !state.config.ads_api_key.is_empty() {
match state.ads.search(&format!("bibcode:{}", params.bibcode), 0, 1, "relevance").await {
Ok(docs) => {
if let Some(doc) = docs.first() {
let standard_paper = convert_ads_doc_to_standard(doc);
// 保存至数据库缓存,并保存引用关联
let _ = save_paper_to_db(&state.db, &standard_paper).await;
if let Some(refs) = &doc.reference {
for ref_bib in refs {
let _ = sqlx::query("INSERT OR IGNORE INTO citations_references (source_bibcode, target_bibcode) VALUES (?, ?)")
.bind(&standard_paper.bibcode)
.bind(ref_bib)
.execute(&state.db)
.await;
}
}
if let Some(cits) = &doc.citation {
for cit_bib in cits {
let _ = sqlx::query("INSERT OR IGNORE INTO citations_references (source_bibcode, target_bibcode) VALUES (?, ?)")
.bind(cit_bib)
.bind(&standard_paper.bibcode)
.execute(&state.db)
.await;
}
}
standard_paper
} else {
return Err((StatusCode::NOT_FOUND, format!("在本地库及 ADS 中均未找到该文献: {}", params.bibcode)));
}
}
Err(e) => {
return Err((StatusCode::INTERNAL_SERVER_ERROR, format!("在线检索文献元数据失败: {}", e)));
}
}
} else {
return Err((StatusCode::NOT_FOUND, format!("本地数据库未收录该文献,且未配置 ADS_API_KEY,无法在线加载: {}", params.bibcode)));
}
}
};
// 加载引用的文献
let refs_rows = sqlx::query("SELECT target_bibcode FROM citations_references WHERE source_bibcode = ?")
.bind(&params.bibcode)
.fetch_all(&state.db)
.await
.unwrap_or_default();
let references: Vec<String> = refs_rows.iter().map(|row| row.get(0)).collect();
// 加载被引用的文献
let cits_rows = sqlx::query("SELECT source_bibcode FROM citations_references WHERE target_bibcode = ?")
.bind(&params.bibcode)
.fetch_all(&state.db)
.await
.unwrap_or_default();
let citations: Vec<String> = cits_rows.iter().map(|row| row.get(0)).collect();
// 加载关联文献的被引数量 (从 SQLite papers 表获取)
let mut citation_counts = std::collections::HashMap::new();
let mut all_related = references.clone();
all_related.extend(citations.clone());
for bib in all_related {
let count_opt: Option<i32> = sqlx::query_scalar("SELECT citation_count FROM papers WHERE bibcode = ?")
.bind(&bib)
.fetch_optional(&state.db)
.await
.unwrap_or_default();
if let Some(c) = count_opt {
citation_counts.insert(bib, c);
}
}
Ok(Json(CitationsResponse {
bibcode: paper.bibcode,
title: paper.title,
citation_count: paper.citation_count,
reference_count: paper.reference_count,
references,
citations,
citation_counts,
}))
}
// ── GET /api/paper ──
#[derive(Serialize)]
pub struct PaperDetailResponse {
pub paper: StandardPaper,
pub english_content: Option<String>,
pub translation_content: Option<String>,
}
// 获取文献标准详情和中英双语内容文件数据
pub async fn get_paper_detail(
State(state): State<Arc<AppState>>,
Query(params): Query<DownloadRequest>,
) -> Result<Json<PaperDetailResponse>, (StatusCode, String)> {
let paper = get_paper_from_db(&state.db, &state.config.library_dir, &params.bibcode)
.await
.map_err(|e| (StatusCode::NOT_FOUND, format!("未找到该文献数据: {}", e)))?;
let (_, _, md_opt, tr_opt) = check_paper_paths_in_db(&state.db, &state.config.library_dir, &params.bibcode)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
.unwrap_or_default();
let english_content = md_opt.and_then(|rel| fs::read_to_string(state.config.library_dir.join(rel)).ok());
let translation_content = tr_opt.and_then(|rel| fs::read_to_string(state.config.library_dir.join(rel)).ok());
Ok(Json(PaperDetailResponse {
paper,
english_content,
translation_content,
}))
}
// ── GET /api/library ──
// 获取本地图书馆文献列表
pub async fn get_library(
State(state): State<Arc<AppState>>,
) -> Result<Json<Vec<StandardPaper>>, (StatusCode, String)> {
let rows = sqlx::query("SELECT bibcode, title, authors, year, pub, keywords, abstract, doi, arxiv_id, citation_count, reference_count, pdf_path, html_path, markdown_path, translation_path, doctype FROM papers ORDER BY created_at DESC")
.fetch_all(&state.db)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("访问本地数据库失败: {}", e)))?;
let mut list = Vec::new();
for r in rows {
let pdf_path: Option<String> = r.get(11);
let html_path: Option<String> = r.get(12);
let markdown_path: Option<String> = r.get(13);
let translation_path: Option<String> = r.get(14);
let doctype_val: Option<String> = r.get(15);
let authors_str: Option<String> = r.get(2);
let authors: Vec<String> = authors_str.and_then(|s| serde_json::from_str(&s).ok()).unwrap_or_default();
let keywords_str: Option<String> = r.get(5);
let keywords: Vec<String> = keywords_str.and_then(|s| serde_json::from_str(&s).ok()).unwrap_or_default();
let pdf_error = pdf_path.as_ref()
.filter(|p| p.starts_with("error:"))
.map(|p| p["error:".len()..].trim().to_string());
let html_error = html_path.as_ref()
.filter(|p| p.starts_with("error:"))
.map(|p| p["error:".len()..].trim().to_string());
list.push(StandardPaper {
bibcode: r.get(0),
title: r.get(1),
authors,
year: r.get(3),
pub_journal: r.get(4),
keywords,
abstract_text: r.get(6),
doi: r.get(7),
arxiv_id: r.get(8),
citation_count: r.get(9),
reference_count: r.get(10),
is_downloaded: pdf_path.as_ref().map(|p| state.config.library_dir.join(p).exists()).unwrap_or(false)
|| html_path.as_ref().map(|p| state.config.library_dir.join(p).exists()).unwrap_or(false),
has_markdown: markdown_path.as_ref().map(|p| state.config.library_dir.join(p).exists()).unwrap_or(false),
has_translation: translation_path.as_ref().map(|p| state.config.library_dir.join(p).exists()).unwrap_or(false),
doctype: doctype_val.unwrap_or_else(|| "article".to_string()),
pdf_error,
html_error,
});
}
Ok(Json(list))
}
// ── POST /api/export ──
#[derive(Deserialize)]
pub struct ExportRequest {
pub bibcodes: Vec<String>,
}
#[derive(Serialize)]
pub struct ExportResponse {
pub bibtex: String,
}
// 批量请求 ADS 接口,获取选中 Bibcode 的标准 BibTeX 引文段落
pub async fn export_citations(
State(state): State<Arc<AppState>>,
Json(req): Json<ExportRequest>,
) -> Result<Json<ExportResponse>, (StatusCode, String)> {
if state.config.ads_api_key.is_empty() {
return Err((StatusCode::BAD_REQUEST, "ADS API key 未在 .env 中配置,无法使用该接口".to_string()));
}
let bibtex = state.ads.export_bibtex(req.bibcodes).await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("批量引文导出失败: {}", e)))?;
Ok(Json(ExportResponse { bibtex }))
}
// ── POST /api/upload ──
// 允许手动上传/推入本地 PDF 或 HTML 文献文件
pub async fn upload_paper_file(
State(state): State<Arc<AppState>>,
mut multipart: axum::extract::Multipart,
) -> Result<Json<StandardPaper>, (StatusCode, String)> {
let mut bibcode = String::new();
let mut file_type = String::new(); // "pdf" 或 "html"
let mut file_bytes = Vec::new();
let mut file_name = String::new();
while let Some(field) = multipart.next_field().await.map_err(|e| {
(StatusCode::BAD_REQUEST, format!("解析文件分块失败: {}", e))
})? {
let name = field.name().unwrap_or("").to_string();
if name == "bibcode" {
bibcode = field.text().await.unwrap_or_default();
} else if name == "type" {
file_type = field.text().await.unwrap_or_default();
} else if name == "file" {
file_name = field.file_name().unwrap_or("").to_string();
file_bytes = field.bytes().await.map_err(|e| {
(StatusCode::INTERNAL_SERVER_ERROR, format!("读取文件字节流失败: {}", e))
})?.to_vec();
}
}
if bibcode.is_empty() {
return Err((StatusCode::BAD_REQUEST, "缺少 bibcode 参数".to_string()));
}
if file_bytes.is_empty() {
return Err((StatusCode::BAD_REQUEST, "上传文件为空或读取失败".to_string()));
}
// 尝试将可能的 DOI 或 arXiv ID 解析为真实的 bibcode
let mut resolved_bibcode = bibcode.clone();
let exists_as_bibcode = sqlx::query("SELECT bibcode FROM papers WHERE bibcode = ?")
.bind(&resolved_bibcode)
.fetch_optional(&state.db)
.await
.unwrap_or(None)
.is_some();
if !exists_as_bibcode {
// 尝试匹配 DOI
let clean_doi = resolved_bibcode
.trim_start_matches("doi:")
.trim_start_matches("DOI:")
.trim_start_matches("https://doi.org/")
.trim_start_matches("http://doi.org/")
.trim();
if let Some(row) = sqlx::query("SELECT bibcode FROM papers WHERE doi = ? OR doi = ? OR LOWER(doi) = LOWER(?)")
.bind(clean_doi)
.bind(&resolved_bibcode)
.bind(clean_doi)
.fetch_optional(&state.db)
.await
.unwrap_or(None)
{
let found: String = row.get(0);
info!("上传接口:通过 DOI 匹配成功,将 '{}' 解析为 bibcode '{}'", bibcode, found);
resolved_bibcode = found;
} else {
// 尝试匹配 arXiv ID
let clean_arxiv = resolved_bibcode
.trim_start_matches("arxiv:")
.trim_start_matches("arXiv:")
.trim_start_matches("ARXIV:")
.trim();
// 移除可能存在的版本号后缀(如 2303.12345v1 -> 2303.12345
let clean_arxiv_no_version = if let Some(pos) = clean_arxiv.find('v') {
if clean_arxiv[pos+1..].chars().all(|c| c.is_ascii_digit()) {
&clean_arxiv[..pos]
} else {
clean_arxiv
}
} else {
clean_arxiv
};
if let Some(row) = sqlx::query(
"SELECT bibcode FROM papers WHERE arxiv_id = ? OR arxiv_id = ? OR arxiv_id LIKE ? OR arxiv_id LIKE ?"
)
.bind(clean_arxiv)
.bind(clean_arxiv_no_version)
.bind(format!("{}%", clean_arxiv_no_version))
.bind(format!("arXiv:{}%", clean_arxiv_no_version))
.fetch_optional(&state.db)
.await
.unwrap_or(None) {
let found: String = row.get(0);
info!("上传接口:通过 arXiv ID 匹配成功,将 '{}' 解析为 bibcode '{}'", bibcode, found);
resolved_bibcode = found;
}
}
}
let bibcode = resolved_bibcode;
// 从数据库读取该文献元数据
let _paper = get_paper_from_db(&state.db, &state.config.library_dir, &bibcode)
.await
.map_err(|e| (StatusCode::NOT_FOUND, format!("未找到该文献记录: {}", e)))?;
// 校验并保存文件
let is_pdf = file_type == "pdf" || file_name.to_lowercase().ends_with(".pdf");
let relative_path = if is_pdf {
crate::services::download::validate_pdf_content(&file_bytes).map_err(|e| {
(StatusCode::BAD_REQUEST, format!("PDF 文件内容校验失败: {}", e))
})?;
let pdf_filename = format!("{}.pdf", bibcode);
let pdf_dest = state.config.library_dir.join("PDF").join(&pdf_filename);
if let Some(parent) = pdf_dest.parent() {
std::fs::create_dir_all(parent).unwrap_or_default();
}
std::fs::write(&pdf_dest, &file_bytes).map_err(|e| {
(StatusCode::INTERNAL_SERVER_ERROR, format!("无法写入 PDF 文件: {}", e))
})?;
format!("PDF/{}", pdf_filename)
} else {
let text_content = String::from_utf8(file_bytes).map_err(|_| {
(StatusCode::BAD_REQUEST, "上传的 HTML 文件不是有效的 UTF-8 文本".to_string())
})?;
crate::services::download::validate_html_content_lenient(&text_content).map_err(|e| {
(StatusCode::BAD_REQUEST, format!("HTML 文件内容校验失败: {}", e))
})?;
let html_filename = format!("{}.html", bibcode);
let html_dest = state.config.library_dir.join("HTML").join(&html_filename);
if let Some(parent) = html_dest.parent() {
std::fs::create_dir_all(parent).unwrap_or_default();
}
std::fs::write(&html_dest, &text_content).map_err(|e| {
(StatusCode::INTERNAL_SERVER_ERROR, format!("无法写入 HTML 文件: {}", e))
})?;
format!("HTML/{}", html_filename)
};
// 更新数据库路径状态
let path_field = if is_pdf { "pdf_path" } else { "html_path" };
let sql = format!("UPDATE papers SET {} = ? WHERE bibcode = ?", path_field);
sqlx::query(&sql)
.bind(&relative_path)
.bind(&bibcode)
.execute(&state.db)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("更新数据库状态失败: {}", e)))?;
// 重新获取最新的文献信息以更新前端界面
let updated_paper = get_paper_from_db(&state.db, &state.config.library_dir, &bibcode)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("重读文献数据失败: {}", e)))?;
Ok(Json(updated_paper))
}
// ── POST /api/no_resource ──
#[derive(Debug, Deserialize)]
pub struct MarkNoResourceRequest {
pub bibcode: String,
pub clear: Option<bool>,
}
pub async fn mark_no_resource(
State(state): State<Arc<AppState>>,
Json(req): Json<MarkNoResourceRequest>,
) -> Result<Json<StandardPaper>, (StatusCode, String)> {
let clear_flag = req.clear.unwrap_or(false);
if clear_flag {
info!("接收到清除文献无资源标记指令,标识符: {}", req.bibcode);
sqlx::query("UPDATE papers SET pdf_path = NULL, html_path = NULL WHERE bibcode = ?")
.bind(&req.bibcode)
.execute(&state.db)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("清除无资源标记失败: {}", e)))?;
} else {
info!("接收到文献无资源标记指令,标识符: {}", req.bibcode);
sqlx::query("UPDATE papers SET pdf_path = 'error:no_resource', html_path = 'error:no_resource' WHERE bibcode = ?")
.bind(&req.bibcode)
.execute(&state.db)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("更新数据库无资源标记失败: {}", e)))?;
}
// 重新获取最新的文献信息以更新前端界面
let updated_paper = get_paper_from_db(&state.db, &state.config.library_dir, &req.bibcode)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("重读文献数据失败: {}", e)))?;
Ok(Json(updated_paper))
}
// ── GET /api/active_bibcode ──
#[derive(Debug, Serialize)]
pub struct ActiveBibcodeResponse {
pub bibcode: Option<String>,
}
pub async fn get_active_bibcode(
State(state): State<Arc<AppState>>,
) -> Json<ActiveBibcodeResponse> {
let active = state.active_bibcode.lock().await;
Json(ActiveBibcodeResponse {
bibcode: active.clone(),
})
}
// ── POST /api/active_bibcode ──
#[derive(Debug, Deserialize)]
pub struct SetActiveBibcodeRequest {
pub bibcode: Option<String>,
}
pub async fn set_active_bibcode(
State(state): State<Arc<AppState>>,
Json(req): Json<SetActiveBibcodeRequest>,
) -> StatusCode {
let mut active = state.active_bibcode.lock().await;
*active = req.bibcode;
tracing::debug!("已更新当前活跃文献 Bibcode 标记为: {:?}", *active);
StatusCode::OK
}
+339
View File
@@ -0,0 +1,339 @@
// src/api/sync.rs
use axum::{
extract::{Query, State},
http::StatusCode,
Json,
};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use sqlx::Row;
use tracing::error;
use super::AppState;
// ── POST /api/sync/meta/run ──
#[derive(Debug, Deserialize)]
pub struct MetaSyncRunRequest {
pub q: String,
pub source: String, // "all" | "ads" | "arxiv"
pub limit: i32,
}
pub async fn run_meta_sync(
State(state): State<Arc<AppState>>,
Json(req): Json<MetaSyncRunRequest>,
) -> Result<StatusCode, (StatusCode, String)> {
// 检查并同步初始化任务状态,防止前端轮询竞态条件
{
let mut status = state.harvest_status.lock().await;
if status.active {
return Err((StatusCode::CONFLICT, "当前已有文献批量同步任务在后台运行中,请勿重复启动".to_string()));
}
status.active = true;
status.query = req.q.clone();
status.source = req.source.clone();
status.synced = 0;
status.total = 0;
}
crate::services::batch_sync::MetaSync::start_harvest(
state.db.clone(),
Arc::new(state.ads.clone()),
Arc::new(state.arxiv.clone()),
req.q,
req.source,
req.limit,
state.harvest_status.clone(),
);
Ok(StatusCode::ACCEPTED)
}
// ── GET /api/sync/meta/count ──
#[derive(Debug, Deserialize)]
pub struct MetaSyncCountRequest {
pub q: String,
pub source: String, // "all" | "ads" | "arxiv"
}
#[derive(Debug, Serialize)]
pub struct MetaSyncCountResponse {
pub total: i32,
}
pub async fn get_meta_sync_count(
State(state): State<Arc<AppState>>,
Query(req): Query<MetaSyncCountRequest>,
) -> Result<Json<MetaSyncCountResponse>, (StatusCode, String)> {
let total = crate::services::batch_sync::MetaSync::get_total_count(
&req.q,
&req.source,
&state.ads,
&state.arxiv,
)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("获取预估文献数失败: {}", e)))?;
Ok(Json(MetaSyncCountResponse { total }))
}
// ── GET /api/sync/meta/status ──
pub async fn get_meta_sync_status(
State(state): State<Arc<AppState>>,
) -> Json<crate::services::batch_sync::MetaSyncStatus> {
let status = state.harvest_status.lock().await;
Json(status.clone())
}
// ── POST /api/sync/asset/run ──
#[derive(Debug, Deserialize)]
pub struct AssetSyncRunRequest {
pub target_phase: String, // "download" | "parse" | "translate"
pub limit_count: Option<i32>, // 批量处理上限,默认 100
pub sort_order: Option<String>, // 处理顺序: "default" | "pub_year_desc" | "created_at_desc"
pub skip_completed: Option<bool>,
pub skip_failed: Option<bool>, // 跳过当前失败 ('error:*')
pub skip_preceding_failed: Option<bool>, // 跳过前置失败
pub skip_preceding_uncompleted: Option<bool>, // 跳过前置未完成
}
struct PaperRecord {
bibcode: String,
pdf_path: Option<String>,
html_path: Option<String>,
markdown_path: Option<String>,
translation_path: Option<String>,
year: String,
created_at: String,
}
pub async fn run_asset_sync(
State(state): State<Arc<AppState>>,
Json(req): Json<AssetSyncRunRequest>,
) -> Result<StatusCode, (StatusCode, String)> {
// 检查是否已经在进行批量处理任务
{
let status = state.process_status.lock().await;
if status.active {
return Err((StatusCode::CONFLICT, "当前已有文献批量任务在后台运行中,请勿重复启动".to_string()));
}
}
let target_phase = req.target_phase.clone();
let action = match target_phase.as_str() {
"download" => crate::services::batch_sync::SyncAction::Download,
"parse" => crate::services::batch_sync::SyncAction::Parse,
"translate" => crate::services::batch_sync::SyncAction::Translate,
_ => return Err((StatusCode::BAD_REQUEST, "不支持的 target_phase 参数值".to_string())),
};
let rows = sqlx::query("SELECT bibcode, pdf_path, html_path, markdown_path, translation_path, year, datetime(created_at, 'localtime') FROM papers")
.fetch_all(&state.db)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("读取数据库失败: {}", e)))?;
let mut records = Vec::new();
for r in rows {
records.push(PaperRecord {
bibcode: r.get(0),
pdf_path: r.get(1),
html_path: r.get(2),
markdown_path: r.get(3),
translation_path: r.get(4),
year: r.get(5),
created_at: r.get(6),
});
}
// 排序
let sort_order = req.sort_order.clone().unwrap_or_else(|| "default".to_string());
if sort_order == "pub_year_desc" {
records.sort_by(|a, b| b.year.cmp(&a.year));
} else if sort_order == "created_at_desc" {
records.sort_by(|a, b| b.created_at.cmp(&a.created_at));
}
let skip_completed = req.skip_completed.unwrap_or(false);
let skip_failed = req.skip_failed.unwrap_or(false);
let skip_preceding_failed = req.skip_preceding_failed.unwrap_or(false);
let skip_preceding_uncompleted = req.skip_preceding_uncompleted.unwrap_or(false);
let is_completed = |path: &Option<String>| -> bool {
if let Some(p) = path {
!p.is_empty() && !p.starts_with("error:") && !p.starts_with("mineru_batch:")
} else {
false
}
};
let is_failed = |path: &Option<String>| -> bool {
if let Some(p) = path {
p.starts_with("error:")
} else {
false
}
};
let is_no_resource = |path: &Option<String>| -> bool {
if let Some(p) = path {
p.starts_with("error:no_resource") || p.starts_with("error:无资源") || p.starts_with("error:无有效全文")
} else {
false
}
};
let mut target_bibcodes = Vec::new();
for rec in records {
let is_no_resource_paper = is_no_resource(&rec.pdf_path) || is_no_resource(&rec.html_path);
if is_no_resource_paper {
continue;
}
let download_completed = is_completed(&rec.pdf_path) || is_completed(&rec.html_path);
let download_failed = is_failed(&rec.pdf_path) || is_failed(&rec.html_path);
let download_uncompleted = !download_completed && !download_failed;
let parse_completed = is_completed(&rec.markdown_path);
let parse_failed = is_failed(&rec.markdown_path);
let parse_uncompleted = !parse_completed && !parse_failed;
let translate_completed = is_completed(&rec.translation_path);
let translate_failed = is_failed(&rec.translation_path);
match target_phase.as_str() {
"download" => {
if skip_completed && download_completed {
continue;
}
if skip_failed && download_failed {
continue;
}
}
"parse" => {
if skip_completed && parse_completed {
continue;
}
if skip_failed && parse_failed {
continue;
}
if skip_preceding_failed && download_failed {
continue;
}
if skip_preceding_uncompleted && download_uncompleted {
continue;
}
}
"translate" => {
if skip_completed && translate_completed {
continue;
}
if skip_failed && translate_failed {
continue;
}
if skip_preceding_failed && parse_failed {
continue;
}
if skip_preceding_uncompleted && parse_uncompleted {
continue;
}
}
_ => {}
}
target_bibcodes.push(rec.bibcode);
}
let limit = req.limit_count.unwrap_or(100) as usize;
if target_bibcodes.len() > limit {
target_bibcodes.truncate(limit);
}
if target_bibcodes.is_empty() {
return Err((StatusCode::OK, "没有需要处理的文献".to_string()));
}
// 启动后台处理
crate::services::batch_sync::AssetSync::start_process(
state.db.clone(),
state.config.clone(),
Arc::new(state.downloader.clone()),
Arc::new(state.qiniu.clone()),
Arc::new(state.dict.clone()),
action,
target_bibcodes,
state.process_status.clone(),
);
Ok(StatusCode::ACCEPTED)
}
// ── POST /api/sync/asset/stop ──
pub async fn stop_asset_sync(
State(state): State<Arc<AppState>>,
) -> StatusCode {
let mut status = state.process_status.lock().await;
if status.active {
status.active = false;
status.add_log("用户手动终止了批量处理任务。".to_string());
}
StatusCode::OK
}
// ── GET /api/sync/queries ──
#[derive(Debug, Serialize, Deserialize)]
pub struct SavedSyncQuery {
pub id: i64,
pub query: String,
pub source: String,
pub limit_count: i32,
pub last_run: String,
}
pub async fn get_sync_queries(
State(state): State<Arc<AppState>>,
) -> Result<Json<Vec<SavedSyncQuery>>, (StatusCode, String)> {
let rows = sqlx::query("SELECT id, query, source, limit_count, datetime(last_run, 'localtime') FROM sync_queries ORDER BY last_run DESC")
.fetch_all(&state.db)
.await
.map_err(|e| {
error!("获取已存同步检索配置失败: {}", e);
(StatusCode::INTERNAL_SERVER_ERROR, format!("获取已存同步检索配置失败: {}", e))
})?;
let mut list = Vec::new();
for r in rows {
list.push(SavedSyncQuery {
id: r.get(0),
query: r.get(1),
source: r.get(2),
limit_count: r.get(3),
last_run: r.get(4),
});
}
Ok(Json(list))
}
// ── DELETE /api/sync/queries/:id ──
pub async fn delete_sync_query(
State(state): State<Arc<AppState>>,
axum::extract::Path(id): axum::extract::Path<i64>,
) -> Result<StatusCode, (StatusCode, String)> {
sqlx::query("DELETE FROM sync_queries WHERE id = ?")
.bind(id)
.execute(&state.db)
.await
.map_err(|e| {
error!("删除同步检索配置失败: {}", e);
(StatusCode::INTERNAL_SERVER_ERROR, format!("删除同步检索配置失败: {}", e))
})?;
Ok(StatusCode::OK)
}
// ── GET /api/sync/asset/status ──
pub async fn get_asset_sync_status(
State(state): State<Arc<AppState>>,
) -> Json<crate::services::batch_sync::AssetSyncStatus> {
let status = state.process_status.lock().await;
Json(status.clone())
}
+501
View File
@@ -0,0 +1,501 @@
// src/bin/health_check.rs
use std::fs;
use std::path::{Path, PathBuf};
use sqlx::{SqlitePool, Row};
use astroresearch::Config;
use tracing::{error, Level};
use tracing_subscriber::FmtSubscriber;
// 检测防爬、验证码、登录墙特征
fn detect_anti_bot(content: &str) -> Option<&'static str> {
let lower = content.to_lowercase();
let cf_patterns = [
("checking your browser", "Cloudflare WAF 浏览器检查"),
("please wait while we verify", "Cloudflare WAF 验证"),
("cf-browser-verification", "Cloudflare WAF 验证特征"),
("cf_chl_opt", "Cloudflare WAF 特征"),
("just a moment", "Cloudflare 正在等待提示"),
("enable javascript and cookies", "Cloudflare JS 挑战"),
("_cf_chl_tk", "Cloudflare Token 特征"),
("awswafintegration", "AWS WAF 拦截"),
("aws waf", "AWS WAF 拦截"),
("captcha", "人机验证码页面"),
("recaptcha", "Google reCAPTCHA"),
("hcaptcha", "hCaptcha 验证"),
("verify you are human", "人机验证提示"),
("robot check", "机器人检测"),
("login required", "出版商登录墙"),
("please log in", "出版商登录墙"),
("subscription required", "出版商订阅/付费墙"),
("access denied", "拒绝访问/付费墙"),
("you do not have access", "无权访问文献"),
("purchase this article", "文章付费墙"),
("sign in to access", "登录以获取访问权限"),
("radware bot manager captcha", "Radware Bot Manager 验证"),
("shieldsquare_styles", "ShieldSquare WAF 拦截"),
];
for &(p, desc) in &cf_patterns {
if lower.contains(p) {
return Some(desc);
}
}
None
}
// 校验 PDF 完整性与是否为虚假内容
fn validate_pdf_content(bytes: &[u8]) -> Result<(), String> {
if !bytes.starts_with(b"%PDF") {
if bytes.starts_with(b"<!") || bytes.starts_with(b"<html") || bytes.starts_with(b"<HTML") {
let scan_len = std::cmp::min(2048, bytes.len());
let text = String::from_utf8_lossy(&bytes[..scan_len]);
if let Some(desc) = detect_anti_bot(&text) {
return Err(format!("虽然文件后缀是 PDF,但实际内容是 HTML(检测到:{}", desc));
}
return Err("虽然文件后缀是 PDF,但实际内容是 HTML 网页,可能是重定向或拦截页面".to_string());
}
return Err("缺少 %PDF 文件头魔数,文件损坏或并非 PDF".to_string());
}
if bytes.len() < 5000 {
return Err(format!("PDF 文件过小(仅 {} 字节),极可能是错误信息页", bytes.len()));
}
let scan_len = std::cmp::min(1024, bytes.len());
let tail = &bytes[bytes.len() - scan_len..];
if !tail.windows(5).any(|w| w == b"%%EOF") {
return Err("PDF 文件未包含尾部 %%EOF 标记,文件已损坏或不完整".to_string());
}
Ok(())
}
// 校验 HTML 内容有效性(过滤假网页、存根页、跳转页及摘要页)
fn validate_html_content(text: &str) -> Result<(), String> {
if let Some(desc) = detect_anti_bot(text) {
return Err(format!("检测到安全拦截或登录限制页:{}", desc));
}
let lower = text.to_lowercase();
// 1. 检查常见的跳转与错误占位特征
if lower.contains("redirecting") || lower.contains("redirect to") || lower.contains("http-equiv=\"refresh\"") || lower.contains("autoredirecttourl") {
return Err("检测到 HTML 重定向跳转页面,而非真实文献正文".to_string());
}
if lower.contains("conversion to html had a fatal error") || lower.contains("no content available") || lower.contains("fatal error and exited abruptly") {
return Err("检测到 ar5iv 转换失败的占位 HTML 页面".to_string());
}
if lower.contains("see pages 1-last of") {
return Err("检测到仅包含 PDF 链接 of 占位 HTML 页面".to_string());
}
// 2. 网页标题精准黑名单校验(防止正文中提及 NSF/VizieR 导致误伤)
if let Some(start_pos) = lower.find("<title") {
if let Some(tag_end) = lower[start_pos..].find('>') {
let title_start = start_pos + tag_end + 1;
if let Some(end_pos) = lower[title_start..].find("</title>") {
let title = &lower[title_start..title_start + end_pos];
if title.contains("nsf award search")
|| title.contains("national science foundation")
|| title.contains("vizier")
|| title.contains("caltechthesis")
|| title.contains("caosp abstract")
|| title.contains("asp conference series")
|| title.contains("aspbooks")
{
return Err(format!("检测到占位网页标题: \"{}\",判定为非正本文献", title.trim()));
}
}
}
}
// 3. 基础字节长度与具体 HTTP 错误特征校验
if text.len() < 2000 {
let error_patterns = [
"404 not found", "403 forbidden", "502 bad gateway",
"500 internal server error", "access denied", "site error"
];
for kw in &error_patterns {
if lower.contains(kw) {
return Err(format!("HTML 包含错误页面特征(包含: {}", kw));
}
}
}
// 4. 结构启发式校验:如果是小于 50KB 的 HTML,必须包含基本的章节或参考文献结构,否则判定为摘要/存根占位页
if text.len() < 50000 {
// 匹配 heading 标签或 Markdown 格式的标题,而不是纯文本中的单词
let has_sections = lower.contains("ltx_title_section")
|| lower.contains("class=\"section\"")
|| lower.contains("## introduction")
|| lower.contains("<h2>introduction")
|| lower.contains("<h3>introduction")
|| lower.contains("class=\"ltx_section\"");
let has_bib = lower.contains("ltx_bibliography")
|| lower.contains("class=\"references\"")
|| lower.contains("<ol class=\"references\"")
|| lower.contains("<ul class=\"references\"")
|| lower.contains("id=\"bib\"")
|| lower.contains("class=\"ltx_bibliography\"");
if !has_sections && !has_bib {
return Err(format!("HTML 长度偏小({} 字节)且缺少正文章节或参考文献,判定为非正本文献", text.len()));
}
}
Ok(())
}
// 递归扫描目录下的所有物理文件
fn scan_directory(dir: &Path, files: &mut Vec<PathBuf>) {
if let Ok(entries) = fs::read_dir(dir) {
for entry in entries.flatten() {
let path = entry.path();
if path.is_dir() {
scan_directory(&path, files);
} else {
files.push(path);
}
}
}
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// 初始化日志
let subscriber = FmtSubscriber::builder()
.with_max_level(Level::INFO)
.finish();
tracing::subscriber::set_global_default(subscriber)?;
let args: Vec<String> = std::env::args().collect();
let fix = args.contains(&"--fix".to_string());
println!("==================================================");
println!(" AstroResearch 馆藏文献健康度检查工具 ");
println!("==================================================");
if fix {
println!("⚠️ 警告:检测到 --fix 参数。程序将自动删除坏文件并将数据库路径重置为 NULL 以便下次重新下载。");
} else {
println!("ℹ️ 提示:当前处于只读扫描模式。如需自动修复坏文件,请附加 '-- --fix' 参数运行。");
}
println!("--------------------------------------------------");
let config = Config::from_env();
let library_dir = &config.library_dir;
println!("本地文献库目录: {:?}", library_dir);
println!("数据库连接串: {}", config.database_url);
let db_path = config.database_url.replace("sqlite://", "");
if !Path::new(&db_path).exists() {
error!("找不到 SQLite 数据库文件: {:?}", db_path);
return Ok(());
}
let pool = SqlitePool::connect(&config.database_url).await?;
println!("成功连接数据库。正在准备进行全面健康检查...");
println!("--------------------------------------------------");
// ─── 阶段 1:磁盘物理文件直接扫描(解决孤儿垃圾文件与坏文件) ───
let mut html_files = Vec::new();
let mut pdf_files = Vec::new();
scan_directory(&library_dir.join("HTML"), &mut html_files);
scan_directory(&library_dir.join("PDF"), &mut pdf_files);
println!("📂 正在扫描物理磁盘文件 (HTML: {} 个, PDF: {} 个)...", html_files.len(), pdf_files.len());
let mut disk_html_invalid = 0;
let mut disk_pdf_invalid = 0;
let mut deleted_files = 0;
let mut db_updated_count = 0;
for path in html_files {
let rel_path = match path.strip_prefix(library_dir) {
Ok(p) => p.to_str().unwrap_or(""),
Err(_) => continue,
};
if let Ok(content) = fs::read_to_string(&path) {
if let Err(e) = validate_html_content(&content) {
disk_html_invalid += 1;
println!(" ❌ 发现磁盘上损坏的 HTML 文件: {:?}", rel_path);
println!(" 原因: {}", e);
if fix {
let _ = fs::remove_file(&path);
deleted_files += 1;
println!(" 🧹 [修复] 已物理删除损坏的文件");
// 检索是否有数据库记录并将其重置
let res = sqlx::query("UPDATE papers SET html_path = NULL WHERE html_path = ? OR html_path = ?")
.bind(rel_path)
.bind(format!("HTML/{}", Path::new(rel_path).file_name().and_then(|f| f.to_str()).unwrap_or("")))
.execute(&pool)
.await;
if let Ok(r) = res {
if r.rows_affected() > 0 {
db_updated_count += r.rows_affected();
println!(" ✅ [修复] 数据库对应状态已重置 (受影响行数: {})", r.rows_affected());
}
}
}
println!("--------------------------------------------------");
}
}
}
for path in pdf_files {
let rel_path = match path.strip_prefix(library_dir) {
Ok(p) => p.to_str().unwrap_or(""),
Err(_) => continue,
};
if let Ok(bytes) = fs::read(&path) {
if let Err(e) = validate_pdf_content(&bytes) {
disk_pdf_invalid += 1;
println!(" ❌ 发现磁盘上损坏的 PDF 文件: {:?}", rel_path);
println!(" 原因: {}", e);
if fix {
let _ = fs::remove_file(&path);
deleted_files += 1;
println!(" 🧹 [修复] 已物理删除损坏的文件");
let res = sqlx::query("UPDATE papers SET pdf_path = NULL WHERE pdf_path = ? OR pdf_path = ?")
.bind(rel_path)
.bind(format!("PDF/{}", Path::new(rel_path).file_name().and_then(|f| f.to_str()).unwrap_or("")))
.execute(&pool)
.await;
if let Ok(r) = res {
if r.rows_affected() > 0 {
db_updated_count += r.rows_affected();
println!(" ✅ [修复] 数据库对应状态已重置 (受影响行数: {})", r.rows_affected());
}
}
}
println!("--------------------------------------------------");
}
}
}
// ─── 阶段 2:数据库记录校验扫描(检测丢失文件、报错记录与孤立 Markdown) ───
println!("🗄️ 正在校验数据库表记录一致性...");
let db_rows = sqlx::query(
"SELECT bibcode, pdf_path, html_path, title, markdown_path, doctype FROM papers"
)
.fetch_all(&pool)
.await?;
let mut db_pdf_missing = 0;
let mut db_html_missing = 0;
let mut db_pdf_err_text = 0;
let mut db_html_err_text = 0;
let mut db_markdown_missing = 0;
let mut db_markdown_orphaned = 0;
let mut db_skip_type_cleaned = 0;
for r in db_rows {
let bibcode: String = r.get(0);
let pdf_path_opt: Option<String> = r.get(1);
let html_path_opt: Option<String> = r.get(2);
let title: String = r.get(3);
let markdown_path_opt: Option<String> = r.get(4);
let doctype_opt: Option<String> = r.get(5);
let mut need_db_fix = false;
let mut pdf_needs_fix = false;
let mut html_needs_fix = false;
let mut markdown_needs_fix = false;
let mut pdf_db_msg = String::new();
let mut html_db_msg = String::new();
let mut markdown_db_msg = String::new();
let doctype_str = doctype_opt.unwrap_or_else(|| "article".to_string()).to_lowercase();
let is_skip_type = doctype_str == "proposal"
|| doctype_str == "abstract"
|| doctype_str == "catalog"
|| doctype_str == "dataset"
|| doctype_str == "software"
|| doctype_str == "circular"
|| doctype_str == "newsletter"
|| doctype_str == "obituary";
if is_skip_type {
let mut has_skip_anomaly = false;
if let Some(ref pdf_p) = pdf_path_opt {
pdf_db_msg = format!("该文献属于跳过类型 [{}],但包含下载/报错路径记录: {}", doctype_str, pdf_p);
need_db_fix = true;
pdf_needs_fix = true;
has_skip_anomaly = true;
}
if let Some(ref html_p) = html_path_opt {
html_db_msg = format!("该文献属于跳过类型 [{}],但包含下载/报错路径记录: {}", doctype_str, html_p);
need_db_fix = true;
html_needs_fix = true;
has_skip_anomaly = true;
}
if let Some(ref md_p) = markdown_path_opt {
markdown_db_msg = format!("该文献属于跳过类型 [{}],但包含解析路径记录: {}", doctype_str, md_p);
need_db_fix = true;
markdown_needs_fix = true;
has_skip_anomaly = true;
}
if has_skip_anomaly {
db_skip_type_cleaned += 1;
}
} else {
if let Some(ref pdf_p) = pdf_path_opt {
if pdf_p.starts_with("error:") {
db_pdf_err_text += 1;
pdf_db_msg = format!("数据库存储了报错字符串: {}", pdf_p);
} else if !library_dir.join(pdf_p).exists() {
db_pdf_missing += 1;
pdf_db_msg = format!("物理 PDF 文件丢失 (路径: {})", pdf_p);
need_db_fix = true;
pdf_needs_fix = true;
}
}
if let Some(ref html_p) = html_path_opt {
if html_p.starts_with("error:") {
db_html_err_text += 1;
html_db_msg = format!("数据库存储了报错字符串: {}", html_p);
} else if !library_dir.join(html_p).exists() {
db_html_missing += 1;
html_db_msg = format!("物理 HTML 文件丢失 (路径: {})", html_p);
need_db_fix = true;
html_needs_fix = true;
}
}
if let Some(ref md_p) = markdown_path_opt {
if !library_dir.join(md_p).exists() {
db_markdown_missing += 1;
markdown_db_msg = format!("物理 Markdown 文件丢失 (路径: {})", md_p);
need_db_fix = true;
markdown_needs_fix = true;
} else {
// 如果 Markdown 物理文件存在,但它既没有有效 PDF 也没有有效 HTML
let has_valid_pdf = pdf_path_opt.as_ref()
.map(|p| !p.starts_with("error:") && library_dir.join(p).exists())
.unwrap_or(false);
let has_valid_html = html_path_opt.as_ref()
.map(|p| !p.starts_with("error:") && library_dir.join(p).exists())
.unwrap_or(false);
if !has_valid_pdf && !has_valid_html {
db_markdown_orphaned += 1;
markdown_db_msg = format!("Markdown 存在且完好,但失去有效 PDF/HTML 数据源,判定为孤立的 Markdown (路径: {})", md_p);
need_db_fix = true;
markdown_needs_fix = true;
}
}
}
}
if need_db_fix {
println!(" ❌ 发现馆藏文献记录损坏/不一致 [{}] 《{}", bibcode, title);
if !pdf_db_msg.is_empty() {
if pdf_needs_fix {
println!(" [异常] PDF 状态: {}", pdf_db_msg);
} else {
println!(" [日志] PDF 历史下载失败原因: {}", pdf_db_msg.replace("数据库存储了报错字符串: ", ""));
}
}
if !html_db_msg.is_empty() {
if html_needs_fix {
println!(" [异常] HTML 状态: {}", html_db_msg);
} else {
println!(" [日志] HTML 历史下载失败原因: {}", html_db_msg.replace("数据库存储了报错字符串: ", ""));
}
}
if !markdown_db_msg.is_empty() {
println!(" [异常] Markdown 状态: {}", markdown_db_msg);
}
if fix {
let mut sql_parts = Vec::new();
if pdf_needs_fix {
sql_parts.push("pdf_path = NULL");
if let Some(ref pdf_p) = pdf_path_opt {
if !pdf_p.starts_with("error:") {
let pdf_abs = library_dir.join(pdf_p);
if pdf_abs.exists() {
let _ = fs::remove_file(&pdf_abs);
deleted_files += 1;
println!(" 🧹 [修复] 已物理删除跳过类型或丢失的 PDF 文件");
}
}
}
}
if html_needs_fix {
sql_parts.push("html_path = NULL");
if let Some(ref html_p) = html_path_opt {
if !html_p.starts_with("error:") {
let html_abs = library_dir.join(html_p);
if html_abs.exists() {
let _ = fs::remove_file(&html_abs);
deleted_files += 1;
println!(" 🧹 [修复] 已物理删除跳过类型或丢失的 HTML 文件");
}
}
}
}
if markdown_needs_fix {
sql_parts.push("markdown_path = NULL");
if let Some(ref md_p) = markdown_path_opt {
let md_abs = library_dir.join(md_p);
if md_abs.exists() {
let _ = fs::remove_file(&md_abs);
deleted_files += 1;
println!(" 🧹 [修复] 已物理删除孤立或跳过类型的 Markdown 文件");
}
}
}
if !sql_parts.is_empty() {
let query_str = format!("UPDATE papers SET {} WHERE bibcode = ?", sql_parts.join(", "));
let res = sqlx::query(&query_str)
.bind(&bibcode)
.execute(&pool)
.await;
if res.is_ok() {
db_updated_count += 1;
println!(" ✅ [修复] 数据库损坏字段已成功重置");
}
}
}
println!("--------------------------------------------------");
}
}
println!("\n==================================================");
println!(" 全面健康度检测扫描报告 ");
println!("==================================================");
println!("磁盘物理损坏统计:");
println!(" - 损坏/假 HTML 文件数: {}", disk_html_invalid);
println!(" - 损坏/假 PDF 文件数: {}", disk_pdf_invalid);
println!("--------------------------------------------------");
println!("数据库一致性统计:");
println!(" - 数据库记录下载失败数 (error:): PDF: {}, HTML: {}", db_pdf_err_text, db_html_err_text);
println!(" - 磁盘文件丢失数 (数据库有记录但文件不存在): PDF: {}, HTML: {}, Markdown: {}", db_pdf_missing, db_html_missing, db_markdown_missing);
println!(" - 孤立无源 Markdown 篇数: {}", db_markdown_orphaned);
println!(" - 需跳过类型但包含下载记录篇数 (已清理/待清理): {}", db_skip_type_cleaned);
println!("--------------------------------------------------");
if fix {
println!("✨ 修复完成!");
println!(" - 共删除磁盘物理损坏/孤立/跳过类型文件: {}", deleted_files);
println!(" - 共重置修复数据库文献字段: {}", db_updated_count);
} else {
let total_issues = disk_html_invalid + disk_pdf_invalid + db_pdf_missing + db_html_missing + db_markdown_missing + db_markdown_orphaned + db_skip_type_cleaned;
if total_issues > 0 {
println!("❌ 警告:共检测出 {} 处坏文件、丢失文件或异常数据库记录。", total_issues);
println!("👉 您可以附加 '-- --fix' 执行一键全面修复:");
println!(" cargo run --bin health_check -- --fix");
} else {
println!("🎉 恭喜!馆藏物理文件及数据库完全健康,未检测到任何损坏或失效记录!");
}
}
println!("==================================================");
Ok(())
}
+5 -1
View File
@@ -94,6 +94,7 @@ async fn main() -> anyhow::Result<()> {
downloader,
harvest_status: Arc::new(tokio::sync::Mutex::new(astroresearch::services::batch_sync::MetaSyncStatus::new())),
process_status: Arc::new(tokio::sync::Mutex::new(astroresearch::services::batch_sync::AssetSyncStatus::new())),
active_bibcode: Arc::new(tokio::sync::Mutex::new(None)),
});
// 7. 设置 Axum 路由、CORS 头以及 React 仪表盘静态资源托管
@@ -105,6 +106,8 @@ async fn main() -> anyhow::Result<()> {
let api_routes = Router::new()
.route("/search", get(handlers::search_papers))
.route("/download", post(handlers::download_paper))
.route("/upload", post(handlers::upload_paper_file))
.route("/no_resource", post(handlers::mark_no_resource))
.route("/parse", post(handlers::parse_paper))
.route("/translate", post(handlers::translate_paper))
.route("/citations", get(handlers::get_citation_network))
@@ -121,7 +124,8 @@ async fn main() -> anyhow::Result<()> {
.route("/sync/asset/stop", post(handlers::stop_asset_sync))
.route("/sync/asset/status", get(handlers::get_asset_sync_status))
.route("/sync/queries", get(handlers::get_sync_queries))
.route("/sync/queries/:id", axum::routing::delete(handlers::delete_sync_query));
.route("/sync/queries/:id", axum::routing::delete(handlers::delete_sync_query))
.route("/active_bibcode", get(handlers::get_active_bibcode).post(handlers::set_active_bibcode));
// 静态文件资源代理托管(当前端打包至 dashboard/dist 后,直接挂载到主域名根路由)
let serve_dir = ServeDir::new("dashboard/dist")
@@ -1,4 +1,4 @@
// src/services/batch_sync.rs
// src/services/batch/asset.rs
use std::sync::Arc;
use std::fs;
use tokio::sync::Mutex;
@@ -7,284 +7,15 @@ use tracing::{info, warn, error};
use sqlx::{SqlitePool, Row};
use crate::Config;
use crate::clients::ads::AdsClient;
use crate::clients::arxiv::ArxivClient;
use crate::clients::qiniu::QiniuClient;
use crate::services::download::Downloader;
use crate::api::handlers::{convert_ads_doc_to_standard, convert_arxiv_to_standard, save_paper_to_db};
// 批量收割进度状态
#[derive(Debug, Clone, Serialize)]
pub struct MetaSyncStatus {
pub active: bool,
pub query: String,
pub source: String,
pub synced: i32,
pub total: i32,
}
impl MetaSyncStatus {
pub fn new() -> Self {
MetaSyncStatus {
active: false,
query: String::new(),
source: String::new(),
synced: 0,
total: 0,
}
}
}
pub struct MetaSync;
impl MetaSync {
// 预估文献总量
pub async fn get_total_count(
query: &str,
source: &str,
ads: &AdsClient,
arxiv: &ArxivClient,
) -> anyhow::Result<i32> {
let mut total = 0;
if source == "all" || source == "ads" {
match ads.get_total_count(query).await {
Ok(count) => {
total += count;
info!("ADS 预估文献总量: {} 篇", count);
}
Err(e) => {
warn!("获取 ADS 预估总量失败: {}", e);
}
}
}
if source == "all" || source == "arxiv" {
match arxiv.get_total_count(query).await {
Ok(count) => {
total += count;
info!("arXiv 预估文献总量: {} 篇", count);
}
Err(e) => {
warn!("获取 arXiv 预估总量失败: {}", e);
}
}
}
Ok(total)
}
// 启动后台收割异步任务
pub fn start_harvest(
db: SqlitePool,
ads: Arc<AdsClient>,
arxiv: Arc<ArxivClient>,
query: String,
source: String,
limit: i32,
status: Arc<Mutex<MetaSyncStatus>>,
) {
let query_clone = query.clone();
let source_clone = source.clone();
tokio::spawn(async move {
info!("启动后台批量收割任务: 查询词='{}', 源='{}', 上限={}", query_clone, source_clone, limit);
// 自动将检索配置存入/更新至 sync_queries 数据库表中进行去重和时间更新
let _ = sqlx::query(
"INSERT INTO sync_queries (query, source, limit_count, last_run) \
VALUES (?, ?, ?, CURRENT_TIMESTAMP) \
ON CONFLICT(query, source, limit_count) DO UPDATE SET last_run=excluded.last_run"
)
.bind(&query_clone)
.bind(&source_clone)
.bind(limit)
.execute(&db)
.await;
// 1. 并行获取两端预估总量
let ads_count_fut = {
let ads = ads.clone();
let query = query_clone.clone();
let is_active = source_clone == "all" || source_clone == "ads";
async move {
if is_active {
ads.get_total_count(&query).await.unwrap_or(0)
} else {
0
}
}
};
let arxiv_count_fut = {
let arxiv = arxiv.clone();
let query = query_clone.clone();
let is_active = source_clone == "all" || source_clone == "arxiv";
async move {
if is_active {
arxiv.get_total_count(&query).await.unwrap_or(0)
} else {
0
}
}
};
let (ads_total, arxiv_total) = tokio::join!(ads_count_fut, arxiv_count_fut);
let total_count = ads_total + arxiv_total;
{
let mut s = status.lock().await;
s.active = true;
s.query = query_clone.clone();
s.source = source_clone.clone();
s.synced = 0;
s.total = total_count;
}
// 计算实际需要收割的总上限,并按比例分配或根据实际匹配量上限控制
let limit_to_harvest = if limit > 0 { std::cmp::min(limit, total_count) } else { total_count };
// 共享的 atomic 计数器,以便两端并行同步时独立累加进度
let synced_counter = Arc::new(std::sync::atomic::AtomicI32::new(0));
// 2. 执行并行的同步子任务
let ads_sync_fut = {
let db = db.clone();
let ads = ads.clone();
let query = query_clone.clone();
let synced_counter = synced_counter.clone();
let status = status.clone();
let is_active = source_clone == "all" || source_clone == "ads";
// 如果是 all 模式,各平台按比例分摊 limit 额度,或者直接限制自身的最大可用量
let ads_limit = if source_clone == "all" {
if ads_total == 0 { 0 } else {
let ratio = ads_total as f32 / total_count as f32;
((limit_to_harvest as f32) * ratio).round() as i32
}
} else {
limit_to_harvest
};
async move {
if !is_active || ads_limit <= 0 {
return;
}
let mut local_synced = 0;
let mut start_offset = 0;
while local_synced < ads_limit {
let chunk_size = std::cmp::min(2000, ads_limit - local_synced);
if chunk_size <= 0 {
break;
}
info!("正在同步 ADS 分批数据: start={}, rows={}", start_offset, chunk_size);
match ads.search(&query, start_offset, chunk_size, "relevance").await {
Ok(docs) => {
if docs.is_empty() {
break;
}
let count = docs.len() as i32;
for doc in docs {
let paper = convert_ads_doc_to_standard(&doc);
let _ = save_paper_to_db(&db, &paper).await;
}
local_synced += count;
start_offset += count;
// 累加全局进度并更新状态
let current_global = synced_counter.fetch_add(count, std::sync::atomic::Ordering::SeqCst) + count;
{
let mut s = status.lock().await;
s.synced = current_global;
}
}
Err(e) => {
error!("批量同步 ADS 数据出错: {}", e);
break;
}
}
}
}
};
let arxiv_sync_fut = {
let db = db.clone();
let arxiv = arxiv.clone();
let query = query_clone.clone();
let synced_counter = synced_counter.clone();
let status = status.clone();
let is_active = source_clone == "all" || source_clone == "arxiv";
let arxiv_limit = if source_clone == "all" {
if arxiv_total == 0 { 0 } else {
let ratio = arxiv_total as f32 / total_count as f32;
((limit_to_harvest as f32) * ratio).round() as i32
}
} else {
limit_to_harvest
};
async move {
if !is_active || arxiv_limit <= 0 {
return;
}
let mut local_synced = 0;
let mut start_offset = 0;
while local_synced < arxiv_limit {
let chunk_size = std::cmp::min(2000, arxiv_limit - local_synced);
if chunk_size <= 0 {
break;
}
info!("正在同步 arXiv 分批数据: start={}, max_results={}", start_offset, chunk_size);
match arxiv.search(&query, start_offset, chunk_size, "relevance").await {
Ok(papers) => {
if papers.is_empty() {
break;
}
let count = papers.len() as i32;
for p in papers {
let paper = convert_arxiv_to_standard(&p);
let _ = save_paper_to_db(&db, &paper).await;
}
local_synced += count;
start_offset += count;
// 累加全局进度并更新状态
let current_global = synced_counter.fetch_add(count, std::sync::atomic::Ordering::SeqCst) + count;
{
let mut s = status.lock().await;
s.synced = current_global;
}
}
Err(e) => {
error!("批量同步 arXiv 数据出错: {}", e);
break;
}
}
// 遵循 arXiv API 3 秒间隔要求
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
}
}
};
// 使用 tokio::join! 并行驱动两端同步任务
tokio::join!(ads_sync_fut, arxiv_sync_fut);
// 4. 收尾并重置状态
let final_synced = synced_counter.load(std::sync::atomic::Ordering::SeqCst);
{
let mut s = status.lock().await;
s.active = false;
s.synced = final_synced;
info!("后台批量收割任务已结束。共成功同步 {} 篇文献。", final_synced);
}
});
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum SyncAction {
Download,
Parse,
Translate,
All,
}
@@ -335,6 +66,7 @@ impl AssetSync {
config: Config,
downloader: Arc<Downloader>,
qiniu: Arc<QiniuClient>,
dict: Arc<crate::services::translation::Dictionary>,
action: SyncAction,
bibcodes: Vec<String>,
status: Arc<Mutex<AssetSyncStatus>>,
@@ -356,6 +88,7 @@ impl AssetSync {
let action_desc = match action {
SyncAction::Download => "下载",
SyncAction::Parse => "解析",
SyncAction::Translate => "翻译",
SyncAction::All => "下载与解析",
};
s.add_log(format!("批量{}任务启动,共 {} 篇文献需处理。", action_desc, total));
@@ -383,13 +116,13 @@ impl AssetSync {
// 1. 获取文献元数据与当前路径状态
let paper_res = sqlx::query(
"SELECT arxiv_id, doi, pdf_path, html_path, markdown_path, doctype FROM papers WHERE bibcode = ?"
"SELECT arxiv_id, doi, pdf_path, html_path, markdown_path, doctype, translation_path FROM papers WHERE bibcode = ?"
)
.bind(&bibcode)
.fetch_optional(&db)
.await;
let (arxiv_id, doi, mut pdf_path, mut html_path, markdown_path, doctype) = match paper_res {
let (arxiv_id, doi, mut pdf_path, mut html_path, markdown_path, doctype, translation_path) = match paper_res {
Ok(Some(row)) => {
let arxiv_id: String = row.get(0);
let doi: String = row.get(1);
@@ -397,7 +130,8 @@ impl AssetSync {
let html_path: Option<String> = row.get(3);
let markdown_path: Option<String> = row.get(4);
let doctype: Option<String> = row.get(5);
(arxiv_id, doi, pdf_path, html_path, markdown_path, doctype)
let translation_path: Option<String> = row.get(6);
(arxiv_id, doi, pdf_path, html_path, markdown_path, doctype, translation_path)
}
_ => {
let mut s = status.lock().await;
@@ -406,9 +140,17 @@ impl AssetSync {
}
};
// 1b. 检查 doctype,如果是 proposal, abstract, catalog, software 等无数字全文的文件,直接跳过处理
// 1b. 检查 doctype,如果是 proposal, abstract, catalog, dataset, software, circular 等无数字全文的文件,直接跳过处理
let doctype_str = doctype.unwrap_or_else(|| "article".to_string()).to_lowercase();
if doctype_str == "proposal" || doctype_str == "abstract" || doctype_str == "catalog" || doctype_str == "software" {
if doctype_str == "proposal"
|| doctype_str == "abstract"
|| doctype_str == "catalog"
|| doctype_str == "dataset"
|| doctype_str == "software"
|| doctype_str == "circular"
|| doctype_str == "newsletter"
|| doctype_str == "obituary"
{
let mut s = status.lock().await;
s.add_log(format!("文献 {} 的类型为 {} (无数字版全文),跳过下载与解析。", bibcode, doctype_str));
// 同样更新处理进度,防止任务进度条卡住
@@ -434,16 +176,22 @@ impl AssetSync {
s.add_log(format!("文献 {} 本地无 PDF/HTML,开始下载...", bibcode));
}
let (downloaded_pdf, downloaded_html) = if !arxiv_id.is_empty() {
let (pdf_res, html_res) = if !arxiv_id.is_empty() {
downloader.download_arxiv_direct(&arxiv_id, &config.library_dir).await
} else {
let doi_opt = if !doi.is_empty() { Some(doi.as_str()) } else { None };
downloader.download_paper(&bibcode, doi_opt, &config.library_dir).await
};
if downloaded_pdf.is_some() || downloaded_html.is_some() {
let pdf_rel = downloaded_pdf.map(|p| p.strip_prefix(&config.library_dir).unwrap_or(&p).to_string_lossy().to_string());
let html_rel = downloaded_html.map(|p| p.strip_prefix(&config.library_dir).unwrap_or(&p).to_string_lossy().to_string());
if pdf_res.is_ok() || html_res.is_ok() {
let pdf_rel = match pdf_res {
Ok(p) => Some(p.strip_prefix(&config.library_dir).unwrap_or(&p).to_string_lossy().to_string()),
Err(e) => Some(format!("error: {}", e)),
};
let html_rel = match html_res {
Ok(p) => Some(p.strip_prefix(&config.library_dir).unwrap_or(&p).to_string_lossy().to_string()),
Err(e) => Some(format!("error: {}", e)),
};
// 更新路径变量与数据库
pdf_path = pdf_rel.clone();
@@ -466,7 +214,24 @@ impl AssetSync {
dl_failed_count += 1;
let mut s = status.lock().await;
s.download_failed = dl_failed_count;
s.add_log(format!("文献 {} 下载失败(PDF 和 HTML 均下载失败)", bibcode));
let pdf_err = match pdf_res {
Err(e) => format!("error: {}", e),
_ => "error: 未知错误".to_string(),
};
let html_err = match html_res {
Err(e) => format!("error: {}", e),
_ => "error: 未知错误".to_string(),
};
s.add_log(format!("文献 {} 下载失败。PDF: {}, HTML: {}", bibcode, pdf_err, html_err));
let _ = sqlx::query("UPDATE papers SET pdf_path = ?, html_path = ? WHERE bibcode = ?")
.bind(&pdf_err)
.bind(&html_err)
.bind(&bibcode)
.execute(&db)
.await;
}
// 每次下载尝试后,加入 3-5 秒随机延迟,防爬防封
@@ -723,6 +488,117 @@ impl AssetSync {
s.parsed += 1;
}
}
// 4. 检查并执行翻译
if action == SyncAction::Translate {
let is_tr_exist = translation_path.as_ref().map(|p| config.library_dir.join(p).exists() && !p.starts_with("error:")).unwrap_or(false);
if !is_tr_exist {
if let Some(md_rel) = &markdown_path {
if !md_rel.starts_with("error:") {
let md_abs = config.library_dir.join(md_rel);
if md_abs.exists() {
{
let mut s = status.lock().await;
s.add_log(format!("文献 {} 开始调用 LLM 翻译...", bibcode));
}
match fs::read_to_string(&md_abs) {
Ok(english_markdown) => {
match crate::services::translation::translate_markdown(&english_markdown, &dict, &config).await {
Ok(translated_markdown) => {
let tr_filename = format!("{}_zh.md", bibcode);
let tr_dest = config.library_dir.join("Translation").join(&tr_filename);
let _ = fs::create_dir_all(tr_dest.parent().unwrap());
if fs::write(&tr_dest, &translated_markdown).is_ok() {
let relative_tr_path = format!("Translation/{}", tr_filename);
let _ = sqlx::query("UPDATE papers SET translation_path = ? WHERE bibcode = ?")
.bind(&relative_tr_path)
.bind(&bibcode)
.execute(&db)
.await;
let mut s = status.lock().await;
s.parsed += 1;
s.add_log(format!("文献 {} 翻译成功!", bibcode));
} else {
let error_msg = "error: 写入翻译文件失败";
let _ = sqlx::query("UPDATE papers SET translation_path = ? WHERE bibcode = ?")
.bind(error_msg)
.bind(&bibcode)
.execute(&db)
.await;
let mut s = status.lock().await;
s.parse_failed += 1;
s.add_log(format!("文献 {} 翻译文件写入失败。", bibcode));
}
}
Err(e) => {
let error_msg = format!("error: {}", e);
let _ = sqlx::query("UPDATE papers SET translation_path = ? WHERE bibcode = ?")
.bind(&error_msg)
.bind(&bibcode)
.execute(&db)
.await;
let mut s = status.lock().await;
s.parse_failed += 1;
s.add_log(format!("文献 {} 翻译失败: {}", bibcode, e));
}
}
}
Err(e) => {
let error_msg = format!("error: 读取英文 Markdown 失败: {}", e);
let _ = sqlx::query("UPDATE papers SET translation_path = ? WHERE bibcode = ?")
.bind(&error_msg)
.bind(&bibcode)
.execute(&db)
.await;
let mut s = status.lock().await;
s.parse_failed += 1;
s.add_log(format!("文献 {} 读取英文 Markdown 失败: {}", bibcode, e));
}
}
} else {
let error_msg = "error: 英文 Markdown 文件不存在";
let _ = sqlx::query("UPDATE papers SET translation_path = ? WHERE bibcode = ?")
.bind(error_msg)
.bind(&bibcode)
.execute(&db)
.await;
let mut s = status.lock().await;
s.parse_failed += 1;
s.add_log(format!("文献 {} 英文 Markdown 文件不存在,无法翻译。", bibcode));
}
} else {
let error_msg = "error: 英文 Markdown 文件处于解析失败状态";
let _ = sqlx::query("UPDATE papers SET translation_path = ? WHERE bibcode = ?")
.bind(error_msg)
.bind(&bibcode)
.execute(&db)
.await;
let mut s = status.lock().await;
s.parse_failed += 1;
s.add_log(format!("文献 {} 英文 Markdown 解析失败,跳过翻译。", bibcode));
}
} else {
let error_msg = "error: 尚未解析英文 Markdown 路径为 NULL";
let _ = sqlx::query("UPDATE papers SET translation_path = ? WHERE bibcode = ?")
.bind(error_msg)
.bind(&bibcode)
.execute(&db)
.await;
let mut s = status.lock().await;
s.parse_failed += 1;
s.add_log(format!("文献 {} 尚未解析英文 Markdown,跳过翻译。", bibcode));
}
} else {
{
let mut s = status.lock().await;
s.add_log(format!("文献 {} 已存在翻译,跳过。", bibcode));
}
let mut s = status.lock().await;
s.parsed += 1;
}
}
}
if !join_handles.is_empty() {
@@ -741,6 +617,7 @@ impl AssetSync {
let action_desc = match action {
SyncAction::Download => "下载",
SyncAction::Parse => "解析",
SyncAction::Translate => "翻译",
SyncAction::All => "下载与解析",
};
s.add_log(format!("批量{}任务顺利完成!", action_desc));
@@ -829,11 +706,13 @@ mod tests {
let qiniu = Arc::new(QiniuClient::new("test_access".to_string(), "test_secret".to_string(), "test_bucket".to_string(), "test_domain".to_string()));
let status = Arc::new(Mutex::new(AssetSyncStatus::new()));
let dict = Arc::new(crate::services::translation::Dictionary::new());
AssetSync::start_process(
pool.clone(),
config,
downloader,
qiniu,
dict,
SyncAction::All,
vec![bibcode.clone()],
status.clone(),
@@ -947,11 +826,13 @@ mod tests {
let qiniu = Arc::new(QiniuClient::new("test_access".to_string(), "test_secret".to_string(), "test_bucket".to_string(), "test_domain".to_string()));
let status = Arc::new(Mutex::new(AssetSyncStatus::new()));
let dict = Arc::new(crate::services::translation::Dictionary::new());
AssetSync::start_process(
pool.clone(),
config,
downloader,
qiniu,
dict,
SyncAction::All,
vec![bib1.clone(), bib2.clone()],
status.clone(),
+273
View File
@@ -0,0 +1,273 @@
// src/services/batch/meta.rs
use std::sync::Arc;
use tokio::sync::Mutex;
use serde::Serialize;
use tracing::{info, warn, error};
use sqlx::SqlitePool;
use crate::clients::ads::AdsClient;
use crate::clients::arxiv::ArxivClient;
use crate::api::handlers::{convert_ads_doc_to_standard, convert_arxiv_to_standard, save_paper_to_db};
// 批量元数据同步进度状态
#[derive(Debug, Clone, Serialize)]
pub struct MetaSyncStatus {
pub active: bool,
pub query: String,
pub source: String,
pub synced: i32,
pub total: i32,
}
impl MetaSyncStatus {
pub fn new() -> Self {
MetaSyncStatus {
active: false,
query: String::new(),
source: String::new(),
synced: 0,
total: 0,
}
}
}
pub struct MetaSync;
impl MetaSync {
// 预估文献总量
pub async fn get_total_count(
query: &str,
source: &str,
ads: &AdsClient,
arxiv: &ArxivClient,
) -> anyhow::Result<i32> {
let mut total = 0;
if source == "all" || source == "ads" {
match ads.get_total_count(query).await {
Ok(count) => {
total += count;
info!("ADS 预估文献总量: {} 篇", count);
}
Err(e) => {
warn!("获取 ADS 预估总量失败: {}", e);
}
}
}
if source == "all" || source == "arxiv" {
match arxiv.get_total_count(query).await {
Ok(count) => {
total += count;
info!("arXiv 预估文献总量: {} 篇", count);
}
Err(e) => {
warn!("获取 arXiv 预估总量失败: {}", e);
}
}
}
Ok(total)
}
// 启动后台元数据同步异步任务
pub fn start_harvest(
db: SqlitePool,
ads: Arc<AdsClient>,
arxiv: Arc<ArxivClient>,
query: String,
source: String,
limit: i32,
status: Arc<Mutex<MetaSyncStatus>>,
) {
let query_clone = query.clone();
let source_clone = source.clone();
tokio::spawn(async move {
info!("启动后台批量元数据同步任务: 查询词='{}', 源='{}', 上限={}", query_clone, source_clone, limit);
// 自动将检索配置存入/更新至 sync_queries 数据库表中进行去重和时间更新
let _ = sqlx::query(
"INSERT INTO sync_queries (query, source, limit_count, last_run) \
VALUES (?, ?, ?, CURRENT_TIMESTAMP) \
ON CONFLICT(query, source, limit_count) DO UPDATE SET last_run=excluded.last_run"
)
.bind(&query_clone)
.bind(&source_clone)
.bind(limit)
.execute(&db)
.await;
// 1. 并行获取两端预估总量
let ads_count_fut = {
let ads = ads.clone();
let query = query_clone.clone();
let is_active = source_clone == "all" || source_clone == "ads";
async move {
if is_active {
ads.get_total_count(&query).await.unwrap_or(0)
} else {
0
}
}
};
let arxiv_count_fut = {
let arxiv = arxiv.clone();
let query = query_clone.clone();
let is_active = source_clone == "all" || source_clone == "arxiv";
async move {
if is_active {
arxiv.get_total_count(&query).await.unwrap_or(0)
} else {
0
}
}
};
let (ads_total, arxiv_total) = tokio::join!(ads_count_fut, arxiv_count_fut);
let total_count = ads_total + arxiv_total;
{
let mut s = status.lock().await;
s.total = total_count;
}
// 计算实际需要元数据同步的总上限,并按比例分配或根据实际匹配量上限控制
let limit_to_harvest = if limit > 0 { std::cmp::min(limit, total_count) } else { total_count };
// 共享的 atomic 计数器,以便两端并行同步时独立累加进度
let synced_counter = Arc::new(std::sync::atomic::AtomicI32::new(0));
// 2. 执行并行的同步子任务
let ads_sync_fut = {
let db = db.clone();
let ads = ads.clone();
let query = query_clone.clone();
let synced_counter = synced_counter.clone();
let status = status.clone();
let is_active = source_clone == "all" || source_clone == "ads";
// 如果是 all 模式,各平台按比例分摊 limit 额度,或者直接限制自身的最大可用量
let ads_limit = if source_clone == "all" {
if ads_total == 0 { 0 } else {
let ratio = ads_total as f32 / total_count as f32;
((limit_to_harvest as f32) * ratio).round() as i32
}
} else {
limit_to_harvest
};
async move {
if !is_active || ads_limit <= 0 {
return;
}
let mut local_synced = 0;
let mut start_offset = 0;
while local_synced < ads_limit {
let chunk_size = std::cmp::min(2000, ads_limit - local_synced);
if chunk_size <= 0 {
break;
}
info!("正在同步 ADS 分批数据: start={}, rows={}", start_offset, chunk_size);
match ads.search(&query, start_offset, chunk_size, "relevance").await {
Ok(docs) => {
if docs.is_empty() {
break;
}
let count = docs.len() as i32;
for doc in docs {
let paper = convert_ads_doc_to_standard(&doc);
let _ = save_paper_to_db(&db, &paper).await;
}
local_synced += count;
start_offset += count;
// 累加全局进度并更新状态
let current_global = synced_counter.fetch_add(count, std::sync::atomic::Ordering::SeqCst) + count;
{
let mut s = status.lock().await;
s.synced = current_global;
}
}
Err(e) => {
error!("批量同步 ADS 数据出错: {}", e);
break;
}
}
}
}
};
let arxiv_sync_fut = {
let db = db.clone();
let arxiv = arxiv.clone();
let query = query_clone.clone();
let synced_counter = synced_counter.clone();
let status = status.clone();
let is_active = source_clone == "all" || source_clone == "arxiv";
let arxiv_limit = if source_clone == "all" {
if arxiv_total == 0 { 0 } else {
let ratio = arxiv_total as f32 / total_count as f32;
((limit_to_harvest as f32) * ratio).round() as i32
}
} else {
limit_to_harvest
};
async move {
if !is_active || arxiv_limit <= 0 {
return;
}
let mut local_synced = 0;
let mut start_offset = 0;
while local_synced < arxiv_limit {
let chunk_size = std::cmp::min(2000, arxiv_limit - local_synced);
if chunk_size <= 0 {
break;
}
info!("正在同步 arXiv 分批数据: start={}, max_results={}", start_offset, chunk_size);
match arxiv.search(&query, start_offset, chunk_size, "relevance").await {
Ok(papers) => {
if papers.is_empty() {
break;
}
let count = papers.len() as i32;
for p in papers {
let paper = convert_arxiv_to_standard(&p);
let _ = save_paper_to_db(&db, &paper).await;
}
local_synced += count;
start_offset += count;
// 累加全局进度并更新状态
let current_global = synced_counter.fetch_add(count, std::sync::atomic::Ordering::SeqCst) + count;
{
let mut s = status.lock().await;
s.synced = current_global;
}
}
Err(e) => {
error!("批量同步 arXiv 数据出错: {}", e);
break;
}
}
// 遵循 arXiv API 3 秒间隔要求
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
}
}
};
// 使用 tokio::join! 并行驱动两端同步任务
tokio::join!(ads_sync_fut, arxiv_sync_fut);
// 4. 收尾并重置状态
let final_synced = synced_counter.load(std::sync::atomic::Ordering::SeqCst);
{
let mut s = status.lock().await;
s.active = false;
s.synced = final_synced;
info!("后台批量元数据同步任务已结束。共成功同步 {} 篇文献。", final_synced);
}
});
}
}
+6
View File
@@ -0,0 +1,6 @@
// src/services/batch/mod.rs
pub mod meta;
pub mod asset;
pub use meta::{MetaSyncStatus, MetaSync};
pub use asset::{SyncAction, AssetSyncStatus, AssetSync};
+398 -126
View File
@@ -90,10 +90,11 @@ fn detect_anti_bot(content: &str, url: Option<&str>) -> Result<()> {
"checking your browser", "please wait while we verify",
"cf-browser-verification", "cf_chl_opt", "just a moment",
"enable javascript and cookies", "_cf_chl_tk",
"awswafintegration", "aws waf",
];
for p in &cf_patterns {
if lower.contains(p) {
anyhow::bail!("检测到 Cloudflare 挑战页面(特征: {}", p);
anyhow::bail!("检测到 Cloudflare 或 AWS WAF 挑战页面(特征: {}", p);
}
}
@@ -125,7 +126,7 @@ fn detect_anti_bot(content: &str, url: Option<&str>) -> Result<()> {
}
/// 校验响应字节是否为有效 PDF(魔数 + 最小大小 + EOF 标记)
fn validate_pdf_content(bytes: &[u8]) -> Result<()> {
pub(crate) fn validate_pdf_content(bytes: &[u8]) -> Result<()> {
if !bytes.starts_with(b"%PDF") {
if bytes.starts_with(b"<!") || bytes.starts_with(b"<html") || bytes.starts_with(b"<HTML") {
let text = String::from_utf8_lossy(&bytes[..bytes.len().min(2048)]);
@@ -145,18 +146,98 @@ fn validate_pdf_content(bytes: &[u8]) -> Result<()> {
Ok(())
}
/// 校验 HTML 内容是否为有效文献页(非错误/登录墙)
fn validate_html_content(text: &str) -> Result<()> {
/// 校验 HTML 内容是否为有效文献页(非错误/登录墙/跳转/摘要占位页
pub(crate) fn validate_html_content(text: &str) -> Result<()> {
detect_anti_bot(text, None)?;
let lower = text.to_lowercase();
// 1. 检查常见的跳转与错误占位特征
if lower.contains("redirecting") || lower.contains("redirect to") || lower.contains("http-equiv=\"refresh\"") || lower.contains("autoredirecttourl") {
anyhow::bail!("检测到 HTML 重定向跳转页面,而非真实文献正文");
}
if lower.contains("conversion to html had a fatal error") || lower.contains("no content available") || lower.contains("fatal error and exited abruptly") {
anyhow::bail!("检测到 ar5iv 转换失败的占位 HTML 页面");
}
if lower.contains("see pages 1-last of") {
anyhow::bail!("检测到仅包含 PDF 链接的占位 HTML 页面");
}
// 2. 网页标题精准黑名单校验(防止正文中提及 NSF/VizieR 导致误伤)
if let Some(start_pos) = lower.find("<title") {
if let Some(tag_end) = lower[start_pos..].find('>') {
let title_start = start_pos + tag_end + 1;
if let Some(end_pos) = lower[title_start..].find("</title>") {
let title = &lower[title_start..title_start + end_pos];
if title.contains("nsf award search")
|| title.contains("national science foundation")
|| title.contains("vizier")
|| title.contains("caltechthesis")
|| title.contains("caosp abstract")
|| title.contains("asp conference series")
|| title.contains("aspbooks")
{
anyhow::bail!("检测到占位网页标题: \"{}\",判定为非正本文献", title.trim());
}
}
}
}
// 3. 基础字节长度与具体 HTTP 错误特征校验
if text.len() < 2000 {
let lower = text.to_lowercase();
for kw in &["error", "404", "not found", "forbidden", "access denied"] {
let error_patterns = [
"404 not found", "403 forbidden", "502 bad gateway",
"500 internal server error", "access denied", "site error"
];
for kw in &error_patterns {
if lower.contains(kw) {
anyhow::bail!("响应是错误页面(包含: {}", kw);
}
}
warn!("HTML 内容较短({} 字节),可能不完整", text.len());
}
// 4. 结构启发式校验:如果是小于 50KB 的 HTML,必须包含基本的章节或参考文献结构,否则判定为摘要/存根占位页
if text.len() < 50000 {
// 匹配 heading 标签或 Markdown 格式的标题,而不是纯文本中的单词
let has_sections = lower.contains("ltx_title_section")
|| lower.contains("class=\"section\"")
|| lower.contains("## introduction")
|| lower.contains("<h2>introduction")
|| lower.contains("<h3>introduction")
|| lower.contains("class=\"ltx_section\"");
let has_bib = lower.contains("ltx_bibliography")
|| lower.contains("class=\"references\"")
|| lower.contains("<ol class=\"references\"")
|| lower.contains("<ul class=\"references\"")
|| lower.contains("id=\"bib\"")
|| lower.contains("class=\"ltx_bibliography\"");
if !has_sections && !has_bib {
anyhow::bail!("HTML 长度偏小({} 字节)且缺少正文章节或参考文献,判定为非正本文献", text.len());
}
}
Ok(())
}
/// 宽松版 HTML 校验,专用于手动上传场景。
/// 用户在浏览器中亲眼确认了文献内容,无需自动下载时的反爬虫/章节启发式检测。
/// 仅做最低限度检查:页面不能过小,不能是纯跳转页。
pub(crate) fn validate_html_content_lenient(text: &str) -> Result<()> {
if text.len() < 500 {
anyhow::bail!("上传的 HTML 文件过小({} 字节),可能是空白或错误页面", text.len());
}
let lower = text.to_lowercase();
// 仅拒绝明确的重定向占位页(通常 body 极短且没有正文)
let is_redirect = lower.contains("http-equiv=\"refresh\"")
|| lower.contains("autoredirecttourl");
if is_redirect && text.len() < 5000 {
anyhow::bail!("检测到 HTML 重定向跳转页面,而非真实文献正文");
}
Ok(())
}
@@ -191,6 +272,45 @@ impl Downloader {
Downloader { client }
}
/// 使用 Obscura 作为后备通道进行下载
async fn download_via_obscura(&self, url: &str, dest_path: &Path, is_pdf: bool) -> Result<()> {
info!("[Obscura 后备通道] 启动下载: {}", url);
if let Some(parent) = dest_path.parent() {
std::fs::create_dir_all(parent)?;
}
let mut cmd = tokio::process::Command::new("bin/obscura");
cmd.arg("fetch").arg(url).arg("--stealth");
if is_pdf {
cmd.arg("--dump").arg("original");
} else {
cmd.arg("--dump").arg("html");
}
cmd.arg("--output").arg(dest_path);
let status = cmd.status().await
.context("启动 Obscura 进程失败,请检查 bin/obscura 是否存在且有执行权限")?;
if !status.success() {
anyhow::bail!("Obscura 进程退出状态非成功: {:?}", status);
}
// 校验下载得到的文件
if is_pdf {
let bytes = tokio::fs::read(dest_path).await?;
validate_pdf_content(&bytes)?;
} else {
let text = tokio::fs::read_to_string(dest_path).await?;
validate_html_content(&text)?;
}
info!("[Obscura 后备通道] 下载并校验成功: {:?}", dest_path);
Ok(())
}
// ─── 辅助工具 ──────────────────────────────────────────────
/// 请求前随机延迟 500-2000ms(模拟人类浏览间隔,降低反爬触发)
@@ -273,38 +393,60 @@ impl Downloader {
let main_url = format!("https://iopscience.iop.org/article/{}", doi);
let pdf_url = format!("https://iopscience.iop.org/article/{}/pdf", doi);
// 步骤 1:访问文章主页,建立 Cookie 会话
debug!("[IOP] 预热主页: {}", main_url);
Self::maybe_delay().await;
match self.client.get(&main_url)
.headers(build_chrome_headers(None))
.send().await
{
Ok(r) => debug!("[IOP] 主页响应: {}", r.status()),
Err(e) => warn!("[IOP] 主页访问失败(继续尝试): {:?}", e),
let res = async {
// 步骤 1:访问文章主页,建立 Cookie 会话
debug!("[IOP] 预热主页: {}", main_url);
Self::maybe_delay().await;
match self.client.get(&main_url)
.headers(build_chrome_headers(None))
.send().await
{
Ok(r) => debug!("[IOP] 主页响应: {}", r.status()),
Err(e) => warn!("[IOP] 主页访问失败(继续尝试): {:?}", e),
}
// 步骤 2:携带 Referer 下载 PDF
debug!("[IOP] 下载 PDF: {}", pdf_url);
Self::maybe_delay().await;
let response = self.client.get(&pdf_url)
.headers(build_chrome_headers(Some(&main_url)))
.send().await
.context("IOP PDF 请求失败")?;
let status = response.status();
if !status.is_success() {
anyhow::bail!("[IOP] 返回 HTTP {}", status);
}
self.stream_download(response, dest_path).await?;
// 步骤 3:校验下载内容
let bytes = tokio::fs::read(dest_path).await?;
validate_pdf_content(&bytes)?;
Ok(())
}.await;
match res {
Ok(()) => {
info!("[IOP] PDF 下载成功: {:?}", dest_path);
Ok(())
}
Err(e) => {
let err_msg = e.to_string();
if err_msg.contains("人机验证")
|| err_msg.contains("挑战页面")
|| err_msg.contains("WAF")
|| err_msg.contains("Cloudflare")
|| err_msg.contains("HTTP 403")
|| err_msg.contains("HTTP 503")
{
warn!("[IOP] 下载触发人机验证或拦截: {}。尝试使用 Obscura 后备通道...", err_msg);
self.download_via_obscura(&pdf_url, dest_path, true).await
} else {
Err(e)
}
}
}
// 步骤 2:携带 Referer 下载 PDF
debug!("[IOP] 下载 PDF: {}", pdf_url);
Self::maybe_delay().await;
let response = self.client.get(&pdf_url)
.headers(build_chrome_headers(Some(&main_url)))
.send().await
.context("IOP PDF 请求失败")?;
let status = response.status();
if !status.is_success() {
anyhow::bail!("[IOP] 返回 HTTP {}", status);
}
self.stream_download(response, dest_path).await?;
// 步骤 3:校验下载内容
let bytes = tokio::fs::read(dest_path).await?;
validate_pdf_content(&bytes)?;
info!("[IOP] PDF 下载成功: {:?}", dest_path);
Ok(())
}
/// Springer/Nature HTML 下载(含会话预热)
@@ -312,74 +454,140 @@ impl Downloader {
let url = format!("https://link.springer.com/article/{}", doi);
info!("[Springer] 下载 HTML: {}", url);
Self::maybe_delay().await;
let response = self.client.get(&url)
.headers(build_browser_headers())
.send().await
.context("Springer HTML 请求失败")?;
let res = async {
Self::maybe_delay().await;
let response = self.client.get(&url)
.headers(build_browser_headers())
.send().await
.context("Springer HTML 请求失败")?;
let status = response.status();
if !status.is_success() {
anyhow::bail!("[Springer] 返回 HTTP {}", status);
let status = response.status();
if !status.is_success() {
anyhow::bail!("[Springer] 返回 HTTP {}", status);
}
self.stream_download(response, dest_path).await?;
let text = tokio::fs::read_to_string(dest_path).await
.context("读取 HTML 文件失败")?;
validate_html_content(&text)?;
Ok(())
}.await;
match res {
Ok(()) => {
info!("[Springer] HTML 下载成功: {:?}", dest_path);
Ok(())
}
Err(e) => {
let err_msg = e.to_string();
if err_msg.contains("人机验证")
|| err_msg.contains("挑战页面")
|| err_msg.contains("WAF")
|| err_msg.contains("Cloudflare")
|| err_msg.contains("HTTP 403")
|| err_msg.contains("HTTP 503")
{
warn!("[Springer] 下载触发人机验证或拦截: {}。尝试使用 Obscura 后备通道...", err_msg);
self.download_via_obscura(&url, dest_path, false).await
} else {
Err(e)
}
}
}
self.stream_download(response, dest_path).await?;
let sniff = Self::read_file_header(dest_path).await?;
let text = String::from_utf8_lossy(&sniff);
validate_html_content(&text)?;
info!("[Springer] HTML 下载成功: {:?}", dest_path);
Ok(())
}
/// 通用 PDF 直链下载(带随机延迟 + 内容校验)
/// 通用 PDF 直链下载(带随机延迟 + 内容校验 + Obscura 后备
async fn download_pdf_direct(&self, url: &str, dest_path: &Path, label: &str) -> Result<()> {
info!("[{}] 下载 PDF: {}", label, url);
Self::maybe_delay().await;
let response = self.client.get(url)
.headers(build_browser_headers())
.send().await
.context(format!("[{}] PDF 请求失败", label))?;
let res = async {
let response = self.client.get(url)
.headers(build_browser_headers())
.send().await
.context(format!("[{}] PDF 请求失败", label))?;
let status = response.status();
if !status.is_success() {
anyhow::bail!("[{}] 返回 HTTP {}", label, status);
let status = response.status();
if !status.is_success() {
anyhow::bail!("[{}] 返回 HTTP {}", label, status);
}
self.stream_download(response, dest_path).await?;
let bytes = tokio::fs::read(dest_path).await?;
validate_pdf_content(&bytes)?;
Ok(())
}.await;
match res {
Ok(()) => {
info!("[{}] PDF 下载成功: {:?}", label, dest_path);
Ok(())
}
Err(e) => {
let err_msg = e.to_string();
if err_msg.contains("人机验证")
|| err_msg.contains("挑战页面")
|| err_msg.contains("WAF")
|| err_msg.contains("Cloudflare")
|| err_msg.contains("HTTP 403")
|| err_msg.contains("HTTP 503")
{
warn!("[{}] 下载触发人机验证或拦截: {}。尝试使用 Obscura 后备通道...", label, err_msg);
self.download_via_obscura(url, dest_path, true).await
} else {
Err(e)
}
}
}
self.stream_download(response, dest_path).await?;
let bytes = tokio::fs::read(dest_path).await?;
validate_pdf_content(&bytes)?;
info!("[{}] PDF 下载成功: {:?}", label, dest_path);
Ok(())
}
/// 通用 HTML 直链下载(带随机延迟 + 反爬检测)
/// 通用 HTML 直链下载(带随机延迟 + 反爬检测 + Obscura 后备
async fn download_html_direct(&self, url: &str, dest_path: &Path, label: &str) -> Result<()> {
info!("[{}] 下载 HTML: {}", label, url);
Self::maybe_delay().await;
let response = self.client.get(url)
.headers(build_browser_headers())
.send().await
.context(format!("[{}] HTML 请求失败", label))?;
let res = async {
let response = self.client.get(url)
.headers(build_browser_headers())
.send().await
.context(format!("[{}] HTML 请求失败", label))?;
let status = response.status();
if !status.is_success() {
anyhow::bail!("[{}] 返回 HTTP {}", label, status);
let status = response.status();
if !status.is_success() {
anyhow::bail!("[{}] 返回 HTTP {}", label, status);
}
self.stream_download(response, dest_path).await?;
let text = tokio::fs::read_to_string(dest_path).await
.context("读取 HTML 文件失败")?;
validate_html_content(&text)?;
Ok(())
}.await;
match res {
Ok(()) => {
info!("[{}] HTML 下载成功: {:?}", label, dest_path);
Ok(())
}
Err(e) => {
let err_msg = e.to_string();
if err_msg.contains("人机验证")
|| err_msg.contains("挑战页面")
|| err_msg.contains("WAF")
|| err_msg.contains("Cloudflare")
|| err_msg.contains("HTTP 403")
|| err_msg.contains("HTTP 503")
{
warn!("[{}] 下载触发人机验证或拦截: {}。尝试使用 Obscura 后备通道...", label, err_msg);
self.download_via_obscura(url, dest_path, false).await
} else {
Err(e)
}
}
}
self.stream_download(response, dest_path).await?;
let sniff = Self::read_file_header(dest_path).await?;
let text = String::from_utf8_lossy(&sniff);
validate_html_content(&text)?;
info!("[{}] HTML 下载成功: {:?}", label, dest_path);
Ok(())
}
// ─── CrossRef 回退通道 ─────────────────────────────────────
@@ -418,7 +626,7 @@ impl Downloader {
/// HTML 下载优先级:
/// 1. 官方 `arxiv.org/html/{id}`2023-12 起支持,质量与 ar5iv 相同,更稳定)
/// 2. ar5iv `ar5iv.labs.arxiv.org/html/{id}`(约 3% 论文转换失败时跳过)
pub async fn download_arxiv_direct(&self, arxiv_id: &str, library_dir: &Path) -> (Option<PathBuf>, Option<PathBuf>) {
pub async fn download_arxiv_direct(&self, arxiv_id: &str, library_dir: &Path) -> (Result<PathBuf, String>, Result<PathBuf, String>) {
// 去除版本号(v1/v2/v3),arxiv.org/html/ 和 ar5iv 均只提供最新渲染版
let clean_id = strip_arxiv_version(arxiv_id);
@@ -426,31 +634,36 @@ impl Downloader {
let pdf_dest = library_dir.join("PDF").join(format!("{}.pdf", arxiv_id));
let html_dest = library_dir.join("HTML").join(format!("{}.html", arxiv_id));
let mut pdf_ok = None;
let mut html_ok = None;
// PDF 下载
match self.download_pdf_direct(&pdf_url, &pdf_dest, "arXiv").await {
Ok(_) => pdf_ok = Some(pdf_dest),
Err(e) => warn!("[arXiv] PDF 下载失败: {:?}", e),
}
let pdf_res = match self.download_pdf_direct(&pdf_url, &pdf_dest, "arXiv").await {
Ok(_) => Ok(pdf_dest),
Err(e) => {
let err_msg = format!("arXiv PDF 下载失败: {}", e);
warn!("{}", err_msg);
Err(err_msg)
}
};
// HTML 下载:官方 arxiv.org/html/ 优先
let official_html_url = format!("https://arxiv.org/html/{}", clean_id);
match self.download_html_direct(&official_html_url, &html_dest, "arXiv-HTML").await {
Ok(_) => html_ok = Some(html_dest.clone()),
let html_res = match self.download_html_direct(&official_html_url, &html_dest, "arXiv-HTML").await {
Ok(_) => Ok(html_dest.clone()),
Err(e) => {
warn!("[arXiv-HTML] 官方 HTML 下载失败,回退 ar5iv: {:?}", e);
// ar5iv 兜底:约 97% 成功率,可能有延迟
let ar5iv_url = format!("https://ar5iv.labs.arxiv.org/html/{}", clean_id);
match self.download_html_direct(&ar5iv_url, &html_dest, "ar5iv").await {
Ok(_) => html_ok = Some(html_dest),
Err(e2) => warn!("[ar5iv] HTML 下载也失败: {:?}", e2),
Ok(_) => Ok(html_dest),
Err(e2) => {
let err_msg = format!("arXiv HTML 下载失败 (官方: {}, ar5iv: {})", e, e2);
warn!("{}", err_msg);
Err(err_msg)
}
}
}
}
};
(pdf_ok, html_ok)
(pdf_res, html_res)
}
/// 下载 arXiv HTML:官方 arxiv.org/html/ 优先,ar5iv 兜底
@@ -477,13 +690,16 @@ impl Downloader {
/// HTML 回退顺序:
/// 1. ADS PUB_HTML 网关(IOP→ 直联 iopsciencearxiv abs → ar5iv
/// 2. ADS EPRINT_HTML 网关(arxiv abs → ar5iv
pub async fn download_paper(&self, bibcode: &str, doi: Option<&str>, library_dir: &Path) -> (Option<PathBuf>, Option<PathBuf>) {
pub async fn download_paper(&self, bibcode: &str, doi: Option<&str>, library_dir: &Path) -> (Result<PathBuf, String>, Result<PathBuf, String>) {
let base = "https://ui.adsabs.harvard.edu/link_gateway";
let pdf_dest = library_dir.join("PDF").join(format!("{}.pdf", bibcode));
let html_dest = library_dir.join("HTML").join(format!("{}.html", bibcode));
let mut pdf_ok: Option<PathBuf> = None;
let mut html_ok: Option<PathBuf> = None;
let mut pdf_res = Err("未尝试任何下载通道".to_string());
let mut html_res = Err("未尝试任何下载通道".to_string());
let mut pdf_errors = Vec::new();
let mut html_errors = Vec::new();
// ── PDF 下载 ───────────────────────────────────────────
info!("[下载] 开始 PDF 下载: {}", bibcode);
@@ -507,11 +723,19 @@ impl Downloader {
self.download_pdf_direct(&resolved, &pdf_dest, "PUB_PDF").await
};
match result {
Ok(_) => { pdf_ok = Some(pdf_dest.clone()); break 'pdf; }
Err(e) => warn!("[PUB_PDF] 下载失败: {:?}", e),
Ok(_) => { pdf_res = Ok(pdf_dest.clone()); break 'pdf; }
Err(e) => {
let msg = format!("PUB_PDF下载失败: {}", e);
warn!("{}", msg);
pdf_errors.push(msg);
}
}
}
Err(e) => warn!("[PUB_PDF] 网关解析失败: {:?}", e),
Err(e) => {
let msg = format!("PUB_PDF网关解析失败: {}", e);
warn!("{}", msg);
pdf_errors.push(msg);
}
}
// 1b. ADS_PDF 网关 (经典 ADS 整合 PDF 直接通道)
@@ -519,11 +743,19 @@ impl Downloader {
match self.resolve_ads_gateway(&gw).await {
Ok(resolved) => {
match self.download_pdf_direct(&resolved, &pdf_dest, "ADS_PDF").await {
Ok(_) => { pdf_ok = Some(pdf_dest.clone()); break 'pdf; }
Err(e) => warn!("[ADS_PDF] 下载失败: {:?}", e),
Ok(_) => { pdf_res = Ok(pdf_dest.clone()); break 'pdf; }
Err(e) => {
let msg = format!("ADS_PDF下载失败: {}", e);
warn!("{}", msg);
pdf_errors.push(msg);
}
}
}
Err(e) => warn!("[ADS_PDF] 网关解析失败: {:?}", e),
Err(e) => {
let msg = format!("ADS_PDF网关解析失败: {}", e);
warn!("{}", msg);
pdf_errors.push(msg);
}
}
// 1c. ADS EPRINT_PDF 网关
@@ -531,29 +763,49 @@ impl Downloader {
match self.resolve_ads_gateway(&gw).await {
Ok(resolved) => {
match self.download_pdf_direct(&resolved, &pdf_dest, "EPRINT_PDF").await {
Ok(_) => { pdf_ok = Some(pdf_dest.clone()); break 'pdf; }
Err(e) => warn!("[EPRINT_PDF] 下载失败: {:?}", e),
Ok(_) => { pdf_res = Ok(pdf_dest.clone()); break 'pdf; }
Err(e) => {
let msg = format!("EPRINT_PDF下载失败: {}", e);
warn!("{}", msg);
pdf_errors.push(msg);
}
}
}
Err(e) => warn!("[EPRINT_PDF] 网关解析失败: {:?}", e),
Err(e) => {
let msg = format!("EPRINT_PDF网关解析失败: {}", e);
warn!("{}", msg);
pdf_errors.push(msg);
}
}
// 1c. CrossRef API 回退(需要 DOI
if let Some(doi_str) = doi {
match self.download_crossref_pdf(doi_str, &pdf_dest).await {
Ok(_) => { pdf_ok = Some(pdf_dest.clone()); break 'pdf; }
Err(e) => warn!("[CrossRef] PDF 下载失败: {:?}", e),
Ok(_) => { pdf_res = Ok(pdf_dest.clone()); break 'pdf; }
Err(e) => {
let msg = format!("CrossRef下载失败: {}", e);
warn!("{}", msg);
pdf_errors.push(msg);
}
}
}
// 1d. ADS SCAN 扫描版文献直接合并下载 PDF(主要针对早期/不可下载直接 PDF 的文献)
let scan_url = format!("https://articles.adsabs.harvard.edu/cgi-bin/nph-iarticle_query?bibcode={}&db_key=AST&data_type=PDF_HIGH", bibcode);
match self.download_pdf_direct(&scan_url, &pdf_dest, "ADS_SCAN").await {
Ok(_) => { pdf_ok = Some(pdf_dest.clone()); }
Err(e) => warn!("[ADS_SCAN] 下载失败: {:?}", e),
Ok(_) => { pdf_res = Ok(pdf_dest.clone()); }
Err(e) => {
let msg = format!("ADS_SCAN下载失败: {}", e);
warn!("{}", msg);
pdf_errors.push(msg);
}
}
}
if pdf_res.is_err() && !pdf_errors.is_empty() {
pdf_res = Err(pdf_errors.join("; "));
}
// ── HTML 下载 ──────────────────────────────────────────
info!("[下载] 开始 HTML 下载: {}", bibcode);
@@ -576,11 +828,19 @@ impl Downloader {
self.download_html_direct(&resolved, &html_dest, "PUB_HTML").await
};
match result {
Ok(_) => { html_ok = Some(html_dest.clone()); break 'html; }
Err(e) => warn!("[PUB_HTML] 下载失败: {:?}", e),
Ok(_) => { html_res = Ok(html_dest.clone()); break 'html; }
Err(e) => {
let msg = format!("PUB_HTML下载失败: {}", e);
warn!("{}", msg);
html_errors.push(msg);
}
}
}
Err(e) => warn!("[PUB_HTML] 网关解析失败: {:?}", e),
Err(e) => {
let msg = format!("PUB_HTML网关解析失败: {}", e);
warn!("{}", msg);
html_errors.push(msg);
}
}
// 2b. ADS EPRINT_HTML 网关(大多数天文论文有 arXiv eprint
@@ -593,15 +853,27 @@ impl Downloader {
self.download_html_direct(&resolved, &html_dest, "EPRINT_HTML").await
};
match result {
Ok(_) => { html_ok = Some(html_dest.clone()); }
Err(e) => warn!("[EPRINT_HTML] 下载失败: {:?}", e),
Ok(_) => { html_res = Ok(html_dest.clone()); }
Err(e) => {
let msg = format!("EPRINT_HTML下载失败: {}", e);
warn!("{}", msg);
html_errors.push(msg);
}
}
}
Err(e) => warn!("[EPRINT_HTML] 网关解析失败: {:?}", e),
Err(e) => {
let msg = format!("EPRINT_HTML网关解析失败: {}", e);
warn!("{}", msg);
html_errors.push(msg);
}
}
}
(pdf_ok, html_ok)
if html_res.is_err() && !html_errors.is_empty() {
html_res = Err(html_errors.join("; "));
}
(pdf_res, html_res)
}
}
@@ -736,7 +1008,7 @@ mod tests {
let temp_dir = std::env::temp_dir();
let (pdf_path, _html_path) = downloader.download_paper(bibcode, None, &temp_dir).await;
assert!(pdf_path.is_some());
assert!(pdf_path.is_ok());
let path = pdf_path.unwrap();
assert!(path.exists());
+5 -1
View File
@@ -2,5 +2,9 @@ pub mod download;
pub mod parser;
pub mod translation;
pub mod query_parser;
pub mod batch_sync;
pub mod batch;
pub mod logging;
pub mod batch_sync {
pub use super::batch::*;
}
+2 -6
View File
@@ -200,9 +200,7 @@ fn postprocess_markdown(text: &str) -> String {
}
}
let mut md = clean_lines.join("\n");
if md.contains("Keywords") {
println!("DEBUG 0: {:?}", md);
}
let div_re = Regex::new(r"</?div[^>]*>").unwrap();
let span_re = Regex::new(r"</?span[^>]*>").unwrap();
@@ -214,9 +212,7 @@ fn postprocess_markdown(text: &str) -> String {
let excessive_newlines = Regex::new(r"\n{4,}").unwrap();
md = excessive_newlines.replace_all(&md, "\n\n\n").to_string();
if md.contains("Keywords") {
println!("DEBUG 1 (excessive): {:?}", md);
}
// 还原被 html2md 自动转义的标题与引用符号
let unescape_h1 = Regex::new(r"\\#\s+").unwrap();