From f3776966bc2266cf5145daa072a2db920bb80065 Mon Sep 17 00:00:00 2001 From: counterweight Date: Tue, 5 Sep 2023 11:47:38 +0200 Subject: [PATCH] A lot of work --- camisatoshi_wordpress_reports/controllers.py | 47 ++++++ camisatoshi_wordpress_reports/order.py | 9 ++ .../report_building.py | 148 ++++++++++++++++++ 3 files changed, 204 insertions(+) create mode 100644 camisatoshi_wordpress_reports/report_building.py diff --git a/camisatoshi_wordpress_reports/controllers.py b/camisatoshi_wordpress_reports/controllers.py index 51f2f13..7a27aa8 100644 --- a/camisatoshi_wordpress_reports/controllers.py +++ b/camisatoshi_wordpress_reports/controllers.py @@ -1,3 +1,4 @@ +from functools import partial from pathlib import Path import datetime import logging @@ -12,6 +13,7 @@ from camisatoshi_wordpress_reports.constants import ( DEFAULT_DOTENV_FILEPATH, bbo_royalty_fee, ) +from camisatoshi_wordpress_reports.report_building import ReportChainBuilder, WoocomerceOrderScope, keep_orders_containing_sku API_CONFIG = dotenv_values( dotenv_path=Path.home() / Path(DEFAULT_DOTENV_FILEPATH) @@ -176,3 +178,48 @@ def generate_sku_report(start_date, end_date, sku): dict_writer = csv.DictWriter(output_file, keys) dict_writer.writeheader() dict_writer.writerows(report) + + +def wip_generate_sku_report(start_date, end_date, sku): + logger.info(f"Fetching orders between {start_date} and {end_date}.") + + report_chain_builder = ReportChainBuilder() + + report_chain_builder.add_order_fetching_step( + wc_order_scope=WoocomerceOrderScope( + after=start_date.isoformat(), + before=end_date.isoformat(), + status="processing,completed" + ) + ) + + report_chain_builder.add_order_filtering_step( + partial( + keep_orders_containing_sku, + sku=sku + ) + ) + + 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["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) diff --git a/camisatoshi_wordpress_reports/order.py b/camisatoshi_wordpress_reports/order.py index a12665e..fd693e9 100644 --- a/camisatoshi_wordpress_reports/order.py +++ b/camisatoshi_wordpress_reports/order.py @@ -114,3 +114,12 @@ 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) diff --git a/camisatoshi_wordpress_reports/report_building.py b/camisatoshi_wordpress_reports/report_building.py new file mode 100644 index 0000000..568b5a4 --- /dev/null +++ b/camisatoshi_wordpress_reports/report_building.py @@ -0,0 +1,148 @@ +from functools import partial + +from woocommerce import API + +from camisatoshi_wordpress_reports.order import Orders, Order + + +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 ReportChain: + def __init__(self): + self.order_fetching_step = None + self.order_filtering_steps = None + self.order_validation_steps = None + self.transformation_steps = None + self.storage_steps = None + + self._orders = None + + def run_chain(self, wc_api_client: API): + self._run_order_fetching_step(wc_api_client) + self._run_order_filtering_steps() + self._run_order_validation_steps() + + def _run_order_fetching_step(self, wc_api_client: API) -> Orders: + self._orders = self.order_fetching_step(wc_api_client) + + def _run_order_filtering_steps(self) -> Orders: + temp_orders = self._orders + + for order_filtering_step in self.order_filtering_steps: + temp_orders = order_filtering_step(temp_orders) + + return 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 ReportChainBuilder: + def __init__(self): + self._wip_report_chain = ReportChain() + + def add_order_fetching_step( + self, wc_order_scope: WoocomerceOrderScope + ) -> "ReportChainBuilder": + """ + Define a scope of orders to get from Woocomerce. + """ + + self._order_scope = wc_order_scope + + self._wip_report_chain.order_fetching_step = partial( + fetch_orders_from_wc, wc_order_scope=wc_order_scope + ) + + return self + + def add_order_filtering_step(self, step) -> "ReportChainBuilder": + + self._wip_report_chain.order_filtering_steps.append(step) + + return self + + def add_order_validation_step(self, step) -> "ReportChainBuilder": + + self._wip_report_chain.order_validation_steps.append(step) + + return self + + def get_report_chain(self) -> ReportChain: + + return self._wip_report_chain