家計をどのように管理するのですか？

特にすべての購入がクレジットカードで行われる場合は特に、予算を正確に追跡することは常に困難でした。理由は簡単です-あなたの目の前に徐々に減少しているお金の束がありません、そしてある時点であなたはこれ以上使うことは何もないことを理解します。購入の大部分がクレジットカードで支払われている場合、残っている金額または使用されている金額を確認する唯一の方法は、インターネットバンクまたはモバイルバンクに行くか、Mintなどの金融アグリゲーターを使用することです。可能ですが、規律が必要であり、同じカードから支払うだけでなく、それを確立することは困難です。

私は、今月まだどれだけのお金があるかについての通知を毎日受け取ったら、このオプションに満足すると思いました。つまり、私は1か月の予算を設定し、何かが私の支出をカウントし、毎日予算の状態に関するレポートを送信します。

最も明らかなオプションは、銀行APIを使用するか、ヘッドレスブラウザを使用してプログラムでインターネット銀行にアクセスすることです。残念ながら、私の銀行のAPIへのアクセスは有料であり、2要素認証のためにインターネット銀行へのアクセスには問題があります。ただし、別のオプションがあります。現在、ほぼすべての銀行が各トランザクションに対してアラートを送信し、トランザクションが完了した時期、量、場所を通知しています。これは、予算を維持するために必要な情報です。それをどのように扱うかを理解することは残っています。

銀行は携帯電話とメールにアラートを送信できます。 SMSメッセージの処理が複雑であるため、携帯電話のオプションは考慮されませんでした。電子メールのオプションも非常に魅力的で、電子メールのソフトウェア処理は数十年前に行われる可能性があります。しかし、今では自宅にあるラップトップを持っているだけで、常にオンになっているわけではありません。つまり、AWSなど、クラウドのどこかで予算を自動化します。

AWSには何が必要ですか？

AWSには多くのサービスがありますが、必要なのは次の3つだけです。Eメールの送受信-SES、それらの処理-Lambda、およびDynamoDBの結果の保存。さらに、バンドルの補助的なもの-SNS、Kinesis、CloudWatch。これが唯一のメッセージ処理オプションではありません：Lambdaの代わりに、RDS（MySQL、PostgreSQL、Oracleなど）にデータを保存できるDynamoDBの代わりにEC2を使用できます。また、真珠とBerkleyDBの小さなサーバーに簡単なスクリプトを書くこともできます。

一般的に、すべての処理はどのように見えますか？取引通知が届き、日付、金額、支払い場所をデータベースに記録し、1日に1回、今月の残高を記載した通知を送信します。アーキテクチャ全体はもう少し複雑で、次のようになります。

手紙はSESに届きます。
SESは、SNSトピックに電子メールを送信します。
Lambda関数ProcessChargeは、SNSを介してレターを受信すると起動され、レターを解析して、トランザクションデータをDynamoDB Transactionsテーブルに書き込みます。
Lambda関数UpdateSummaryは、Transactionsテーブルへの書き込み後にトリガーとして機能し、Summaryテーブルの予算の現在の状態に関するデータを更新します。

これらの手順をより詳細に検討してください。

メールを受け取る

SESとも呼ばれるシンプルなメールサービスは、手紙を送受信するためのサービスです。レターを受信したら、実行するアクションを指定できます。S3にレターを保存し、Lambda関数を呼び出し、SNSなどにレターを送信します。手紙を受け取るには、ドメインをバインドする必要があります。つまり、MXドメインレコードでSESサーバーを指定する必要があります。当時は自分のドメインを持っていなかったので、これが別のAWSサービスRoute 53を使用して登録する正当な理由であると判断しました。同じ場所、Route 53でもホストしました。

ドメインをSESにバインドする場合、その検証が必要です。これを行うために、SESはDNSゾーン（MXおよびTXT）にいくつかのレコードを追加するように要求し、それらの存在を確認します。ドメインがRoute 53でホストされている場合、これはすべて自動的に行われます。ドメインが検証されたら、メール受信のルールの構成に進むことができます。私の唯一のルールは非常に単純です：ドメインのccalert @アドレスにすべての文字をSNSのccalertsトピックに送信します。

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" } }

レター処理

SNSトピックで新しいレターが公開されると、Lambda関数ProcessChargeが呼び出されます。彼女は2つのことを行う必要があります。手紙を解析し、データベースにデータを保存します。

 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

その中で、不必要な文字を削除し、正規表現を使用して、メッセージにトランザクションに関する情報が含まれているかどうかを確認し、含まれている場合はそれを断片に分割します。目的のテキストは次のとおりです。

Amazon.comでの100.00ドル（$ USD）の請求は、2017年7月19日1:55:52 PM EDTに承認されました。

残念ながら、Python標準ライブラリはタイムゾーンをほとんど認識していないため、EDT（東部夏時間）はその中には含まれていません。したがって、EDTを数値指定-0400に置き換え、メインタイムゾーンESTに対しても同じことを行います。その後、トランザクションの日付と時刻を解析し、DynamoDBでサポートされる標準ISO 8601形式に変換できます。

このメソッドは、トランザクション量、ストア名、および日付を含むハッシュテーブルを返します。このデータは、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テーブルにデータを保存し、一意のトランザクション識別子を生成します。

予算の更新

ここで、より詳細に、つまり、予算の状態がどのように監視されているのかを詳しく説明します。いくつかの値を自分で定義します。

予算-その月の予算の大きさ。
合計-月あたりの支出額。
利用可能-残り、（予算-合計）;

いつでも、これらすべての値を知りたいです。これを行うには2つの方法があります。

予算の状態を知る必要があるたびに、トランザクションが合計されて合計が取得され、利用可能=（予算-合計）になります。
新しいトランザクションが記録されるたびに、合計が更新されます。予算の状態を調べる必要がある場合、available =（budget-total）が実行されます。

どちらのアプローチにも長所と短所があり、選択はシステムの要件と制限に大きく依存します。最初のアプローチは、データを非正規化せず、トランザクション量を個別に保存するという点で優れています。一方、合計はリクエストごとに考慮される必要があります。私のボリュームではこれは問題になりませんが、私の場合はDynamoDBによる制限があります。 N個のトランザクションの合計を計算するには、N個のレコードを読み取る必要があります。つまり、N個の読み取りキャパシティーユニットを使用します。明らかに、これは非常にスケーラブルなソリューションではなく、数十のトランザクションでも困難（または高コスト）を引き起こします。

2番目のアプローチを使用すると、合計は各トランザクションの後に更新され、常に最新の状態になります。これにより、すべてのトランザクションの合計が回避されます。私の場合、このアプローチはより合理的であるように思えました。再度、さまざまな方法で実装します。

ProcessChargeの同じLambda関数で各トランザクションを記録した後、合計を更新します。
Transactionsテーブルに新しい要素を追加した後、トリガーの合計を更新します。

マルチスレッドの観点を含め、トリガーでの更新はより実用的であるため、Lambda関数UpdateSummaryを作成しました。

 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の興味深い機能が役割を果たしました。ISO8601標準に準拠し、Python自体が生成した-HH：MM形式のタイムゾーンを持つ日付/時刻を解析できません（上記のコード、parse_content（））メソッド内。したがって、必要なタイムゾーンを、彼が理解している-HHMM形式に置き換えるだけです。サードパーティのライブラリを使用して、より美しくすることが可能であり、将来のために残しておくことができました。おそらく、私のPythonの貧弱な知識がまだ影響を与えている-このプロジェクトは、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（）メソッドを使用して現在の期間の概要を読み込みます。このメソッドでは、合計を更新する必要があります。要約がまだ存在しない場合は、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（period）'が使用され、新しいセキュリティ情報が存在しない場合にのみ保存されます。したがって、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 +：請求済み'

ほとんどの場合、これはセキュリティで保護された更新には十分ですが、控えめに行動することを決定し、要約が別のスレッドで更新されなかった場合にのみ記録が行われるという条件を追加しました。

ConditionExpression = '#total =：total'、

totalはDynamoDBのキーワードであるため、DynamoDB式で使用するには、シノニムを作成する必要があります。

ExpressionAttributeNames = {

「#total」：「total」

}

これにより、トランザクションの処理と予算の更新のプロセスが完了します。

期間	予算	合計
2017-07	1000	500

予算通知

システムの最後の部分は、予算の状態の通知です。冒頭で書いたように、私は1日に1回通知を受け取るだけで十分です。ただし、各トランザクションの後、または費用/残高のしきい値の後に通知することを妨げるものは何もありません。通知メールを送信するためのアーキテクチャは非常にシンプルで、次のようになります。

CloudWatchタイマーは1日に1回実行され、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!")

まとめ

それだけですこれで、各トランザクションの後、受信レターが処理されて予算が更新され、1日に1回、予算の状態に関する通知とともにレターが送信されます。私はまだ機能を追加する計画があります。たとえば、カテゴリ別に費用を分類し、通知に最近のトランザクションのリストを含めます。何か価値のあることが起こったら、別の記事で共有します。ご質問、コメント、編集がある場合は、コメントをお待ちしています。

AWS SES、Lambda、DynamoDB（およびRoute53）を備えたシンプルな家族予算トラッカー