1mod wire;
16
17use std::sync::Arc;
18use std::time::Duration;
19
20use aspero_core::http::{escape_lucene, path_segment, status_error};
21use aspero_core::{
22 Capabilities, Cursor, Doi, Error, Field, FieldSet, Filter, Hit, HttpClient, HttpRequest,
23 HttpResponse, IdKind, Lookup, Page, Provider, ProviderId, ResiliencePolicy, Response, Result,
24 SearchMode, SearchQuery, SourceRecord, WorkId,
25};
26use async_trait::async_trait;
27
28use crate::wire::WireList;
29
30pub const ID: ProviderId = ProviderId::from_static("datacite");
31
32pub const DEFAULT_BASE_URL: &str = "https://api.datacite.org";
33
34pub(crate) const ARXIV_DOI_PREFIX: &str = "10.48550";
35
36const MAX_BATCH: usize = 50;
38
39const MAX_PAGE: u32 = 1000;
41
42const PASSTHROUGH: &[&str] = &["client-id", "prefix", "affiliation-id", "has-citations"];
44
45#[derive(Debug, Clone)]
46#[non_exhaustive]
47pub struct DataciteConfig {
48 pub base_url: String,
49}
50
51impl Default for DataciteConfig {
52 fn default() -> Self {
53 DataciteConfig {
54 base_url: DEFAULT_BASE_URL.to_owned(),
55 }
56 }
57}
58
59pub struct Datacite {
60 config: DataciteConfig,
61 http: Arc<dyn HttpClient>,
62 capabilities: Capabilities,
63}
64
65impl Datacite {
66 pub fn new(config: DataciteConfig, http: Arc<dyn HttpClient>) -> Self {
67 let mut capabilities = Capabilities::default();
68 capabilities.lookup_kinds = vec![IdKind::Doi, IdKind::Arxiv];
69 capabilities.max_batch = MAX_BATCH;
70 capabilities.fields = FieldSet::ALL.difference(FieldSet::of(&[Field::OpenAccess]));
72 capabilities.search = true;
73 capabilities.search_filters = vec![Filter::Years, Filter::WorkType];
74 Datacite {
75 config,
76 http,
77 capabilities,
78 }
79 }
80
81 pub fn default_policy() -> ResiliencePolicy {
83 let mut policy = ResiliencePolicy::default();
84 policy.min_interval = Duration::from_millis(200);
85 policy
86 }
87
88 fn url(&self, rest: &str, params: &[(&str, String)]) -> Result<String> {
89 let base = format!("{}/dois{rest}", self.config.base_url.trim_end_matches('/'));
90 let url = if params.is_empty() {
91 url::Url::parse(&base)
92 } else {
93 url::Url::parse_with_params(&base, params)
94 };
95 url.map(String::from)
96 .map_err(|e| Error::BadRequest(format!("invalid DataCite base URL: {e}")))
97 }
98
99 async fn get(&self, url: String) -> Result<HttpResponse> {
100 let req = HttpRequest::get(url).header("accept", "application/vnd.api+json");
101 let resp = self.http.send(req).await?;
102 check(&resp)?;
103 Ok(resp)
104 }
105
106 async fn lookup_one(&self, doi: &Doi) -> Result<Option<SourceRecord>> {
107 let resp = match self
108 .get(self.url(&format!("/{}", path_segment(doi.as_str())), &[])?)
109 .await
110 {
111 Ok(resp) => resp,
112 Err(Error::NotFound) => return Ok(None),
113 Err(e) => return Err(e),
114 };
115 let body: serde_json::Value = serde_json::from_slice(&resp.body).map_err(Error::decode)?;
116 let data = body.get("data").ok_or_else(|| Error::decode(MissingData))?;
117 wire::to_record(data, resp.received_at)
118 .map(Some)
119 .map_err(Error::decode)
120 }
121
122 async fn list(&self, params: &[(&str, String)]) -> Result<(Vec<Hit>, wire::WireMeta)> {
123 let resp = self.get(self.url("", params)?).await?;
124 let list: WireList = serde_json::from_slice(&resp.body).map_err(Error::decode)?;
125 let hits = list
126 .data
127 .iter()
128 .filter_map(|v| or_skip(wire::to_record(v, resp.received_at)))
129 .map(|r| Hit::new(r, None))
130 .collect();
131 Ok((hits, list.meta))
132 }
133}
134
135fn or_skip(record: serde_json::Result<SourceRecord>) -> Option<SourceRecord> {
138 record
139 .inspect_err(|e| {
140 tracing::warn!(
141 target: "aspero::decode",
142 provider = %ID, error = %e, "skipped a record that doesn't decode"
143 );
144 })
145 .ok()
146}
147
148#[derive(Debug, thiserror::Error)]
149#[error("DataCite response has no `data`")]
150struct MissingData;
151
152fn to_doi(id: &WorkId) -> Result<Doi> {
153 match id {
154 WorkId::Doi(doi) => Ok(doi.clone()),
155 WorkId::Arxiv(arxiv) => Ok(format!("{ARXIV_DOI_PREFIX}/arxiv.{}", arxiv.id()).parse()?),
156 _ => Err(Error::Unsupported("DataCite lookup of non-DOI identifiers")),
157 }
158}
159
160fn check(resp: &HttpResponse) -> Result<()> {
161 if resp.is_success() {
162 return Ok(());
163 }
164 let message = message(resp);
165 Err(match resp.status {
166 400 | 422 => Error::BadRequest(message.unwrap_or_else(|| "rejected by DataCite".into())),
167 401 | 403 => Error::AuthRejected { message },
168 _ => status_error(resp, message),
169 })
170}
171
172fn message(resp: &HttpResponse) -> Option<String> {
175 let json: serde_json::Value = serde_json::from_slice(&resp.body).ok()?;
176 let errors = json.get("errors")?;
177 let first = errors.as_array().and_then(|a| a.first()).unwrap_or(errors);
178 first.get("title")?.as_str().map(str::to_owned)
179}
180
181#[async_trait]
182impl Provider for Datacite {
183 fn id(&self) -> ProviderId {
184 ID
185 }
186
187 fn capabilities(&self) -> &Capabilities {
188 &self.capabilities
189 }
190
191 async fn lookup(&self, ids: &[WorkId], _fields: FieldSet) -> Result<Response<Vec<Lookup>>> {
192 let dois = ids.iter().map(to_doi).collect::<Result<Vec<_>>>()?;
193 if let [doi] = dois.as_slice() {
194 let record = self.lookup_one(doi).await?;
195 return Ok(Response::new(vec![Lookup {
196 id: ids[0].clone(),
197 record,
198 }]));
199 }
200 let mut found = Vec::new();
201 for chunk in dois.chunks(MAX_BATCH) {
202 let terms: Vec<String> = chunk
203 .iter()
204 .map(|d| format!("\"{}\"", d.as_str().replace('"', "\\\"")))
205 .collect();
206 let params = [
207 ("query", format!("doi:({})", terms.join(" OR "))),
208 ("page[size]", chunk.len().to_string()),
209 ];
210 let (hits, _) = self.list(¶ms).await?;
211 found.extend(hits.into_iter().map(|h| h.record));
212 }
213 let lookups = ids
214 .iter()
215 .zip(&dois)
216 .map(|(id, doi)| Lookup {
217 id: id.clone(),
218 record: found
219 .iter()
220 .find(|r| {
221 r.ids
222 .iter()
223 .any(|i| matches!(i, WorkId::Doi(d) if d == doi))
224 })
225 .cloned(),
226 })
227 .collect();
228 Ok(Response::new(lookups))
229 }
230
231 async fn search(&self, query: &SearchQuery) -> Result<Response<Page>> {
232 if query.mode != SearchMode::Keywords {
233 return Err(Error::Unsupported("DataCite has no citation matcher"));
234 }
235 let mut clauses = Vec::new();
236 if !query.text.trim().is_empty() {
237 clauses.push(format!("({})", escape_lucene(query.text.trim())));
238 }
239 if let Some(years) = query.years {
240 let end = |y: Option<i16>| y.map_or("*".to_owned(), |y| y.to_string());
241 clauses.push(format!(
242 "publicationYear:[{} TO {}]",
243 end(years.from),
244 end(years.to)
245 ));
246 }
247 let mut params: Vec<(&str, String)> = Vec::new();
248 for (key, value) in &query.extra {
249 match key.as_str() {
250 "query" => clauses.push(format!("({value})")),
252 k if PASSTHROUGH.contains(&k) => params.push((k, value.clone())),
253 other => {
254 return Err(Error::BadRequest(format!(
255 "unknown DataCite parameter {other:?} (supported: query, {})",
256 PASSTHROUGH.join(", ")
257 )));
258 }
259 }
260 }
261 if !clauses.is_empty() {
262 params.insert(0, ("query", clauses.join(" AND ")));
263 }
264 if let Some(t) = &query.work_type {
265 let id = wire::resource_type_id(t).ok_or_else(|| {
266 Error::BadRequest(format!(
267 "DataCite has no resource type matching {:?}",
268 t.as_csl()
269 ))
270 })?;
271 params.push(("resource-type-id", id.to_owned()));
272 }
273 let page_number: u32 = match &query.page.cursor {
274 None => 1,
275 Some(c) => {
276 c.0.parse()
277 .map_err(|_| Error::BadRequest(format!("invalid DataCite cursor {:?}", c.0)))?
278 }
279 };
280 params.push(("sort", "relevance".into()));
281 params.push(("page[number]", page_number.to_string()));
282 params.push((
283 "page[size]",
284 query.page.limit.clamp(1, MAX_PAGE).to_string(),
285 ));
286
287 let (hits, meta) = self.list(¶ms).await?;
288 let more = meta.total_pages.is_some_and(|pages| page_number < pages) && !hits.is_empty();
289 let next = more.then(|| Cursor((page_number + 1).to_string()));
290 Ok(Response::new(Page::new(hits, meta.total, next)))
291 }
292}