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 API_CONFIG = dotenv_values( dotenv_path=Path.home() / Path(".camisatoshi-wordpress-reports/.env") ) WC_API = API( url=API_CONFIG["URL"], consumer_key=API_CONFIG["CONSUMER_KEY"], consumer_secret=API_CONFIG["CONSUMER_SECRET"], 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']}") try: api_reported_version = WC_API.get("").json()["namespace"] except: raise ConnectionError("There was an issue connecting to the WooCommerce API.") logger.info(f"Informed version of the API: {API_CONFIG['VERSION']}") logger.info(f"Version reported by the API itself: {api_reported_version}") logger.info("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 ] ) 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.") # Fetch orders: # - Between specific dates # - That contain the hardcoded products # - That have been paid, hence status is either processing or completed # - That have not been settled yet (is_settled_with_um = 0) 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) * 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_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.") # Fetch orders: # - Between specific dates # - That contain the hardcoded products # - That have been paid, hence status is either processing or completed # - That have not been settled yet (is_settled_with_um = 0) 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) * 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("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": 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)