"""Ingest a monthly "עו"ד על נדל"ן" bulletin into the digests radar (X12). A bulletin PDF is multi-topic: it EXPLODES into several digest rows — one per case-law pointer (digest_kind='decision') and one per article (digest_kind= 'article'), all tagged publication='עו"ד על נדל"ן' to distinguish them from the daily "כל יום" issues. This reuses the existing radar (no parallel corpus — G2): the case pointers join search_digests / the /digests page and autolink to the underlying ruling exactly like a daily digest; articles are deep-context only. LOCAL-ONLY (LLM split + embedding) — host scripts/MCP, never the container path. Idempotent: each item's content_hash (hash of its analysis_text) is the dedup key, so re-running a bulletin skips already-ingested items. """ from __future__ import annotations import logging from pathlib import Path from legal_mcp.services import db, embeddings, extractor from legal_mcp.services import bulletin_splitter, digest_library logger = logging.getLogger(__name__) PUBLICATION = 'עו"ד על נדל"ן' SOURCE_FIRM = "צבי שוב + רונית אלפר, עורכי דין" async def _store_and_embed(digest_row: dict) -> None: """Compute + store the single radar embedding for a freshly created item.""" emb_text = digest_library._embedding_text(digest_row) if not emb_text: return try: vecs = await embeddings.embed_texts([emb_text], input_type="document") if vecs: await db.store_digest_embedding(digest_row["id"], vecs[0]) except Exception as e: # §6 — surfaced, not swallowed logger.warning("bulletin item embedding failed for %s: %s", digest_row.get("id"), e) async def _create_item(*, analysis_text: str, kind: str, concept_tag: str, headline: str, summary: str, citation: str, court: str, practice_area: str, subject_tags: list[str], src: str) -> dict | None: """Create one digest row from a bulletin item. Returns the row, or None if it already exists (idempotent skip) or the insert raced on content_hash.""" content_hash = db._content_hash(analysis_text) if await db.get_digest_by_content_hash(content_hash): return None try: return await db.create_digest( analysis_text=analysis_text, publication=PUBLICATION, source_firm=SOURCE_FIRM, concept_tag=concept_tag, headline_holding=headline, summary=summary, underlying_citation=citation, underlying_court=court, practice_area=practice_area, subject_tags=subject_tags, source_document_path=src, extraction_status="completed", digest_kind=kind, ) except Exception as e: # uq_digests_content_hash race (concurrent run) → treat as already-present. if "uq_digests_content_hash" in str(e): return None raise async def ingest_bulletin(file_path: str, model: str | None = None) -> dict: """Split a bulletin PDF into digest rows (case pointers + articles). Returns counts: {cases, articles, created, skipped, linked}. Idempotent. """ path = str(file_path) raw_text, _pages, _meta = await extractor.extract_text(path) split = await bulletin_splitter.split(raw_text, model=model) cases, articles = split.get("cases", []), split.get("articles", []) out = {"file": Path(path).name, "cases": len(cases), "articles": len(articles), "created": 0, "skipped": 0, "linked": 0} for c in cases: # analysis_text bundles the pointer's substance → stable per-item hash. atext = "\n".join(p for p in ( c["concept_tag"], c["headline_holding"], c["summary"], c["underlying_citation"] ) if p).strip() row = await _create_item( analysis_text=atext, kind="decision", concept_tag=c["concept_tag"], headline=c["headline_holding"], summary=c["summary"], citation=c["underlying_citation"], court=c["underlying_court"], practice_area=c["practice_area"], subject_tags=c["subject_tags"], src=path, ) if row is None: out["skipped"] += 1 continue out["created"] += 1 await _store_and_embed(row) linked = await digest_library.try_autolink(row["id"], c["underlying_citation"]) if linked: out["linked"] += 1 for a in articles: # The article body is the substance; prefix authors into the summary. body = a["body"] or a["summary"] summary = (f"מאת {a['authors']}. " if a["authors"] else "") + (a["summary"] or "") atext = "\n".join(p for p in (a["title"], summary, body) if p).strip() row = await _create_item( analysis_text=atext, kind="article", concept_tag=a["title"], headline=a["title"], summary=summary, citation="", court="", practice_area=a["practice_area"], subject_tags=a["subject_tags"], src=path, ) if row is None: out["skipped"] += 1 continue out["created"] += 1 await _store_and_embed(row) return out