ksiemgowy package

Submodules

ksiemgowy.current_report_builder module

Ksiemgowy’s reporting module. Generates a dictionary that details information about organization’s current financial status.

class ksiemgowy.current_report_builder.T_CURRENT_REPORT(*args, **kwargs)

Bases: dict

balance_so_far: float

balances_by_account_labels: Dict[str, float]

dues_last_updated: str

dues_num_subscribers: int

dues_total_lastmonth: float

extra_monthly_reservations: int

monthly: ksiemgowy.current_report_builder.T_MONTHLY_REPORT

class ksiemgowy.current_report_builder.T_MONTHLY_REPORT(*args, **kwargs)

Bases: dict

Bilans: Dict[str, Dict[str, float]]

Przychody: Dict[str, Dict[str, float]]

Saldo: Dict[str, Dict[str, float]]

Wydatki: Dict[str, Dict[str, float]]

ksiemgowy.current_report_builder.apply_d33tah_dues(monthly_income: Dict[str, Dict[str, float]], balances_by_account_labels: Dict[str, float], first_200pln_d33tah_due_date: datetime.datetime, last_200pln_d33tah_due_date: datetime.datetime) → None

Applies dues paid by d33tah to monthly_income. This is here because the banking system didn’t notify about self-transfers, so they needed to be added explicitly.

ksiemgowy.current_report_builder.apply_expenses(expenses: Iterable[ksiemgowy.mbankmail.MbankAction], balances_by_account_labels: Dict[str, float], account_labels: Dict[str, str], categories: List[ksiemgowy.config.CategoryCriteria]) → Tuple[datetime.datetime, Dict[str, Dict[str, float]]]

Apply all expenses both to balances_by_account_labels and monthly_expenses. Returns newly built monthly_expenses.

ksiemgowy.current_report_builder.apply_global_corrections(corrections_by_label: Dict[str, float], balances_by_account_labels: Dict[str, float]) → None

Apply a specified set of corrections to balances_by_account_labels.

ksiemgowy.current_report_builder.apply_monthly_corrections(monthly_income_corrections: Dict[str, Dict[str, float]], monthly_expense_corrections: Dict[str, Dict[str, float]], monthly_income: Dict[str, Dict[str, float]], monthly_expenses: Dict[str, Dict[str, float]]) → None

Apply a specified set of corrections to monthly_income and monthly_expenses.

ksiemgowy.current_report_builder.apply_positive_transfers(now: datetime.datetime, last_updated: datetime.datetime, positive_actions: Iterable[ksiemgowy.mbankmail.MbankAction], balances_by_account_labels: Dict[str, float], account_labels: Dict[str, str]) → Tuple[float, int, datetime.datetime, Dict[str, Dict[str, float]]]

Apply all positive transfers both to balances_by_account_labels and monthly_income. Returns newly built monthly_expenses, as well as total money raised and current information about the number of members who paid dues and the datestamp of due last paid.

ksiemgowy.current_report_builder.build_extra_monthly_reservations(now: datetime.datetime, extra_monthly_reservations_started_date: datetime.datetime) → int

Returns all extra monthly reservations collected until now. On 24 November 2020, we agreed that we’ll be continuing to increase our reserves by 200 PLN each month.

ksiemgowy.current_report_builder.build_monthly_balance(months: Set[str], monthly_income: Dict[str, Dict[str, Union[float, int]]], monthly_expenses: Dict[str, Dict[str, float]]) → Dict[str, Dict[str, Union[float, int]]]

Calculates balances for each of the months - the final amount of money on all of our accounts at the end of the month.

ksiemgowy.current_report_builder.build_monthly_final_balance(months: Set[str], monthly_income: Dict[str, Dict[str, float]], monthly_expenses: Dict[str, Dict[str, float]]) → Tuple[Dict[str, Dict[str, float]], float]

Calculates monthly final balances, given all of the actions - an amount that specifies whether we accumulated more than we spent, or otherwise.

ksiemgowy.current_report_builder.determine_category(action: ksiemgowy.mbankmail.MbankAction, categories: List[ksiemgowy.config.CategoryCriteria]) → str

Given an incoming action, determine what label to assign to it.

ksiemgowy.current_report_builder.get_current_report(now: datetime.datetime, expenses: Iterable[ksiemgowy.mbankmail.MbankAction], positive_actions: Iterable[ksiemgowy.mbankmail.MbankAction], report_builder_config: ksiemgowy.config.ReportBuilderConfig) → ksiemgowy.current_report_builder.T_CURRENT_REPORT

Module’s entry point. Given time, expenses, income and corrections, generates a monthly summary of actions that happened on the accounts.

ksiemgowy.homepage_updater module

Downloads the most current uploaded financial report and compares it to the one that can be generated based on the latest data. If they differ, uploads a new one to the website.

ksiemgowy.homepage_updater.deserialize(serialized_string: str) → ksiemgowy.current_report_builder.T_CURRENT_REPORT

De-serializes a given string, returning it in a format generated and expected by Ksiemgowy.

ksiemgowy.homepage_updater.do_states_differ(remote_state: Optional[ksiemgowy.current_report_builder.T_CURRENT_REPORT], current_report: ksiemgowy.current_report_builder.T_CURRENT_REPORT) → bool

Returns true if current report differs from the remote one. Does not compare keys which are not in current report.

ksiemgowy.homepage_updater.get_remote_state_dues(remote_state_path: pathlib.Path) → Optional[ksiemgowy.current_report_builder.T_CURRENT_REPORT]

Reads remote state from a specified file, then de-serializes it and returns in a format that’s similar to the one generated by Ksiemgowy.

ksiemgowy.homepage_updater.git_cloned(deploy_key_path: str, git_url: str) → Generator[Dict[str, str], None, None]

Sets up SSH authorization to a Git repository using SSH deploy keys, then clones the repository and sets up git identify for it. Acts as a context manager - once it exits, the source tree gets deleted.

Yields enviroment variables needed to be able to commit and push to git.

ksiemgowy.homepage_updater.is_local_state_newer(remote_state: ksiemgowy.current_report_builder.T_CURRENT_REPORT, current_report: ksiemgowy.current_report_builder.T_CURRENT_REPORT) → bool

Compares remote state and current report, using time criteria.

ksiemgowy.homepage_updater.maybe_update(database: ksiemgowy.models.KsiemgowyDB, homepage_updater_config: ksiemgowy.config.HomepageUpdaterConfig, report_builder_config: ksiemgowy.config.ReportBuilderConfig) → None

Submodule’s entry point. Checks out the repository, operates on it and cleans up the checked out tree.

ksiemgowy.homepage_updater.maybe_update_dues(database: ksiemgowy.models.KsiemgowyDB, git_env: Dict[str, str], dues_file_path: str, report_builder_config: ksiemgowy.config.ReportBuilderConfig) → ksiemgowy.current_report_builder.T_CURRENT_REPORT

Generates the current report, retrieves the one that’s accessible online and if the current one is later, updates the remote state.

ksiemgowy.homepage_updater.serialize(dict_to_serialize: ksiemgowy.current_report_builder.T_CURRENT_REPORT) → str

Serializes a given object, returning it in a format generated and expected by Ksiemgowy.

ksiemgowy.homepage_updater.set_up_git_identity(username: str, email: str, cwd: str) → None

Sets up git identity required in order to be able to commit: an e-mail and user name.

ksiemgowy.homepage_updater.ssh_agent_import_key_and_build_env_and_setup_git(deploy_key_path: str) → Dict[str, str]

Creates an SSH agent session and adds a specified key to it. Returns environment variables needed to use the SSH agent.

ksiemgowy.homepage_updater.update_git_remote_state(filepath: pathlib.Path, new_state: ksiemgowy.current_report_builder.T_CURRENT_REPORT, env: Dict[str, str]) → None

Updates remote state by writing to a file, creating a git commit and pushing it to the repository.

ksiemgowy.homepage_updater.upload_to_graphite(graphite_host: str, graphite_port: int, current_report: ksiemgowy.current_report_builder.T_CURRENT_REPORT) → None

Uploads metrics to Graphite server.

ksiemgowy.homepage_updater.upload_value_to_graphite(host: Tuple[str, int], metric: str, value: str) → None

Uploads a single metric to a Graphite server.

ksiemgowy.mbankmail module

Parses mbank daily notification e-mails.

class ksiemgowy.mbankmail.MbankAction(sender_acc_no: str, recipient_acc_no: str, amount_pln: float, in_person: str, in_desc: str, balance: float, timestamp: str, action_type: str)

Bases: object

A container for all transfers, positive or negative.

action_type: str

amount_pln: float

anonymized(mbank_anonymization_key: bytes) → ksiemgowy.mbankmail.MbankAction

Anonymizes all potentially sensitive fields using mbank_anonymization_key as cryptographic pepper.

asdict(*, dict_factory=<class 'dict'>)

Return the fields of a dataclass instance as a new dictionary mapping field names to field values.

Example usage:

@dataclass class C:

x: int y: int

c = C(1, 2) assert asdict(c) == {‘x’: 1, ‘y’: 2}

If given, ‘dict_factory’ will be used instead of built-in dict. The function applies recursively to field values that are dataclass instances. This will also look into built-in containers: tuples, lists, and dicts.

balance: float

get_timestamp() → datetime.datetime

Returns timestamp. This is there because we currently store the timestamp as string for rather random reasons.

sender_acc_no: str

in_desc: str

in_person: str

recipient_acc_no: str

timestamp: str

ksiemgowy.mbankmail.anonymize(hashed_string: str, mbank_anonymization_key: bytes) → str

Anonymizes an input string using mbank_anonymization_key as cryptographic pepper.

ksiemgowy.mbankmail.main(input_fpath: str, mode: str, loglevel: str) → None

Entry point for the submodule, used for diagnostics. Reads data from input_fpath, then runs either parse_mbank_html or parse_mbank_email, depending on the mode.

ksiemgowy.mbankmail.parse_args() → Dict[str, str]

Parses command-line arguments and returns them in a form usable as **kwargs.

ksiemgowy.mbankmail.parse_mbank_email(msg: email.message.Message) → Dict[str, List[ksiemgowy.mbankmail.MbankAction]]

Finds attachment with mBank account update in an .eml mBank email, then behaves like parse_mbank_html.

ksiemgowy.mbankmail.parse_mbank_html(mbank_html: bytes) → Dict[str, List[ksiemgowy.mbankmail.MbankAction]]

Parses mBank .htm attachment file and generates a list of actions that were derived from it.

ksiemgowy.models module

This module describes data structures used in ksiemgowy.

class ksiemgowy.models.KsiemgowyDB(database_uri: str)

Bases: object

A class that groups together all models that describe the state of ksiemgowy.

add_expense(bank_action: ksiemgowy.mbankmail.MbankAction) → None

Adds an expense to the database.

add_positive_transfer(positive_action: ksiemgowy.mbankmail.MbankAction) → None

Adds a positive transfer to the database.

get_email_for_sender_acc_no(sender_acc_no: str) → Optional[str]

Returns an e-mail address for a given sender_acc_no.

get_potentially_overdue_accounts(now: datetime.datetime) → Dict[str, str]

Returns a list of accounts that might be overdue and can be notified.

list_expenses() → Iterator[ksiemgowy.mbankmail.MbankAction]

Returns a generator that lists all expenses transfers that were observed so far.

list_positive_transfers() → Iterator[ksiemgowy.mbankmail.MbankAction]

Returns a generator that lists all positive transfers that were observed so far.

mark_imap_id_already_handled(imap_id: str) → None

Marks a given IMAP ID as already processed by ksiemgowy.

postpone_next_notification(sender_acc_no: str, now: datetime.datetime) → None

Postpone next overdue notification for an account with a given sender_acc_no.

was_imap_id_already_handled(imap_id: str) → bool

Tells whether a given IMAP ID was already processed by ksiemgowy.

Module contents