o (yi@@sFddlmZddlZddlZddlZddlZddlmZmZmZddl m Z ddl Z ddl m ZddlmZmZmZmZmZmZmZmZmZddlmZmZmZmZdd lmZdd l m!Z!m"Z"dd l#m$Z$dd l%m&Z&ee'Z(e d Z)dZ*ej+,ej+-e.dZ/dZ0dZ1dZ2dZ3ddZ4ddZ5ddZ6ej7dddddZ8dS) ) getLoggerN)datetimetimezone timedelta) load_dotenv celery_app) sanitize_col_namesave_watermarkDB_PATHget_system_stateset_system_staterecreate_deals_table create_deals_table_if_not_exists recreate_user_restrictions_tableget_deal_count)fetch_deals_for_dealsfetch_product_batch validate_asinfetch_seller_data) TokenManager)_process_single_dealclean_numeric_values)get_seller_info_for_single_deal)clear_analysis_cachez2.9-persistent-statedealsz headers.jsonbackfill_deals_locki/ zbackfill_state.jsoncCs(tdddtjd}|t|d}|S)zFConverts Keepa time (minutes since 2011-01-01) to ISO 8601 UTC string.ir )tzinfo)minutes)rrutcr isoformat) keepa_minutes keepa_epoch dt_objectr&1/var/www/agentarbitrage/keepa_deals/backfiller.py_convert_keepa_time_to_iso*sr(c Cstd}|dur!zt|WSty td|dYdSwtjtrt dz3t td#}t |}| dd}t d |d t||WdWS1sXwYWdSt jttfy~}ztd |WYd}~dSd}~wwdS) z Loads the last completed page from the system_state table. Migrates from legacy JSON file if DB entry is missing. backfill_pageNzInvalid backfill_page in DB: z. Defaulting to 0.rz5Backfill state missing in DB. Checking legacy file...rlast_completed_pagez"Found legacy backfill state: page z. Migrating to DB.z%Error loading legacy backfill state: )r int ValueErrorloggererrorospathexistsSTATE_FILE_LEGACYinfoopenjsonloadgetsave_backfill_stateJSONDecodeErrorFileNotFoundError)valfdatapageer&r&r'load_backfill_state0s2       $ rAcCs$tdt|td|ddS)z8Saves the last completed page to the system_state table.r)z/--- Backfill state saved. Last completed page:  ---N)rstrr.r4) page_numberr&r&r'r9Nsr9z%keepa_deals.backfiller.backfill_deals)nameFc. st|rJtdtdtjtr?ztttdtdWnt y>}zt d|WYd}~nd}~wwt t tdt jtjj}|jttd}|jdd sft d dSzzatd ttd }td }|stdWW|tddSt|}|t}td|d tddkrttdd}t } | |krtd| d|dntd|dt!||d|d\} } } |"| | rd| vs| ddstdnd d!| ddD} td"t#| d#|d|dkr&| r&| d$d$}|r&t%|}t&|t'dt#| t(D]}| ||t(}|s>q.td%|t(d&d't#| t(d&t(d(|dd)d!|D}t)z.t*+t,}|-}d*.d+t#|}|/d,t0d-|d.|d/d0|1D|2Wnt3y}zt d1|WYd}~nd}~wwfd2d!|D}i}|rtd3t#|d4|4d5t#|t5||d6d&d5d7\}} } }|r|"||rd8|vr|6d9d:|d8Dt7}|rEtd;t#|d<|4d&t#||4d5t#|t5||d6d&d5d7\}} } }|r1|"||rEd8|vrE|6d=d:|d8Dtd>t#|d?g}|D]B}|d@}||vraqT||}|6|t8|||}t9|||} | rt:| } t;?| dA<dB| dC<|@| tABd&qT|rtdDt#|dEd}!zzt*+t,}!|!-}"dFd!|D}#|#rdG.d+gt#|#}$d,t0d-|$d.}%|"/|%|#dHd0|"1Dt#}&t#|#|&}'tdI|'dJ|&dKtCtD }(tEF|(Wdn 1s wYdLd!D})|)GdAdCgdG.d+gt#|)}dMd!|)D}*dNt0dOdG.|*dP|d.}+fdQd!|D},|"H|+|,|!ItdRt#|dSddTlJmK}-dUd!|D}|rr|-jLdV|gdW|-LdXtdYt#|dZWnt*jMy}z tjd[|dd\d}~wwW|!r|!2n |!r|!2wwtABd]q.t||d&7}tABd&qtd^Wnt3y}ztjd_|dd\WYd}~nd}~wwW|tddSW|tddS|tdw)`Nz6Reset requested. Clearing backfill state and database.rzRemoved legacy state file .z$Could not remove legacy state file: z>Database has been reset (Deals and User Restrictions cleared).)timeoutF)blockingzD--- Task: backfill_deals is already running. Skipping execution. ---z$--- Task: backfill_deals started --- KEEPA_API_KEY XAI_TOKENz KEEPA_API_KEY not set. Aborting.z+--- Task: backfill_deals lock released. ---z --- Resuming backfill from page rBTbackfill_limit_enabledtruebackfill_limit_counti z'--- Artificial backfill limit reached (z >= z). Stopping backfill. ---zFetching page z of deals...)use_deal_settings token_managerrdrz)No more deals found. Pagination complete.cSsg|] }t|dr|qSasin)rr8.0dr&r&r' sz"backfill_deals..zFound z deals on page lastUpdatez--- Processing chunk r /z on page cSsg|]}|dqSrQr&rSr&r&r'rV,?zSELECT ASIN FROM z WHERE ASIN IN ()cSh|]}|dqSrr&)rTr*r&r&r' rYz!backfill_deals..z Failed to check existing ASINs: csg|]}|vr|qSr&r&)rTa)existing_asinsr&r'rVzFetching full history for z NEW deals...rim)dayshistoryoffersproductscSi|]}|d|qSrQr&rTpr&r&r' z"backfill_deals..zFetching lightweight stats for z EXISTING deals...cSrgrQr&rhr&r&r'rjrkzFetched product data for z ASINs in chunk.rR last_seen_utc backfillersourcez Upserting z. processed deals from chunk into the database.cSs g|] }|dr|dqSASINr8rTrowr&r&r'rVs z, cSr]r^r&rrr&r&r'r_rYz--- Upsert Stats: z New Deals, z Refreshed Deals ---cSsg|]}t|qSr&)r rThr&r&r'rV"rYcSsg|]}d|dqS)"r&)rTcolr&r&r'rV&rbzINSERT OR REPLACE INTO z (z ) VALUES (cs6g|]tfddDddfqS)c3s|]}|VqdS)Nrqrtrsr&r' (sz,backfill_deals...rlrn)tupler8)rT) headers_datarxr'rV(s6zSuccessfully upserted z deals.rcSsg|] }d|vr|dqSror&rSr&r&r'rV.sz4keepa_deals.sp_api_tasks.check_restriction_for_asins)argsz+keepa_deals.simple_task.update_recent_dealsz#--- Triggered downstream tasks for z ASINs. ---z&Database error while upserting deals: )exc_info<z&--- Task: backfill_deals finished. ---z5An unexpected error occurred in backfill_deals task: )Nrr.r4r9r0r1r2r3removeOSErrorwarningrrredisRedisfrom_urlceleryconf broker_urllockLOCK_KEY LOCK_TIMEOUTacquirergetenvr/releaser sync_tokensrAr r,rrupdate_after_calllenr8r(r rangeDEALS_PER_CHUNKsetsqlite3connectr cursorjoinexecute TABLE_NAMEfetchallclose Exceptionrequest_permission_for_callrupdatelistrrrrnowrr!r"appendtimesleepr5 HEADERS_PATHr6r7extend executemanycommitworkerr send_taskError).resetr@ redis_clientrapi_key xai_api_keyrOr? limit_count current_count deal_response_ tokens_left deals_on_page newest_tswm_isoi chunk_deals all_asins conn_checkc_check placeholders new_asinsall_fetched_products prod_respt_left existing_listrows_to_upsertdealrR product_dataseller_data_cache processed_rowconnrasin_list_upsertplaceholders_check query_checkcount_refreshed count_newr= db_columnsquoted_columnsquerydata_to_insertrr&)rar{r'backfill_dealsSsP          J    6  #             >  r)F)9loggingrr0r6rrrrrdotenvrrrrrdb_utilsr r r r rrrrr keepa_apirrrrrOr processingrr seller_inforstable_calculationsr__name__r.BACKFILLER_VERSIONrr1rdirname__file__rrrrr3r(rAr9taskrr&r&r&r's:   ,