cmb_debit.py

import sys
from typing import Optional

from beancount.ingest import importer
sys.path.append("./")
import csv
from datetime import datetime
from datetime import timedelta

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

def get_account_by_guess(
    from_user: Optional[str], description: str, time=None
) -> Optional[str]:
    return None

def normalize_money(input: str) -> str:
    return input.replace("￥", "").replace(",", "").strip()


def normalize_description(input: str) -> str:
    return (
        input.replace("CHN", "")
        .replace("－", "-")
        .replace(" - ", "-")
        .replace("(还款)", "")
        .replace("（特约）", "")
        .replace("(消费)", "")
        .replace("(赠送)", "")
        .replace("支付宝-", "")
        .replace("财付通-", "")
        .replace("返 现", "返现")
        .replace("上海拉扎斯信息科技有限公司", "饿了么")
        .replace("上海钧正网络科技有限公司", "哈啰出行")
        .replace("上海寻梦信息技术有限公司", "拼多多")
        .replace("\xa0", " ")
        .strip()
    )

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

one_day = timedelta(days=1)


class CMBDebitCSVImporter(importer.ImporterProtocol):
    """
    招行网银专业版，信用卡-账户管理-自助对账

    bean-extract -e ledger/main.bean scripts/imports/cmb_debit.py documents/CMB_xxxx.csv
    """

    def identify(self, file):
        return "CMB" in file.name

    def file_name(self, file):
        return f"cmb.{file.name.split('-')[-1]}"

    def file_account(self, _):
        return Account招商储蓄卡

    def extract(self, file, existing_entries=None):
        lines = []
        with open(file.name, encoding="gbk") as f:
            first_line = f.readline().strip()
            if not first_line.startswith("# 招商银行交易记录"):
                raise ValueError("Not cmb csv")
            for line in f:
                line = line.strip()
                if line.startswith("#"):
                    continue
                if line.strip(",").strip() == "":
                    continue
                lines.append(line)

        transactions = []
        last_time = None
        last_balance = None
        for row in csv.DictReader(lines):
            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["交易备注"])
            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")
            transactions.append(entry)

        return transactions


from smart_importer import PredictPayees, PredictPostings, apply_hooks
from smart_importer.detector import DuplicateDetector

CONFIG = [
    apply_hooks(
        CMBDebitCSVImporter(),
        [DuplicateDetector(), PredictPayees(), PredictPostings()],
    )
]