App Daemon MINT Api Scraper
Repo:
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 
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 …
AppDaemon.yaml
My AppDaemon configuration as follows:
system_packages:
- chromium-chromedriver
- chromium
python_packages:
- 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 Mint.com 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:
mint_scraper:
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 mint_scraper.py 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."""
self.email = 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"""
logger.info("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:
logger.info("Mint DATA is more than 4 hours old - refreshing accounts...")
raw_data = self.scrape()
else:
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."""
logger.info("Initializeing MINT Api")
mint = Mint(
email=self.email,
password=self.password,
mfa_method="soft-token",
mfa_token=self.mfa_token,
headless=True,
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,
)
logger.info("Querying MINT Api")
raw_data = mint.get_account_data(limit=5000)
logger.info("Writing mint data to disk")
self.write_data_to_disk(raw_data)
return raw_data
def _parse_mint_data(self, raw_data):
"""Prase out the mint data adding a few "extra" stuff"""
logger.info("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(
x,
sensor_suffix="balance",
object_id=f'mint {x["fiName"]} {x["name"]} balance',
state_topic=f'mint/data/{x["fiName"]}/{x["name"]}_{x["id"]}'.replace(" ", "_").lower(),
state_class="measurement",
value_template="{{value_json.value}}",
unit_of_measurement=x["currency"],
json_attributes_template="{{value_json | tojson}}",
json_attributes_topic="/mint/data/attributes",
# force_update=True,
icon=self._get_icon(x),
),
"discovery_topic_update": f'homeassistant/sensor/mint_{x["id"]}/last_update/config',
"discovery_payload_update": self._build_discovery_payload(
x,
sensor_suffix="updated",
state_topic=f'mint/data/{x["fiName"]}/{x["name"]}_{x["id"]}'.replace(" ", "_").lower(),
device_class="timestamp",
object_id=f'mint {x["fiName"]} {x["name"]} last update',
value_template="{{ value_json.metaData.lastUpdatedDate | as_datetime }}",
icon="mdi:update",
),
"discovery_topic_error": f'homeassistant/binary_sensor/mint_{x["id"]}/error/config',
"discovery_payload_error": self._build_discovery_payload(
x,
sensor_suffix="error",
entity_category="diagnostic",
state_topic=f'mint/data/{x["fiName"]}/{x["name"]}_{x["id"]}'.replace(" ", "_").lower(),
sensor_type="binary_sensor",
object_id=f'mint {x["fiName"]} {x["name"]} error',
value_template="{{value_json.isError }}",
payload_on="true",
payload_off="false",
icon="mdi:alert-circle",
),
"state_payload": x,
}
for x in raw_data
# Only get banking data
if x["type"] == "BankAccount"
]
def _build_discovery_payload(
self,
account_data,
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
account_data["fiLoginId"]
],
"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:
logger.info("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
else:
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"
else:
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:
mint_output.write(json.dumps(raw_data))
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."""
self._check_args()
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)
scraper.scrape_or_load()
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"]
my_scraper.scrape_or_load()
def callback_send_data(self, cb_args):
self.log(f"callback_send_data::CB_ARGS: {cb_args}")
my_scraper: MintScraper = cb_args["scraper"]
self.send_mqtt_data(scraper=my_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)