feat: add fetch module for RDF over HTTP with fallbacks

gtfierro · aider-chat-bot · gtfierro · commit 87efbba0bdef · 2025-09-30T09:21:59.000-06:00
Co-authored-by: aider (openai/gpt-5) &lt;aider@aider.chat&gt;
diff --git a/lib/src/fetch.rs b/lib/src/fetch.rs
@@ -0,0 +1,254 @@
+use crate::errors::OfflineRetrievalError;
+use anyhow::{anyhow, Result};
+use chrono::prelude::*;
+use oxigraph::io::RdfFormat;
+use reqwest::blocking::Client;
+use reqwest::header::{HeaderMap, HeaderValue, ACCEPT, CONTENT_TYPE, LINK};
+use std::time::Duration;
+
+#[derive(Debug, Clone)]
+pub struct FetchOptions {
+    pub offline: bool,
+    pub timeout: Duration,
+    pub accept_order: Vec<&'static str>,
+    pub extension_candidates: Vec<&'static str>,
+}
+
+impl Default for FetchOptions {
+    fn default() -> Self {
+        Self {
+            offline: false,
+            timeout: Duration::from_secs(30),
+            accept_order: vec![
+                "text/turtle",
+                "application/rdf+xml",
+                "application/n-triples",
+            ],
+            extension_candidates: vec![".ttl", ".rdf", ".owl", "index.ttl", "index.rdf"],
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct FetchResult {
+    pub bytes: Vec<u8>,
+    pub format: Option<RdfFormat>,
+    pub final_url: String,
+    pub content_type: Option<String>,
+}
+
+fn detect_format(ct: &str) -> Option<RdfFormat> {
+    let ct = ct.split(';').next().unwrap_or("").trim().to_ascii_lowercase();
+    match ct.as_str() {
+        "text/turtle" | "application/x-turtle" => Some(RdfFormat::Turtle),
+        "application/rdf+xml" => Some(RdfFormat::RdfXml),
+        "application/n-triples" | "application/ntriples" | "text/plain" => Some(RdfFormat::NTriples),
+        _ => None,
+    }
+}
+
+fn build_accept(accept_order: &[&'static str]) -> String {
+    if accept_order.is_empty() {
+        return "*/*".to_string();
+    }
+    let mut parts = Vec::new();
+    let mut q = 1.0f32;
+    for (i, t) in accept_order.iter().enumerate() {
+        parts.push(format!("{t}; q={:.1}", q));
+        let next = 1.0f32 - 0.1f32 * (i as f32 + 1.0f32);
+        q = if next < 0.1 { 0.1 } else { next };
+    }
+    parts.push("*/*; q=0.1".to_string());
+    parts.join(", ")
+}
+
+fn build_extension_candidates(orig: &str, exts: &[&str]) -> Vec<String> {
+    let mut cands = Vec::new();
+    if orig.ends_with('/') {
+        for e in exts {
+            cands.push(format!("{orig}{e}"));
+        }
+        return cands;
+    }
+    // split path
+    let slash_pos = orig.rfind('/').map(|i| i + 1).unwrap_or(0);
+    let (prefix, filename) = orig.split_at(slash_pos);
+    if let Some(dot) = filename.rfind('.') {
+        let stem = &filename[..dot];
+        let base = format!("{prefix}{stem}");
+        for rep in [".ttl", ".rdf", ".owl"] {
+            cands.push(format!("{base}{rep}"));
+        }
+    } else {
+        for rep in [".ttl", ".rdf", ".owl"] {
+            cands.push(format!("{orig}{rep}"));
+        }
+    }
+    cands
+}
+
+fn parse_link_alternates(headers: &HeaderMap, accept_order: &[&'static str]) -> Vec<String> {
+    let mut out = Vec::new();
+    if let Some(link_val) = headers.get(LINK) {
+        if let Ok(link_str) = link_val.to_str() {
+            for part in link_str.split(',') {
+                let part = part.trim();
+                if !part.contains("rel=\"alternate\"") {
+                    continue;
+                }
+                // Try to extract type and URL
+                let has_rdf_type = accept_order
+                    .iter()
+                    .any(|typ| part.contains(&format!("type=\"{}\"", typ)));
+                if !has_rdf_type {
+                    continue;
+                }
+                if let Some(start) = part.find('<') {
+                    if let Some(end) = part[start + 1..].find('>') {
+                        let url = &part[start + 1..start + 1 + end];
+                        out.push(url.to_string());
+                    }
+                }
+            }
+        }
+    }
+    out
+}
+
+fn try_get(
+    url: &str,
+    client: &Client,
+    accept: &str,
+) -> Result<(Vec<u8>, Option<String>, Option<String>, String, reqwest::StatusCode)> {
+    let resp = client.get(url).header(ACCEPT, accept).send()?;
+    let status = resp.status();
+    let final_url = resp.url().to_string();
+    let ct = resp
+        .headers()
+        .get(CONTENT_TYPE)
+        .and_then(|h| h.to_str().ok())
+        .map(|s| s.to_string());
+    let link = resp
+        .headers()
+        .get(LINK)
+        .and_then(|h| h.to_str().ok())
+        .map(|s| s.to_string());
+    let bytes = resp.bytes()?.to_vec();
+    Ok((bytes, ct, link, final_url, status))
+}
+
+pub fn fetch_rdf(url: &str, opts: &FetchOptions) -> Result<FetchResult> {
+    if opts.offline {
+        return Err(anyhow!(OfflineRetrievalError {
+            file: url.to_string()
+        }));
+    }
+    let client = Client::builder().timeout(opts.timeout).build()?;
+    let accept = build_accept(&opts.accept_order);
+
+    // First attempt
+    let (bytes, ct, link, final_url, status) = try_get(url, &client, &accept)?;
+
+    // If success and looks RDF by Content-Type, return
+    if status.is_success() {
+        if let Some(ref cts) = ct {
+            if let Some(fmt) = detect_format(cts) {
+                return Ok(FetchResult {
+                    bytes,
+                    format: Some(fmt),
+                    final_url,
+                    content_type: ct,
+                });
+            }
+        }
+        // Unknown or HTML content-type: fall through to alternates with Link header hints
+    }
+
+    // Try Link: rel="alternate" first if present
+    if let Some(link_header) = link {
+        let mut headers = HeaderMap::new();
+        headers.insert(
+            LINK,
+            HeaderValue::from_str(&link_header).unwrap_or(HeaderValue::from_static("")),
+        );
+        for alt in parse_link_alternates(&headers, &opts.accept_order) {
+            let (b2, ct2, _link2, fu2, st2) = try_get(&alt, &client, &accept)?;
+            if st2.is_success() {
+                let fmt = ct2.as_deref().and_then(detect_format);
+                return Ok(FetchResult {
+                    bytes: b2,
+                    format: fmt,
+                    final_url: fu2,
+                    content_type: ct2,
+                });
+            }
+        }
+    }
+
+    // Status-based or type-based fallbacks
+    if !status.is_success() || ct.as_deref().map(|s| s.contains("html")).unwrap_or(true) {
+        for candidate in build_extension_candidates(url, &opts.extension_candidates) {
+            let (b2, ct2, _link2, fu2, st2) = try_get(&candidate, &client, &accept)?;
+            if st2.is_success() {
+                let fmt = ct2.as_deref().and_then(detect_format);
+                return Ok(FetchResult {
+                    bytes: b2,
+                    format: fmt,
+                    final_url: fu2,
+                    content_type: ct2,
+                });
+            }
+        }
+    }
+
+    // As a last resort, if the original was successful but with unknown CT, return it.
+    if status.is_success() {
+        let fmt = ct.as_deref().and_then(detect_format);
+        return Ok(FetchResult {
+            bytes,
+            format: fmt,
+            final_url,
+            content_type: ct,
+        });
+    }
+
+    Err(anyhow!(
+        "Failed to retrieve RDF from {} (HTTP {}) and fallbacks",
+        url,
+        status
+    ))
+}
+
+pub fn head_last_modified(url: &str, opts: &FetchOptions) -> Result<Option<DateTime<Utc>>> {
+    if opts.offline {
+        return Err(anyhow!(OfflineRetrievalError {
+            file: url.to_string()
+        }));
+    }
+    let client = Client::builder().timeout(opts.timeout).build()?;
+    let accept = build_accept(&opts.accept_order);
+    let resp = client.head(url).header(ACCEPT, accept).send()?;
+    if !resp.status().is_success() {
+        return Ok(None);
+    }
+    if let Some(h) = resp.headers().get("Last-Modified") {
+        if let Ok(s) = h.to_str() {
+            if let Ok(dt) = DateTime::parse_from_rfc2822(s) {
+                return Ok(Some(dt.with_timezone(&Utc)));
+            }
+        }
+    }
+    Ok(None)
+}
+
+pub fn head_exists(url: &str, opts: &FetchOptions) -> Result<bool> {
+    if opts.offline {
+        return Err(anyhow!(OfflineRetrievalError {
+            file: url.to_string()
+        }));
+    }
+    let client = Client::builder().timeout(opts.timeout).build()?;
+    let accept = build_accept(&opts.accept_order);
+    let resp = client.head(url).header(ACCEPT, accept).send()?;
+    Ok(resp.status().is_success())
+}
diff --git a/lib/src/io.rs b/lib/src/io.rs
@@ -27,6 +27,30 @@ pub struct StoreStats {
     pub num_triples: usize,
 }
 
+fn load_staging_store_from_bytes(bytes: &[u8], preferred: Option<RdfFormat>) -> Result<Store> {
+    // Try preferred first, then fall back to other formats with a fresh store each time
+    let mut candidates = vec![RdfFormat::Turtle, RdfFormat::RdfXml, RdfFormat::NTriples];
+    if let Some(p) = preferred {
+        candidates.retain(|f| *f != p);
+        candidates.insert(0, p);
+    }
+    for fmt in candidates {
+        let store = Store::new()?;
+        let staging_graph = NamedNode::new_unchecked("temp:graph");
+        let parser = RdfParser::from_format(fmt)
+            .with_default_graph(GraphNameRef::NamedNode(staging_graph.as_ref()))
+            .without_named_graphs();
+        if store
+            .bulk_loader()
+            .load_from_reader(parser, std::io::Cursor::new(bytes))
+            .is_ok()
+        {
+            return Ok(store);
+        }
+    }
+    Err(anyhow!("Failed to parse RDF bytes in any supported format"))
+}
+
 /// A helper function to read an ontology from a location, add it to a store,
 /// and return the parsed ontology metadata. This is used by multiple GraphIO implementations.
 fn add_ontology_to_store(
@@ -36,27 +60,23 @@ fn add_ontology_to_store(
     offline: bool,
     strict: bool,
 ) -> Result<Ontology> {
-    // Read source into bytes and detect format
-    let (bytes, format) = match &location {
-        OntologyLocation::File(path) => get_file_contents(path)?,
+    // Read source into bytes (via fetch for URLs) and load into a staging store with format fallback
+    let t0 = Instant::now();
+    let staging_graph = NamedNode::new_unchecked("temp:graph");
+    let tmp_store = match &location {
+        OntologyLocation::File(path) => {
+            let (bytes, format) = get_file_contents(path)?;
+            load_staging_store_from_bytes(&bytes, format)?
+        }
         OntologyLocation::Url(url) => {
             if offline {
                 return Err(Error::new(OfflineRetrievalError { file: url.clone() }));
             }
-            get_url_contents(url.as_str())?
+            let opts = crate::fetch::FetchOptions::default();
+            let fetched = crate::fetch::fetch_rdf(url.as_str(), &opts)?;
+            load_staging_store_from_bytes(&fetched.bytes, fetched.format)?
         }
     };
-
-    // Parse once into a temporary, isolated store to discover ontology metadata
-    let tmp_store = Store::new()?;
-    let staging_graph = NamedNode::new_unchecked("temp:graph");
-    let parser = RdfParser::from_format(format.unwrap_or(RdfFormat::Turtle))
-        .with_default_graph(GraphNameRef::NamedNode(staging_graph.as_ref()))
-        .without_named_graphs();
-    let t0 = Instant::now();
-    tmp_store
-        .bulk_loader()
-        .load_from_reader(parser, bytes.as_slice())?;
     debug!(
         "Loaded {} into staging store in {:?}",
         location.as_str(),
@@ -180,14 +200,9 @@ pub trait GraphIO: Send + Sync {
                 modified
             }
             OntologyLocation::Url(url) => {
-                let response = reqwest::blocking::Client::new().head(url).send()?;
-                let url_last_modified = response.headers().get("Last-Modified");
-                match url_last_modified {
-                    Some(date) => {
-                        let date = date.to_str()?;
-                        let date = DateTime::parse_from_rfc2822(date)?;
-                        date.with_timezone(&Utc)
-                    }
+                let opts = crate::fetch::FetchOptions::default();
+                match crate::fetch::head_last_modified(url, &opts)? {
+                    Some(dt) => dt,
                     None => Utc::now(),
                 }
             }
diff --git a/lib/src/lib.rs b/lib/src/lib.rs
@@ -101,6 +101,7 @@ pub mod io;
 pub mod ontology;
 pub mod options;
 pub mod policy;
+pub mod fetch;
 #[macro_use]
 pub mod util;
 pub mod transform;
diff --git a/lib/src/ontology.rs b/lib/src/ontology.rs
@@ -276,12 +276,8 @@ impl Ontology {
         match &self.location {
             Some(OntologyLocation::File(p)) => p.exists(),
             Some(OntologyLocation::Url(u)) => {
-                // check if the URL is reachable
-                let res = reqwest::blocking::get(u);
-                match res {
-                    Ok(r) => r.status().is_success(),
-                    Err(_) => false,
-                }
+                let opts = crate::fetch::FetchOptions::default();
+                crate::fetch::head_exists(u, &opts).unwrap_or(false)
             }
             None => false,
         }
diff --git a/lib/src/util.rs b/lib/src/util.rs
