Properly compute amount owed to UM

README.md

@@ -8,3 +8,33 @@ This repository hosts a Python CLI app that can be used to generate reports by f
 1. Install the package 
 2. In the home directory of the running user, create a folder named `.camisatoshi-wordpress-reports`.
 3. Copy the provided `.env-example` file in that directory and fill it with the required params from Woocomerce.
+
+
+## How to run
+
+### Check that the API is reachable
+
+```shell
+camisatoshi-wordpress-reports check-health
+```
+
+### Make a report for the UM agreement between two dates
+
+```shell
+camisatoshi-wordpress-reports generate-um-report --start-date "2023-08-01T00:00:00" --end-date "2023-09-01T00:00:00"c
+```
+
+This will generate a file named `report.csv` in the current working directory.
+
+
+### Make a simple report for the sales of some SKU
+
+```shell
+camisatoshi-wordpress-reports generate-sku-report --start-date "2023-08-01T00:00:00" --end-date "2023-09-01T00:00:00" --sku TEE-05-BBO-BLACK
+```
+
+This will generate a file named `report.csv` in the current working directory.
+
+
+## Open issues
+- Pagination is not being managed. The moment we have more than 100 orders, we are gonna run into issues.

camisatoshi_wordpress_reports/cli.py

@@ -1,5 +1,9 @@
+import datetime
+
 import typer
 
+from typing_extensions import Annotated
+
 import camisatoshi_wordpress_reports.controllers as controllers
 
 app = typer.Typer()
@@ -11,6 +15,16 @@ def check_health():
 
 
 @app.command()
-def show_orders():
-    controllers.show_orders()
+def generate_sku_report(
+    start_date: Annotated[datetime.datetime, typer.Option(prompt=True)],
+    end_date: Annotated[datetime.datetime, typer.Option(prompt=True)],
+    sku: Annotated[str, typer.Option(prompt=True)]
+):
+    controllers.generate_sku_report(start_date, end_date, sku)
 
+@app.command()
+def generate_um_report(
+    start_date: Annotated[datetime.datetime, typer.Option(prompt=True)],
+    end_date: Annotated[datetime.datetime, typer.Option(prompt=True)],
+):
+    controllers.generate_um_report(start_date, end_date)

camisatoshi_wordpress_reports/constants.py

@@ -0,0 +1,28 @@
+from types import SimpleNamespace
+
+### Config
+
+DEFAULT_DOTENV_FILEPATH = ".camisatoshi-wordpress-reports/.env"
+
+### Order keys
+
+order_keys = SimpleNamespace()
+order_keys.meta_data = "meta_data"
+order_keys.total = "total"
+order_keys.line_items = "line_items"
+
+order_keys.line_item_keys = SimpleNamespace()
+order_keys.line_item_keys.sku = "sku"
+order_keys.line_item_keys.quantity = "quantity"
+order_keys.line_item_keys.total = "total"
+
+custom_meta_data_keys = SimpleNamespace()
+custom_meta_data_keys.is_settled_um = "is_settled_um"
+custom_meta_data_keys.sats_received = "sats_received"
+
+
+### Other
+
+um_first_agreement_percentage = 0.5
+bbo_royalty_fee = 0.2
+

camisatoshi_wordpress_reports/controllers.py

@@ -1,10 +1,20 @@
 from pathlib import Path
+import datetime
+import logging
+import csv
 
 from dotenv import dotenv_values
 from woocommerce import API
 
+from camisatoshi_wordpress_reports.order import Order, Orders
+from camisatoshi_wordpress_reports.constants import (
+    um_first_agreement_percentage,
+    DEFAULT_DOTENV_FILEPATH,
+    bbo_royalty_fee,
+)
+
 API_CONFIG = dotenv_values(
-    dotenv_path=Path.home() / Path(".camisatoshi-wordpress-reports/.env")
+    dotenv_path=Path.home() / Path(DEFAULT_DOTENV_FILEPATH)
 )
 WC_API = API(
     url=API_CONFIG["URL"],
@@ -13,26 +23,156 @@ WC_API = API(
     version=API_CONFIG["VERSION"],
 )
 
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger()
+
 
 def check_health():
-    print(f"Connecting to the configured woocomerce at {API_CONFIG['URL']}")
+    logger.info(
+        f"Connecting to the configured WooCommerce at {API_CONFIG['URL']}"
+    )
 
     try:
         api_reported_version = WC_API.get("").json()["namespace"]
     except:
         raise ConnectionError(
-            "There was an issue connecting to the woocomerce API."
+            "There was an issue connecting to the WooCommerce API."
         )
 
-    print(f"Informed version of the API: {API_CONFIG['VERSION']}")
-    print(f"Version reported by the API itself: {api_reported_version}")
+    logger.info(f"Informed version of the API: {API_CONFIG['VERSION']}")
+    logger.info(f"Version reported by the API itself: {api_reported_version}")
 
-    print("Connection successful. The API is reachable.")
+    logger.info("Connection successful. The API is reachable.")
 
 
-def show_orders():
-    print(
-        WC_API.get(
-            "orders", params={"after": "2023-07-21T18:02:23+00:00"}
-        ).json()
+def generate_um_report(
+    start_date: datetime.datetime, end_date: datetime.datetime
+) -> None:
+    logger.info(f"Fetching orders between {start_date} and {end_date}.")
+
+    orders_in_date_range = WC_API.get(
+        endpoint="orders",
+        params={
+            "after": start_date.isoformat(),
+            "before": end_date.isoformat(),
+            "per_page": 100,
+            "status": "processing,completed",
+        },
+    ).json()
+    orders_in_date_range = Orders(
+        [
+            Order.from_api_response(order_raw_data)
+            for order_raw_data in orders_in_date_range
+        ]
     )
+    logger.info(f"Received {len(orders_in_date_range)} orders.")
+
+    relevant_sku = "TEE-05-BBO-BLACK"
+    logger.info(f"Filtering by SKU: {relevant_sku}")
+    relevant_orders = orders_in_date_range.filter_orders_by_sku(
+        sku=relevant_sku
+    )
+    logger.info(f"Kept {len(relevant_orders)} orders.")
+
+    logger.info(
+        "Checking if all orders have the sats_received entry filled in."
+    )
+    orders_without_sats_received = (
+        relevant_orders.filter_orders_without_sats_received()
+    )
+    if orders_without_sats_received:
+        logger.warning(
+            f"There are {len(orders_without_sats_received)} orders without a properly filled sats_received entry."
+        )
+        logger.warning(f"See details below.")
+        logger.warning(orders_without_sats_received)
+        raise ValueError(
+            "Not all orders have sats_received. Can't compute sats owed without that."
+        )
+    logger.info("Success, all orders have sats_received filled in.")
+
+    logger.info("Removing settled orders.")
+    unsettled_orders = relevant_orders.filter_unsettled_orders()
+    logger.info(f"Kept {len(unsettled_orders)} unsettled orders.")
+
+    logger.info("Order filtering finished.")
+
+    logger.info(
+        f"Relevant orders: {[order['id'] for order in unsettled_orders]}."
+    )
+    report = []
+    for order in unsettled_orders:
+        report.append(
+            {
+                "order_id": order["id"],
+                "sku": relevant_sku,
+                "units_sold": order.units_of_sku(relevant_sku),
+                "eur_income": order.sales_of_sku(relevant_sku),
+                "sats_income": order.sats_received_for_sku(relevant_sku),
+                "sats_owed_to_um": (
+                    order.sats_received_for_sku(relevant_sku)
+                    * (1 - bbo_royalty_fee)
+                )
+                * um_first_agreement_percentage,
+            }
+        )
+    logger.info("Report generated.")
+    logger.info(report)
+
+    keys = report[0].keys()
+
+    with open("report.csv", "w", newline="") as output_file:
+        dict_writer = csv.DictWriter(output_file, keys)
+        dict_writer.writeheader()
+        dict_writer.writerows(report)
+
+
+def generate_sku_report(start_date, end_date, sku):
+    logger.info(f"Fetching orders between {start_date} and {end_date}.")
+
+    orders_in_date_range = WC_API.get(
+        endpoint="orders",
+        params={
+            "after": start_date.isoformat(),
+            "before": end_date.isoformat(),
+            "per_page": 100,
+            "status": "processing,completed",
+        },
+    ).json()
+    orders_in_date_range = Orders(
+        [
+            Order.from_api_response(order_raw_data)
+            for order_raw_data in orders_in_date_range
+        ]
+    )
+    logger.info(f"Received {len(orders_in_date_range)} orders.")
+
+    logger.info(f"Filtering by SKU: {sku}")
+    relevant_orders = orders_in_date_range.filter_orders_by_sku(sku=sku)
+    logger.info(f"Kept {len(relevant_orders)} orders.")
+
+    logger.info("Order filtering finished.")
+
+    logger.info(
+        f"Relevant orders: {[order['id'] for order in relevant_orders]}."
+    )
+
+    report = []
+    for order in relevant_orders:
+        report.append(
+            {
+                "order_id": order["id"],
+                "sku": sku,
+                "units_sold": order.units_of_sku(sku),
+                "eur_income": order.sales_of_sku(sku),
+            }
+        )
+    logger.info("Report generated.")
+    logger.info(report)
+
+    keys = report[0].keys()
+
+    with open("report.csv", "w", newline="") as output_file:
+        dict_writer = csv.DictWriter(output_file, keys)
+        dict_writer.writeheader()
+        dict_writer.writerows(report)

camisatoshi_wordpress_reports/order.py

@@ -0,0 +1,116 @@
+from typing import Dict, Collection
+
+from camisatoshi_wordpress_reports.constants import order_keys, custom_meta_data_keys
+from camisatoshi_wordpress_reports.utils import safe_zero_division
+
+
+class Order:
+    def __init__(self, raw_data: Dict):
+        self.raw_data = raw_data
+
+    def __getitem__(self, item):
+        return self.raw_data[item]
+
+    @property
+    def meta_data_entries(self):
+        return {
+            meta_data_entry["key"]: meta_data_entry["value"]
+            for meta_data_entry in self.raw_data[order_keys.meta_data]
+        }
+
+    def units_of_sku(self, sku: str) -> int:
+        units = 0
+        for line in self[order_keys.line_items]:
+            if line[order_keys.line_item_keys.sku] == sku:
+                units += line[order_keys.line_item_keys.quantity]
+        return units
+
+    def sales_of_sku(self, sku: str) -> float:
+        sales = 0
+        for line in self[order_keys.line_items]:
+            if line[order_keys.line_item_keys.sku] == sku:
+                sales += float(line[order_keys.line_item_keys.total])
+        return sales
+
+    def sats_received_for_sku(self, sku: str) -> float:
+        total_order_eur = float(self[order_keys.total])
+        eur_of_sku = self.sales_of_sku(sku)
+
+        monetary_weight_of_sku_in_order = safe_zero_division(eur_of_sku, total_order_eur)
+        total_order_sats_received = float(
+            self.meta_data_entries[custom_meta_data_keys.sats_received]
+        )
+        sats_received_for_sku = (
+            monetary_weight_of_sku_in_order * total_order_sats_received
+        )
+
+        return sats_received_for_sku
+
+    def contains_sku(self, sku: str) -> bool:
+        for item in self[order_keys.line_items]:
+            if item[order_keys.line_item_keys.sku] == sku:
+                return True
+        return False
+
+    def contains_meta_data_entry(self, meta_data_entry_key: str) -> bool:
+        if meta_data_entry_key in self.meta_data_entries.keys():
+            return True
+        return False
+
+    def is_settled_with_um(self):
+        is_settled = self.meta_data_entries.get(
+            custom_meta_data_keys.is_settled_um, None
+        )
+
+        return bool(is_settled)
+
+    @classmethod
+    def from_api_response(cls, raw_data) -> "Order":
+        return Order(raw_data)
+
+
+class Orders:
+    def __init__(self, orders: Collection[Order]):
+        self._orders = orders
+
+    def __len__(self):
+        return len(self._orders)
+
+    def __iter__(self):
+        self._index = 0
+        return self
+
+    def __next__(self):
+        if self._index < len(self._orders):
+            next_order = self._orders[self._index]
+            self._index += 1
+            return next_order
+        raise StopIteration
+
+    def filter_orders_by_sku(self, sku: str) -> "Orders":
+        filtered_orders = []
+
+        for order in self:
+            if order.contains_sku(sku):
+                filtered_orders.append(order)
+
+        return Orders(filtered_orders)
+
+    def filter_orders_without_sats_received(self) -> "Orders":
+        orders_without_sats_received = []
+
+        for order in self:
+            if not order.contains_meta_data_entry(custom_meta_data_keys.sats_received):
+                orders_without_sats_received.append(order)
+                continue
+
+        return Orders(orders_without_sats_received)
+
+    def filter_unsettled_orders(self) -> "Orders":
+        unsettled_orders = []
+
+        for order in self:
+            if not order.is_settled_with_um():
+                unsettled_orders.append(order)
+
+        return Orders(unsettled_orders)

camisatoshi_wordpress_reports/utils.py

@@ -0,0 +1,2 @@
+def safe_zero_division(n, d):
+    return n / d if d else 0