126 lines
4.1 KiB
Python
126 lines
4.1 KiB
Python
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: 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 = [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.")
|
|
logger.info("Success, all orders have sats_received filled in.")
|
|
|
|
logger.info("Removing settled orders.")
|
|
unsettled_orders = filter_settled_orders(relevant_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)
|
|
|
|
# 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
|
|
|
|
|
|
def filter_settled_orders(orders: List[Order]) -> List[Order]:
|
|
unsettled_orders = []
|
|
|
|
for order in orders:
|
|
if not order.is_settled_with_um():
|
|
unsettled_orders.append(order)
|
|
|
|
return unsettled_orders
|