use std::path::PathBuf;
use crate::error::{Error, Result};
use crate::fetcher::{default_cache_dir, resolved_base_url, CachedFetcher};
use crate::parquet_io::{read_dividends, DivRow};
use crate::record::{DivEvent, DividendSnapshot};
pub struct Divkit {
fetcher: CachedFetcher,
}
impl Divkit {
pub fn new() -> Self {
let http = reqwest::Client::builder()
.user_agent("divkit/0.1 (+https://github.com/userFRM/divkit)")
.timeout(std::time::Duration::from_secs(30))
.build()
.unwrap_or_else(|_| reqwest::Client::new());
Self {
fetcher: CachedFetcher::new(http, resolved_base_url(), default_cache_dir()),
}
}
pub fn with_base_url(mut self, url: impl Into<String>) -> Self {
self.fetcher.set_base_url(url.into());
self
}
pub fn with_cache_dir(mut self, dir: PathBuf) -> Self {
self.fetcher.set_cache_dir(dir);
self
}
pub fn with_mirror_url(mut self, url: Option<String>) -> Self {
self.fetcher.set_mirror_url(url);
self
}
pub async fn dividends(&self, ticker: &str) -> Result<Vec<DivEvent>> {
let rows = self.load_all_rows().await?;
Ok(filter_ticker(&rows, ticker)
.into_iter()
.map(row_to_event)
.collect())
}
pub async fn dividend_snapshot(&self, ticker: &str) -> Result<DividendSnapshot> {
let rows = self.load_all_rows().await?;
let matching = filter_ticker(&rows, ticker);
if matching.is_empty() {
return Err(Error::NotFound(format!("no dividend data for {ticker}")));
}
let cik = matching[0].cik;
let events: Vec<DivEvent> = matching.into_iter().map(row_to_event).collect();
Ok(DividendSnapshot::from_events(
ticker.to_uppercase(),
cik,
events,
))
}
pub async fn annual_dividend(&self, ticker: &str) -> Result<Option<f64>> {
let rows = self.load_all_rows().await?;
let matching = filter_ticker(&rows, ticker);
if matching.is_empty() {
return Ok(None);
}
let cik = matching[0].cik;
let events: Vec<DivEvent> = matching.into_iter().map(row_to_event).collect();
let snap = DividendSnapshot::from_events(ticker.to_uppercase(), cik, events);
Ok(Some(snap.annual_amount()))
}
pub fn dividends_blocking(&self, ticker: &str) -> Result<Vec<DivEvent>> {
block(self.dividends(ticker))
}
pub fn dividend_snapshot_blocking(&self, ticker: &str) -> Result<DividendSnapshot> {
block(self.dividend_snapshot(ticker))
}
pub fn annual_dividend_blocking(&self, ticker: &str) -> Result<Option<f64>> {
block(self.annual_dividend(ticker))
}
async fn load_all_rows(&self) -> Result<Vec<DivRow>> {
let shard_keys = self.discover_shards().await?;
let mut all_rows = Vec::new();
for key in shard_keys {
let bytes = self.fetcher.fetch(&key).await?;
let rows = read_dividends(&bytes)?;
all_rows.extend(rows);
}
Ok(all_rows)
}
async fn discover_shards(&self) -> Result<Vec<String>> {
let manifest_url = format!("{}/manifest.json", self.fetcher.base_url);
let resp = self
.fetcher
.http
.get(&manifest_url)
.send()
.await
.map_err(Error::Http)?;
if !resp.status().is_success() {
return Err(Error::Other(format!(
"manifest.json: HTTP {} {}",
resp.status().as_u16(),
resp.status().canonical_reason().unwrap_or("")
)));
}
let manifest: serde_json::Value = resp.json().await.map_err(Error::Http)?;
let obj = manifest
.as_object()
.ok_or_else(|| Error::Other("manifest.json is not a JSON object".into()))?;
let mut keys: Vec<String> = obj
.keys()
.filter(|k| is_dividend_shard(k))
.map(|k| k.trim_end_matches(".parquet").to_string())
.collect();
keys.sort();
Ok(keys)
}
}
impl Default for Divkit {
fn default() -> Self {
Self::new()
}
}
pub async fn annual_dividend_for(ticker: &str) -> Result<Option<f64>> {
Divkit::new().annual_dividend(ticker).await
}
pub async fn dividends_for(ticker: &str) -> Result<Vec<DivEvent>> {
Divkit::new().dividends(ticker).await
}
pub async fn dividend_snapshot_for(ticker: &str) -> Result<DividendSnapshot> {
Divkit::new().dividend_snapshot(ticker).await
}
fn is_dividend_shard(name: &str) -> bool {
let Some(rest) = name.strip_prefix("dividends-") else {
return false;
};
let Some(year_str) = rest.strip_suffix(".parquet") else {
return false;
};
!year_str.is_empty() && year_str.bytes().all(|b| b.is_ascii_digit())
}
fn filter_ticker<'a>(rows: &'a [DivRow], target: &str) -> Vec<&'a DivRow> {
let upper = target.to_uppercase();
rows.iter()
.filter(|r| r.ticker.as_deref().map(|t| t.to_uppercase()) == Some(upper.clone()))
.collect()
}
fn row_to_event(row: &DivRow) -> DivEvent {
DivEvent {
period_start: row.period_start,
period_end: row.period_end,
amount: row.amount,
concept: row.concept,
accn: row.accn.clone(),
form: row.form.clone(),
}
}
fn block<F: std::future::Future<Output = Result<T>>, T>(fut: F) -> Result<T> {
match tokio::runtime::Handle::try_current() {
Ok(handle) => tokio::task::block_in_place(|| handle.block_on(fut)),
Err(_) => {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(Error::Io)?;
rt.block_on(fut)
}
}
}