def convert_food_to_retail_search_product()

in projects/conversational-commerce-agent/data-ingestion/food_to_retail_search.py [0:0]

• 98 lines of code
• 9 McCabe index (conditional complexity)

def convert_food_to_retail_search_product(
    input_file:str,
    output_file:str,
    project_number:str,
    branch:str="0") -> str:

    """
    Transforms a Flipkart JSONL file to
    Google Cloud Retail Search Product Schema.

    Args:
      input_file: Path to the input Flipkart JSONL file.
      output_file: Path to the output JSONL file.
      project_number: Google Cloud Project number.
      branch: Retail Search Branch Id. defaults to 0
    Returns:
      Path to the output JSONL file.
    """

    processed_products = ""
    with open(input_file, "r", encoding="utf-8") as infile:
        with open(output_file, "w", encoding="utf-8") as outfile:
            for line in infile:
                try:
                    source_obj = json.loads(line)
                    target_obj = {}

                    # Required fields
                    target_obj["title"] = source_obj.get(
                        "menuItemName", "Unknown Product"
                    )

                    if "menuItemImageUrl" not in source_obj:
                        logging.warning(
                            (
                                "[Warning]Product doed not"
                                "have a product url:%s"
                            ),
                            target_obj["title"]
                        )
                        continue

                    if target_obj["title"] in processed_products:
                        continue
                    else:
                        processed_products += f"""|{target_obj["title"]}"""

                    target_obj["categories"] = [source_obj["menuItemCategory"]]

                    target_obj["id"] = f"{uuid.uuid4()}"
                    target_obj["name"] = (
                    f"projects/{project_number}/locations/global/catalogs/"
                    f"""default_catalog/branches/{branch}"""
                    f"""/products/{target_obj["id"]}"""
                    )
                    target_obj["primaryProductId"] = target_obj["id"]
                    target_obj["type"] = "PRIMARY"  # Assuming all are primary

                    target_obj["description"] = source_obj.get(
                                "menuItemDescription", target_obj["title"]
                            )
                    if target_obj["description"] == target_obj["title"]:
                        print(f"""{target_obj["title"]} has no description.""")

                    target_desc = target_obj["description"]
                    if len(target_desc) >= 5000: # Max description
                        target_obj["description"] = target_desc[:5000]

                    target_obj["languageCode"] = "en-us" # Default language

                    source_images = source_obj["menuItemImageUrl"]
                    if source_images:
                        target_obj["images"] = [
                            {"uri": source_images}
                        ]
                    else:
                        logging.error(
                        "[Error]product does not have images:%s",
                        target_obj["title"],)
                        continue

                    target_obj["uri"] = source_obj.get("menuItemImageUrl", None)

                    # Price Information
                    source_obj["discounted_price"]= 0
                    source_obj["retail_price"]= 0
                    try:
                        item_price = float(
                            source_obj.get(
                                "menuItemCurrentPrice",
                                "0").replace("$", "")
                            )
                        target_obj["priceInfo"] = {
                            "currencyCode": "INR",
                            "price": item_price,
                            "originalPrice": item_price,
                            "priceRange": {},
                        }
                    except ValueError as e:
                        print(source_obj.get(
                                "menuItemCurrentPrice",
                                "0").replace("$", "")
                            )
                        logging.error(e)
                        logging.error("Unable to parse price for %s",
                                    target_obj["title"])
                        continue

                    target_obj["attributes"] = update_attributes(
                        source_obj=source_obj
                    )

                    # Availability
                    target_obj["availability"] = "IN_STOCK"
                    target_obj["availableQuantity"] = 0
                    target_obj["fulfillmentInfo"] = [
                        {
                            "type": "custom-type-1",
                            "placeIds": ["mobile", "www"]
                         }
                    ]
                    target_obj["retrievableFields"] = (
                        "name,title,brands,uri,categories,"
                        "priceInfo,description"
                        )
                    outfile.write(json.dumps(target_obj) + "\n")

                except json.JSONDecodeError as e:
                    logging.error("""