"""Stable, filtered document pagination over revision-pinned Parquet.""" from __future__ import annotations import hashlib,json from datetime import timezone from pathlib import Path from duckdb_runtime import query_rows def check_exchange(value): if value not in ("ASX","NZX"):raise ValueError("invalid exchange") return value def fingerprint(filters):return hashlib.sha256(json.dumps(filters,sort_keys=True,separators=(",", ":"),default=str).encode()).hexdigest() def artifacts(cache_dir,table): files=sorted(str(path) for path in (Path(cache_dir)/"data/v1"/table).rglob("*.parquet")) if not files:raise RuntimeError("No %s Parquet artifacts"%table) return files def iso(value): if value is None:return None if hasattr(value,"astimezone"):return value.astimezone(timezone.utc).isoformat().replace("+00:00","Z") return value.isoformat() def search_documents(cache_dir,*,exchange="ASX",ticker=None,instrument_id=None,doc_type=None,date_from=None,date_to=None,direction="desc",limit=25,after=None): if direction not in {"asc","desc"}:raise ValueError("invalid direction") conditions=[f"d.exchange='{check_exchange(exchange)}'"];params=[artifacts(cache_dir,"documents")] if ticker:conditions.append("d.ticker=?");params.append(ticker) if doc_type:conditions.append("d.doc_type=?");params.append(doc_type) if date_from:conditions.append("d.announcement_date>=?::DATE");params.append(str(date_from)) if date_to:conditions.append("d.announcement_date<=?::DATE");params.append(str(date_to)) if instrument_id: conditions.append("EXISTS (SELECT 1 FROM read_parquet(?) e WHERE e.document_id=d.document_id AND e.instrument_id=?)");params.extend([artifacts(cache_dir,"events"),instrument_id]) sort_time="coalesce(d.announced_at,cast(d.announcement_date as TIMESTAMPTZ))";op=">" if direction=="asc" else "<" if after: conditions.append(f"({sort_time} {op} ?::TIMESTAMPTZ OR ({sort_time}=?::TIMESTAMPTZ AND d.document_id {op} ?))");params.extend([after[0],after[0],after[1]]) sql=f"""SELECT d.document_id,d.exchange,d.ticker,d.source_document_id,d.title,d.doc_type,d.doc_type_source,d.doc_type_confidence,d.announced_at,d.announcement_date,d.raw_artifact_key,d.raw_size_bytes,d.markdown_artifact_key,d.markdown_size_bytes,d.publication_state FROM read_parquet(?) d WHERE {' AND '.join(conditions)} ORDER BY {sort_time} {direction.upper()},d.document_id {direction.upper()} LIMIT ?""" params.append(limit+1);rows=query_rows(sql,params);names=["document_id","exchange","ticker","source_document_id","title","doc_type","doc_type_source","doc_type_confidence","announced_at","announcement_date","raw_artifact_key","raw_size_bytes","markdown_artifact_key","markdown_size_bytes","publication_state"] items=[] for row in rows[:limit]: item=dict(zip(names,row));item["announced_at"]=iso(item["announced_at"]);item["announcement_date"]=iso(item["announcement_date"]);items.append(item) next_values=[items[-1]["announced_at"] or items[-1]["announcement_date"]+"T00:00:00Z",items[-1]["document_id"]] if len(rows)>limit else None return items,next_values def get_document(cache_dir,document_id): paths=artifacts(cache_dir,"documents");rows=query_rows("SELECT document_id,exchange,ticker,source_document_id,title,doc_type,doc_type_source,doc_type_confidence,announced_at,announcement_date,raw_artifact_key,raw_size_bytes,markdown_artifact_key,markdown_size_bytes,publication_state FROM read_parquet(?) WHERE document_id=? LIMIT 1",[paths,document_id]);row=rows[0] if rows else None if not row:return None names=["document_id","exchange","ticker","source_document_id","title","doc_type","doc_type_source","doc_type_confidence","announced_at","announcement_date","raw_artifact_key","raw_size_bytes","markdown_artifact_key","markdown_size_bytes","publication_state"];item=dict(zip(names,row));item["announced_at"]=iso(item["announced_at"]);item["announcement_date"]=iso(item["announcement_date"]);return item