"""Orchestrates the change-detection layers (hash -> structured diff -> text diff -> scoring) into a single `DetectedChange` row, or nothing if the change isn't real/meaningful/novel enough to record. Cross-source corroboration (Task C in the spec's LLM analysis section) is Phase 7 scope - `independent_source_count` is always 1 here. See KNOWN_LIMITATIONS.md. """ from __future__ import annotations import re import uuid from datetime import UTC, datetime, timedelta from sqlalchemy.ext.asyncio import AsyncSession from app.change_detection.extractors import extract_prices, mentions_leadership_title from app.change_detection.scoring import classify_severity, compute_confidence, compute_significance from app.change_detection.structured_diff import StructuredDiff, diff_item_sets from app.change_detection.text_diff import TextDiffResult, bounded_text_diff from app.models.company import Company from app.models.detected_change import DetectedChange from app.models.enums import ChangeType, SourceType from app.models.snapshot import Snapshot from app.models.source import Source from app.repositories.detected_change_repository import DetectedChangeRepository from app.repositories.source_repository import SnapshotRepository _COOLDOWN = timedelta(hours=24) _MIN_CONTENT_DIFF_RATIO = 0.05 _EXTRACTION_CONFIDENCE: dict[ChangeType, float] = { ChangeType.LEADERSHIP_CHANGE: 0.6, ChangeType.FILING_NEW: 0.95, ChangeType.PRICE_CHANGE: 0.6, ChangeType.NEW_DOCUMENT: 0.85, ChangeType.REMOVED_DOCUMENT: 0.85, ChangeType.CONTENT_MODIFIED: 0.7, } async def detect_change_for_source( db: AsyncSession, source: Source, company: Company, current_snapshot: Snapshot, monitoring_run_id: uuid.UUID, ) -> DetectedChange | None: snapshot_repo = SnapshotRepository(db) previous = await snapshot_repo.get_previous(source.id, current_snapshot) if previous is None: return None # Baseline snapshot - nothing to compare against yet. # Layer 1: exact hash comparison. if previous.hash == current_snapshot.hash: return None structured_diff = diff_item_sets( previous.structured_summary.get("urls", []), current_snapshot.structured_summary.get("urls", []), ) text_diff = bounded_text_diff(previous.text_summary or "", current_snapshot.text_summary or "") change_type = _classify_change(source, structured_diff, text_diff) if change_type is None: return None # The hash differed, but only noise (Layer 3 already strips it). # Always drawn from the actual new text (not the URL/title evidence used # for display) - that's what's meaningful to compare against the user's # stated focus, regardless of which change_type it triggered. focus_evidence_text = "\n".join(text_diff.added_lines) raw_diff = { "structured_added": structured_diff.added, "structured_removed": structured_diff.removed, "text_diff_ratio": text_diff.diff_ratio, "text_added_lines": text_diff.added_lines, "text_removed_lines": text_diff.removed_lines, } change_repo = DetectedChangeRepository(db) since = datetime.now(UTC) - _COOLDOWN recent = await change_repo.get_recent_for_source_and_type(source.id, change_type, since) if recent is not None and recent.raw_diff == raw_diff: return None # Exact repeat within the cooldown window - nothing new to report. is_repeat = recent is not None significance = compute_significance( change_type=change_type, source_trust_score=source.trust_score, independent_source_count=1, focus_match=_matches_focus(company.monitoring_focus, focus_evidence_text), is_repeat=is_repeat, diff_ratio=text_diff.diff_ratio if change_type is ChangeType.CONTENT_MODIFIED else None, ) confidence = compute_confidence( extraction_confidence=_EXTRACTION_CONFIDENCE[change_type], source_trust_score=source.trust_score, independent_source_count=1, ) severity = classify_severity(significance, confidence) change = await change_repo.create( company_id=company.id, source_id=source.id, monitoring_run_id=monitoring_run_id, previous_snapshot_id=previous.id, current_snapshot_id=current_snapshot.id, change_type=change_type, raw_diff=raw_diff, significance_score=significance, confidence_score=confidence, severity=severity, summary=_build_summary(change_type, structured_diff, text_diff), ) await db.commit() return change def _classify_change( source: Source, structured_diff: StructuredDiff, text_diff: TextDiffResult ) -> ChangeType | None: added_text = "\n".join(text_diff.added_lines) if mentions_leadership_title(added_text): return ChangeType.LEADERSHIP_CHANGE if source.source_type is SourceType.SEC_EDGAR and structured_diff.added: return ChangeType.FILING_NEW price_delta = extract_prices(added_text) | extract_prices("\n".join(text_diff.removed_lines)) if price_delta: return ChangeType.PRICE_CHANGE if structured_diff.added: return ChangeType.NEW_DOCUMENT if structured_diff.removed: return ChangeType.REMOVED_DOCUMENT if text_diff.diff_ratio >= _MIN_CONTENT_DIFF_RATIO: return ChangeType.CONTENT_MODIFIED return None def _matches_focus(monitoring_focus: str | None, evidence_text: str) -> bool: if not monitoring_focus: return False focus_words = {w.lower() for w in re.findall(r"[a-zA-Z]{5,}", monitoring_focus)} evidence_words = {w.lower() for w in re.findall(r"[a-zA-Z]{5,}", evidence_text)} return bool(focus_words & evidence_words) def _build_summary( change_type: ChangeType, structured_diff: StructuredDiff, text_diff: TextDiffResult ) -> str: if change_type is ChangeType.NEW_DOCUMENT: n = len(structured_diff.added) return f"{n} new item{'s' if n != 1 else ''} detected" if change_type is ChangeType.REMOVED_DOCUMENT: n = len(structured_diff.removed) return f"{n} item{'s' if n != 1 else ''} removed" if change_type is ChangeType.PRICE_CHANGE: return "Pricing information changed" if change_type is ChangeType.LEADERSHIP_CHANGE: return "Possible leadership change mentioned" if change_type is ChangeType.FILING_NEW: n = len(structured_diff.added) return f"{n} new regulatory filing{'s' if n != 1 else ''}" return f"Content changed ({text_diff.diff_ratio:.0%} different)"