(h: ddlmZmZddlmZmZmZmZddlmZddl m Z ddl m Z ddl Z ddlmZddlmZmZdd lmZmZmZdd lmZdd lmZmZdd lmZdd lmZmZmZm Z ddl!m"Z"ddl#m$Z$ddl%m&Z&m'Z'm(Z(m)Z)ddl*m+Z+ddl,m-Z-m.Z.m/Z/ddl0m1Z1m2Z2ddl3m4Z4ddl5m6Z6ddl7m8Z8ddl9m:Z:ddl;mZ>m?Z?m@Z@ddlAmBZBddlCZCddlDZDddlEZEddlFmGZGdedeeHeIffdZJdedeeHd eHdee fd!ZKd"eHd#e dee fd$ZLd%ee d&eeHe fddfd'ZMd%ee dee fd(ZNd)ee dee fd*ZOe'fd+eeHd%ee d,eHd-eHd.eeHddf d/ZPy)0)datetimetimezone)DictListOptionalTuple)UUID)ThreadPoolExecutor)DecimalN)r) HTTPExceptionstatus)and_funcor_)IntegrityError)Sessionaliased) nullslast)RollupLogCreateRollupLogUpdateRollupLogResponse RollupOutput)settings) APIException) API_PREFIXESEXPORT_DATA_SUB_FOLDERRP_HEADER_LISTsize_conversions)QueryPaginator) RollupType RollupStatus Separator)create_rollup_logupdate_rollup_log)create_rollup_output) get_rates)get_vintage_map)get_bottle_size_map)get_retailers_map) get_codes_from_web_crawler_filesget_web_crawler_by_codeget_latest_file_content)get_mongo_id_by_wine_id)Pathdbreturnc K t|ttjt j t jtjd{}t|d{}t|d{}t|d{}t|dd{}t|d{}g}t|xsd}d} |D]} t!||j"t%d| d|  d{} t'|| d{} | rwt(j*} |j-| j/}t1|| j"d d{}|j3t5||| | |||d{| d z } t7||d{t9|d{}|j;dt j }d|j=dd}t?||t@t(jBjDtFd{t!||j"t%tjHt j t jd| d{}tKjL|S7O7?7/77 7777X7777 77?#tN$r}t!|j"t%tjPt j t jtS| d{7tUddtS|itVjXdYd}~yd}~wwxYww)a< 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. ) rollup_type date_startr )r/payloadNHISTORY)r/typerz Processing /)message)r/ rollup_log_idr4)r/code)r/web_crawler_idr6c"|jxsdS)N) wine_alert_id)xs \/var/www/html/wine-match-dev/backend/winematch-backend/src/apps/rollup/services/rp_rollup.pyz#execute_rp_rollup..}s!//*?R)keyForSalez%y.%m.%d.%H%M%Sz.txt) file_namerollup_output_list header_row separatordownload_location)r date_endr8rF)r rKr8execute_rp_rollup exceptionz"Error executing RP rollup process.)moduleerror status_coder8)-r#rr RPrnowrutcr!RUNNINGr'r(r)r*r&lenr$idrr+r"PIPEgetlowerr,extendprocess_rollup_outputsequalize_price"filter_by_median_price_per_wine_idsortstrftimecreate_rp_rollup_filerTABvaluerSUCCESSrmodel_validate ExceptionFAILEDstrrr HTTP_500_INTERNAL_SERVER_ERROR)r/ new_rollup vintage_mapbottle_size_map retailers_mapweb_crawler_codescurrency_ratesrG total_recordsprocessing_countr:updated_rollup_logcrawlerinput_delimiterretailerhistory_matches filtered_listrRrFfinished_rollupes rArLrL,s2u -#&MM#<< 5#++  ,B// 3B 77/33 #Cby"YY )},-.3! % "D(9(mm')*:);1]OL(" 4rEEG"+..(,,TZZ\:(?2V]V`V`gp(q"q"))0'' #'   ! A "D/@@@@ASTT  @Allncll+<=>dC $,%mm))4    !2$--##++!hll3#  !  !//@@ 073Z-"F #r AT     $--##**!hll3A    &A'==8    s<M7AKJ,K0J/1KJ2KJ5K+J8,K>J;?AKJ>KKAK,K-#KK K/K 0KK A3K6K7AKKK+M7,K/K2K5K8K;K>KKKK K KKK M4AM/<L?=-M/*M7/M44M7rursc. Kfd tj fd|Dd{}|Dcgc]}|| } }t| D cic]} | j| jf| c} j } | S7Vcc}wcc} ww)a 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. c K |jdd}|j}t|dddd{}t| |d{S77#t$r}t d|Yd}~yd}~wwxYww)Nr>r )r/wine_id) web_crawlerrtrowrjrkmongo_idzError processing line: )replacesplitr-r%reprint) linerrrxrkrrr/rsrtrjs rA process_linez,process_rollup_outputs..process_lines <<"-D**_-C4CF3BKPPH-#!' /!  Q  +A3/ 0 sPB9A AA AA BA A B)A<7B<BBc3.K|] }|yw)N).0rrs rA z)process_rollup_outputs..s$TD\$%7$TsN)asynciogatherlistactual_retailer_descriptionpricevalues) r/rursrrrtrjrkresultsresultnew_listitem unique_listrs ` ````` @rAr[r[s.&NN$TO$TUUG&-C60BCHCYabQU994::FLbiiklK  VDbs2)BB BB B  B B2B B bottle_sizercK|r|y td}|}|j}|tvrt||}|jtdS#t$r}t d|Yd}~yd}~wwxYww)z6 Standardizes prices to a 750 ml bottle size. N7500.01zError standardizing price: )r rYrquantizerer)rr standard_size new_pricerxs rAstandardize_pricers %-   !'')  * *(5e5.5r<Nr) r}appendr itemsrUlistingrfilter_by_median_pricerZ)rGrgroupedrr} low_filter wine_listrvs rAr]r]sFG"&,,$" ' !!GG % &J%mmo) y>Q  "+||(<(<AUAUYcAcM #9"CCM }  ! MM- (!)$ M Ds$A.C!00C C!1C2C!C!rcKg}td}g}|D]Y}|j|j|k\s |js-|jtt |j[t |dkDr |j t |dz}t |dzdk7r||}n||||dz ztdz }|tdz}|tdz }|Dcgc]:}|jr*|j |j|kr|j|k\r|<}}|S|Dcgc]+}|jr|jr|j|k\r|-}}|Scc}w#t$rU} td| |Dcgc]+}|jr|jr|j|k\r|-ncc}w}}Yd} ~ |Sd} ~ wwxYwcc}ww) z 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. rNrr<24z)Warning: Error calculating median price: ) r rrrrgrUr^rer) rrvr price_listr mid_numbermedian max_price min_pricerxs rArr=sMJJB  ,  J . LL   gc$*>*>&?@ A B  :  OO ZA-J:"a'#J/$Z0:j1n3MMQXY\Q]] -I-I"+ ))5))Y6))Y6 M*  ' <zError creating RP rollup file z: )rjoin to_stringr rQospathmakedirsdirnameaiofilesopenwriter.chmodstatS_IRUSRS_IWUSRS_IRGRPS_IROTHrer) rHrGrIrFrJline_terminatorsbr file_pathfilerxs rAr`r`|sY(A  )..,>?) TF IIf&&y*--@?R S TGGLL!2I>  BGGOOI.>==wG * *4**RWWR[) ) ) * * YdllT\\9DLLH4<<WX  * ) * * * * A .ykA3?@@AsF/CFE-F$E3<E/=E3 F E1 AF,F/-F/E31F3F9E< :FF F,F'"F/'F,,F/)Qrrtypingrrrruuidr concurrent.futuresr decimalr rfastapir r sqlalchemyrrrsqlalchemy.excrsqlalchemy.ormrrsqlalchemy.sql.expressionrsrc.apps.rollup.schemas.rolluprrrrsrc.core.configrsrc.core.exceptionsrsrc.utils.constantsrrrrsrc.utils.paginationrsrc.utils.enumsr r!r"#src.apps.rollup.services.rollup_logr#r$&src.apps.rollup.services.rollup_outputr%#src.apps.currency.services.currencyr&&src.apps.wine.vintage.services.vintager'.src.apps.wine.bottle_size.services.bottle_sizer((src.apps.wine.retailer.services.retailerr))src.apps.web_crawler.services.web_crawlerr*r+r, src.apps.wine.wine.services.winer-rrrpathlibr.rganyrLr[rr\r]rr`rrCrArs'..1)&&)+/ %,ff/??H9BNF E B B  #s(^B H00#Y00 ,0fW'AR,.T,-?.QUVY[bVbQc.hl.0)lAS)X\]iXj)X<D,><4 CU<H(> +AS +A\*+A+A +A  } +A  +ArC