legal-ai/mcp-server/src/legal_mcp/services/bulletin_library.py

"""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