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

Compare commits

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

No commits in common. "flexible_report_building" and "master" have entirely different histories.
flexible_r ... master

7 changed files with 80 additions and 306 deletions
Download patch file Download diff file Expand all files Collapse all files

3
README.md
View file

@ -21,7 +21,7 @@ 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"
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.
@ -38,4 +38,3 @@ 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.
- Some reports will break if the SKU has absolutely no orders. But well, if that happens, there is simply no report to build, so the only serious improvement would be dropping an informative error message.

6
camisatoshi_wordpress_reports/cli.py
View file

@ -5,10 +5,7 @@ import typer
from typing_extensions import Annotated
import camisatoshi_wordpress_reports.controllers as controllers
import camisatoshi_wordpress_reports.logging as our_logging
our_logging.set_config_level()
app = typer.Typer()
@ -21,11 +18,10 @@ def check_health():
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)],
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)],

29
camisatoshi_wordpress_reports/constants.py
View file

@ -6,24 +6,23 @@ DEFAULT_DOTENV_FILEPATH = ".camisatoshi-wordpress-reports/.env"
### Order keys
ORDER_KEYS = SimpleNamespace()
ORDER_KEYS.id = "id"
ORDER_KEYS.meta_data = "meta_data"
ORDER_KEYS.total = "total"
ORDER_KEYS.line_items = "line_items"
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"
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"
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_PERCENTAGE = 0.2
BBO_SKUS = ["TEE-05-BBO-BLACK", "SUD-01-BBO-BLACK"]
um_first_agreement_percentage = 0.5
bbo_royalty_fee = 0.2

112
camisatoshi_wordpress_reports/controllers.py
View file

@ -1,4 +1,3 @@
from functools import partial
from pathlib import Path
import datetime
import logging
@ -9,16 +8,9 @@ from woocommerce import API
from camisatoshi_wordpress_reports.order import Order, Orders
from camisatoshi_wordpress_reports.constants import (
ORDER_KEYS,
UM_FIRST_AGREEMENT_PERCENTAGE,
um_first_agreement_percentage,
DEFAULT_DOTENV_FILEPATH,
BBO_ROYALTY_FEE_PERCENTAGE,
BBO_SKUS,
)
from camisatoshi_wordpress_reports.report_building import (
OrderObtentionChainBuilder,
WoocomerceOrderScope,
keep_orders_containing_sku,
bbo_royalty_fee,
)
API_CONFIG = dotenv_values(
@ -31,6 +23,7 @@ WC_API = API(
version=API_CONFIG["VERSION"],
)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()
@ -74,14 +67,10 @@ def generate_um_report(
)
logger.info(f"Received {len(orders_in_date_range)} orders.")
relevant_skus = [
"TEE-05-BBO-BLACK",
"SUD-01-BBO-BLACK",
"TEE-09-SIMPLY-BITCOIN",
]
logger.info(f"Filtering by SKUs: {relevant_skus}")
relevant_orders = orders_in_date_range.filter_orders_by_skus(
skus=relevant_skus
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.")
@ -112,35 +101,21 @@ def generate_um_report(
f"Relevant orders: {[order['id'] for order in unsettled_orders]}."
)
report = []
for relevant_sku in relevant_skus:
logger.debug(f"Reporting SKU {relevant_sku}")
for order in unsettled_orders:
if not order.contains_sku(relevant_sku):
continue
logger.debug(f"Reporting for order {order[ORDER_KEYS.id]}")
# A few helper variables to make the last variable more understandable
sats_received_for_sku = order.sats_received_for_sku(relevant_sku)
bbo_fee_if_applicable = (BBO_ROYALTY_FEE_PERCENTAGE * (relevant_sku in BBO_SKUS))
discount_factor_for_bbo_skus = 1 - bbo_fee_if_applicable
# We owe UM his percentages of the sats received after discounting
# royalties paid to BBO for the BBO products.
sats_owed_to_um = (
(sats_received_for_sku * discount_factor_for_bbo_skus)
* UM_FIRST_AGREEMENT_PERCENTAGE
)
report.append(
{
"order_id": order[ORDER_KEYS.id],
"sku": relevant_sku,
"units_sold": order.units_of_sku(relevant_sku),
"eur_income": order.sales_of_sku(relevant_sku),
"sats_income": sats_received_for_sku,
"sats_owed_to_um": sats_owed_to_um,
}
)
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)
@ -155,32 +130,41 @@ def generate_um_report(
def generate_sku_report(start_date, end_date, sku):
logger.info(f"Fetching orders between {start_date} and {end_date}.")
report_chain_builder = OrderObtentionChainBuilder()
report_chain_builder.add_order_fetching_step(
wc_order_scope=WoocomerceOrderScope(
after=start_date.isoformat(),
before=end_date.isoformat(),
status="processing,completed",
)
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.")
report_chain_builder.add_order_filtering_step(
partial(keep_orders_containing_sku, sku=sku)
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_chain = report_chain_builder.get_report_chain()
relevant_orders = report_chain.run_chain(WC_API)
report = []
for order in relevant_orders:
report.append(
{
"order_id": order[ORDER_KEYS.id],
"order_id": order["id"],
"sku": sku,
"units_sold": order.units_of_skus(sku),
"eur_income": order.sales_of_skus(sku),
"units_sold": order.units_of_sku(sku),
"eur_income": order.sales_of_sku(sku),
}
)
logger.info("Report generated.")

5
camisatoshi_wordpress_reports/logging.py
View file

@ -1,5 +0,0 @@
import logging
def set_config_level():
logging.basicConfig(level=logging.DEBUG)

66
camisatoshi_wordpress_reports/order.py
View file

@ -1,10 +1,6 @@
import logging
from typing import Dict, Collection
from camisatoshi_wordpress_reports.constants import (
ORDER_KEYS,
CUSTOM_META_DATA_KEYS,
)
from camisatoshi_wordpress_reports.constants import order_keys, custom_meta_data_keys
from camisatoshi_wordpress_reports.utils import safe_zero_division
@ -19,32 +15,30 @@ class Order:
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]
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]
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])
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])
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
)
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]
self.meta_data_entries[custom_meta_data_keys.sats_received]
)
sats_received_for_sku = (
monetary_weight_of_sku_in_order * total_order_sats_received
@ -53,8 +47,8 @@ class Order:
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:
for item in self[order_keys.line_items]:
if item[order_keys.line_item_keys.sku] == sku:
return True
return False
@ -65,7 +59,7 @@ class Order:
def is_settled_with_um(self):
is_settled = self.meta_data_entries.get(
CUSTOM_META_DATA_KEYS.is_settled_um, None
custom_meta_data_keys.is_settled_um, None
)
return bool(is_settled)
@ -93,9 +87,6 @@ class Orders:
return next_order
raise StopIteration
def __repr__(self):
return str([f"Order {order['id']}" for order in self])
def filter_orders_by_sku(self, sku: str) -> "Orders":
filtered_orders = []
@ -105,26 +96,11 @@ class Orders:
return Orders(filtered_orders)
def filter_orders_by_skus(self, skus: Collection[str]) -> "Orders":
filtered_orders = []
for order in self:
order_contains_at_least_one_of_the_skus = None # Guilty until proven innocent
for sku in skus:
if order.contains_sku(sku):
order_contains_at_least_one_of_the_skus = True
if order_contains_at_least_one_of_the_skus:
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
):
if not order.contains_meta_data_entry(custom_meta_data_keys.sats_received):
orders_without_sats_received.append(order)
continue
@ -138,15 +114,3 @@ class Orders:
unsettled_orders.append(order)
return Orders(unsettled_orders)
def filter_orders_by_metadata_value(self, key, value):
orders_with_metadata_value = []
for order in self:
if (
order.contains_meta_data_entry(key)
and order.meta_data_entries[key] == value
):
orders_with_metadata_value.append(order)
return Orders(orders_with_metadata_value)

163
camisatoshi_wordpress_reports/report_building.py
View file

@ -1,163 +0,0 @@
import logging
from functools import partial
from woocommerce import API
from camisatoshi_wordpress_reports.order import Orders, Order
logger = logging.getLogger()
class MetadataFilter:
def __init__(self, key, value, operator):
self.key = key
self.value = value
self.operator = operator
def order_satisfies_filter(self, order: Order) -> bool:
return self.operator(order.meta_data_entries[self.key], self.value)
class WoocomerceOrderScope:
serializable_fields = ("after", "before", "status")
def __init__(self, after=None, before=None, status=None):
self.after = after
self.before = before
self.status = status
def serialize_for_api_request(self) -> dict:
params = {}
for field in self.serializable_fields:
if hasattr(self, field):
params[field] = self.__getattribute__(field)
params["per_page"] = 100
return params
def fetch_orders_from_wc(
wc_api_client: API, wc_order_scope: WoocomerceOrderScope
) -> Orders:
params = wc_order_scope.serialize_for_api_request()
orders_in_scope = wc_api_client.get(
endpoint="orders", params=params
).json()
orders_in_scope = Orders(
[
Order.from_api_response(order_raw_data)
for order_raw_data in orders_in_scope
]
)
return orders_in_scope
def keep_orders_containing_sku(orders: Orders, sku: str) -> Orders:
return orders.filter_orders_by_sku(sku)
def keep_orders_containing_metadata_value(
orders: Orders, metadata_key: str, metadata_value
) -> Orders:
return orders.filter_orders_by_metadata_value(
key=metadata_key, value=metadata_value
)
def validate_orders_satisfy_metadata_filter(
orders: Orders, metadata_filter: MetadataFilter
) -> bool:
for order in orders:
if not metadata_filter.order_satisfies_filter(order):
return False
return True
class OrderObtentionChain:
def __init__(self):
self.order_fetching_step = None
self.order_filtering_steps = []
self.order_validation_steps = []
self._orders = None
def run_chain(self, wc_api_client: API):
logger.info("Running report chain.")
self._run_order_fetching_step(wc_api_client)
self._run_order_filtering_steps()
self._run_order_validation_steps()
logger.info(f"Report chain ran. Returning {len(self._orders)} orders.")
return self._orders
def _run_order_fetching_step(self, wc_api_client: API) -> Orders:
logger.debug("Running order fetching step.")
self._orders = self.order_fetching_step(wc_api_client)
logger.debug(f"Received {len(self._orders)} orders.")
def _run_order_filtering_steps(self) -> Orders:
logger.debug(
f"Running {len(self.order_filtering_steps)} order filtering steps."
)
temp_orders = self._orders
for order_filtering_step in self.order_filtering_steps:
logger.debug(f"{len(temp_orders)} before filtering.")
temp_orders = order_filtering_step(temp_orders)
logger.debug(f"{len(temp_orders)} after filtering.")
self._orders = temp_orders
def _run_order_validation_steps(self):
for order_validation_step in self.order_validation_steps:
if not order_validation_step(self._orders):
raise ValueError("Error during order validation step.")
class OrderObtentionChainBuilder:
def __init__(self):
self._wip_order_obtention_chain = OrderObtentionChain()
def add_order_fetching_step(
self, wc_order_scope: WoocomerceOrderScope
) -> "OrderObtentionChainBuilder":
"""
Define a scope of orders to get from Woocomerce.
"""
self._order_scope = wc_order_scope
self._wip_order_obtention_chain.order_fetching_step = partial(
fetch_orders_from_wc, wc_order_scope=wc_order_scope
)
return self
def add_order_filtering_step(self, step) -> "OrderObtentionChainBuilder":
self._wip_order_obtention_chain.order_filtering_steps.append(step)
return self
def add_order_validation_step(self, step) -> "OrderObtentionChainBuilder":
self._wip_order_obtention_chain.order_validation_steps.append(step)
return self
def get_report_chain(self) -> OrderObtentionChain:
return self._wip_order_obtention_chain