From 63a8c2910509b341998d23ccb81c4a69015fdfd4 Mon Sep 17 00:00:00 2001 From: C <0xicl33n@gmail.com> Date: Sun, 4 Jan 2026 10:40:43 -0800 Subject: [PATCH] reporting stats --- src/stats.py | 298 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 298 insertions(+) create mode 100644 src/stats.py diff --git a/src/stats.py b/src/stats.py new file mode 100644 index 0000000..986f8da --- /dev/null +++ b/src/stats.py @@ -0,0 +1,298 @@ +""" +Statistics gathering functions for daily_stats table. +Queries Square catalog to collect metrics about LEGO inventory quality. +""" + +import json +from typing import Dict, Any +from square import AsyncSquare +import get_sq_env + +# Initialize Square client +client = AsyncSquare( + token=get_sq_env.SQUARE_TOKEN, + environment=get_sq_env.environment +) + +LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID + + +async def get_daily_stats() -> Dict[str, Any]: + """ + Gather all metrics for daily_stats table in a single pass. + + Returns: + dict: { + "total_items": int, + "enriched_items": int, + "missing_upcs": int, + "missing_images": int, + "total_catalog_value": float + } + """ + # Get all items in LEGO category + result = await client.catalog.search( + object_types=["ITEM"], + query={ + "exact_query": { + "attribute_name": "category_id", + "attribute_value": LEGO_CAT_ID, + } + }, + ) + + result_json = json.loads(result.model_dump_json()) + objects = result_json.get("objects", []) + + # Initialize counters + total_items = 0 + enriched_items = 0 + missing_upcs = 0 + missing_images = 0 + total_catalog_value = 0.0 + + for obj in objects: + total_items += 1 + item_data = obj.get("item_data", {}) + + # Check if item has all enrichment data + has_upc = False + has_image = False + has_proper_name = False + item_price = 0.0 + + # Check variations for UPC and price + variations = item_data.get("variations", []) + if variations: + var_data = variations[0].get("item_variation_data", {}) + + # Check UPC + if var_data.get("upc"): + has_upc = True + else: + missing_upcs += 1 + + # Get price (stored in cents) + price_money = var_data.get("price_money", {}) + if price_money and price_money.get("amount"): + item_price = price_money["amount"] / 100.0 # Convert cents to dollars + total_catalog_value += item_price + else: + missing_upcs += 1 + + # Check for image + if item_data.get("image_ids"): + has_image = True + else: + missing_images += 1 + + # Check if name follows format: "12345 - THEME Set Name" + name = item_data.get("name", "") + if " - " in name and name.split()[0].isdigit() and len(name.split()[0]) == 5: + has_proper_name = True + + # Item is "enriched" if it has UPC, image, and proper name format + if has_upc and has_image and has_proper_name: + enriched_items += 1 + + return { + "total_items": total_items, + "enriched_items": enriched_items, + "missing_upcs": missing_upcs, + "missing_images": missing_images, + "total_catalog_value": round(total_catalog_value, 2), + } + + +async def get_total_items() -> int: + """ + Count total items in LEGO category. + + Returns: + int: Total number of items + """ + result = await client.catalog.search( + object_types=["ITEM"], + query={ + "exact_query": { + "attribute_name": "category_id", + "attribute_value": LEGO_CAT_ID, + } + }, + ) + + result_json = json.loads(result.model_dump_json()) + return len(result_json.get("objects", [])) + + +async def get_enriched_count() -> int: + """ + Count items with complete enrichment (UPC + image + proper name format). + + Returns: + int: Number of fully enriched items + """ + result = await client.catalog.search( + object_types=["ITEM"], + query={ + "exact_query": { + "attribute_name": "category_id", + "attribute_value": LEGO_CAT_ID, + } + }, + ) + + result_json = json.loads(result.model_dump_json()) + objects = result_json.get("objects", []) + + enriched_count = 0 + + for obj in objects: + item_data = obj.get("item_data", {}) + + # Check UPC + has_upc = False + variations = item_data.get("variations", []) + if variations: + var_data = variations[0].get("item_variation_data", {}) + has_upc = bool(var_data.get("upc")) + + # Check image + has_image = bool(item_data.get("image_ids")) + + # Check name format: "12345 - THEME Set Name" + name = item_data.get("name", "") + has_proper_name = False + if " - " in name: + first_part = name.split()[0] + if first_part.isdigit() and len(first_part) == 5: + has_proper_name = True + + if has_upc and has_image and has_proper_name: + enriched_count += 1 + + return enriched_count + + +async def get_missing_upcs() -> int: + """ + Count items without UPC barcodes. + + Returns: + int: Number of items missing UPCs + """ + result = await client.catalog.search( + object_types=["ITEM"], + query={ + "exact_query": { + "attribute_name": "category_id", + "attribute_value": LEGO_CAT_ID, + } + }, + ) + + result_json = json.loads(result.model_dump_json()) + objects = result_json.get("objects", []) + + missing_count = 0 + + for obj in objects: + item_data = obj.get("item_data", {}) + variations = item_data.get("variations", []) + + if not variations: + missing_count += 1 + continue + + var_data = variations[0].get("item_variation_data", {}) + if not var_data.get("upc"): + missing_count += 1 + + return missing_count + + +async def get_missing_images() -> int: + """ + Count items without images. + + Returns: + int: Number of items missing images + """ + result = await client.catalog.search( + object_types=["ITEM"], + query={ + "exact_query": { + "attribute_name": "category_id", + "attribute_value": LEGO_CAT_ID, + } + }, + ) + + result_json = json.loads(result.model_dump_json()) + objects = result_json.get("objects", []) + + missing_count = 0 + + for obj in objects: + item_data = obj.get("item_data", {}) + if not item_data.get("image_ids"): + missing_count += 1 + + return missing_count + + +async def get_total_catalog_value() -> float: + """ + Calculate total dollar value of all items in LEGO category. + + Returns: + float: Total catalog value in dollars + """ + result = await client.catalog.search( + object_types=["ITEM"], + query={ + "exact_query": { + "attribute_name": "category_id", + "attribute_value": LEGO_CAT_ID, + } + }, + ) + + result_json = json.loads(result.model_dump_json()) + objects = result_json.get("objects", []) + + total_value = 0.0 + + for obj in objects: + item_data = obj.get("item_data", {}) + variations = item_data.get("variations", []) + + if variations: + var_data = variations[0].get("item_variation_data", {}) + price_money = var_data.get("price_money", {}) + + if price_money and price_money.get("amount"): + # Price is stored in cents, convert to dollars + total_value += price_money["amount"] / 100.0 + + return round(total_value, 2) + + +# Example usage +if __name__ == "__main__": + import asyncio + + async def main(): + print("Gathering daily statistics for LEGO catalog...") + + stats = await get_daily_stats() + + print(f"\n=== Daily Stats ===") + print(f"Total Items: {stats['total_items']}") + print(f"Enriched Items: {stats['enriched_items']}") + print(f"Missing UPCs: {stats['missing_upcs']}") + print(f"Missing Images: {stats['missing_images']}") + print(f"Total Catalog Value: ${stats['total_catalog_value']:,.2f}") + print(f"Completion Rate: {(stats['enriched_items'] / stats['total_items'] * 100):.1f}%") + + asyncio.run(main())