ϟi_0dZddlmZddlZddlZddlmZddlmZmZm Z ddl Z ddl Z ejdde ejdd ejd d ejd d ejdd dZdZdZdZd!dZd"dZd#d$dZdZ d% d&dZ d'dZdZdZdZd#d(dZd)dZd*d+dZdZ d# d,d Zy)-ug MODULE 6: PostgreSQL Store — Genesis KB Ingestion Pipeline =========================================================== Handles page metadata tracking, ingestion logging, and content hash lookup for the platform KB pipeline. Design decisions: - customer_id=None is stored as '' (empty string) for global KB pages. This avoids the NULL != NULL edge case in PostgreSQL unique constraints while keeping the API clean (callers pass None for global). - All functions accept an optional `conn` parameter; if not provided, a new connection is created and closed after use (auto-commit pattern). - UNIQUE constraint on (platform, customer_id, url) works cleanly because customer_id is never NULL in the DB — it's always at least ''. # VERIFICATION_STAMP # Story: 6.01–6.05 # Verified By: parallel-builder # Verified At: 2026-02-26 # Tests: 13/13 # Coverage: 100% ) annotationsN)datetime)ListOptionalTuplePG_HOSTz(postgresql-genesis-u50607.vm.elestio.appPG_PORT25432PG_USERpostgresPG_PASSzCiBjh6LM7Yuqkq-jo2r7eQDwPG_DB)hostportuserpassworddbnamea CREATE TABLE IF NOT EXISTS platform_kb_pages ( id SERIAL PRIMARY KEY, platform VARCHAR(50) NOT NULL, customer_id VARCHAR(100) NOT NULL DEFAULT '', url TEXT NOT NULL, title TEXT, content_hash VARCHAR(64) NOT NULL, chunk_count INTEGER DEFAULT 0, last_ingested TIMESTAMPTZ DEFAULT NOW(), metadata JSONB DEFAULT '{}', UNIQUE(platform, customer_id, url) ); CREATE INDEX IF NOT EXISTS idx_pkp_platform ON platform_kb_pages(platform); CREATE INDEX IF NOT EXISTS idx_pkp_customer ON platform_kb_pages(customer_id); a} CREATE TABLE IF NOT EXISTS platform_kb_ingestion_log ( id SERIAL PRIMARY KEY, platform VARCHAR(50) NOT NULL, customer_id VARCHAR(100) NOT NULL DEFAULT '', started_at TIMESTAMPTZ DEFAULT NOW(), completed_at TIMESTAMPTZ, pages_fetched INTEGER DEFAULT 0, pages_changed INTEGER DEFAULT 0, chunks_created INTEGER DEFAULT 0, vectors_upserted INTEGER DEFAULT 0, errors INTEGER DEFAULT 0, status VARCHAR(20) DEFAULT 'running', metadata JSONB DEFAULT '{}' ); c4tjditS)z3Return a new PostgreSQL connection using PG_CONFIG.)psycopg2connect PG_CONFIGr)/mnt/e/genesis-system/core/kb/pg_store.pyget_connectionrWs    (i ((rc||StS)u;Convert None → '' so the UNIQUE constraint works cleanly.)_GLOBAL_CUSTOMER_ID) customer_ids r_normalize_customer_idr \s%1;J7JJrc:|du}|r t} |j5}|jt|jtddd|j |r|j yy#1swY-xYw#|r|j wwxYw)u( Create platform_kb_pages and platform_kb_ingestion_log tables if they do not already exist. Idempotent — safe to call on every startup. Parameters ---------- conn : psycopg2 connection, optional If not supplied, a new connection is created, used, and closed. N)rcursorexecuteCREATE_PAGES_TABLE_SQLCREATE_LOG_TABLE_SQLcommitclose)conn_owncurs r ensure_schemar+as 4BBa INSERT INTO platform_kb_pages (platform, customer_id, url, title, content_hash, chunk_count, last_ingested, metadata) VALUES (%(platform)s, %(customer_id)s, %(url)s, %(title)s, %(content_hash)s, %(chunk_count)s, NOW(), %(metadata)s) ON CONFLICT (platform, customer_id, url) DO UPDATE SET title = EXCLUDED.title, content_hash = EXCLUDED.content_hash, chunk_count = EXCLUDED.chunk_count, last_ingested = NOW(), metadata = EXCLUDED.metadata RETURNING id; c t|}tj|xsi} |j5} | j t ||||||| d| j } |j| dcdddS#1swYyxYw)a$ Upsert page metadata into platform_kb_pages. On conflict (same platform + customer_id + url), all mutable fields are updated. customer_id=None is treated as global (stored as ''). Returns ------- int The database row id of the inserted / updated page. platformrurltitle content_hash chunk_countmetadatarN)r jsondumpsr"r#UPSERT_PAGE_SQLfetchoner&) r(r.r/r0r1r2rr3cid meta_jsonr*rows r upsert_pager;s* ! -C 8>r*I # $" ,*%  lln 1vs ABB c >|sy|j5}|D]L\}}}}}}} t|} tj| xsi} |j t || ||||| dN|j dddt|S#1swYt|SxYw)z Batch upsert multiple pages in a single transaction. Each tuple in *pages* is: (platform, url, title, content_hash, chunk_count, customer_id, metadata) Returns the number of rows upserted. rr-N)r"r r4r5r#r6r&len) r(pagesr*r.r/r0r1r2rr3r8r9s rupsert_pages_batchr?s  #',  $XsE< ((5C 8>r2I KK (#&"$0#. )    #$ u:%$ u:s A"B  Bz INSERT INTO platform_kb_ingestion_log (platform, customer_id, status) VALUES (%(platform)s, %(customer_id)s, 'running') RETURNING id; aq UPDATE platform_kb_ingestion_log SET completed_at = NOW(), pages_fetched = %(pages_fetched)s, pages_changed = %(pages_changed)s, chunks_created = %(chunks_created)s, vectors_upserted = %(vectors_upserted)s, errors = %(errors)s, status = %(status)s, metadata = %(metadata)s WHERE id = %(log_id)s; a  SELECT id, platform, customer_id, started_at, completed_at, pages_fetched, pages_changed, chunks_created, vectors_upserted, errors, status, metadata FROM platform_kb_ingestion_log WHERE platform = %(platform)s ORDER BY started_at DESC LIMIT %(limit)s; ct|}|j5}|jt||d|j }|j |dcdddS#1swYyxYw)z Record the start of an ingestion run. Returns ------- int The log row id, to be passed to log_ingestion_complete(). r.rrN)r r"r# LOG_START_SQLr7r&)r(r.rr8r*r:s rlog_ingestion_startrCs\ ! -C # M#MNlln 1v s >A$$A-c|j5}|jt||jdd|jdd|jdd|jdd|jdd|jddt j |jd id |j d d d y #1swYy xYw) ax Finalise an ingestion log row with statistics. Parameters ---------- conn Active psycopg2 connection. log_id ID returned by log_ingestion_start(). stats Dict with keys: pages_fetched, pages_changed, chunks_created, vectors_upserted, errors, status ('completed' | 'failed'), and optionally metadata (dict). pages_fetchedr pages_changedchunks_createdvectors_upsertederrorsstatus completedr3)log_idrErFrGrHrIrJr3N)r"r#LOG_COMPLETE_SQLgetr4r5r&)r(rLstatsr*s rlog_ingestion_completerP s #  $)IIoq$A$)IIoq$A$)II.>$B$)II.@!$D$)IIh$:$)IIh $D$(JJuyyR/H$I   s B3C  Cc|jtjj5}|j t ||d|j }dddDcgc] }t|c}S#1swY"xYwcc}w)z Return the most recent ingestion runs for a platform, newest first. Returns ------- list[dict] Each dict has keys matching the platform_kb_ingestion_log columns. )cursor_factory)r.limitN)r"rextrasRealDictCursorr#LOG_HISTORY_SQLfetchalldict)r(r.rSr*rowsrs rget_ingestion_historyr[-sl HOO$B$B Cs O(U%KL||~" "DG "" #s*A6!B6A?zt SELECT url, content_hash FROM platform_kb_pages WHERE platform = %(platform)s AND customer_id = %(customer_id)s; ct|}|j5}|jt||d|j }dddDcic]\}}|| c}}S#1swYxYwcc}}w)a Return a {url: content_hash} mapping for all pages belonging to the given platform (and optionally scoped to a customer). Used by the orchestrator for change-detection: only pages whose hash differs from the stored value need re-ingestion. rAN)r r"r#GET_HASHES_SQLrW)r(r.rr8r*rYr/r1s rget_content_hashesr^Hsm ! -C # N#$NO||~8< <"3#|C  << =s*A$ A0$A-)returnzpsycopg2.extensions.connection)r Optional[str]r_str)N)r_None)NN)r.rar/rar0rar1rar2intrr`r3zOptional[dict]r_rc)r>zCList[Tuple[str, str, str, str, int, Optional[str], Optional[dict]]]r_rc)r.rarr`r_rc)rLrcrOrXr_rb) )r.rarSrcr_z list[dict])r.rarr`r_zdict[str, str]) __doc__ __future__rr4osrtypingrrrrpsycopg2.extrasgetenvrcrrr$r%rr r+r6r;r?rBrMrVrCrPr[r]r^rrrrksz.# (( BIIi!K L   )W- . BIIi , )%?@bii,   "$) K 6 ."&#'' '  '  '  ''' 'T! N! !P "@ #&"&=== =r