counterweight/camisatoshi-wordpress-reports
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: counterweight:master
Branches Tags
counterweight:master
counterweight:flexible_report_building
counterweight:um_report
counterweight:health_check
..
compare: counterweight:health_check
Branches Tags
counterweight:flexible_report_building
counterweight:master
counterweight:um_report
counterweight:health_check

No commits in common. "master" and "health_check" have entirely different histories.
master ... health_che

6 changed files with 14 additions and 344 deletions
Download patch file Download diff file Expand all files Collapse all files

32
README.md
View file

@ -7,34 +7,4 @@ 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.
3. Copy the provided `.env-example` file in that directory and fill it with the required params from Woocomerce.

18
camisatoshi_wordpress_reports/cli.py
View file

@ -1,9 +1,5 @@
import datetime
import typer
from typing_extensions import Annotated
import camisatoshi_wordpress_reports.controllers as controllers
app = typer.Typer()
@ -15,16 +11,6 @@ def check_health():
@app.command()
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)
def show_orders():
controllers.show_orders()
@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)

28
camisatoshi_wordpress_reports/constants.py
View file

@ -1,28 +0,0 @@
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

162
camisatoshi_wordpress_reports/controllers.py
View file

@ -1,20 +1,10 @@
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(DEFAULT_DOTENV_FILEPATH)
dotenv_path=Path.home() / Path(".camisatoshi-wordpress-reports/.env")
)
WC_API = API(
url=API_CONFIG["URL"],
@ -23,156 +13,26 @@ WC_API = API(
version=API_CONFIG["VERSION"],
)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()
def check_health():
logger.info(
f"Connecting to the configured WooCommerce at {API_CONFIG['URL']}"
)
print(f"Connecting to the configured woocomerce at {API_CONFIG['URL']}")
try:
api_reported_version = WC_API.get("").json()["namespace"]
except:
raise ConnectionError(
"There was an issue connecting to the WooCommerce API."
"There was an issue connecting to the woocomerce API."
)
logger.info(f"Informed version of the API: {API_CONFIG['VERSION']}")
logger.info(f"Version reported by the API itself: {api_reported_version}")
print(f"Informed version of the API: {API_CONFIG['VERSION']}")
print(f"Version reported by the API itself: {api_reported_version}")
logger.info("Connection successful. The API is reachable.")
print("Connection successful. The API is reachable.")
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
]
def show_orders():
print(
WC_API.get(
"orders", params={"after": "2023-07-21T18:02:23+00:00"}
).json()
)
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)

116
camisatoshi_wordpress_reports/order.py
View file

@ -1,116 +0,0 @@
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)

2
camisatoshi_wordpress_reports/utils.py
View file

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