Как контролировать семейный бюджет? 
Мне всегда было трудно придерживаться бюджета, особенно сейчас, когда все мои покупки совершаются по кредитной карте.
Причина проста – перед глазами нет пачки денег, которая постепенно тает, и в какой-то момент ты понимаешь, что тратить уже нечего.
Если большая часть ваших покупок оплачивается кредитной картой, то единственный способ узнать, сколько денег осталось или сколько потрачено, — это зайти в интернет-банкинг или мобильный банкинг, либо воспользоваться финансовыми агрегаторами, например Mint, который вам также необходимо войти в систему и проверить свой баланс.
Это возможно, но требует дисциплины, а когда с одной карты платишь не ты один, установить сложно.
Я подумал, что буду рад варианту, если каждый день буду получать уведомление о том, сколько денег у меня еще осталось в этом месяце.
То есть я бы ставил бюджет на месяц, а что-то считало бы мои расходы и отправляло каждый день отчет о состоянии бюджета.
Самый очевидный вариант — использовать API банка или получить доступ к его интернет-банку программно с помощью какого-нибудь headless-браузера.
К сожалению, доступ к API моего банка платный, а зайти в интернет-банк проблематично из-за двухфакторной аутентификации.
Однако есть и другой вариант. Почти все банки сегодня отправляют оповещения о каждой транзакции, сообщая, когда, сколько и где была совершена транзакция.
Именно та информация, которая вам нужна для управления своим бюджетом.
Остаётся только придумать, как его обработать.
Мой банк может отправлять оповещения на мобильный телефон и электронную почту.
Вариант использования мобильного телефона не рассматривался из-за сложности обработки SMS-сообщений.
Вариант с электронной почтой выглядит очень заманчиво; программную обработку электронной почты можно было сделать десятилетия назад. Но сейчас у меня дома только ноутбук, который не всегда включен, а значит автоматизировать бюджет будем где-нибудь в облаке, например AWS. Что нам нужно в AWS? У AWS много сервисов, но нам нужны только три: для получения и отправки писем — SES, для их обработки — Lambda и для хранения результата DynamoDB. Плюс еще парочка вспомогательных для подключения — SNS, Kinesis, CloudWatch. Это не единственный вариант обработки сообщений: вместо Lambda вы можете использовать EC2, вместо DynamoDB вы можете хранить данные в RDS (MySQL, PostgreSQL, Oracle,.
), а можете даже написать простой скрипт на своем небольшом сервер с использованием Perl и BerkleyDB. Как вообще выглядит весь процесс? Приходит письмо о транзакции, мы фиксируем дату, сумму и место платежа в базе данных и раз в день отправляем письмо с остатком за данный месяц.
Вся архитектура немного сложнее и выглядит так: 
- Письмо приходит в СЭС.
- СЭС отправляет письмо в тему SNS.
- Функция ProcessCharge Lambda запускается при приходе письма через SNS, анализирует письмо и записывает данные транзакции в таблицу DynamoDB Transactions.
- Функция UpdateSummary Lambda срабатывает как триггер после записи в таблицу «Транзакции» и обновляет текущий статус бюджета в сводной таблице.
Получение письма Simple Email Service, также известный как SES, — это сервис для получения и отправки писем.
При получении письма вы можете указать, какое действие следует выполнить: сохранить письмо в S3, вызвать лямбда-функцию, отправить письмо в SNS и другие.
Для получения писем необходимо привязать свой домен, а именно указать SES-сервер в MX-записи домена.
Своего домена у меня на тот момент не было, и я решил, что это веская причина зарегистрировать его, используя другой сервис AWS — Route 53. Там же, в Route 53, я его хостил.
При привязке домена к SES его необходимо подтвердить.
Для этого SES просит добавить в зону DNS некоторые записи (MX и TXT), а затем проверяет их наличие.
Если домен размещен на Route 53, то все это делается автоматически.
Когда домен проверен, можно переходить к настройке правил получения почты.
Единственное мое правило очень простое: все письма, приходящие на адрес ccalert@ нашего домена, отправлять в тему ccalerts в соц:
Обработка писем Когда новое электронное письмо публикуется в теме SNS, вызывается функция ProcessCharge Lambda. Ей нужно сделать два действия — разобрать письмо и сохранить данные в базе данных.aws> ses describe-receipt-rule --rule-set-name "ccalerts" --rule-name "ccalert" { "Rule": { "Name": "ccalert", "Recipients": [ "ccalert@=censored=” ], "Enabled": true, "ScanEnabled": true, "Actions": [ { "SNSAction": { "TopicArn": "arn:aws:sns:us-west-2:=censored=:ccalerts", "Encoding": "UTF-8" } } ], "TlsPolicy": "Optional" } }
from __future__ import print_function
import json
import re
import uuid
from datetime import datetime
import boto3
def lambda_handler(event, context):
message = json.loads(event['Records'][0]['Sns']['Message'])
print("Processing email {}".
format(message['mail']))
content = message['content']
trn = parse_content(content)
if trn is not None:
print("Transaction: %s" % trn)
process_transaction(trn)
За парсинг отвечает метод parse_content(): def parse_content(content):
content = content.replace("=\r\n", "")
match = re.search(r'A charge of \(\$USD\) (\d+\.
\d+) at (.
+?) has been authorized on (\d+/\d+/\d+ \d+:\d+:\d+ \S{2} \S+?)\.
', content, re.M)
if match:
print("Matched %s" % match.group(0))
date = match.group(3)
# replace time zone with hour offset because Python can't parse it
date = date.replace("EDT", "-0400")
date = date.replace("EST", "-0500")
dt = datetime.strptime(date, "%m/%d/%Y %I:%M:%S %p %z")
return {'billed': match.group(1), 'merchant': match.group(2), 'datetime': dt.isoformat()}
else:
print("Didn't match")
return None
В нем мы удаляем ненужные символы и с помощью регулярного выражения проверяем, содержит ли письмо информацию о транзакции, и если есть, разбиваем его на части.
Текст поиска выглядит следующим образом:
Комиссия в размере 100 долларов США на Amazon.com была разрешена 19 июля 2017 г.К сожалению, стандартная библиотека Python знает несколько часовых поясов, и EDT (восточное летнее время) не входит в их число.в 13:55:52 по восточному времени.
Поэтому мы заменяем EDT на числовое обозначение -0400 и делаем то же самое для основного часового пояса EST. Затем мы можем проанализировать дату и время транзакции и преобразовать их в стандартный формат ISO 8601, поддерживаемый DynamoDB. Метод возвращает хеш-таблицу с суммой транзакции, названием магазина, датой и временем.
Эти данные передаются в методprocess_transaction: def process_transaction(trn):
ddb = boto3.client('dynamodb')
trn_id = uuid.uuid4().
hex
ddb.put_item(
TableName='Transactions',
Item={
'id': {'S': trn_id},
'datetime': {'S': trn['datetime']},
'merchant': {'S': trn['merchant']},
'billed': {'N': trn['billed']}
})
В нем мы храним данные в таблице Transactions, генерируя уникальный идентификатор транзакции.
Обновление бюджета Здесь хотелось бы остановиться подробнее, а именно на том, как контролируется состояние бюджета.
Определим для себя несколько значений:
- бюджет — размер бюджета на месяц;
- итого — сумма расходов в месяц;
- доступный - баланс, (бюджет - итог);
Это можно сделать двумя способами:
- Каждый раз, когда вам нужно узнать состояние бюджета, транзакции суммируются, чтобы получить итог, а затем доступность = (бюджет — итог).
- Каждый раз, когда записывается новая транзакция, общая сумма обновляется.
Когда вам нужно узнать состояние бюджета, делается доступность = (бюджет – итого).
Первый подход хорош тем, что не денормализует данные, сохраняя сумму транзакции отдельно.
С другой стороны, при нем сумму необходимо рассчитывать по каждому запросу.
Для моих томов это не будет проблемой, но в моем случае у меня есть ограничение, вызванное DynamoDB. Чтобы вычислить количество N транзакций, вам нужно прочитать N записей, что означает затрату N единиц мощности чтения.
Очевидно, что это не очень масштабируемое решение и вызовет сложность (или высокую стоимость) даже при нескольких десятках транзакций.
При использовании второго подхода сумма обновляется после каждой транзакции и всегда актуальна, что позволяет избежать суммирования всех транзакций.
В моем случае такой подход показался мне более рациональным.
Опять же, это можно реализовать по-разному:
- Обновите общую сумму после записи каждой транзакции в той же функции ProcessCharge Lambda.
- Обновить итоговую сумму в триггере после добавления нового элемента в таблицу «Транзакции».
from __future__ import print_function
from datetime import datetime
import boto3
def lambda_handler(event, context):
for record in event['Records']:
if record['eventName'] != 'INSERT':
print("Unsupported event {}".
format(record))
return
trn = record['dynamodb']['NewImage']
print(trn)
process_transaction(trn)
Нас интересуют только события о добавлении элементов в таблицу; все остальные игнорируются.
def process_transaction(trn):
period = get_period(trn)
if period is None:
return
billed = trn['billed']['N']
# update total for current period
update_total(period, billed)
print("Transaction processed")
В функцииprocess_transaction() мы вычисляем период в формате год-месяц, к которому принадлежит транзакция, и вызываем метод общего обновления.
def get_period(trn):
try:
# python cannot parse -04:00, it needs -0400
dt = trn['datetime']['S'].
replace("-04:00", "-0400") dt = dt.replace("-05:00", "-0500") dt = dt.replace("-07:00", "-0700") dt = datetime.strptime(dt, "%Y-%m-%dT%H:%M:%S%z") return dt.strftime("%Y-%m") except ValueError as err: print("Cannot parse date {}: {}".
format(trn['datetime']['S'], err))
return None
Этот код очень далек от совершенства, и в этом сыграла свою роль интересная особенность Python: он не может парсить дату/время с часовым поясом в формате -HH:MM, который соответствует стандарту ISO 8601, и который Python генерируется сам (код выше, в методе parse_content()).
Поэтому я просто заменяю нужные мне часовые пояса на тот формат ЧЧММ, который он понимает. Можно было использовать стороннюю библиотеку и сделать покрасивее, оставлю это на будущее.
Возможно, сказывается и мое плохое знание Python — этот проект — мой первый опыт разработки на нем.
Всего обновлений: def update_total(period, billed):
ddb = boto3.client('dynamodb')
response = load_summary(ddb, period)
print("Summary: {}".
format(response))
if 'Item' not in response:
create_summary(ddb, period, billed)
else:
total = response['Item']['total']['N']
update_summary(ddb, period, total, billed)
В этом методе мы с помощью метода load_summary() загружаем сводку (Summary) за текущий период, итоговую сумму в которой нам необходимо обновить.
Если сводка еще не существует, мы создаем ее в методе create_summary(); если он существует, мы обновляем его в update_summary().
def load_summary(ddb, period):
print("Loading summary for period {}".
format(period))
return ddb.get_item(
TableName = 'Summary',
Key = {
'period': {'S': period}
},
ConsistentRead = True
)
Поскольку обновление сводки может выполняться из нескольких потоков, мы используем последовательное чтение, которое более затратно, но гарантирует, что мы получим последнее записанное значение.
def create_summary(ddb, period, total):
print("Creating summary for period {} with total {}".
format(period, total))
ddb.put_item(
TableName = 'Summary',
Item = {
'period': {'S': period},
'total': {'N': total},
'budget': {'N': "0"}
},
ConditionExpression = 'attribute_not_exists(period)'
)
При создании новой сводки по той же причине возможной записи из нескольких потоков используется условная запись, ConditionExpression = 'attribute_not_exists(период)' , который сохранит новое резюме, только если оно не существует. Итак, если кому-то удалось создать сводку между моментом, когда мы пытались загрузить ее в load_summary() и ее там не было, и когда мы попытались создать ее в create_summary(), наш вызов put_item() завершится ошибкой с ошибкой исключение, и вся функция Lambda будет перезапущена.
def update_summary(ddb, period, total, billed):
print("Updating summary for period {} with total {} for billed {}".
format(period, total, billed))
ddb.update_item(
TableName = 'Summary',
Key = {
'period': {'S': period}
},
UpdateExpression = 'SET #total = #total + :billed',
ConditionExpression = '#total = :total',
ExpressionAttributeValues = {
':billed': {'N': billed},
':total': {'N': total}
},
# total is a reserved word so we create an alias #total to use it in expression
ExpressionAttributeNames = {
'#total': 'total'
}
)
Обновление общего значения в сводке выполняется внутри DynamoDB:
UpdateExpression = 'SET #total = #total + :billed'Скорее всего, этого достаточно для безопасного обновления, но я решил действовать консервативно и добавил условие, что запись должна происходить только в том случае, если сводка не обновлялась в другом потоке и она все еще содержит то значение, которое у нас есть:
ConditionExpression = '#total = :total',Поскольку итоговое значение — это ключевое слово для DynamoDB, для его использования в выражениях DynamoDB необходимо создать синоним:
ExpressionAttributeNames = { '#total': 'всего' }На этом процесс обработки транзакций и обновления бюджета завершен:
| период | бюджет | общий |
|---|---|---|
| 2017-07 | 1000 | 500 |
Как я писал в самом начале, мне достаточно получать уведомление раз в сутки, что я и реализовал.
Однако ничто не мешает вам уведомлять после каждой транзакции или после достижения некоторых пороговых значений расходов/баланса.
Архитектура отправки уведомления по электронной почте довольно проста и выглядит следующим образом: 
- Таймер CloudWatch срабатывает один раз в день и вызывает функцию DailyNotification Lambda.
- DailyNotification загружает данные из сводной таблицы DynamoDB и вызывает SES для отправки электронного письма.
from __future__ import print_function
from datetime import date
import boto3
def lambda_handler(event, context):
ddb = boto3.client('dynamodb')
current_date = date.today()
print("Preparing daily notification for {}".
format(current_date.isoformat()))
period = current_date.strftime("%Y-%m")
response = load_summary(ddb, period)
print("Summary: {}".
format(response))
if 'Item' not in response:
print("No summary available for period {}".
format(period))
return
summary = response['Item']
total = summary['total']['N']
budget = summary['budget']['N']
send_email(total, budget)
def load_summary(ddb, period):
print("Loading summary for period {}".
format(period))
return ddb.get_item(
TableName = 'Summary',
Key = {
'period': {'S': period}
},
ConsistentRead = True
)
Сначала пытаемся загрузить сводку за текущий период, а если ее нет, то выходим.
Если есть, готовим и отправляем письмо: def send_email(total, budget):
sender = "Our Budget <ccalert@==censored==>"
recipients = [“==censored==“]
charset = "UTF-8"
available = float(budget) - float(total)
today = date.today().
strftime("%Y-%m-%d") message = ''' As of {0}, available funds are ${1:.
2f}.
This month budget is ${2:.
2f}, spendings so far totals ${3:.
2f}.
More details coming soon!''' subject = "How are we doing?" textbody = message.format(today, float(available), float(budget), float(total)) print("Sending email: {}".
format(textbody)) client = boto3.client('ses', region_name = 'us-west-2') try: response = client.send_email( Destination = { 'ToAddresses': recipients }, Message = { 'Body': { 'Text': { 'Charset': charset, 'Data': textbody, }, }, 'Subject': { 'Charset': charset, 'Data': subject, }, }, Source = sender, ) # Display an error if something goes wrong. except Exception as e: print("Couldn't send email: {}".
format(e))
else:
print("Email sent!")
Нижняя граница Вот и все.
Теперь после каждой транзакции входящее письмо обрабатывается и бюджет обновляется, а письмо отправляется раз в день с уведомлением о состоянии бюджета.
У меня еще есть планы добавить функционал, например, классификацию расходов по категориям и включение в уведомление списка последних транзакций; если я получу что-то стоящее, я поделюсь этим в другой статье.
Если у вас есть какие-либо вопросы, комментарии или исправления, пожалуйста, оставляйте их в комментариях.
Теги: #aws #aws лямбда #dynamodb #NoSQL #aws ses #бюджет #программирование #NoSQL #Amazon Web Services
-
Руководство По Покупке Ноутбука – Часть 3
19 Oct, 24 -
Архитектурный 3D-Рендеринг
19 Oct, 24 -
Еженедельный Подкаст «Умпутун» — Серия 233
19 Oct, 24
