cmb.py

import csv
import re
from collections import namedtuple
from datetime import date
from datetime import datetime
from datetime import timedelta
from typing import List

from beancount.core import data
from beancount.core.amount import Amount
from beancount.core.amount import Decimal
from beancount.core.data import Transaction
from beancount.core.number import D
from beancount.core.number import ZERO
from dateutil.parser import parse as dateparse

from . import NotSuitable
from common.accounts import get_account_by_guess
from common.deduplicate import Deduplicate
from common.utils import normalize_description
from common.utils import normalize_money

Account招商 = "Liabilities:CreditCards:CMB"
Account招商储蓄卡 = "Assets:Bank:CMB"

one_day = timedelta(days=1)


class CMBDebitCSV:
    """
    招行网银专业版，信用卡-账户管理-自助对账
    """

    def __init__(self, filename, entries, option_map):
        if not filename.endswith("csv"):
            raise NotSuitable("Not cmb")
        try:
            with open(filename, encoding="gbk") as f:
                first_line = f.readline().strip()
                if first_line != "# 招商银行交易记录":
                    raise NotSuitable("Not cmb csv")
                lines = [
                    line for line in f if line.strip() and not line.startswith("#")
                ]
                print(lines)
                self.reader = csv.DictReader(lines)
        except:
            raise NotSuitable("Not cmb")
        self.deduplicate = Deduplicate(entries, option_map)

    def parse(self):
        transactions = []
        last_time = None
        last_balance = None
        for row in self.reader:
            print(row)
            transaction_time = datetime.strptime(
                row["交易日期"].strip() + row["交易时间"].strip(), "%Y%m%d%H:%M:%S"
            )

            if last_balance and transaction_time.date() != last_time.date():
                txn_balance = data.Balance(
                    account=Account招商储蓄卡,
                    amount=Amount(
                        D(last_balance),
                        "CNY",
                    ),
                    meta=data.new_metadata(".", 1000),
                    tolerance=None,
                    diff_amount=None,
                    date=last_time.date() + one_day,
                )
                transactions.append(txn_balance)
                last_time = transaction_time
                last_balance = row["余额"].strip()
            elif not last_time or transaction_time > last_time:
                last_time = transaction_time
                last_balance = row["余额"].strip()
            description = normalize_description(row["交易备注"])
            print(f"Importing {description} at {transaction_time}")
            account = get_account_by_guess(
                normalize_description(row["交易类型"]),
                description=description,
                time=transaction_time,
            )
            flag = "*"
            meta = data.new_metadata("beancount/core/testing.beancount", 12345, {})
            entry = Transaction(
                meta,
                transaction_time.date(),
                flag,
                description,
                None,
                data.EMPTY_SET,
                data.EMPTY_SET,
                [],
            )
            number = normalize_money(row["支出"]) or f"-{normalize_money(row['收入'])}"
            if account:
                data.create_simple_posting(entry, account, None, None)
            data.create_simple_posting(entry, Account招商储蓄卡, -Decimal(number), "CNY")
            if not self.deduplicate.find_duplicate(
                entry,
                -Decimal(number),
                None,
                Account招商储蓄卡,
            ):
                transactions.append(entry)

        self.deduplicate.apply_beans()
        return transactions