diff --git a/backend/app/main.py b/backend/app/main.py index 66168de..1072743 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1401,6 +1401,14 @@ async def _run_repo_scan_task( except ValueError: epsilon = 0.15 + + # CORRECT WAY: + if not disable_dedup and SENTENCE_TRANSFORMERS_AVAILABLE: + # The assertion must be safely inside the block body + assert all(isinstance(f, Finding) for f in findings), \ + f"Expected Finding objects, got {set(type(f).__name__ for f in findings)}" + + findings = deduplicate(findings, epsilon) if not disable_dedup and SENTENCE_TRANSFORMERS_AVAILABLE: findings = deduplicate(findings, epsilon) diff --git a/backend/app/ml/deduplicator.py b/backend/app/ml/deduplicator.py index fae1036..1ee1cb7 100644 --- a/backend/app/ml/deduplicator.py +++ b/backend/app/ml/deduplicator.py @@ -1,5 +1,10 @@ from collections import defaultdict +from typing import Union, List, Dict, Any +from sklearn.cluster import DBSCAN +from app.models import Finding + + from sklearn.cluster import DBSCAN @@ -28,6 +33,11 @@ def embed_findings(findings): def deduplicate( + + findings: List[Union[Finding, Dict[str, Any]]], + epsilon: float = 0.15, +) -> List[Union[Finding, Dict[str, Any]]]: + findings: list[dict], epsilon: float = 0.15, ) -> list[dict]: @@ -73,3 +83,6 @@ def deduplicate( results.append(representative) return results + + return results + diff --git a/backend/app/ml/embedder.py b/backend/app/ml/embedder.py index f05ad71..141683d 100644 --- a/backend/app/ml/embedder.py +++ b/backend/app/ml/embedder.py @@ -1,18 +1,54 @@ + +import logging + import numpy as np try: from sentence_transformers import SentenceTransformer MODEL = SentenceTransformer("all-MiniLM-L6-v2") + +except Exception: + MODEL = None + logging.getLogger(__name__).warning( + "Failed to load sentence-transformers model: all-MiniLM-L6-v2", + exc_info=True + ) + + +def _extract_text(finding) -> str: + """Safely extracts title and description from either a Pydantic Finding object or a raw dict.""" + if isinstance(finding, dict): + return f"{finding.get('title', '')} {finding.get('description', '')}".strip() + return f"{getattr(finding, 'title', '')} {getattr(finding, 'description', '')}".strip() + + +def embed_findings(findings: list) -> np.ndarray: + except ImportError: MODEL = None def embed_findings(findings: list[dict]) -> np.ndarray: + """ Convert findings into embeddings. Each finding is converted to: + + "{title} {description}" + + Returns: + np.ndarray of shape (n, 384) + """ + if MODEL is None: + raise RuntimeError( + "sentence-transformers is not installed or failed to initialize. " + "Install it using: pip install sentence-transformers" + ) + + texts = [_extract_text(finding) for finding in findings] + "{rule_id} {message} {file_path}" Returns: