from pathlib import Path import datetime from typing import List, Dict import logging from dotenv import dotenv_values from woocommerce import API 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 woocomerce at {API_CONFIG['URL']}" ) try: api_reported_version = WC_API.get("").json()["namespace"] except: raise ConnectionError( "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}") 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() 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.") # 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[Dict], skus: List[str]): filtered_orders = [] for order in orders: for item in order["line_items"]: if item["sku"] in skus: filtered_orders.append(order) return filtered_orders