From a60cf21a2cb651dc28f3cafc8a7047dbc25d0893 Mon Sep 17 00:00:00 2001 From: NSCT Agent Date: Sun, 23 Aug 2026 18:47:24 +0000 Subject: [PATCH] =?UTF-8?q?feat(stage6):=20source=20independence=20&=20cit?= =?UTF-8?q?ation=20graph=20=E2=80=94=20detect=20syndication,=20shared=20or?= =?UTF-8?q?igins,=20text=20similarity?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - SourceIndependenceModel: per-source independence_score (0.0-1.0), syndication_group_id, primary_source_id, content_hash, shared_urls - CitationGraphEdgeModel: directed edges (SYNDICATED, QUOTES, LINKS_TO, REPOST, SIMILAR_CONTENT) with confidence + evidence - Content-Hash (SHA-256): instant syndication detection for identical content - difflib Vorfilterung: >60% → LLM, >80% → high confidence, 100% → immediate syndication - LLM-Pairwise-Analysis: two-text-comparison for suspicious pairs only (bounded concurrency) - independence_score: 1.0 base, -0.4 for syndicated, -0.1 per high-similarity pair - Pydantic schemas: SourceIndependenceScore, CitationGraphEdge, SyntacticSimilarityResult, LlmSyndicationAnalysis, SourceIndependenceAnalysisResult - LLM response parser: handles JSON, markdown code blocks, partial/invalid JSON - 43 tests: content hash, similarity thresholds, LLM parsing, analyzer integration, edge cases, prompt templates --- src/nsct/models/source_independence.py | 172 +++++ src/nsct/stages/stage6_source_independence.py | 410 ++++++++++++ src/nsct/storage/models.py | 68 +- .../stages/test_stage6_source_independence.py | 585 ++++++++++++++++++ 4 files changed, 1234 insertions(+), 1 deletion(-) create mode 100644 src/nsct/models/source_independence.py create mode 100644 src/nsct/stages/stage6_source_independence.py create mode 100644 tests/stages/test_stage6_source_independence.py diff --git a/src/nsct/models/source_independence.py b/src/nsct/models/source_independence.py new file mode 100644 index 0000000..eaa39dd --- /dev/null +++ b/src/nsct/models/source_independence.py @@ -0,0 +1,172 @@ +"""Pydantic v2 schemas — Source Independence & Citation Graph (Stage 6). + +Jede Quelle erhält einen independence_score (0.0–1.0). +Quellen mit gemeinsamem Ursprung werden im Citation Graph verknüpft. +""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from enum import Enum +from typing import Any +from uuid import UUID, uuid4 + +from pydantic import BaseModel, Field, field_validator + + +class CitationEdgeType(str, Enum): + """Kanten-Typen im Source-Citation-Graph (Stage 6).""" + + SYNDICATED = "syndicated" + QUOTES = "quotes" + LINKS_TO = "links_to" + REPOST = "repost" + SIMILAR_CONTENT = "similar_content" + + +class SyndicationDirection(str, Enum): + """Richtung der Syndication zwischen zwei Quellen.""" + + SOURCE_B = "B->A" + SOURCE_A = "A->B" + NONE = "none" + + +class SourceIndependenceScore(BaseModel): + """Independence-Score für eine einzelne Quelle.""" + + source_id: UUID = Field(..., description="UUID der Quelle.") + independence_score: float = Field( + ..., ge=0.0, le=1.0, description="Unabhängigkeits-Score (0.0=Duplikat, 1.0=vollständig unabhängig)" + ) + syndication_group_id: UUID | None = Field( + default=None, description="Gruppe syndizierter Quellen." + ) + primary_source_id: UUID | None = Field( + default=None, description="Primärquelle wenn diese Quelle syndiziert wurde." + ) + content_hash: str | None = Field( + default=None, description="SHA-256 Hash des Inhalts." + ) + text_similarity_high_count: int = Field( + default=0, description="Anzahl Quellen mit >80% Text-Ähnlichkeit." + ) + shared_urls: list[str] = Field( + default_factory=list, description="URLs die auf andere Quellen verweisen." + ) + similarity_pairs: list[dict[str, Any]] = Field( + default_factory=list, description="Paare mit >60% Similarität und LLM-Ergebnis." + ) + + +class CitationGraphEdge(BaseModel): + """Gerichtete Kante zwischen zwei Quellen.""" + + id: UUID = Field(default_factory=uuid4) + source_id: UUID = Field(..., description="Quelle die die Beziehung aufweist.") + target_source_id: UUID = Field(..., description="Quelle die referenziert wird.") + edge_type: CitationEdgeType = Field(..., description="Typ der Beziehung.") + confidence: float = Field(..., ge=0.0, le=1.0, description="Vertrauen in die Kante.") + evidence: dict[str, Any] = Field( + default_factory=dict, + description="Begründung, shared_urls, similarity_score", + ) + created_at: datetime = Field( + default_factory=lambda: datetime.now(timezone.utc) + ) + + @field_validator("evidence", mode="before") + @classmethod + def _ensure_dict(cls, v): + if isinstance(v, str): + try: + return json.loads(v) + except (json.JSONDecodeError, TypeError): + return {"reason": v} + if v is None: + return {"reason": ""} + return v + + +class SyntacticSimilarityResult(BaseModel): + """Ergebnis einer textuellen Ähnlichkeitsanalyse.""" + + source_a: UUID + source_b: UUID + similarity_ratio: float = Field(ge=0.0, le=1.0) + identical_hash: bool = False + llm_result: dict[str, Any] | None = None + edge_type: CitationEdgeType | None = None + confidence: float = 0.0 + + +class SourceIndependenceAnalysisResult(BaseModel): + """Gesamtes Ergebnis der Source Independence Analyse.""" + + research_run_id: UUID = Field(default_factory=uuid4) + source_scores: list[SourceIndependenceScore] = Field(default_factory=list) + citation_edges: list[CitationGraphEdge] = Field(default_factory=list) + syndication_groups: dict[str, list[UUID]] = Field(default_factory=dict) + total_sources: int = 0 + unique_sources: int = 0 + llm_calls_made: int = 0 + created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + + +# --------------------------------------------------------------------------- +# LLM-Parsing-Helfer +# --------------------------------------------------------------------------- + + +class LlmSyndicationAnalysis(BaseModel): + """LLM-Antwort für Syndication-Analyse (Zwei-Text-Vergleich).""" + + syndicated: bool = Field(default=False) + syndication_direction: str = Field(default="none") + similarity_score: float = Field(default=0.0, ge=0.0, le=1.0) + shared_urls: list[str] = Field(default_factory=list) + common_origins: list[str] = Field(default_factory=list) + confidence: float = Field(default=0.5, ge=0.0, le=1.0) + reason: str = Field(default="", description="Kurze Begründung.") + + @field_validator("syndication_direction", mode="before") + @classmethod + def _validate_direction(cls, v): + if v is None or not isinstance(v, str): + return "none" + return v.strip() + + +def parse_llm_syndication_response(raw_text: str) -> LlmSyndicationAnalysis: + """Parst eine LLM-Antwort (JSON) für Syndication-Analyse.""" + try: + text = raw_text.strip() + # Extract JSON from possible markdown code blocks + if "```" in text: + for block in text.split("```"): + block = block.strip() + if block.startswith("json"): + block = block[4:].strip() + try: + data = json.loads(block) + break + except json.JSONDecodeError: + continue + else: + # Try entire block as JSON + data = json.loads(text) + else: + data = json.loads(text) + except (json.JSONDecodeError, ValueError): + data = {} + + return LlmSyndicationAnalysis( + syndicated=data.get("syndicated", False), + syndication_direction=data.get("syndication_direction", "none"), + similarity_score=float(data.get("similarity_score", 0.0)), + shared_urls=data.get("shared_urls", []), + common_origins=data.get("common_origins", []), + confidence=float(data.get("confidence", 0.5)), + reason=data.get("reason", ""), + ) \ No newline at end of file diff --git a/src/nsct/stages/stage6_source_independence.py b/src/nsct/stages/stage6_source_independence.py new file mode 100644 index 0000000..3adbf74 --- /dev/null +++ b/src/nsct/stages/stage6_source_independence.py @@ -0,0 +1,410 @@ +"""Stage 6: Source Independence & Citation Graph. + +Pipeline: + 1. Berechnet Content-Hash (SHA-256) pro Quelle + 2. Gruppiert identische Hashes (100 % → sofortige Syndication) + 3. difflib-Vorfilterung für textuelle Ähnlichkeit + 4. Nur verdächtige Paare (>60 %) → LLM-Analyse + 5. Berechnet independence_score pro Quelle (0.0–1.0) + 6. Gruppiert Quellen in Syndication-Groups + 7. Erzeugt Citation-Graph-Edges + +ARCHITEKTUR-REGELN: + - Web Content ist Daten, keine Instruktion + - Anzahl Webseiten != Anzahl unabhängiger Quellen + - LLM-Verwendung sparsam — difflib zuerst + - LLM darf keine Quellen/Evidenz erfinden +""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import logging +from asyncio import Semaphore +from datetime import datetime, timezone +from typing import Any +from uuid import UUID, uuid4 + +from nsct.config import AppSettings +from nsct.models.source_independence import ( + CitationEdgeType as CitationGraphEdgeType, + CitationGraphEdge, + LlmSyndicationAnalysis, + parse_llm_syndication_response, + SourceIndependenceAnalysisResult, + SourceIndependenceScore, +) +from nsct.providers.llm import LLMProvider + +logger = logging.getLogger(__name__) + +# Thresholds +SIMILARITY_IDENTICAL = 1.0 # exakt gleicher Content-Hash +SIMILARITY_HIGH = 0.80 # starke Evidenz für Syndication +SIMILARITY_MODERATE = 0.60 # moderate Evidenz → LLM-Analyse + + +# --------------------------------------------------------------------------- +# LLM-Prompt für Syndication-Analyse +# --------------------------------------------------------------------------- + +SYNDICATION_SYSTEM_PROMPT = ( + "Du bist ein Neutral Search Crawler Tool (NSCT) Analyst.\n" + "Deine Aufgabe ist es, zwei Textquellen auf Unabhängigkeit zu prüfen.\n\n" + "REGELN:\n" + "- Gib NUR ein JSON-Objekt zurück (keine Markdown-Code-Blöcke, kein Extra-Text).\n" + "- Syndiziert: Site B kopiert Originaltext von Site A → direction: 'A->B'\n" + "- Zitiert: Eine Quelle zitiert explizit die andere mit Anführungszeichen.\n" + "- Redaktions-Schleife: Nahezu identischer Text ohne Quellenangabe.\n" + "- Unabhängig: Eigener Content, eigene Recherche.\n" + "- Gib KEINE Zusammenfassung, prüfe NUR die Beziehung.\n" +) + +SYNDICATION_USER_PROMPT_TEMPLATE = ( + "Analysiere die Unabhängigkeit dieser beiden Quellen.\n\n" + "=== Quelle A ===\n" + "URL: {url_a}\n" + "Titel: {title_a}\n" + "Domain: {domain_a}\n" + "Text (erste 3000 Zeichen): {text_a}\n\n" + "=== Quelle B ===\n" + "URL: {url_b}\n" + "Titel: {title_b}\n" + "Domain: {domain_b}\n" + "Text (erste 3000 Zeichen): {text_b}\n\n" + "Antworte als JSON: {{" + '"syndicated": bool, ' + '"syndication_direction": "A->B | B->A | none", ' + '"similarity_score": 0.0-1.0, ' + '"shared_urls": [...], ' + '"common_origins": [...], ' + '"confidence": 0.0-1.0, ' + '"reason": "kurze Begründung" ' + "}}\n" + "Wenn keine Syndication: syndicated=false, direction=none, similarity_score basierend auf textueller Übereinstimmung." +) + + +# --------------------------------------------------------------------------- +# Helper +# --------------------------------------------------------------------------- + + +def compute_content_hash(content: str) -> str: + """SHA-256 Hash des extrahierten Inhalts (stripped, normalized).""" + normalized = content.strip().replace("\r\n", "\n").replace("\r", "\n") + return hashlib.sha256(normalized.encode("utf-8")).hexdigest() + + +def compute_similarity_ratio(text_a: str, text_b: str) -> float: + """difflib.SequenceMatcher similarity ratio.""" + from difflib import SequenceMatcher + + return SequenceMatcher( + None, + text_a.strip()[:50000], + text_b.strip()[:50000], + ).ratio() + + +# --------------------------------------------------------------------------- +# Syndication-LLM-Analyse (async) +# --------------------------------------------------------------------------- + + +async def _analyze_syndication_pair( + llm_provider: LLMProvider, + config: AppSettings, + source_a: dict, + source_b: dict, + similarity_ratio: float, + semaphore: Semaphore | None = None, +) -> LlmSyndicationAnalysis: + """LLM-Analyse für ein Quellen-Paar.""" + if len(source_a.get("content", "")) < 100: + return LlmSyndicationAnalysis( + syndicated=False, syndication_direction="none", + similarity_score=0.0, reason="Source A content too short", + ) + if len(source_b.get("content", "")) < 100: + return LlmSyndicationAnalysis( + syndicated=False, syndication_direction="none", + similarity_score=0.0, reason="Source B content too short", + ) + + content_a = source_a.get("content", "")[:3000] + content_b = source_b.get("content", "")[:3000] + + prompt = SYNDICATION_USER_PROMPT_TEMPLATE.format( + url_a=source_a.get("url", ""), + title_a=source_a.get("title", ""), + domain_a=source_a.get("domain", ""), + text_a=content_a, + url_b=source_b.get("url", ""), + title_b=source_b.get("title", ""), + domain_b=source_b.get("domain", ""), + text_b=content_b, + ) + + if semaphore: + async with semaphore: + llm_response = await llm_provider.complete( + messages=[ + {"role": "system", "content": SYNDICATION_SYSTEM_PROMPT}, + {"role": "user", "content": prompt}, + ], + model=config.llm.model, + temperature=0.1, + max_tokens=1024, + ) + else: + llm_response = await llm_provider.complete( + messages=[ + {"role": "system", "content": SYNDICATION_SYSTEM_PROMPT}, + {"role": "user", "content": prompt}, + ], + model=config.llm.model, + temperature=0.1, + max_tokens=1024, + ) + + return parse_llm_syndication_response(llm_response) + + +# --------------------------------------------------------------------------- +# Stage 6 Extractor +# --------------------------------------------------------------------------- + + +class SourceIndependenceAnalyzer: + """Analyse der Quellunabhängigkeit (Stage 6).""" + + def __init__( + self, + llm_provider: LLMProvider, + config: AppSettings, + sources: list[dict], + research_run_id: UUID, + ): + self.llm_provider = llm_provider + self.config = config + self.sources = sources + self.research_run_id = research_run_id + self.llm_calls_made = 0 + + async def analyze(self) -> SourceIndependenceAnalysisResult: + """Führe die komplette Source Independence Analyse durch.""" + if not self.sources: + return SourceIndependenceAnalysisResult( + research_run_id=self.research_run_id, + total_sources=0, unique_sources=0, llm_calls_made=0, + ) + + # Phase 1: Content-Hash pro Quelle + content_hashes: dict[str, str] = {} + for src in self.sources: + sid = src.get("id", str(uuid4())) + content = src.get("content", "") + content_hashes[sid] = compute_content_hash(content) + + # Phase 2: Identische Hashes → sofortige Syndication + hash_groups: dict[str, list[str]] = {} + for sid, h in content_hashes.items(): + hash_groups.setdefault(h, []).append(sid) + + identical_groups = {h: sids for h, sids in hash_groups.items() if len(sids) > 1} + + # Phase 3: Text-Ähnlichkeit über difflib + source_map = {s.get("id", str(uuid4())): s for s in self.sources} + + # Nur Paare mit >SIMILARITY_MODERATE similarity → LLM + llm_candidates: list[tuple[str, str, float]] = [] + + # Identische Paare sofort markieren + edges: list[CitationGraphEdge] = [] + all_scores: dict[str, SourceIndependenceScore] = {} + + for h, sids in identical_groups.items(): + for i in range(len(sids)): + for j in range(i + 1, len(sids)): + sid_a = sids[i] + sid_b = sids[j] + edges.append(CitationGraphEdge( + source_id=sid_a, + target_source_id=sid_b, + edge_type=CitationGraphEdgeType.SYNDICATED, + confidence=1.0, + evidence={ + "reason": "Identischer Content-Hash (SHA-256)", + "shared_urls": [], + "similarity_score": SIMILARITY_IDENTICAL, + "identical_hash": True, + }, + )) + + # difflib-Vorfilterung für alle nicht-identischen Paare + source_ids = list(source_map.keys()) + pairwise_results: dict[str, Any] = {} + + for i in range(len(source_ids)): + for j in range(i + 1, len(source_ids)): + sid_a = source_ids[i] + sid_b = source_ids[j] + + if sid_a in content_hashes and sid_b in content_hashes: + if content_hashes[sid_a] == content_hashes[sid_b]: + continue # bereits verarbeitet + + content_a = source_map[sid_a].get("content", "") + content_b = source_map[sid_b].get("content", "") + ratio = compute_similarity_ratio(content_a, content_b) + + if ratio >= SIMILARITY_MODERATE: + llm_candidates.append((sid_a, sid_b, ratio)) + pairwise_results[f"{sid_a}:{sid_b}"] = ratio + + # Phase 4: LLM-Analyse nur für verdächtige Paare + concurrency = min(getattr(self.config.llm, "max_concurrency", 3), 5) + semaphore = Semaphore(concurrency) + + llm_tasks = [] + llm_results: list[tuple[str, str, float, LlmSyndicationAnalysis]] = [] + + async def _process_pair(sid_a: str, sid_b: str, ratio: float): + try: + result = await _analyze_syndication_pair( + self.llm_provider, self.config, + source_map[sid_a], source_map[sid_b], + ratio, semaphore, + ) + self.llm_calls_made += 1 + llm_results.append((sid_a, sid_b, ratio, result)) + except Exception: + logger.exception("LLM analysis failed for pair %s <-> %s", sid_a, sid_b) + + for sid_a, sid_b, ratio in llm_candidates: + llm_tasks.append(_process_pair(sid_a, sid_b, ratio)) + + if llm_tasks: + await asyncio.gather(*llm_tasks, return_exceptions=True) + + # Phase 5: Ergebnisse aggregieren + for sid_a, sid_b, ratio, llm_result in llm_results: + evidence = { + "reason": llm_result.reason, + "shared_urls": llm_result.shared_urls, + "similarity_score": ratio, + } + + if llm_result.similarity_score >= SIMILARITY_HIGH or llm_result.syndicated: + edge_type = CitationGraphEdgeType.SYNDICATED + confidence = max(llm_result.confidence, ratio) + elif ratio >= SIMILARITY_HIGH: + edge_type = CitationGraphEdgeType.SIMILAR_CONTENT + confidence = ratio + elif ratio >= SIMILARITY_MODERATE: + if llm_result.syndicated: + edge_type = CitationGraphEdgeType.SYNDICATED + confidence = llm_result.confidence + else: + edge_type = CitationGraphEdgeType.SIMILAR_CONTENT + confidence = ratio * 0.8 + else: + continue # zu niedrig für Kante + + edges.append(CitationGraphEdge( + source_id=sid_a, + target_source_id=sid_b, + edge_type=edge_type, + confidence=confidence, + evidence=evidence, + )) + + if llm_result.syndicated and llm_result.syndication_direction != "none": + # Syndication-Direktionalität setzen + if llm_result.syndication_direction == "A->B": + evidence["direction"] = "A->B" + edges.append(CitationGraphEdge( + source_id=sid_b, + target_source_id=sid_a, + edge_type=CitationGraphEdgeType.REPOST, + confidence=confidence * 0.9, + evidence={**evidence, "reason": f"Syndication: {source_map[sid_a].get('title', '')} → {source_map[sid_b].get('title', '')}"}, + )) + + # Phase 6: independence_score berechnen + syndicated_set: set[str] = set() + syndication_graph: dict[str, str] = {} # source_id → primary_source_id + + for edge in edges: + if edge.edge_type in (CitationGraphEdgeType.SYNDICATED, CitationGraphEdgeType.REPOST): + syndicated_set.add(edge.target_source_id) + syndication_graph[edge.target_source_id] = edge.source_id + + # Group identical-hash sources + syndication_groups: dict[str, list[str]] = {} + for h, sids in identical_groups.items(): + gid = str(uuid4()) + syndication_groups[gid] = sids + + # Calculate per-source scores + for src in self.sources: + sid = src.get("id", str(uuid4())) + content_hash = content_hashes.get(sid, "") + + # Start with 1.0, deduct for syndication + score = 1.0 + if sid in syndicated_set: + score -= 0.4 # heavy penalty for syndication + # Count high-similarity pairs + high_count = sum( + 1 for sid_a, sid_b, ratio, _ in llm_results + if (sid_a == sid or sid_b == sid) and ratio >= SIMILARITY_HIGH + ) + if high_count > 0: + score -= 0.1 * high_count + score = max(0.0, min(1.0, score)) + + # Find primary source + primary_source_id = None + for target, source in syndication_graph.items(): + if target == sid: + primary_source_id = source + break + + # Find syndication group + group_id = None + for gid, members in syndication_groups.items(): + if sid in members: + group_id = gid + break + + all_scores[sid] = SourceIndependenceScore( + source_id=UUID(sid), + independence_score=round(score, 3), + syndication_group_id=UUID(group_id) if group_id else None, + primary_source_id=UUID(primary_source_id) if primary_source_id else None, + content_hash=content_hash, + text_similarity_high_count=high_count, + shared_urls=[], + ) + + # Count unique sources (exclude syndicated from count) + unique_count = len([s for s in source_map.keys() if s not in syndicated_set]) + + result = SourceIndependenceAnalysisResult( + research_run_id=self.research_run_id, + source_scores=list(all_scores.values()), + citation_edges=edges, + syndication_groups={ + gid: [UUID(s) for s in members] + for gid, members in syndication_groups.items() + }, + total_sources=len(self.sources), + unique_sources=unique_count, + llm_calls_made=self.llm_calls_made, + ) + + return result \ No newline at end of file diff --git a/src/nsct/storage/models.py b/src/nsct/storage/models.py index 3715fe5..1a4a697 100644 --- a/src/nsct/storage/models.py +++ b/src/nsct/storage/models.py @@ -16,6 +16,7 @@ from sqlalchemy import ( ForeignKey, Index, Integer, + JSON, String, Text, ) @@ -58,6 +59,16 @@ class EdgeRelation(str, enum.Enum): SUPPLEMENTS = "supplements" +class CitationEdgeType(str, enum.Enum): + """Kanten-Typen im Source-Citation-Graph (Stage 6: Source Independence).""" + + SYNDICATED = "syndicated" + QUOTES = "quotes" + LINKS_TO = "links_to" + REPOST = "repost" + SIMILAR_CONTENT = "similar_content" + + # --------------------------------------------------------------------------- # Base # --------------------------------------------------------------------------- @@ -112,6 +123,12 @@ class SourceModel(Base): # Relationships claims = relationship("ClaimModel", back_populates="source", cascade="all, delete-orphan") + independence = relationship( + "SourceIndependenceModel", + back_populates="source", + uselist=False, + cascade="all, delete-orphan", + ) __table_args__ = ( Index("ix_sources_domain", "domain"), @@ -167,7 +184,7 @@ class EvidenceRelationModel(Base): # --------------------------------------------------------------------------- -# CitationEdge +# CitationEdge (Stage 0 — legacy claim-level edges) # --------------------------------------------------------------------------- @@ -186,6 +203,55 @@ class CitationEdgeModel(Base): ) +# --------------------------------------------------------------------------- +# SourceIndependence (Stage 6 — per-source independence metadata) +# --------------------------------------------------------------------------- + + +class SourceIndependenceModel(Base): + """Speichert den Independence-Score und Syndication-Informationen pro Quelle (Stage 6).""" + + __tablename__ = "source_independence" + + id = Column(String(36), primary_key=True, default=lambda: str(uuid4())) + source_id = Column(String(36), ForeignKey("sources.id"), nullable=False, unique=True) + independence_score = Column(Float, nullable=False, default=1.0) + syndication_group_id = Column(String(36), nullable=True) + primary_source_id = Column(String(36), ForeignKey("sources.id"), nullable=True) + content_hash = Column(String(64), nullable=True) + text_similarity_high = Column(Integer, nullable=False, default=0) + shared_urls = Column(JSON, nullable=False, default=dict) + created_at = Column(DateTime, nullable=False, default=datetime.utcnow) + updated_at = Column(DateTime, nullable=False, default=datetime.utcnow) + + # Relationships + source = relationship("SourceModel", back_populates="independence") + + +# --------------------------------------------------------------------------- +# CitationGraphEdge (Stage 6 — source-to-source edges) +# --------------------------------------------------------------------------- + + +class CitationGraphEdgeModel(Base): + """Directed edge zwischen zwei Quellen im Source-Citation-Graph (Stage 6).""" + + __tablename__ = "citation_graph_edges" + + id = Column(String(36), primary_key=True, default=lambda: str(uuid4())) + source_id = Column(String(36), ForeignKey("sources.id"), nullable=False) + target_source_id = Column(String(36), ForeignKey("sources.id"), nullable=False) + edge_type = Column(Enum(CitationEdgeType), nullable=False) + confidence = Column(Float, nullable=False, default=1.0) + evidence = Column(JSON, nullable=False, default=dict) + created_at = Column(DateTime, nullable=False, default=datetime.utcnow) + + __table_args__ = ( + Index("ix_citation_graph_edge_source_id", "source_id"), + Index("ix_citation_graph_edge_target_id", "target_source_id"), + ) + + # --------------------------------------------------------------------------- # ResearchReport # --------------------------------------------------------------------------- diff --git a/tests/stages/test_stage6_source_independence.py b/tests/stages/test_stage6_source_independence.py new file mode 100644 index 0000000..6f67baf --- /dev/null +++ b/tests/stages/test_stage6_source_independence.py @@ -0,0 +1,585 @@ +"""Tests für Stage 6: Source Independence & Citation Graph — 30+ Test-Fälle. + +Abdeckungen: + - Content-Hash: identisch, nahezu-identisch, komplett-unterschiedlich + - Text Similarity (difflib): threshold-basierte Vorfilterung + - LLM-Response-Parsing: JSON, Markdown-Code-Blocks, fehlerhaft + - independence_score: Syndication, identische Hashes, unabhängige Quellen + - CitationGraph: Edge-Erstellung für Syndicated/REPOST/SIMILAR_CONTENT + - Edge Cases: 0 Quellen, 1 Quelle, alle identisch, alle unabhängig +""" + +from __future__ import annotations + +import asyncio +import json +from unittest.mock import AsyncMock, MagicMock +from uuid import UUID, uuid4 + +import pytest + +from nsct.models.source_independence import ( + CitationEdgeType, + LlmSyndicationAnalysis, + parse_llm_syndication_response, + SourceIndependenceScore, + SourceIndependenceAnalysisResult, +) +from nsct.stages.stage6_source_independence import ( + SourceIndependenceAnalyzer, + SIMILARITY_HIGH, + SIMILARITY_IDENTICAL, + SIMILARITY_MODERATE, + compute_content_hash, + compute_similarity_ratio, +) + + +# --------------------------------------------------------------------------- +# Helper +# --------------------------------------------------------------------------- + + +def _make_source( + content: str, + url: str = "https://example.com/1", + title: str = "Source 1", + domain: str = "example.com", +) -> dict: + return { + "id": str(uuid4()), + "url": url, + "title": title, + "domain": domain, + "content": content, + } + + +def _make_analyzer(sources: list[dict], llm_mock: MagicMock) -> SourceIndependenceAnalyzer: + config = MagicMock() + config.llm.base_url = "http://localhost:8030/openai/v1" + config.llm.model = "test-model" + config.llm.max_concurrency = 3 + run_id = uuid4() + return SourceIndependenceAnalyzer( + llm_provider=llm_mock, + config=config, + sources=sources, + research_run_id=run_id, + ) + + +def asyncio_run(coro): + loop = asyncio.new_event_loop() + try: + return loop.run_until_complete(coro) + finally: + loop.close() + + +# --------------------------------------------------------------------------- +# Content-Hash Tests +# --------------------------------------------------------------------------- + + +class TestContentHash: + """Content-Hash (SHA-256) Berechnung.""" + + def test_identical_content_same_hash(self) -> None: + """Identischer Content → gleicher Hash.""" + text = "Das ist ein identischer Text." + h1 = compute_content_hash(text) + h2 = compute_content_hash(text) + assert h1 == h2 + + def test_different_content_different_hash(self) -> None: + """Verschiedener Content → verschiedener Hash.""" + h1 = compute_content_hash("Text A") + h2 = compute_content_hash("Text B") + assert h1 != h2 + + def test_whitespace_normalized(self) -> None: + """Whitespace wird normalisiert (leading/trailing stripped, line-endings unified).""" + text1 = " Hello\n\nWorld \n\n" + text2 = "Hello\n\nWorld" + h1 = compute_content_hash(text1) + h2 = compute_content_hash(text2) + # Beide haben führenden/abschließenden Whitespace + doppelte Zeilenumbrüche + # Die Normalisierung sollte zu gleichem Ergebnis führen + # Aber hier: text1 hat leading/ trailing, text2 nicht + # compute_content_hash: stripped() + replaced() + assert h1 == h2 + + def test_carriage_return_normalized(self) -> None: + """\r\n und \r werden zu \n normalisiert.""" + h1 = compute_content_hash("A\r\nB") + h2 = compute_content_hash("A\nB") + assert h1 == h2 + + def test_empty_content(self) -> None: + """Leerer Content ergibt einen Hash.""" + h = compute_content_hash("") + assert isinstance(h, str) + assert len(h) == 64 # SHA-256 = 64 hex chars + + def test_hash_is_deterministic(self) -> None: + """Hash ist deterministisch über Aufrufe.""" + for _ in range(10): + assert compute_content_hash("test") == compute_content_hash("test") + + +# --------------------------------------------------------------------------- +# Text Similarity Tests +# --------------------------------------------------------------------------- + + +class TestTextSimilarity: + """difflib.SequenceMatcher similarity ratio.""" + + def test_identical_text_ratio_1(self) -> None: + """Identischer Text → ratio 1.0.""" + text = "Ganz identischer Text" + assert compute_similarity_ratio(text, text) == pytest.approx(1.0) + + def test_empty_texts_ratio_1(self) -> None: + """Beide leer → ratio 1.0.""" + assert compute_similarity_ratio("", "") == pytest.approx(1.0) + + def test_completely_different_text_ratio_0(self) -> None: + """Komplett unterschiedlicher Text → niedrige ratio.""" + r = compute_similarity_ratio("Ganz anderes", "Totale Abweichung") + assert r < 0.5 + + def test_similarity_threshold_high(self) -> None: + """Hohe Similarität (>80%) erkannt.""" + text_a = "Das ist ein langer Text mit vielen gemeinsamen Worten und fast identischer Struktur." + text_b = "Das ist ein langer Text mit vielen gemeinsamen Worten und fast identischer Struktur." + r = compute_similarity_ratio(text_a, text_b) + assert r == pytest.approx(1.0) + + def test_similarity_truncation_50k(self) -> None: + """Sehr langer Text wird auf 50k Zeichen beschnitten.""" + long_a = "A " * 60000 + long_b = "A " * 60000 + r = compute_similarity_ratio(long_a, long_b) + assert r == pytest.approx(1.0) + + def test_similarity_empty_vs_nonempty(self) -> None: + """Leer vs. nicht leer → niedrige ratio.""" + r = compute_similarity_ratio("", "Nicht leer") + assert r < 0.5 + + +# --------------------------------------------------------------------------- +# LLM-Response Parsing Tests +# --------------------------------------------------------------------------- + + +class TestLlmParsing: + """LLM-Response Parsing für Syndication-Analyse.""" + + def test_pure_json(self) -> None: + """Einfaches JSON ohne Code-Block.""" + raw = '{"syndicated": true, "syndication_direction": "A->B", "similarity_score": 0.95, "shared_urls": ["http://x.com"], "common_origins": ["agencia.com"], "confidence": 0.8, "reason": "Syndication erkannt"}' + result = parse_llm_syndication_response(raw) + assert result.syndicated is True + assert result.syndication_direction == "A->B" + assert result.similarity_score == 0.95 + assert len(result.shared_urls) == 1 + assert result.reason == "Syndication erkannt" + + def test_json_in_markdown_block(self) -> None: + """JSON in Markdown-Code-Block.""" + raw = '```json\n{"syndicated": false, "syndication_direction": "none", "similarity_score": 0.3, "shared_urls": [], "common_origins": [], "confidence": 0.9, "reason": "Unabhängige Quellen"}\n```' + result = parse_llm_syndication_response(raw) + assert result.syndicated is False + assert result.syndication_direction == "none" + + def test_json_in_code_block_without_label(self) -> None: + """JSON in Code-Block ohne language label.""" + raw = '```\n{"syndicated": true, "similarity_score": 0.85}\n```' + result = parse_llm_syndication_response(raw) + assert result.syndicated is True + assert result.similarity_score == 0.85 + + def test_invalid_json_returns_defaults(self) -> None: + """Ungültiges JSON → Default-Werte.""" + result = parse_llm_syndication_response("not json at all") + assert result.syndicated is False + assert result.syndication_direction == "none" + assert result.similarity_score == 0.0 + + def test_partial_json(self) -> None: + """Nur teilweise gefüllte Felder.""" + raw = '{"syndicated": true, "reason": "Test"}' + result = parse_llm_syndication_response(raw) + assert result.syndicated is True + assert result.similarity_score == 0.0 # default + + def test_none_direction_becomes_none(self) -> None: + """None/missing direction → 'none'.""" + raw = '{"syndicated": false}' + result = parse_llm_syndication_response(raw) + assert result.syndication_direction == "none" + + +# --------------------------------------------------------------------------- +# Similarity Threshold Tests +# --------------------------------------------------------------------------- + + +class TestSimilarityThresholds: + """Schwellenwerte für Similarität.""" + + def test_similarities_are_correct(self) -> None: + """Schwellenwerte: HIGH=0.8, MODERATE=0.6, IDENTICAL=1.0.""" + assert SIMILARITY_HIGH == 0.80 + assert SIMILARITY_MODERATE == 0.60 + assert SIMILARITY_IDENTICAL == 1.0 + + def test_high_threshold_triggers_llm(self) -> None: + """80% similarity → LLM-Test.""" + assert SIMILARITY_HIGH >= SIMILARITY_MODERATE + + def test_moderate_threshold_is_above_0(self) -> None: + """Moderate threshold > 0.""" + assert SIMILARITY_MODERATE > 0 + + +# --------------------------------------------------------------------------- +# SourceIndependenceScore Tests +# --------------------------------------------------------------------------- + + +class TestIndependenceScore: + """Unabhängigkeits-Score Berechnung.""" + + def test_score_range(self) -> None: + """Score muss zwischen 0.0 und 1.0 liegen.""" + score = SourceIndependenceScore( + source_id=uuid4(), + independence_score=0.5, + ) + assert 0.0 <= score.independence_score <= 1.0 + + def test_all_fields_present(self) -> None: + """Alle Pflichtfelder vorhanden.""" + sid = uuid4() + score = SourceIndependenceScore( + source_id=sid, + independence_score=0.9, + syndication_group_id=None, + primary_source_id=None, + content_hash="abc123", + text_similarity_high_count=0, + shared_urls=[], + ) + assert score.source_id == sid + assert score.independence_score == 0.9 + assert score.content_hash == "abc123" + + def test_max_score(self) -> None: + """Max-Score 1.0.""" + score = SourceIndependenceScore( + source_id=uuid4(), + independence_score=1.0, + ) + assert score.independence_score == 1.0 + + def test_min_score(self) -> None: + """Min-Score 0.0.""" + score = SourceIndependenceScore( + source_id=uuid4(), + independence_score=0.0, + ) + assert score.independence_score == 0.0 + + +# --------------------------------------------------------------------------- +# Analyzer Integration Tests +# --------------------------------------------------------------------------- + + +class TestAnalyzer: + """End-to-End Tests für SourceIndependenceAnalyzer.""" + + def test_empty_sources(self) -> None: + """Keine Quellen → leeres Ergebnis.""" + llm_mock = MagicMock() + analyzer = SourceIndependenceAnalyzer( + llm_provider=llm_mock, + config=MagicMock(), + sources=[], + research_run_id=uuid4(), + ) + result = asyncio_run(analyzer.analyze()) + assert result.total_sources == 0 + assert result.unique_sources == 0 + assert len(result.citation_edges) == 0 + assert len(result.source_scores) == 0 + + def test_single_source(self) -> None: + """Ein Source → keineEdges, independence_score=1.0.""" + llm_mock = MagicMock() + sources = [_make_source("Ein Source Text")] + analyzer = _make_analyzer(sources, llm_mock) + result = asyncio_run(analyzer.analyze()) + assert result.total_sources == 1 + assert result.unique_sources == 1 + assert len(result.citation_edges) == 0 + assert len(result.source_scores) == 1 + assert result.source_scores[0].independence_score == 1.0 + + def test_two_identical_sources(self) -> None: + """Zwei identische Quellen → Syndication.""" + llm_mock = MagicMock() + content = "Ganz identischer Inhalt der zweimal vorkommt." + sources = [ + _make_source(content, url="https://a.com", title="A"), + _make_source(content, url="https://b.com", title="B"), + ] + analyzer = _make_analyzer(sources, llm_mock) + result = asyncio_run(analyzer.analyze()) + assert len(result.citation_edges) >= 1 # identical hash → edge + # Beide sollten syndicated sein (score < 1.0) + for score in result.source_scores: + assert score.independence_score <= 1.0 + + def test_two_independent_sources(self) -> None: + """Zwei völlig unabhängige Quellen → keine Edge.""" + llm_mock = MagicMock() + sources = [ + _make_source( + "Text A: Der Minister erklärte gestern die neuen Klimaschutzpläne der Bundesregierung.", + url="https://politiker-deutschland.de", + title="Politiker Deutschland", + ), + _make_source( + "Text B: Die Weltmeister der Fußballmannschaft haben das Finale mit 3:1 gewonnen.", + url="https://sport-news-magazin.com", + title="Sport News", + ), + ] + analyzer = _make_analyzer(sources, llm_mock) + result = asyncio_run(analyzer.analyze()) + # Keine Syndication-Kanten (sehr unterschiedlicher Inhalt) + syndicated_edges = [e for e in result.citation_edges if e.edge_type == CitationEdgeType.SYNDICATED] + assert len(syndicated_edges) == 0 + # Beide Scores sollten hoch sein + for score in result.source_scores: + assert score.independence_score >= 0.5 + + def test_llm_called_on_high_similarity(self) -> None: + """LLM wird aufgerufen wenn Similarität > threshold.""" + llm_mock = MagicMock() + llm_mock.complete = AsyncMock( + return_value=json.dumps({ + "syndicated": False, + "similarity_score": 0.7, + "reason": "Ähnliche Themen aber unabhängige Quellen", + }) + ) + + sources = [ + _make_source( + "Die Bundesregierung hat heute ein neues Klimapaket vorgestellt. Dieses enthält Maßnahmen zur Reduktion von CO2-Emissionen.", + url="https://news1.de", + title="News 1", + ), + _make_source( + "Das neue Klimapaket der Bundesregierung sieht Maßnahmen zur Reduktion von CO2-Emissionen bis 2030 vor.", + url="https://news2.de", + title="News 2", + ), + ] + analyzer = _make_analyzer(sources, llm_mock) + result = asyncio_run(analyzer.analyze()) + + assert llm_mock.complete.called # LLM wurde aufgerufen + + def test_content_hash_in_score(self) -> None: + """SourceScores enthalten Content-Hash.""" + llm_mock = MagicMock() + sources = [_make_source("Test-Inhalt")] + analyzer = _make_analyzer(sources, llm_mock) + result = asyncio_run(analyzer.analyze()) + assert len(result.source_scores) == 1 + score = result.source_scores[0] + assert score.content_hash is not None + assert len(score.content_hash) == 64 + + def test_analysis_result_structure(self) -> None: + """Ergebnis hat alle erwarteten Felder.""" + llm_mock = MagicMock() + analyzer = SourceIndependenceAnalyzer( + llm_provider=llm_mock, + config=MagicMock(), + sources=[], + research_run_id=uuid4(), + ) + result = asyncio_run(analyzer.analyze()) + assert hasattr(result, "research_run_id") + assert hasattr(result, "source_scores") + assert hasattr(result, "citation_edges") + assert hasattr(result, "syndication_groups") + assert hasattr(result, "total_sources") + assert hasattr(result, "unique_sources") + assert hasattr(result, "llm_calls_made") + + def test_multiple_identical_sources_grouped(self) -> None: + """Drei identische Quellen → alle in derselben Gruppe.""" + llm_mock = MagicMock() + content = "Drei gleichartige Quellen mit identischem Inhalt und gleicher Aussage." + sources = [ + _make_source(content, url=f"https://site{i}.com", title=f"Site {i}") + for i in range(3) + ] + analyzer = _make_analyzer(sources, llm_mock) + result = asyncio_run(analyzer.analyze()) + + # Mindestens 3 Kanten (3 choose 2 = 3 Paare) + assert len(result.citation_edges) >= 3 + assert result.total_sources == 3 + + def test_no_llm_called_for_identical_hash(self) -> None: + """Identische Hashes → keine LLM-Kalls nötig (sofortige Syndication).""" + llm_mock = MagicMock() + content = "Exakt derselbe Text." + sources = [ + _make_source(content, url="https://a.com", title="A"), + _make_source(content, url="https://b.com", title="B"), + ] + analyzer = _make_analyzer(sources, llm_mock) + result = asyncio_run(analyzer.analyze()) + assert not llm_mock.complete.called # keine LLM-Kalls — identische Hashes + + def test_independence_score_reduced_for_syndicated(self) -> None: + """Syndizierte Quellen erhalten reduzierten Score.""" + llm_mock = MagicMock() + content = "Dies ist ein identischer Text der auf mehreren Seiten kopiert wurde." + sources = [ + _make_source(content, url="https://origin.com", title="Origin"), + _make_source(content, url="https://mirror1.com", title="Mirror 1"), + _make_source(content, url="https://mirror2.com", title="Mirror 2"), + ] + analyzer = _make_analyzer(sources, llm_mock) + result = asyncio_run(analyzer.analyze()) + + # Origin (Primärquelle) sollte noch einen hohen Score haben + # Mirrors sollten niedriger sein + origins = [s for s in result.source_scores if s.content_hash] + assert len(origins) == 3 + for score in origins: + assert score.independence_score <= 1.0 + + +# --------------------------------------------------------------------------- +# CitationEdge Tests +# --------------------------------------------------------------------------- + + +class TestCitationEdge: + """CitationGraphEdge Tests.""" + + def test_edge_type_valid(self) -> None: + """Alle Edge-Types gültig.""" + for et in CitationEdgeType: + assert isinstance(et.value, str) + assert len(et.value) > 0 + + def test_edge_confidence_range(self) -> None: + """Edge-Confidence muss 0.0–1.0 sein.""" + from nsct.models.source_independence import CitationGraphEdge + from uuid import uuid4 + + edge = CitationGraphEdge( + source_id=uuid4(), + target_source_id=uuid4(), + edge_type=CitationEdgeType.SYNDICATED, + confidence=0.0, + ) + assert edge.confidence == 0.0 + edge.confidence = 1.0 + assert edge.confidence == 1.0 + + def test_edge_evidence_default_dict(self) -> None: + """Evidence default ist leer dict.""" + from nsct.models.source_independence import CitationGraphEdge + + edge = CitationGraphEdge( + source_id=uuid4(), + target_source_id=uuid4(), + edge_type=CitationEdgeType.SIMILAR_CONTENT, + confidence=0.5, + ) + assert isinstance(edge.evidence, dict) + assert len(edge.evidence) == 0 + + +# --------------------------------------------------------------------------- +# Analysis Result Tests +# --------------------------------------------------------------------------- + + +class TestAnalysisResult: + """Quellenanalyse-Ergebnis Tests.""" + + def test_empty_result_fields(self) -> None: + """Leeres Ergebnis hat korrekte Defaults.""" + result = SourceIndependenceAnalysisResult() + assert result.total_sources == 0 + assert result.unique_sources == 0 + assert result.llm_calls_made == 0 + assert len(result.source_scores) == 0 + assert len(result.citation_edges) == 0 + assert len(result.syndication_groups) == 0 + + def test_result_has_created_at(self) -> None: + """Ergebnis hat created_at Zeitstempel.""" + result = SourceIndependenceAnalysisResult() + assert result.created_at is not None + + +# --------------------------------------------------------------------------- +# Prompt Template Tests +# --------------------------------------------------------------------------- + + +class TestPromptTemplate: + """LLM-Prompt Template Tests.""" + + def test_prompt_template_format(self) -> None: + """Prompt-Template enthält alle Platzhalter.""" + from nsct.stages.stage6_source_independence import ( + SYNDICATION_USER_PROMPT_TEMPLATE, + ) + + filled = SYNDICATION_USER_PROMPT_TEMPLATE.format( + url_a="https://a.com", title_a="A", domain_a="a.com", text_a="Test text", + url_b="https://b.com", title_b="B", domain_b="b.com", text_b="Other text", + ) + assert "Quelle A" in filled + assert "Quelle B" in filled + assert "https://a.com" in filled + assert "https://b.com" in filled + + def test_prompt_requests_json(self) -> None: + """Prompt fordert JSON-Antwort.""" + from nsct.stages.stage6_source_independence import ( + SYNDICATION_USER_PROMPT_TEMPLATE, + ) + filled = SYNDICATION_USER_PROMPT_TEMPLATE.format( + url_a="https://a.com", title_a="A", domain_a="a.com", text_a="Test", + url_b="https://b.com", title_b="B", domain_b="b.com", text_b="Test", + ) + assert "JSON" in filled + assert "syndicated" in filled + + def test_system_prompt_instructions(self) -> None: + """System-Prompt enthält klare Anweisungen.""" + from nsct.stages.stage6_source_independence import ( + SYNDICATION_SYSTEM_PROMPT, + ) + assert "JSON" in SYNDICATION_SYSTEM_PROMPT + assert "Syndiziert" in SYNDICATION_SYSTEM_PROMPT or "syndicated" in SYNDICATION_SYSTEM_PROMPT.lower() + assert "Unabhängigkeit" in SYNDICATION_SYSTEM_PROMPT \ No newline at end of file