"""Projection of the Lefevre infospace into a content-explorer presentation.""" from __future__ import annotations import sys from pathlib import Path import json import yaml from infospace_bench.markdown_adapter import extract_section_text, parse_markdown_artifact from infospace_bench.presentation import display_title, project_infospace from infospace_bench.semantics import slugify ROOT = Path(__file__).resolve().parents[1] LEFEVRE = ROOT / "infospaces" / "lefevre-reminiscences-of-a-stock-operator" CONTENT_EXPLORER = Path("/home/worsch/content-explorer") sys.path.insert(0, str(CONTENT_EXPLORER)) from content_explorer.contract import validate_documents # noqa: E402 def test_lefevre_projection_matches_stored_artifacts_and_contract() -> None: first = project_infospace(LEFEVRE) second = project_infospace(LEFEVRE) snapshot = first.snapshot manifest = first.manifest assert [item["id"] for item in snapshot["entities"]] == [ item["id"] for item in second.snapshot["entities"] ] assert [item["id"] for item in snapshot["relations"]] == [ item["id"] for item in second.snapshot["relations"] ] index = yaml.safe_load((LEFEVRE / "artifacts" / "index.yaml").read_text(encoding="utf-8")) indexed = index["artifacts"] source_rows = [item for item in indexed if item["kind"] == "source"] entity_rows = [item for item in indexed if item["kind"] == "entity"] generated_rows = [item for item in indexed if item["kind"] == "generated"] relation_rows = [item for item in indexed if item["kind"] == "relation"] assert len(entity_rows) == 14 assert len(snapshot["entities"]) == 14 assert {item["kindId"] for item in snapshot["entities"]} >= { "instrument", "strategy", "market", "event", "evidence-bearing-claim", "institution", "error", "trader", } assert {item["id"] for item in snapshot["kinds"]} == { item["kindId"] for item in snapshot["entities"] } assert {item["id"] for item in snapshot["kinds"]}.isdisjoint( {"source", "entity", "relation", "generated"} ) book = source_rows[0]["provenance"]["book_metadata"] assert len(snapshot["sources"]) == 1 work = snapshot["sources"][0] assert work["license"] == book["rights"] == "Public domain in the USA." assert work["uri"] == book["source_url"] assert work["title"] == book["title"] attribution = work["attribution"] assert "provenance:" in attribution assert len(source_rows) == 6 for source in source_rows: provenance = source["provenance"] assert provenance["digest"] in attribution assert provenance["original_path"] in attribution assert provenance["chunk_id"] in attribution generated_titles = {item["title"] for item in generated_rows} generated_ids = {item["id"] for item in generated_rows} by_path = {item["properties"]["artifact_path"]: item for item in snapshot["entities"]} assert len(by_path) == 14 for row in entity_rows: projected = by_path[row["path"]] parsed = parse_markdown_artifact(LEFEVRE / row["path"]) heading = next(item.text for item in parsed.headings if item.level == 1) definition = extract_section_text(parsed, "Definition").strip() category = extract_section_text(parsed, "Category").strip() assert projected["title"] == display_title(heading) assert projected["summary"] == definition assert projected["tldr"] assert definition.startswith(projected["tldr"]) assert projected["kindId"] == slugify(category) assert projected["id"] == ( f"urn:ce:lefevre-reminiscences-of-a-stock-operator:" f"{projected['kindId']}:{slugify(projected['title'])}" ) assert projected["sourceRefs"] == [work["id"]] assert projected["properties"]["workflow_id"] == row["provenance"]["workflow_id"] assert projected["properties"]["input_artifact_id"] == row["provenance"]["input_artifact_id"] assert projected["title"] not in generated_titles assert projected["id"] not in generated_ids block_count = 0 for row in relation_rows: text = (LEFEVRE / row["path"]).read_text(encoding="utf-8") block_count += text.count("\n## Subject") + int(text.startswith("## Subject")) assert block_count == 27 assert len(first.resolved_relations) + len(first.quarantined_relations) == block_count assert first.quarantined_relations assert len(snapshot["relations"]) == len(first.resolved_relations) entities_by_id = {item["id"]: item for item in snapshot["entities"]} relations_by_id = {item["id"]: item for item in snapshot["relations"]} for record in first.resolved_relations: projected = relations_by_id[record.relation_id] assert projected["predicate"] == record.relation_type.replace("_", "-") assert " " not in projected["predicate"] if record.predicate_text != projected["predicate"]: assert projected["predicate"] != record.predicate_text assert slugify(record.subject) == slugify(entities_by_id[record.from_id]["title"]) assert slugify(record.object) == slugify(entities_by_id[record.to_id]["title"]) assert projected["from"] == record.from_id assert projected["to"] == record.to_id assert projected["sourceRefs"] == [work["id"]] resolved_keys = {(item.artifact_id, item.block_index) for item in first.resolved_relations} for record in first.quarantined_relations: assert (record.artifact_id, record.block_index) not in resolved_keys subject_hits = [ item for item in snapshot["entities"] if slugify(item["title"]) == slugify(record.subject) ] object_hits = [ item for item in snapshot["entities"] if slugify(item["title"]) == slugify(record.object) ] assert not (len(subject_hits) == 1 and len(object_hits) == 1) assert manifest["schemaVersion"] == "0.1" assert manifest["id"] == "lefevre-reminiscences-of-a-stock-operator" assert manifest["panels"] == ["summary", "relations", "evidence", "claims"] assert manifest["factPolicy"] == "show-all" assert set(manifest["kindViews"]) == {item["id"] for item in snapshot["kinds"]} for view in manifest["kindViews"].values(): assert view["columns"] assert "facets" in view assert set(manifest["home"]["featuredKinds"]) == {item["id"] for item in snapshot["kinds"]} assert set(manifest["relationLabels"]) == {item["predicate"] for item in snapshot["relations"]} for wording in manifest["relationLabels"].values(): assert wording["forward"] assert wording["inverse"] export_dir = LEFEVRE / "exports" / "content-explorer" assert json.loads((export_dir / "snapshot.json").read_text(encoding="utf-8")) == snapshot assert json.loads((export_dir / "manifest.json").read_text(encoding="utf-8")) == manifest errors = validate_documents(snapshot, manifest, require_size=True) assert errors == [] raw_errors = validate_documents(index, index, require_size=True) assert raw_errors note = (LEFEVRE / "exports" / "content-explorer" / "PROJECTION.md").read_text(encoding="utf-8") assert "lefevre-reminiscences-of-a-stock-operator" in note assert "content-explorer only presents" in note.lower() assert "infospace-bench still owns" in note.lower() print("ENTITY_IDS " + ",".join(item["id"] for item in snapshot["entities"])) print("RELATION_IDS " + ",".join(item["id"] for item in snapshot["relations"])) print("VALIDATOR_ERRORS 0") print("RAW_ARTIFACTS_ERRORS " + str(len(raw_errors)))