from datetime import datetime, timezone from typing import Dict, List, Optional, Tuple from uuid import UUID from concurrent.futures import ThreadPoolExecutor from decimal import Decimal import asyncio from datetime import datetime from fastapi import HTTPException, status from sqlalchemy import and_, func, or_ from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session, aliased from sqlalchemy.sql.expression import nullslast from src.apps.rollup.schemas.rollup import ( RollupLogCreate, RollupLogUpdate, RollupLogResponse, RollupOutput ) from src.core.config import settings from src.core.exceptions import APIException from src.utils.constants import API_PREFIXES, EXPORT_DATA_SUB_FOLDER, RP_HEADER_LIST, size_conversions from src.utils.pagination import QueryPaginator from src.utils.enums import RollupType, RollupStatus, Separator from src.apps.rollup.services.rollup_log import ( create_rollup_log, update_rollup_log ) from src.apps.rollup.services.rollup_output import create_rollup_output from src.apps.currency.services.currency import get_rates from src.apps.wine.vintage.services.vintage import get_vintage_map from src.apps.wine.bottle_size.services.bottle_size import get_bottle_size_map from src.apps.wine.retailer.services.retailer import get_retailers_map from src.apps.web_crawler.services.web_crawler import ( get_codes_from_web_crawler_files, get_web_crawler_by_code, get_latest_file_content ) from src.apps.wine.wine.services.wine import get_mongo_id_by_wine_id import aiofiles import os import stat from pathlib import Path async def execute_rp_rollup( db: Session ) -> Dict[str, any]: """ Execute the RP rollup process based on the provided payload. Args: db (Session): SQLAlchemy session for database interaction. payload (RollupOutput): The input data for the rollup process. Returns: Dict[str, any]: A dictionary containing the results of the rollup process. """ try: # Create a new rollup log entry new_rollup = await create_rollup_log( db=db, payload=RollupLogCreate( rollup_type=RollupType.RP, date_start=datetime.now(timezone.utc), status=RollupStatus.RUNNING ) ) # fetch necessary mappings vintage_map = await get_vintage_map(db) bottle_size_map = await get_bottle_size_map(db) retailers_map = await get_retailers_map(db) # Fetch web crawler codes web_crawler_codes = await get_codes_from_web_crawler_files(db=db, type="HISTORY") # Fetch currency rates currency_rates = await get_rates(db) rollup_output_list = [] total_records = len(web_crawler_codes) or 0 # used to track progress processing_count = 0 for code in web_crawler_codes: # Process each code and create RollupOutput objects updated_rollup_log = await update_rollup_log( db=db, rollup_log_id=new_rollup.id, payload=RollupLogUpdate( message=f"Processing {processing_count}/{total_records}", ) ) # get web_crawler from code crawler = await get_web_crawler_by_code(db=db, code=code) # print(f"Processing code: {code}, Crawler found: {crawler.code is not None}") # Debug print if crawler: input_delimiter = Separator.PIPE retailer = retailers_map.get(code.lower()) history_matches = await get_latest_file_content(db=db, web_crawler_id=crawler.id, type="HISTORY") rollup_output_list.extend( await process_rollup_outputs( db, history_matches, input_delimiter, crawler, retailer, vintage_map, bottle_size_map ) ) processing_count += 1 await equalize_price(rollup_output_list, currency_rates) filtered_list = await filter_by_median_price_per_wine_id(rollup_output_list) # sort by wine_alert_id filtered_list.sort(key=lambda x: (x.wine_alert_id or "")) # print(f"Total Rollup Outputs: {len(rollup_output_list)}, After Filtering: {len(filtered_list)}") # Get the current date and time now = datetime.now() # Format the filename file_name = f"ForSale{now.strftime('%y.%m.%d.%H%M%S')}.txt" # Write the filtered rollup output to a text file await create_rp_rollup_file( file_name=file_name, rollup_output_list=filtered_list, header_row=RP_HEADER_LIST, separator=Separator.TAB.value, download_location=EXPORT_DATA_SUB_FOLDER ) # Update the rollup log with the completed status and output file details finished_rollup = await update_rollup_log( db=db, rollup_log_id=new_rollup.id, payload=RollupLogUpdate( status=RollupStatus.SUCCESS, date_end=datetime.now(timezone.utc), message=None, file_name=file_name ) ) return RollupLogResponse.model_validate(finished_rollup) except Exception as e: await update_rollup_log( db=db, rollup_log_id=new_rollup.id, payload=RollupLogUpdate( status=RollupStatus.FAILED, date_end=datetime.now(timezone.utc), message=str(e) ) ) APIException( module="execute_rp_rollup", error={"exception": str(e)}, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, message="Error executing RP rollup process.", ) async def process_rollup_outputs( db: Session, history_matches: List[str], input_delimiter: str, crawler, retailer, vintage_map, bottle_size_map ) -> List[RollupOutput]: """ Processes rollup outputs in parallel, removing nulls and duplicates. Args: history_matches (List[str]): List of history match lines. input_delimiter (str): Delimiter for splitting lines. crawler: Crawler object. retailer: Retailer object. vintage_map: Map of vintages. bottle_size_map: Map of bottle sizes. Returns: List[RollupOutput]: Processed rollup outputs. """ async def process_line(line): try: line = line.replace("\u0000", "") row = line.split(input_delimiter) mongo_id = await get_mongo_id_by_wine_id(db=db, wine_id=row[0][:12]) return await create_rollup_output( web_crawler=crawler, retailer=retailer, row=row, vintage_map=vintage_map, bottle_size_map=bottle_size_map, mongo_id=mongo_id ) except Exception as e: # Log the error and continue processing other lines print(f"Error processing line: {e}") return None # Use asyncio.gather for parallel processing results = await asyncio.gather(*(process_line(line) for line in history_matches)) # Remove nulls and duplicates new_list = [result for result in results if result is not None] unique_list = list({(item.actual_retailer_description, item.price): item for item in new_list}.values()) return unique_list async def standardize_price(bottle_size: str, price: Decimal) -> Optional[Decimal]: """ Standardizes prices to a 750 ml bottle size. """ if not bottle_size or price is None: return None try: standard_size = Decimal('750') new_price = price bottle_size = bottle_size.lower() if bottle_size in size_conversions: new_price = size_conversions[bottle_size](price) return new_price.quantize(Decimal('0.01')) # Round to 2 decimal places except Exception as e: print(f"Error standardizing price: {e}") return None async def equalize_price(rollup_output_list: List[RollupOutput], currency_rates: Dict[str, Decimal]) -> None: """ Equalizes the price of rollup outputs by converting to USD and standardizing to 750 ml. """ for output in rollup_output_list: if output.price is not None: try: # Convert price to USD using currency rates - divide by rate with CEILING rounding currency = output.currency or "USD" currency_rate = currency_rates.get(currency, Decimal('1.0')) # Use ROUND_CEILING and scale of 2 decimal places like the original price_in_usd = Decimal(str(output.price)).quantize( Decimal('0.01'), rounding='ROUND_CEILING' ) / currency_rate # Standardize to 750ml bottle size output.equalized_price = await standardize_price(output.bottle_size, price_in_usd) except Exception as e: print(f"Warning: Error equalizing price for output: {e}") output.equalized_price = None async def filter_by_median_price_per_wine_id(rollup_output_list: List[RollupOutput]) -> List[RollupOutput]: """ Filters rollup outputs by median price per wine ID. Args: rollup_output_list (List[RollupOutput]): List of rollup outputs to filter. Returns: List[RollupOutput]: Filtered list of rollup outputs. """ result = [] # Group by wine_id grouped = {} for item in rollup_output_list: wine_id = item.wine_id or "" if wine_id not in grouped: grouped[wine_id] = [] grouped[wine_id].append(item) # Filter each group of wine_id low_filter = Decimal('5.5') for wine_id, wine_list in grouped.items(): filtered_list = [] if len(wine_list) == 1: # If only 1 wine in a group, no need to calculate & filter by median # Ignore records with price less than 5.5 $ # IMPORTANT: If the listing is FALSE, include every record (Ignore the price validation) filtered_list = [ item for item in wine_list if not item.listing or (item.equalized_price and item.equalized_price >= low_filter) ] else: # Filter by median price filtered_list = await filter_by_median_price(wine_list) if len(filtered_list) > 0: result.extend(filtered_list) return result async def filter_by_median_price(wine_list: List[RollupOutput]) -> List[RollupOutput]: """ Filters a list of wines by median price logic. Args: wine_list (List[RollupOutput]): List of wines with the same wine_id. Returns: List[RollupOutput]: Filtered list based on median price logic. """ filtered_list = [] # Ignore records with price less than 5.5 $ low_filter = Decimal('5.5') price_list = [] for item in wine_list: if (item.equalized_price is not None and item.equalized_price >= low_filter and item.listing): price_list.append(Decimal(str(item.equalized_price))) if len(price_list) > 0: try: # Calculate median price_list.sort() mid_number = len(price_list) // 2 if len(price_list) % 2 != 0: median = price_list[mid_number] else: median = (price_list[mid_number] + price_list[mid_number - 1]) / Decimal('2') # Ignore records with price > median * 4 # Ignore records with price < median / 4 # IMPORTANT: If the listing is FALSE, include every record (Ignore the price validation) max_price = median * Decimal('4') min_price = median / Decimal('4') filtered_list = [ item for item in wine_list if (not item.listing or (item.equalized_price is not None and item.equalized_price <= max_price and item.equalized_price >= min_price)) ] except Exception as e: print(f"Warning: Error calculating median price: {e}") # Fallback to basic filter if median calculation fails filtered_list = [ item for item in wine_list if not item.listing or (item.equalized_price and item.equalized_price >= low_filter) ] else: # If no valid prices found, apply basic filter filtered_list = [ item for item in wine_list if not item.listing or (item.equalized_price and item.equalized_price >= low_filter) ] return filtered_list async def create_rp_rollup_file( header_row: List[str], rollup_output_list: List[RollupOutput], separator: str, file_name: str, download_location: Optional[str] = EXPORT_DATA_SUB_FOLDER ) -> None: """ Creates an RP rollup file with the given data and sets file permissions. Args: header_row (List[str]): The header row for the file. rollup_output_list (List[RollupOutput]): The list of rollup outputs to write. separator (str): The separator to use in the file. file_name (str): The name of the file to create. download_location (str): The directory where the file will be saved. Returns: None """ try: # Prepare the file content line_terminator = "\n" sb = [] # Append header row sb.append(separator.join(header_row) + line_terminator) # Append rollup output rows for output in rollup_output_list: sb.append(output.to_string(separator, RollupType.RP) + line_terminator) # Create the file path file_path = os.path.join(download_location, file_name) os.makedirs(os.path.dirname(file_path), exist_ok=True) # Write to the file asynchronously async with aiofiles.open(file_path, mode='w', encoding='utf-8') as file: await file.write(''.join(sb)) # Set file permissions Path(file_path).chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH) except Exception as e: print(f"Error creating RP rollup file {file_name}: {e}")