personal/monzo-ingestion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: personal:3df05b2d9cf65363f39c4abd52553e2258d8a43e
Branches Tags
personal:main
personal:tracing-export-attempt
..
compare: personal:89d0d12e26039c7ab6668afc811bc6457623d181
Branches Tags
personal:main
personal:tracing-export-attempt

No commits in common. "3df05b2d9cf65363f39c4abd52553e2258d8a43e" and "89d0d12e26039c7ab6668afc811bc6457623d181" have entirely different histories.
3df05b2d9c ... 89d0d12e26

5 changed files with 116 additions and 174 deletions
Show Stats Expand all files Collapse all files

2
entity/src/transaction.rs
View File

@ -16,8 +16,6 @@ pub struct Model {
pub notes: Option<String>,
pub receipt: Option<String>,
pub description: Option<String>,
#[sea_orm(unique)]
pub identity_hash: Option<i64>,
}
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]

6
migration/src/lib.rs
View File

@ -1,7 +1,6 @@
pub use sea_orm_migration::prelude::*;
pub mod m20230904_141851_create_monzo_tables;
mod m20240529_195030_add_transaction_identity_hash;
pub struct Migrator;
@ -12,9 +11,6 @@ impl MigratorTrait for Migrator {
}
fn migrations() -> Vec<Box<dyn MigrationTrait>> {
vec![
Box::new(m20230904_141851_create_monzo_tables::Migration),
Box::new(m20240529_195030_add_transaction_identity_hash::Migration),
]
vec![Box::new(m20230904_141851_create_monzo_tables::Migration)]
}
}

34
migration/src/m20240529_195030_add_transaction_identity_hash.rs
View File

@ -1,34 +0,0 @@
use sea_orm_migration::prelude::*;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager.alter_table(
TableAlterStatement::new()
.table(Transaction::Table)
.add_column(
ColumnDef::new(Transaction::IdentityHash)
.big_integer()
.unique_key(),
).to_owned()
).await
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager.alter_table(
TableAlterStatement::new()
.table(Transaction::Table)
.drop_column(Transaction::IdentityHash)
.to_owned()
).await
}
}
#[derive(DeriveIden)]
enum Transaction {
Table,
IdentityHash,
}

11
src/ingestion/db.rs
View File

@ -7,7 +7,6 @@ use sea_orm::{
ConnectionTrait, DatabaseBackend, DatabaseTransaction, DbErr, QueryFilter, QueryTrait,
Statement,
};
use crate::ingestion::ingestion_logic::MonzoRow;
pub struct Insertion {
pub transaction: transaction::ActiveModel,
@ -53,13 +52,9 @@ async fn update_expenditures(
// trying to move failures earlier and improve reporting.
pub async fn insert(
db: &DatabaseConnection,
monzo_rows: Vec<MonzoRow>,
insertions: Vec<Insertion>,
) -> Result<Vec<String>, AppError> {
let mut new_transaction_ids = Vec::new();
let insertions = monzo_rows
.into_iter()
.map(MonzoRow::into_insertion)
.collect::<Result<Vec<_>, _>>()?;
for insertions in insertions.chunks(400) {
let tx = db.begin().await?;
@ -82,12 +77,10 @@ async fn update_transactions(
insertions: &[Insertion],
tx: &DatabaseTransaction,
) -> Result<Vec<String>, AppError> {
let insert =
transaction::Entity::insert_many(insertions.iter().map(|i| &i.transaction).cloned())
.on_conflict(
OnConflict::columns([transaction::Column::Id, transaction::Column::IdentityHash])
OnConflict::column(transaction::Column::Id)
.update_columns(transaction::Column::iter())
.to_owned(),
)

205
src/ingestion/ingestion_logic.rs
View File

@ -1,14 +1,13 @@
use std::hash::Hash;
use crate::ingestion::db::Insertion;
use anyhow::{anyhow, Context};
use anyhow::Context;
use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime};
use csv::StringRecord;
use entity::expenditure::ActiveModel;
use entity::transaction;
use num_traits::FromPrimitive;
use sea_orm::prelude::Decimal;
use sea_orm::ActiveValue::*;
use sea_orm::IntoActiveModel;
use serde_json::Value;
#[allow(dead_code)]
mod headings {
@ -30,22 +29,6 @@ mod headings {
pub const CATEGORY_SPLIT: usize = 15;
}
#[derive(Debug, Eq, PartialEq, Hash)]
pub struct MonzoRow {
category_split: Option<String>,
primary_category: String,
total_amount: Decimal,
receipt: Option<String>,
notes: Option<String>,
emoji: Option<String>,
description: Option<String>,
transaction_type: String,
title: String,
timestamp: NaiveDateTime,
transaction_id: String,
}
impl MonzoRow {
fn parse_section(monzo_transaction_id: &str, section: &str) -> anyhow::Result<ActiveModel> {
let mut components = section.split(':');
let category: String = components
@ -62,78 +45,25 @@ impl MonzoRow {
transaction_id: monzo_transaction_id.to_string(),
category,
amount,
}.into_active_model())
}
.into_active_model())
}
/// Compute a hash of this row, returning the number as an i64 to be used as a unique constraint
/// in the database.
pub fn compute_hash(&self) -> i64 {
use std::collections::hash_map::DefaultHasher;
use std::hash::Hasher;
let mut hasher = DefaultHasher::new();
self.hash(&mut hasher);
hasher.finish() as i64
}
pub fn into_insertion(self) -> Result<Insertion, anyhow::Error> {
let expenditures: Vec<_> = match self.category_split {
Some(split) if !split.is_empty() => split
.split(',')
.map(|section| Self::parse_section(&self.transaction_id, section))
.collect::<Result<Vec<_>, anyhow::Error>>()?,
_ => vec![entity::expenditure::Model {
category: self.primary_category.clone(),
amount: self.total_amount,
transaction_id: self.transaction_id.clone(),
}
.into_active_model()],
};
Ok(Insertion {
transaction: transaction::Model {
id: self.transaction_id,
transaction_type: self.transaction_type,
timestamp: self.timestamp,
title: self.title,
emoji: self.emoji,
notes: self.notes,
receipt: self.receipt,
total_amount: self.total_amount,
description: self.description,
identity_hash: Some(self.compute_hash()),
}.into_active_model(),
contained_expenditures: expenditures,
})
}
}
fn json_opt(value: &Value) -> Option<String> {
fn json_opt(value: &serde_json::Value) -> Option<String> {
match value {
Value::String(string) if string.is_empty() => None,
Value::String(string) => Some(string.to_string()),
serde_json::Value::String(string) if string.is_empty() => None,
serde_json::Value::String(string) => Some(string.to_string()),
_ => None,
}
}
fn json_required_str(value: &Value, label: &str) -> anyhow::Result<String> {
match value {
Value::String(string) if string.is_empty() => Err(anyhow!("{} is empty", label)),
Value::String(string) => Ok(string.to_string()),
_ => Err(anyhow!("{} is not a string", label)),
}
}
pub fn from_json_row(row: Vec<serde_json::Value>) -> anyhow::Result<Insertion> {
use serde_json::Value;
let monzo_transaction_id = row[headings::TRANSACTION_ID]
.as_str()
.context("No transaction id")?
.to_string();
fn parse_timestamp(date: &str, time: &str) -> anyhow::Result<NaiveDateTime> {
let date = NaiveDate::parse_from_str(date, "%Y-%m-%d")?;
let time = NaiveTime::parse_from_str(time, "%H:%M:%S")?;
Ok(date.and_time(time))
}
pub fn from_json_row(row: Vec<Value>) -> anyhow::Result<MonzoRow> {
let date = DateTime::parse_from_rfc3339(row[headings::DATE].as_str().context("No date")?)
.context("Failed to parse date")?;
@ -143,21 +73,54 @@ pub fn from_json_row(row: Vec<Value>) -> anyhow::Result<MonzoRow> {
let timestamp = date.date_naive().and_time(time);
let title = row[headings::NAME]
.as_str()
.context("No title")?
.to_string();
let monzo_transaction_type = row[headings::TYPE]
.as_str()
.context("No transaction type")?
.to_string();
let description = json_opt(&row[headings::DESCRIPTION]);
let emoji = json_opt(&row[headings::EMOJI]);
let notes = json_opt(&row[headings::NOTES_AND_TAGS]);
let receipt = json_opt(&row[headings::RECEIPT]);
let total_amount = Decimal::from_f64(row[headings::AMOUNT].as_f64().context("No amount")?)
.context("Failed to parse date")?;
Ok(MonzoRow {
transaction_id: json_required_str(&row[headings::TRANSACTION_ID], "Transaction ID")?,
title: json_required_str(&row[headings::NAME], "Title")?,
transaction_type: json_required_str(&row[headings::TYPE], "Transaction type")?,
description: json_opt(&row[headings::DESCRIPTION]),
emoji: json_opt(&row[headings::EMOJI]),
notes: json_opt(&row[headings::NOTES_AND_TAGS]),
receipt: json_opt(&row[headings::RECEIPT]),
primary_category: json_required_str(&row[headings::CATEGORY], "Primary Category")?,
category_split: json_opt(&row[headings::CATEGORY_SPLIT]),
total_amount,
timestamp,
let expenditures: Vec<_> = match row.get(headings::CATEGORY_SPLIT) {
Some(Value::String(split)) if !split.is_empty() => split
.split(',')
.map(|section| parse_section(&monzo_transaction_id, section))
.collect::<Result<Vec<_>, anyhow::Error>>()?,
_ => vec![entity::expenditure::Model {
category: row[headings::CATEGORY]
.as_str()
.context("No context")?
.to_string(),
amount: total_amount,
transaction_id: monzo_transaction_id.clone(),
}
.into_active_model()],
};
Ok(Insertion {
transaction: transaction::ActiveModel {
id: Set(monzo_transaction_id),
transaction_type: Set(monzo_transaction_type),
timestamp: Set(timestamp),
title: Set(title),
emoji: Set(emoji),
notes: Set(notes),
receipt: Set(receipt),
total_amount: Set(total_amount),
description: Set(description),
},
contained_expenditures: expenditures,
})
}
@ -168,7 +131,9 @@ fn csv_opt(s: &str) -> Option<String> {
}
}
pub fn from_csv_row(row: StringRecord) -> anyhow::Result<MonzoRow> {
pub fn from_csv_row(row: StringRecord) -> anyhow::Result<Insertion> {
let monzo_transaction_id = row[headings::TRANSACTION_ID].to_string();
let date = NaiveDate::parse_from_str(&row[headings::DATE], "%d/%m/%Y")
.context("Failed to parse date from csv")?;
@ -177,17 +142,41 @@ pub fn from_csv_row(row: StringRecord) -> anyhow::Result<MonzoRow> {
let timestamp = NaiveDateTime::new(date, time);
Ok(MonzoRow {
timestamp,
transaction_id: row[headings::TRANSACTION_ID].to_string(),
title: row[headings::NAME].to_string(),
transaction_type: row[headings::TYPE].to_string(),
description: csv_opt(&row[headings::DESCRIPTION]),
emoji: csv_opt(&row[headings::EMOJI]),
notes: csv_opt(&row[headings::NOTES_AND_TAGS]),
receipt: csv_opt(&row[headings::RECEIPT]),
total_amount: row[headings::AMOUNT].parse::<Decimal>()?,
category_split: csv_opt(&row[headings::CATEGORY_SPLIT]),
primary_category: row[headings::CATEGORY].to_string(),
let title = row[headings::NAME].to_string();
let monzo_transaction_type = row[headings::TYPE].to_string();
let description = csv_opt(&row[headings::DESCRIPTION]);
let emoji = csv_opt(&row[headings::EMOJI]);
let notes = csv_opt(&row[headings::NOTES_AND_TAGS]);
let receipt = csv_opt(&row[headings::RECEIPT]);
let total_amount = row[headings::AMOUNT].parse::<Decimal>()?;
let expenditures: Vec<_> = match row.get(headings::CATEGORY_SPLIT) {
Some(split) if !split.is_empty() => split
.split(',')
.map(|section| parse_section(&monzo_transaction_id, section))
.collect::<Result<Vec<_>, anyhow::Error>>()?,
_ => vec![entity::expenditure::Model {
transaction_id: monzo_transaction_id.clone(),
category: row[headings::CATEGORY].to_string(),
amount: total_amount,
}
.into_active_model()],
};
Ok(Insertion {
transaction: transaction::ActiveModel {
id: Set(monzo_transaction_id),
transaction_type: Set(monzo_transaction_type),
timestamp: Set(timestamp),
title: Set(title),
emoji: Set(emoji),
notes: Set(notes),
receipt: Set(receipt),
total_amount: Set(total_amount),
description: Set(description),
},
contained_expenditures: expenditures,
})
}