reporting stats

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