""" 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())