Li!dZddlmZddlZddlZddlZddlZddlZddlZddl m Z m Z m Z ddl m Z mZmZddlmZddlmZmZddlmZmZej0eZdd Zddd Zdd Zed k(rej<ej>ej@d eZ!e!jEZ#ejHee#jJZ&e'ejPe&de)yy)u Cron Scheduler — KB Ingestion Pipeline Module 11 ================================================= Runs nightly re-ingestion for all active platforms. Usage: python3 -m core.kb.cron python3 -m core.kb.cron --dry-run ) annotationsN)datetimetimezone timedelta)AnyDictOptional)ingest_platform)list_platforms get_platform)get_ingestion_historyget_connectioncpt||d}|sy|djd}|ytjtj }t |trtj|}|j |jtj }||z jdz }||k\S)uS Check if a platform needs re-ingestion based on last ingestion time. Parameters ---------- platform_name: The platform key (e.g. "telnyx", "hubspot"). refresh_hours: Minimum hours that must have elapsed since the last ingestion. If hours_since_last_run >= refresh_hours the platform is stale. conn: Active psycopg2 connection passed to get_ingestion_history(). Returns ------- bool True — platform should be re-ingested (never run, or stale). False — platform was ingested recently enough, skip it. )limitTr started_at)tzinfoi) r getrnowrutc isinstancestr fromisoformatrreplace total_seconds) platform_name refresh_hoursconnhistorylast_runr hours_sinces %/mnt/e/genesis-system/core/kb/cron.py_should_refreshr#$s($D-qAG qz~~l+H ,,x|| $C(C ))(3##8<<#8>002T9K - ''c Ktj}tjtjj }|dddiddd}t }t||d<d} t}|D]&}t|}|,tjd||dxxdz cc<d |d |<=t||j|}|s,tjd ||dxxdz cc<d |d |<|r,tjd ||d xxdz cc<d|d |<tjd| t|d{} |d xxdz cc<| |d |<tjd|| j!dd| j!dd) | |j)t+tj|z d|d<|S7#t"$rE} |dxxdz cc<t%| |d} | |d |<tj'd|| Yd} ~ d} ~ wwxYw#t"$rYwxYw#|! |j)w#t"$rYwwxYwwxYww)u Run nightly re-ingestion for all registered platforms. For each platform: 1. Check last ingestion time from PG via get_ingestion_history(). 2. If hours since last ingestion >= platform.refresh_hours → ingest. 3. Otherwise skip (platform is fresh). 4. Collect stats from each ingestion; aggregate into a summary result. Parameters ---------- dry_run: When True, check which platforms need refresh but do NOT ingest. "platforms_ingested" will count platforms that *would* be ingested, and all platform results for those will be "would_ingest". Returns ------- dict with keys: run_at — ISO 8601 timestamp of when this run started platforms_checked — total platforms evaluated platforms_ingested — platforms that were (or would be) ingested platforms_skipped — platforms that were up-to-date platform_results — {platform_name: stats_dict | "skipped" | "would_ingest"} errors — count of platforms that raised an exception total_duration_seconds — wall-clock duration of the whole run rg)run_atplatforms_checkedplatforms_ingestedplatforms_skippedplatform_resultserrorstotal_duration_secondsr'Nu6Platform '%s' listed but config not found — skippingr)rskippedr*u#Platform '%s' is fresh — skippingu&DRY RUN — would ingest platform '%s'r( would_ingestzIngesting platform '%s'z+Platform '%s' ingested: %d pages, %d chunks pages_fetchedchunks_createdr+)errorplatformz"Platform '%s' ingestion failed: %sr,)timerrrr isoformatr lenrr loggerwarningr#rinfor r Exceptionrr1closeround) dry_runt_startr&resultplatform_namesrnameconfigshouldstatsexc error_infos r"nightly_ingestionrGSs8iikG \\(,, ' 1 1 3F"%F$%N"%n"5F  D2"' ND!$'F~WY]^*+q0+3<)*40$T6+?+?FF A4H*+q0+3<)*40 DdK+,1,3A)*40 KK14 8 N-d33+,1,38)*40 AIIoq1II.2 ;' NV    (-TYY[7-BA'FF #$ M14 Nx A% '*3xTB 3=)*40 A4MM  N         sA*I'-CH?GGAG H?$I''H07&I'G H-(:H("H?(H--H?0 H<9I';H<<I'?I$II$ I I$I  I$$I'c\tjdd}|jddd|S)Nzpython3 -m core.kb.cronu7KB Nightly Cron Scheduler — re-ingest stale platforms)prog descriptionz --dry-run store_truez=Check which platforms need refresh without actually ingesting)actionhelp)argparseArgumentParser add_argument)parsers r" _build_parserrRs=  $ $ &MF  L Mr$__main__u2%(asctime)s %(levelname)s %(name)s — %(message)s)levelstreamformat)r=)indentdefault)rrrintreturnbool)F)r=r\r[zDict[str, Any])r[zargparse.ArgumentParser)*__doc__ __future__rrNasynciojsonloggingsysr4rrrtypingrrr core.kb.orchestratorr core.kb.platform_registryr r core.kb.pg_storer r getLogger__name__r7r#rGrR basicConfigINFOstderrrQ parse_argsargsrunr= run_resultprintdumpsrr$r"rss#  22&&1BB   8 $'(^dX  zGllzzC _F    D.t||DEJ *$**Z3 78r$