MINT Financial Scraper + MQTT

App Daemon MINT Api Scraper


So for some reason I thought it would be nice to get my account balances into Home Assistant rather than having to log into lots of banks. The thought process is I’ll eventually write some automations if one of my checking accounts gets too low so I can move money from a high yield savings account into my everyday checking account or something like that. This code is pretty rough at the moment and ultimately I think I’ll end up moving it into HACS or something - but it does seem to work. So if anybody wants to play around I thought I’d post it here.

Looking for feedback and any help if somebody wants to toss the effort in.

Anyways without giving out too much of my personal information :slight_smile:

When its running via MQTT Discovery it will create a set of sensors with banking information - when it was lasted updated - and an error sensor in case mint is reporting something wrong.


In any case … the info is below and as follows …


My AppDaemon configuration as follows:

  - chromium-chromedriver
  - chromium
  - mintapi
init_commands: []

App Secrets

In your secrets.yaml you’ll need to add the following values:

mint_mfa_token: <SEED_FOR_MFA_TOKEN>
mint_password: <PASSWORD_FOR_MINT>
mint_email: <EMAIL_FOR_MINT>

Details about the mint_mfa_token are available in the backing lib: GitHub - mintapi/mintapi: an unofficial screen-scraping "API" for but in summary - when you add a MFA token to your Mint account it will print out a “seed” you can use to manually setup an authenticator - and thats the value you want to capture.

MQTT Plugin

I’ve only tested this by disabling the HASS plugin and leaving just MQTT enabled. If anybody has help about how to get my code to work in multi-plugin mode let me know!

So this is my appdaemon.yaml file

App Definition

In your apps.yaml add the following:

  module: mint_scraper
  class: MintScrapperApp
  mint_mfa_token: !secret mint_mfa_token
  mint_password: !secret mint_password
  mint_email: !secret mint_email

App Code

Create a file called in /appdaemon/apps with the following “stuff”:

from mintapi.api import Mint
import mqttapi as mqtt
import json
import logging
import datetime
from dateutil.parser import isoparse
import time
from os.path import exists

# logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("mintapi")

class MintScraper:
    """Define a mint scraper wrapper"""

    def __init__(self, email: str, password: str, mfa_token: str) -> None:
        """Initialize the mint acocunt scraper.""" = email
        self.password = password
        self.mfa_token = mfa_token
        self.mint_data = []

    def load_raw_scrape_data(self, file_name):
        """Load data and output the data age""""Opening Mint data: %s", file_name)
        with open(file_name) as file:
            raw_data = json.load(file)

        return raw_data

    def scrape_or_load(self):
        """Decides whether to scrape or load the data from the data file."""

        if exists("mint.json"):
            raw_data = self.load_raw_scrape_data("mint.json")

            # Calculate the age of the raw_data
            max_time = 0
            for entry in raw_data:
                timestamp: float = isoparse(entry["metaData"]["lastUpdatedDate"]).timestamp()
                max_time = max(max_time, timestamp)

            age = time.time() - max_time
            age_in_hours = divmod(age, 3600)[0]

            if age_in_hours > 4:
      "Mint DATA is more than 4 hours old - refreshing accounts...")
                raw_data = self.scrape()
            raw_data = self.scrape()
        # Parse raw data
        self.mint_data = self._parse_mint_data(raw_data=raw_data)

    def scrape(self) -> list[any]:
        """Scrape MINT Accounts and return the results.""""Initializeing MINT Api")
        mint = Mint(
            wait_for_sync=False,  # not options.no_wait_for_sync,
            wait_for_sync_timeout=300,  # options.wait_for_sync_timeout,
            fail_if_stale=False,  # options.fail_if_stale,
            use_chromedriver_on_path=True,  # options.use_chromedriver_on_path,
            beta=False,  # options.beta,
        )"Querying MINT Api")
        raw_data = mint.get_account_data(limit=5000)"Writing mint data to disk")
        return raw_data

    def _parse_mint_data(self, raw_data):
        """Prase out the mint data adding a few "extra" stuff""""Parsing MINT data")
        return [
                "state_topic": f'mint/data/{x["fiName"]}/{x["name"]}_{x["id"]}'.replace(" ", "_").lower(),
                "discovery_topic_balance": f'homeassistant/sensor/mint_{x["id"]}/account_balance/config',
                "discovery_payload_balance": self._build_discovery_payload(
                    object_id=f'mint {x["fiName"]} {x["name"]} balance',
                    state_topic=f'mint/data/{x["fiName"]}/{x["name"]}_{x["id"]}'.replace(" ", "_").lower(),
                    json_attributes_template="{{value_json | tojson}}",
                    # force_update=True,
                "discovery_topic_update": f'homeassistant/sensor/mint_{x["id"]}/last_update/config',
                "discovery_payload_update": self._build_discovery_payload(
                    state_topic=f'mint/data/{x["fiName"]}/{x["name"]}_{x["id"]}'.replace(" ", "_").lower(),
                    object_id=f'mint {x["fiName"]} {x["name"]} last update',
                    value_template="{{ value_json.metaData.lastUpdatedDate | as_datetime }}",
                "discovery_topic_error": f'homeassistant/binary_sensor/mint_{x["id"]}/error/config',
                "discovery_payload_error": self._build_discovery_payload(
                    state_topic=f'mint/data/{x["fiName"]}/{x["name"]}_{x["id"]}'.replace(" ", "_").lower(),
                    object_id=f'mint {x["fiName"]} {x["name"]} error',
                    value_template="{{value_json.isError }}",
                "state_payload": x,
            for x in raw_data
            # Only get banking data
            if x["type"] == "BankAccount"

    def _build_discovery_payload(
        sensor_suffix: str,
        state_topic: str = "",
        entity_category: str | None = None,
        state_class: str | None = None,
        sensor_type: str = "sensor",
        object_id: str | None = None,
        expire_after: str | None = None,
        force_update: bool = False,
        payload_on: str | bool | None = None,
        payload_off: str | bool | None = None,
        device_class: str | None = None,
        unit_of_measurement: str | None = None,
        value_template: str = "",
        json_attributes_template: str | None = None,
        json_attributes_topic: str | None = None,
        icon: str | None = None,
        unique_id = f'{account_data["id"]}_{sensor_suffix}'.replace(" ", "_")

        discovery_payload = {
            "device": {
                "identifiers": [
                    # Bank name
                "manufacturer": "Mint Scraper",
                "model": "Bank Account",
                "name": f"{account_data['fiName']}",
                "sw_version": "",
            "name": account_data["name"].capitalize() + " " + sensor_suffix,
            "unique_id": unique_id,
            "state_topic": state_topic,
            "value_template": value_template,
            "force_update": force_update,

        # set things if they exist:

        if unit_of_measurement:
            discovery_payload["unit_of_measurement"] = unit_of_measurement

        if icon:
  "Processing icon %s", icon)
            discovery_payload["icon"] = icon

        if payload_off:
            discovery_payload["payload_off"] = payload_off

        if entity_category:
            discovery_payload["entity_category"] = entity_category

        if object_id:
            discovery_payload["object_id"] = object_id

        if state_class:
            discovery_payload["state_class"] = state_class
        if expire_after:
            discovery_payload["expire_after"] = expire_after
        if payload_on:
            discovery_payload["payload_on"] = payload_on
        if device_class:
            discovery_payload["device_class"] = device_class

        if json_attributes_template:
            if json_attributes_topic:
                discovery_payload["json_attributes_topic"] = json_attributes_topic
                discovery_payload["json_attributes_topic"] = state_topic

            discovery_payload["json_attributes_template"] = json_attributes_template

        # Return data
        return discovery_payload

    def _get_icon(self, account_type: str) -> str:
        if account_type["bankAccountType"] == "CHECKING":
            return "mdi:checkbook"
            return "mdi:piggy-bank"

    def write_data_to_disk(self, raw_data):
        """Write the current set of data to disk."""

        with open("mint.json", "w") as mint_output:

class MintScrapperApp(mqtt.Mqtt):
    def _check_args(self) -> bool:
        """Verify the right arguments are there"""

        ret = True
        self.log("-- Verifying configuration data")

        if "mint_mfa_token" not in self.args:
            self.log("Missing argument: `mint_mfa_token'")
            ret = False
        if "mint_password" not in self.args:
            self.log("Missing argument: `mint_password'")
            ret = False
        if "mint_email" not in self.args:
            self.log("Missing argument: `mint_email'")
            ret = False

        if not ret:
            self.log("-- Invalid configuration data")

        return ret

    def initialize(self):
        """Iniitalize the Scraping App."""
        MFA_TOKEN = self.args["mint_mfa_token"]
        MINT_PASSWORD = self.args["mint_password"]
        MINT_EMAIL = self.args["mint_email"]

        self.log("-- Initializing Mint API")
        scraper = MintScraper(email=MINT_EMAIL, password=MINT_PASSWORD, mfa_token=MFA_TOKEN)

        self.log("-- Registering Callback: callback_get_data")
        self.run_hourly(self.callback_get_data, datetime.time(0, 0, 0), scraper=scraper)

        self.log("::mintapi... Detected %d accounts", len(scraper.mint_data))

        self.log("-- Initializing MQTT")
        self.mqtt = self.get_plugin_api("MQTT")

        if self.mqtt.is_client_connected():
            self.log("-- Registering Callback: callback_send_data")
            self.run_minutely(self.callback_send_data, datetime.time(0, 0, 0), scraper=scraper)

    def callback_get_data(self, cb_args):
        self.log(f"callback_get_data::CB_ARGS: {cb_args}")
        my_scraper: MintScraper = cb_args["scraper"]

    def callback_send_data(self, cb_args):
        self.log(f"callback_send_data::CB_ARGS: {cb_args}")
        my_scraper: MintScraper = cb_args["scraper"]

    def _convert_bool_to_string(self, obj):
        """Converts json bool values into string representations"""
        if isinstance(obj, bool):
            return str(obj).lower()
        if isinstance(obj, (list, tuple)):
            return [self._convert_bool_to_string(item) for item in obj]
        if isinstance(obj, dict):
            return {
                self._convert_bool_to_string(key): self._convert_bool_to_string(value) for key, value in obj.items()
        return obj

    def send_mqtt_data(self, scraper: MintScraper):
        self.log("send_mqtt_data::Sending discovery packets via MQTT")

        for entry in scraper.mint_data:
            # Process discovery messages and topics
            for item in ["balance", "update", "error"]:
                topic = entry[f"discovery_topic_{item}"]
                payload = json.dumps(self._convert_bool_to_string(entry[(f"discovery_payload_{item}")]))
                self.mqtt_publish(topic, payload)

            # Process state data
            topic = entry["state_topic"]
            payload = json.dumps(self._convert_bool_to_string(entry["state_payload"]))

            self.log("send_mqtt_data::Publishing State data")
            self.mqtt_publish(topic, payload)