from pathlib import Path import datetime from typing import List, Dict import logging from dotenv import dotenv_values from woocommerce import API from camisatoshi_wordpress_reports.order import Order 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: str, end_date: str): start_date = datetime.datetime.strptime(start_date, "%Y-%m-%d").isoformat() end_date = datetime.datetime.strptime(end_date, "%Y-%m-%d").isoformat() logger.info(f"Fetching orders between {start_date} and {end_date}.") orders_in_date_range = WC_API.get( endpoint="orders", params={ "after": start_date, "before": end_date, "per_page": 100, "status": "processing,completed", }, ).json() orders_in_date_range = [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.") skus_to_keep = ["TEE-05-BBO-BLACK"] logger.info(f"Filtering by SKUs: {skus_to_keep}") relevant_orders = filter_orders_by_sku( orders_in_date_range, skus=skus_to_keep ) 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 = find_orders_without_sats_received(relevant_orders) 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.") # 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) # Print to screen: # - Orders that do not have the `sats_received` metadata informed # - The unit count for each product # - The sales sum for each product # - The sats sum for each product # - The corresponding payment owed to UM # - The list of order ids that have been taken into account # Update orders: # - Add metadata entry: is_settled_with_um: 1 def filter_orders_by_sku(orders: List[Order], skus: List[str]) -> List[Order]: filtered_orders = [] for order in orders: for sku in skus: if order.contains_sku(sku): filtered_orders.append(order) return filtered_orders def find_orders_without_sats_received(orders: List[Order]) -> List[Order]: orders_without_sats_received = [] for order in orders: if not order.contains_meta_data_entry("sats_received"): orders_without_sats_received.append(order) continue return orders_without_sats_received