Insert PiTaPa CSV to GnuCash

insert-pitapa.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import argparse
import csv
import sqlite3
import sys
import datetime
import pytz
import collections
import re
import uuid

DATE = re.compile(r'(\d\d\d\d)/(\d\d)/(\d\d)')

def _main():
    parser = argparse.ArgumentParser(description="Search Credit Info")
    parser.add_argument('gnucash')
    parser.add_argument('pitapa', type=argparse.FileType('r', encoding='shift-jis'))
    options = parser.parse_args()

    gnucash = sqlite3.connect(options.gnucash)
    pitapa_reader = csv.reader(options.pitapa)
    mode = None

    header = next(pitapa_reader)
    if header != ['ご利用日', '入場時刻', '出場時刻', 'ご利用内容', 'ご利用額（円）', '備考']:
        print('This is not pitapa CSV')
        exit(1)
    

    entries = collections.defaultdict(list)

    postpay_month = None
    postpay_amount = 0
    discount_amount = 0
    
    for row in pitapa_reader:
        if len(row) < 4: continue
        if row[3] == '＜交通ご利用明細＞':
            mode = 'CREDIT'
        elif row[3] == '＜その他＞':
            mode = 'OTHER'
        elif row[3] == '≪プリペイドご利用≫':
            mode = 'PREPAID'

        date = DATE.match(row[0])
        if date:
           entries[mode].append(row)
           postpay_month = date

        if row[3] == '＠ポストペイ対象普通運賃合計':
            postpay_amount += int(row[4])
        if row[3] == '＠割引額（利用額割引）':
            discount_amount += int(row[4])
            

    #print(entries)

    prepaid_account, credit_account = get_pitapa_account(gnucash)
    transportation_account = get_transportation_account(gnucash)
    cash_account = get_cash_account(gnucash)
    credit2_account = get_named_account(gnucash, '三井住友カード')
    other_income_account = get_named_account(gnucash, 'その他の収入')
    jpy = get_jpy(gnucash)

    print(transportation_account)
    print(cash_account)
    print(prepaid_account)
    print(credit_account)
    print(jpy)

    for one in entries['CREDIT']:
        day = datetime.datetime.strptime(one[0], '%Y/%m/%d')
        description = one[3] + ' ' + one[1] + '-' + one[2]
        yen = int(one[4])
        insert_entry(gnucash, credit_account, transportation_account, jpy, description, yen, day, 1)

    for one in entries['OTHER']:
        if not one[3].startswith('オートチャージ'): continue
        day = datetime.datetime.strptime(one[0], '%Y/%m/%d')
        description = one[3] + ' ' + one[1] + '-' + one[2]
        yen = int(one[4])
        insert_entry(gnucash, credit2_account, prepaid_account, jpy, description, yen, day, 1)

    for one in entries['PREPAID']:
        day = datetime.datetime.strptime(one[0], '%Y/%m/%d')
        description = one[3] + ' ' + one[1] + '-' + one[2]
        yen = -int(one[4])

        if yen > 0:
            insert_entry(gnucash, prepaid_account, transportation_account, jpy, description, yen, day, 1)
        else:
            insert_entry(gnucash, cash_account, prepaid_account, jpy, description, -yen, day, 1)


    paydate = datetime.date(int(postpay_month.group(1)), int(postpay_month.group(2)) + 1, 1) - datetime.timedelta(days = 1)
    print(paydate)
    if postpay_amount > 0:
        insert_entry(gnucash, credit2_account, credit_account, jpy, '交通ご利用', postpay_amount, paydate, 1)
    if discount_amount < 0:
        insert_entry(gnucash, transportation_account, credit2_account, jpy, '交通ご利用割引', -discount_amount, paydate, 1)
    
    gnucash.commit()


def insert_entry(gnucash, from_account, to_account, currency, description, yen, date, count):
    cur = gnucash.execute('SELECT guid FROM transactions WHERE post_date = ? AND description = ?', (date.strftime('%Y%m%d')+'105900', description))
    transactions = list(cur)
    #if transactions:
    #    print('Already inserted', transactions)

    count -= len(transactions)
    

    for i in range(count):
        now = datetime.datetime.utcnow()
        transaction_guid = uuid.uuid4().hex
        split1_guid = uuid.uuid4().hex
        split2_guid = uuid.uuid4().hex
    
        gnucash.execute('INSERT INTO transactions(guid,currency_guid,num,post_date,enter_date,description) VALUES (?,?,?,?,?,?)',
                        (transaction_guid, currency[0], '', date.strftime('%Y%m%d')+'105900', now.strftime('%Y%m%d%H%M%S'), description))
        gnucash.execute('INSERT INTO splits(guid,tx_guid,account_guid,memo,"action",reconcile_state,reconcile_date,value_num,value_denom,quantity_num,quantity_denom,lot_guid) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)', (split1_guid, transaction_guid, to_account[0], '', '', 'n', now.strftime('%Y%m%d%H%M%S'), yen, 1, yen, 1, None))
        gnucash.execute('INSERT INTO splits(guid,tx_guid,account_guid,memo,"action",reconcile_state,reconcile_date,value_num,value_denom,quantity_num,quantity_denom,lot_guid) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)', (split2_guid, transaction_guid, from_account[0], '', '', 'n', now.strftime('%Y%m%d%H%M%S'), -yen, 1, -yen, 1, None))
        gnucash.execute('INSERT INTO slots(obj_guid,name,slot_type,int64_val,string_val,double_val,timespec_val,guid_val,numeric_val_num,numeric_val_denom,gdate_val) VALUES (?,?,?,?,?,?,?,?,?,?,?)', (transaction_guid, 'date-posted', 10, 0, None, 0, None, None, 0, 1, now.strftime('%Y%m%d')))
        gnucash.execute('INSERT INTO slots(obj_guid,name,slot_type,int64_val,string_val,double_val,timespec_val,guid_val,numeric_val_num,numeric_val_denom,gdate_val) VALUES (?,?,?,?,?,?,?,?,?,?,?)', (transaction_guid, 'notes', 4, 0, '', 0, None, None, 0, 1, None))
        print('insert', from_account[1], to_account[1], currency[1], description, yen, date, count)


def get_pitapa_account(gnucash):
    cur = gnucash.execute('SELECT guid, name, account_type FROM accounts WHERE name LIKE \'PiTaPa%\'')
    account_list = [x for x in cur]
    prepaid_account = [x for x in account_list if x[2] == 'ASSET']
    credit_account = [x for x in account_list if x[2] == 'CREDIT']

    if len(prepaid_account) != 1:
        print('invalid prepaid pitapa account', prepaid_account)
        exit(1)
    else:
        prepaid_account = prepaid_account[0]

    if len(credit_account) != 1:
        print('invalid credit pitapa account', credit_account)
        exit(1)
    else:
        credit_account = credit_account[0]

    return (prepaid_account, credit_account)


def get_transportation_account(gnucash):
    return get_named_account(gnucash, '交通機関')


def get_cash_account(gnucash):
    return get_named_account(gnucash, '現金')


def get_named_account(gnucash, name):
    cur = gnucash.execute('SELECT guid, name, account_type FROM accounts WHERE name = ?', (name,))
    account_list = [x for x in cur]
    if len(account_list) != 1:
        print('Invalid cash account', account_list)
        exit(1)
    return account_list[0]


def get_jpy(gnucash):
    cur = gnucash.execute('SELECT guid, namespace, mnemonic, fullname FROM commodities WHERE namespace == \'CURRENCY\' AND mnemonic == \'JPY\'')
    curlist = list(cur)
    if len(curlist) != 1:
        print('Invalid number of jpy currency', curlist)
        exit(1)
    return curlist[0]


if __name__ == '__main__':
    _main()