New API access feature for reading and writing Beancount data easily

curl https://api.beanhub.io/v1/repos/USERNAME/REPO_NAME/entries \
  -H "Access-Token: YOUR_ACCESS_TOKEN_HERE"

curl https://api.beanhub.io/v1/repos/USERNAME/REPO_NAME/forms/FORM_NAME \
  -H "Access-Token: YOUR_ACCESS_TOKEN_HERE" \
  -d '{"data": {"date": "2023-08-28", "hours": "15"}}'

curl https://api.beanhub.io/v1/repos/USERNAME/REPO_NAME/forms/FORM_NAME \
  -H "Access-Token: YOUR_ACCESS_TOKEN_HERE" \
  -d '{
    "data": [
      {"date": "2023-08-26", "hours": "15"},
      {"date": "2023-08-27", "hours": "13"},
      {"date": "2023-08-28", "hours": "14"}
    ],
    "commit": {
      "message": "Add project XYZ contracting hours for 2023-08-26 ~ 2023-08-28"
    }
  }'

import datetime
import decimal
import logging
import os
import pathlib
import re
import urllib.parse

import click
import requests
# This module is our internal library for generating invoice PDF file.
# You need to make one for yourself.
# Please checkout https://py-pdf.github.io/fpdf2/ to learn how to
# generate a PDF file.
from make_invoice import ItemLine
from make_invoice import make_invoice

BEANHUB_USERNAME = os.environ["BEANHUB_USERNAME"]
BEANHUB_REPO_NAME = os.environ["BEANHUB_REPO_NAME"]
BEANHUB_ACCESS_TOKEN = os.environ["BEANHUB_ACCESS_TOKEN"]
BEANHUB_BASE_API_URL = (
    f"https://api.beanhub.io/v1/repos/{BEANHUB_USERNAME}/{BEANHUB_REPO_NAME}/"
)


def fetch_entries():
    logger = logging.getLogger(__name__)

    # Notice: ideally we should be able to filter entries based on account and the date
    #         range, or other conditions with URL parameters. But the API feature
    #         is not available yet at this moment. So let's fetch everything and filter
    #         instead for now, given the frequency of running this script is like
    #         bi-weekly, it shouldn't be too bad for now.
    entries_url = urllib.parse.urljoin(BEANHUB_BASE_API_URL, "entries")
    commit_hexsha = None
    page = 0
    while True:
        resp = requests.get(
            entries_url + f"?page={page}",
            headers={"Access-Token": BEANHUB_ACCESS_TOKEN},
        )
        resp.raise_for_status()
        payload = resp.json()
        pagination = payload["pagination"]
        current_page = pagination["page"]
        total_pages = pagination["total_pages"]
        logger.info("Processing entries page %s / %s", current_page, total_pages)
        yield from payload["entries"]
        current_commit_hex_sha = payload["commit"]["hexsha"]
        if commit_hexsha is not None and current_commit_hex_sha != commit_hexsha:
            # Ensure that we are fetching data from the same commit otherwise we may
            # see inconsistent data
            raise ValueError(
                f"Inconsistent commit hexsha {commit_hexsha} and {current_commit_hex_sha}"
            )
        if current_page >= total_pages:
            break
        commit_hexsha = current_commit_hex_sha
        page += 1


def get_account_posting(
    entry: dict,
    narration_pattern: str,
    account: str,
) -> tuple[datetime.date, str, dict] | None:
    if entry["entry_type"] != "transaction":
        return None
    narration = entry["narration"]
    if not re.match(narration_pattern, narration):
        return None

    entry_date = datetime.datetime.strptime(entry["date"], "%Y-%m-%d").date()
    for posting in entry["postings"]:
        if posting["account"] == account:
            return entry_date, entry["narration"], posting


@click.command()
@click.argument("invoice_number")
@click.argument("output_file", type=click.Path())
@click.option(
    "-s",
    "--start-week-date",
    type=click.DateTime(formats=["%Y-%m-%d"]),
    default=str(datetime.date.today() - datetime.timedelta(days=7)),
)
def main(
    invoice_number: str, output_file: pathlib.Path, start_week_date: datetime.datetime
):
    logger = logging.getLogger(__name__)
    start_week_date = start_week_date.date()
    if start_week_date.weekday():
        # find monday of this week
        start_week_date -= datetime.timedelta(days=start_week_date.weekday())
    # two weeks from now
    end_week_date = start_week_date + datetime.timedelta(days=14)
    logger.info("Generating invoice between %s to %s", start_week_date, end_week_date)
    items: list[ItemLine] = []
    for entry in fetch_entries():
        item = get_account_posting(
            entry,
            narration_pattern="Hours spent on the software development project for client XYZ",
            account="Assets:AccountsReceivable:Contracting:XYZ",
        )
        if item is None:
            continue
        date, narration, posting = item
        if date < start_week_date:
            continue
        if date >= end_week_date:
            continue
        items.append(
            ItemLine(
                date=date,
                desc=narration,
                quantity=posting["units"]["number"],
                price=posting["price"]["number"],
            )
        )
        logger.info("Found transaction %s %r with posting %s", date, narration, posting)
    make_invoice(invoice_number=invoice_number, output_file=output_file, items=items)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    main()