From 2f7769e2c3b96120b0cd3892e31f954e59cd5d2b Mon Sep 17 00:00:00 2001 From: C <0xicl33n@gmail.com> Date: Mon, 5 Jan 2026 15:27:34 -0800 Subject: [PATCH] imporerror --- src/stats.py | 103 +++++++++++++++++++++++++-------------------------- 1 file changed, 51 insertions(+), 52 deletions(-) diff --git a/src/stats.py b/src/stats.py index 986f8da..d159fc3 100644 --- a/src/stats.py +++ b/src/stats.py @@ -6,13 +6,10 @@ 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 +from . import get_sq_env # Initialize Square client -client = AsyncSquare( - token=get_sq_env.SQUARE_TOKEN, - environment=get_sq_env.environment -) +client = AsyncSquare(token=get_sq_env.SQUARE_TOKEN, environment=get_sq_env.environment) LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID @@ -20,7 +17,7 @@ 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, @@ -40,38 +37,38 @@ async def get_daily_stats() -> Dict[str, Any]: } }, ) - + 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"): @@ -79,22 +76,22 @@ async def get_daily_stats() -> Dict[str, Any]: 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, @@ -107,7 +104,7 @@ async def get_daily_stats() -> Dict[str, Any]: async def get_total_items() -> int: """ Count total items in LEGO category. - + Returns: int: Total number of items """ @@ -120,7 +117,7 @@ async def get_total_items() -> int: } }, ) - + result_json = json.loads(result.model_dump_json()) return len(result_json.get("objects", [])) @@ -128,7 +125,7 @@ async def get_total_items() -> int: async def get_enriched_count() -> int: """ Count items with complete enrichment (UPC + image + proper name format). - + Returns: int: Number of fully enriched items """ @@ -141,25 +138,25 @@ async def get_enriched_count() -> int: } }, ) - + 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 @@ -167,17 +164,17 @@ async def get_enriched_count() -> int: 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 """ @@ -190,31 +187,31 @@ async def get_missing_upcs() -> int: } }, ) - + 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 """ @@ -227,24 +224,24 @@ async def get_missing_images() -> int: } }, ) - + 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 """ @@ -257,42 +254,44 @@ async def get_total_catalog_value() -> float: } }, ) - + 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}%") - + print( + f"Completion Rate: {(stats['enriched_items'] / stats['total_items'] * 100):.1f}%" + ) + asyncio.run(main())