Web-based Calendar Aggregator https://ticker.c3d2.de/
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

317 lines
10 KiB

use std::str::FromStr;
use std::mem::replace;
use std::io::Read;
use std::sync::RwLock;
use std::collections::{HashMap, VecDeque};
use chrono::{Duration, offset::Utc};
use reqwest::header::{IF_NONE_MATCH, IF_MODIFIED_SINCE, ETAG, LAST_MODIFIED, USER_AGENT};
use diesel::{Connection, pg::PgConnection, prelude::*};
use rrule::RRule;
use libticker::{
config::{Config, CalendarOptions},
schema::{self, calendars::dsl::calendars},
model::{Calendar, NewCalendar, Event},
ics::{Parser, Object, Timestamp, GetValue},
};
fn extract_vevent_objs(results: &mut Vec<Object>, mut obj: Object) {
let children = replace(&mut obj.children, vec![]);
if obj.name == "VEVENT" {
results.push(obj);
}
for child in children.into_iter() {
extract_vevent_objs(results, *child);
}
}
const RRULE_LOOKBACK: i64 = 30;
const RRULE_LOOKAHEAD: i64 = 366;
fn obj_to_events(calendar: String, obj: &Object) -> Vec<Event> {
if obj.name != "VEVENT" {
return vec![];
}
let (dtstart, dtstart_str): (Timestamp, &str) = match (obj.get("DTSTART"), obj.get("DTSTART")) {
(Some(dtstart), Some(dtstart_str)) => (dtstart, dtstart_str),
_ => return vec![],
};
let dtstart = dtstart.or_hms(0, 0, 0);
let dtend: Option<Timestamp> = obj.get("DTEND");
let dtend = dtend.map(|time| time.or_hms(23, 59, 59));
let summary: &str = match obj.get("SUMMARY") {
Some(summary) => summary,
None => return vec![],
};
let uid = obj.get("UID").unwrap_or("");
let dtstamp = obj.get("DTSTAMP").unwrap_or("");
let location = obj.get("LOCATION");
let url = obj.get("URL");
let generate_event = |dtstart| {
let id = format!("{}{}{}", uid, dtstart, dtstamp);
Event {
calendar: calendar.clone(),
id,
dtstart,
dtend,
summary: summary.to_owned(),
location: location.clone(),
url: url.clone(),
}
};
let rrule_str: Option<&str> = obj.get("rrule");
let rrule = rrule_str.and_then(
|rrule_str| RRule::from_str(
&format!("DTSTART:{}\nRRULE:{}", dtstart_str, rrule_str)
).map_err(|e| println!("Error parsing RRULE: {}", e))
.ok()
);
match rrule {
Some(rrule) => {
let now = Utc::now();
let start = now - Duration::days(RRULE_LOOKBACK);
let end = now + Duration::days(RRULE_LOOKAHEAD);
println!("rrule {}:\n{:?}", rrule_str.unwrap_or(""), rrule.into_iter()
.skip_while(|d| *d < start)
.take_while(|d| *d <= end)
.map(|dtstart| dtstart.naive_utc())
.collect::<Vec<_>>()
);
rrule.into_iter()
.skip_while(|d| *d < start)
.take_while(|d| *d <= end)
.map(|dtstart| dtstart.naive_utc())
.map(generate_event)
.collect()
}
None =>
vec![generate_event(dtstart)],
}
}
pub struct Resources {
db_url: String,
http_client: reqwest::Client,
queue: RwLock<VecDeque<(String, CalendarOptions)>>,
}
impl Resources {
pub fn new(db_url: String, queue: VecDeque<(String, CalendarOptions)>) -> Self {
let http_client = reqwest::Client::new();
let queue = RwLock::new(queue);
Resources { db_url, http_client, queue }
}
fn next(&self) -> Option<(String, CalendarOptions)> {
let mut queue = self.queue.write().unwrap();
queue.pop_front()
}
fn worker_fetch(&self, cal_id: &str, cal_opts: &CalendarOptions, etag: Option<String>, last_modified: Option<String>)
-> Result<(Option<String>, Option<String>, HashMap<String, Event>), Error>
{
let mut req = self.http_client.get(&cal_opts.url)
.header(USER_AGENT, "Ticker/0.0.0");
match etag {
Some(etag) => req = req.header(IF_NONE_MATCH, etag),
None => (),
}
match last_modified {
Some(last_modified) => req = req.header(IF_MODIFIED_SINCE, last_modified),
None => (),
}
let mut res = req.send()?;
if res.status() != 200 {
let msg = format!("HTTP {}", res.status());
return Err(Error::Message(msg));
}
let etag = res.headers().get(ETAG)
.and_then(|v| v.to_str().ok())
.map(|s| s.to_owned());
let last_modified = res.headers().get(LAST_MODIFIED)
.and_then(|v| v.to_str().ok())
.map(|s| s.to_owned());
let mut p = Parser::new();
let mut buf = [0; 1024];
let mut events = HashMap::new();
loop {
match res.read(&mut buf)? {
len if len > 0 => {
let data = &buf[..len];
p.feed(data, |obj| {
let mut objs = vec![];
extract_vevent_objs(&mut objs, obj);
for obj in objs {
for event in obj_to_events(cal_id.to_owned(), &obj) {
events.insert(event.id.clone(), event);
}
}
});
}
_ => break,
}
}
Ok((etag, last_modified, events))
}
fn worker_job(&self, db: &PgConnection) -> Result<bool, Error> {
let (cal_id, cal_opts) = match self.next() {
Some(next) => next,
None => return Ok(true),
};
let (etag, last_modified) = db.transaction::<_, Error, _>(|| {
let cals = calendars
.filter(schema::calendars::dsl::id.eq(cal_id.clone()))
.limit(1)
.load::<Calendar>(db)?;
match cals.first() {
None => {
let cal = NewCalendar {
id: cal_id.clone(),
url: cal_opts.url.clone(),
last_fetch: Some(Utc::now().naive_utc()),
};
diesel::insert_into(calendars)
.values(&cal)
.execute(db)?;
Ok((None, None))
}
Some(cal) => {
diesel::update(calendars)
.filter(schema::calendars::dsl::id.eq(cal_id.clone()))
.set(schema::calendars::dsl::last_fetch.eq(Utc::now().naive_utc()))
.execute(db)?;
let result =
// Use ETag/Last-Modified only if URL hasn't changed
if cal.url == cal_opts.url {
(cal.etag.clone(), cal.last_modified.clone())
} else {
(None, None)
};
Ok(result)
}
}
})?;
let (etag, last_modified, events) =
self.worker_fetch(&cal_id, &cal_opts, etag, last_modified)
.map_err(|e| {
let msg = format!("{}", e);
println!("[{}] {}", cal_id, msg);
let _ = diesel::update(calendars)
.filter(schema::calendars::dsl::id.eq(cal_id.clone()))
.set(schema::calendars::dsl::error_message.eq(msg))
.execute(db);
e
})?;
println!("[{}] {} events", cal_id, events.len());
db.transaction::<_, Error, _>(|| {
diesel::update(calendars)
.filter(schema::calendars::dsl::id.eq(cal_id.clone()))
.set((schema::calendars::dsl::last_success.eq(Utc::now().naive_utc()),
schema::calendars::dsl::etag.eq(etag),
schema::calendars::dsl::last_modified.eq(last_modified),
schema::calendars::dsl::error_message.eq(None as Option<String>)))
.execute(db)?;
diesel::delete(schema::events::dsl::events)
.filter(schema::events::dsl::calendar.eq(cal_id))
.execute(db)?;
for event in events.values() {
diesel::insert_into(schema::events::dsl::events)
.values(event)
.execute(db)?;
}
Ok(())
})?;
Ok(false)
}
fn worker_loop(&self) -> Result<(), Error> {
let db = PgConnection::establish(&self.db_url)?;
let mut done = false;
while ! done {
done = self.worker_job(&db)
.map_err(|e| println!("{}", e))
.unwrap_or(false);
}
Ok(())
}
}
#[derive(Debug)]
pub enum Error {
DbConnection(diesel::ConnectionError),
Db(diesel::result::Error),
Http(reqwest::Error),
Io(std::io::Error),
Message(String),
}
impl From<diesel::ConnectionError> for Error {
fn from(e: diesel::ConnectionError) -> Self {
Error::DbConnection(e)
}
}
impl From<diesel::result::Error> for Error {
fn from(e: diesel::result::Error) -> Self {
Error::Db(e)
}
}
impl From<reqwest::Error> for Error {
fn from(e: reqwest::Error) -> Self {
Error::Http(e)
}
}
impl From<std::io::Error> for Error {
fn from(e: std::io::Error) -> Self {
Error::Io(e)
}
}
impl std::fmt::Display for Error {
fn fmt(&self, fmt: &mut std::fmt::Formatter) -> Result<(), std::fmt::Error> {
match self {
Error::DbConnection(e) => e.fmt(fmt),
Error::Db(e) => e.fmt(fmt),
Error::Http(e) => e.fmt(fmt),
Error::Io(e) => e.fmt(fmt),
Error::Message(e) => e.fmt(fmt),
}
}
}
fn main() {
let config = Config::read_yaml_file("config.yaml");
let res = Resources::new(
config.db_url,
config.calendars.into_iter()
.collect()
);
crossbeam::scope(|s| {
let cpus = num_cpus::get();
let workers: Vec<_> = (0..cpus)
.map(|_| s.spawn(|_| {
res.worker_loop()
}))
.collect();
for worker in workers.into_iter() {
worker.join().expect("worker join").expect("worker");
}
}).unwrap();
}