feat: Implement self-learning conference acronym recognition (fixes #70) (#86)

coding-ai-assistant[bot] · florath · web-flow · commit 3e4155935547 · 2025-11-22T17:27:15.000+01:00
Add self-learning acronym cache that builds conference/journal acronym
mappings over time based on user's domain-specific usage.

Database:
- Add conference_acronyms table to cache.db with acronym, full_name, source

Acronym Storage:
- Extract acronyms from parenthetical references in BibTeX entries
- Extract acronyms from OpenAlex display_name responses
- Warn when overwriting existing acronym mapping

Acronym Expansion:
- Detect standalone acronyms in input (2-10 chars, mostly uppercase)
- Look up expansion in cache and add as alias for searching
- Fall back to expanded name if initial query yields no results

User Feedback:
- CLI shows 'Note: Expanded acronym' message
- JSON output includes acronym_expansion_used field

Users querying ICML will get legitimate results after the system has seen
the full name with acronym in any BibTeX file or OpenAlex response.

Co-authored-by: florath-ai-assistant[bot] &lt;Andreas.Florath@telekom.de&gt;
diff --git a/src/aletheia_probe/backends/openalex_analyzer.py b/src/aletheia_probe/backends/openalex_analyzer.py
@@ -94,6 +94,9 @@ async def _query_api(self, query_input: QueryInput) -> BackendResult:
                         response_time=response_time,
                     )
 
+                # Store acronym mapping if display_name contains acronym in parentheses
+                self._store_acronym_from_openalex(openalex_data)
+
                 # Route to appropriate assessment based on publication type
                 source_type = openalex_data.get("source_type", "").lower()
                 if source_type == "conference":
@@ -134,6 +137,29 @@ async def _query_api(self, query_input: QueryInput) -> BackendResult:
                 response_time=response_time,
             )
 
+    def _store_acronym_from_openalex(self, openalex_data: dict[str, Any]) -> None:
+        """Extract and store acronym mapping from OpenAlex display_name.
+
+        OpenAlex sometimes includes acronyms in parentheses in the display_name field.
+        For example: "International Conference on Machine Learning (ICML)"
+
+        Args:
+            openalex_data: Raw data from OpenAlex API
+        """
+        from ..normalizer import InputNormalizer
+
+        display_name = openalex_data.get("display_name")
+        if not display_name:
+            return
+
+        # Use the normalizer's acronym extraction logic
+        normalizer = InputNormalizer()
+        acronyms = normalizer._extract_acronyms(display_name)
+
+        if acronyms:
+            # Store the mapping for each extracted acronym
+            normalizer._store_acronym_mappings_from_text(display_name, acronyms)
+
     def _analyze_journal_patterns(
         self, openalex_data: dict[str, Any]
     ) -> dict[str, Any]:
diff --git a/src/aletheia_probe/cache.py b/src/aletheia_probe/cache.py
@@ -97,6 +97,16 @@ def _init_database(self) -> None:
                     UNIQUE(journal_id, source_id)
                 );
 
+                -- Conference/journal acronym mappings (self-learning cache)
+                CREATE TABLE IF NOT EXISTS conference_acronyms (
+                    acronym TEXT PRIMARY KEY COLLATE NOCASE,
+                    full_name TEXT NOT NULL,
+                    source TEXT,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    last_used_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+                CREATE INDEX IF NOT EXISTS idx_acronyms_full_name ON conference_acronyms(full_name);
+
                 -- Source metadata (replaces JSON metadata)
                 CREATE TABLE IF NOT EXISTS source_metadata (
                     id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -1088,6 +1098,97 @@ def cleanup_expired_article_retractions(self) -> int:
             conn.commit()
             return cursor.rowcount
 
+    # Acronym management methods
+
+    def get_full_name_for_acronym(self, acronym: str) -> str | None:
+        """
+        Look up the full name for a conference/journal acronym.
+
+        Args:
+            acronym: The acronym to look up (e.g., 'ICML', 'CVPR')
+
+        Returns:
+            Full name if found in cache, None otherwise
+        """
+        with sqlite3.connect(self.db_path) as conn:
+            conn.row_factory = sqlite3.Row
+            cursor = conn.cursor()
+
+            cursor.execute(
+                """
+                SELECT full_name FROM conference_acronyms
+                WHERE acronym = ? COLLATE NOCASE
+                """,
+                (acronym.strip(),),
+            )
+
+            row = cursor.fetchone()
+            if row:
+                # Update last_used_at timestamp
+                cursor.execute(
+                    """
+                    UPDATE conference_acronyms
+                    SET last_used_at = CURRENT_TIMESTAMP
+                    WHERE acronym = ? COLLATE NOCASE
+                    """,
+                    (acronym.strip(),),
+                )
+                conn.commit()
+                return str(row["full_name"])
+            return None
+
+    def store_acronym_mapping(
+        self, acronym: str, full_name: str, source: str = "unknown"
+    ) -> None:
+        """
+        Store an acronym to full name mapping in the cache.
+
+        If the acronym already exists with a different full_name, logs a warning
+        and overwrites with the new mapping.
+
+        Args:
+            acronym: The acronym (e.g., 'ICML')
+            full_name: The full conference/journal name
+            source: Source of the mapping ('bibtex_extraction', 'openalex_response', 'manual')
+        """
+        from .logging_config import get_status_logger
+
+        status_logger = get_status_logger()
+
+        acronym = acronym.strip()
+        full_name = full_name.strip()
+
+        with sqlite3.connect(self.db_path) as conn:
+            conn.row_factory = sqlite3.Row
+            cursor = conn.cursor()
+
+            # Check for existing mapping
+            cursor.execute(
+                """
+                SELECT full_name FROM conference_acronyms
+                WHERE acronym = ? COLLATE NOCASE
+                """,
+                (acronym,),
+            )
+
+            existing = cursor.fetchone()
+            if existing and existing["full_name"] != full_name:
+                status_logger.warning(
+                    f"Acronym '{acronym}' already maps to '{existing['full_name']}', "
+                    f"overwriting with '{full_name}'"
+                )
+
+            # Insert or replace the mapping
+            cursor.execute(
+                """
+                INSERT OR REPLACE INTO conference_acronyms
+                (acronym, full_name, source, created_at, last_used_at)
+                VALUES (?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
+                """,
+                (acronym, full_name, source),
+            )
+            conn.commit()
+
 
 # Global cache manager instance with factory pattern
 _cache_manager_instance: CacheManager | None = None
diff --git a/src/aletheia_probe/cli.py b/src/aletheia_probe/cli.py
@@ -380,6 +380,13 @@ async def _async_assess_publication(
                 label = "Journal"
 
             print(f"{label}: {result.input_query}")
+
+            # Show acronym expansion note if applicable
+            if result.acronym_expansion_used and result.acronym_expanded_from:
+                print(
+                    f"Note: Expanded acronym '{result.acronym_expanded_from}' using cached mapping"
+                )
+
             print(f"Assessment: {result.assessment.upper()}")
             print(f"Confidence: {result.confidence:.2f}")
             print(f"Overall Score: {result.overall_score:.2f}")
diff --git a/src/aletheia_probe/dispatcher.py b/src/aletheia_probe/dispatcher.py
@@ -88,6 +88,8 @@ async def assess_journal(self, query_input: QueryInput) -> AssessmentResult:
                 metadata=None,
                 reasoning=["No backends available for assessment"],
                 processing_time=time.time() - start_time,
+                acronym_expanded_from=query_input.acronym_expanded_from,
+                acronym_expansion_used=bool(query_input.acronym_expanded_from),
             )
 
         self.status_logger.info(
@@ -102,6 +104,46 @@ async def assess_journal(self, query_input: QueryInput) -> AssessmentResult:
             query_input, backend_results, time.time() - start_time
         )
 
+        # Acronym fallback: If initial query yields no confident results and input looks
+        # like an acronym with a cached expansion, retry with the expanded name
+        if self._should_try_acronym_fallback(assessment_result, query_input):
+            from .cache import CacheManager
+            from .normalizer import InputNormalizer
+
+            normalizer = InputNormalizer()
+            cache = CacheManager()
+
+            # Check if input is acronym-like and has expansion
+            if normalizer._is_standalone_acronym(query_input.raw_input):
+                expanded_name = cache.get_full_name_for_acronym(query_input.raw_input)
+
+                if expanded_name:
+                    self.status_logger.info(
+                        f"No confident results for '{query_input.raw_input}'. "
+                        f"Retrying with expanded name: '{expanded_name}'"
+                    )
+
+                    # Create new query input with expanded name
+                    from .normalizer import input_normalizer
+
+                    expanded_query = input_normalizer.normalize(expanded_name)
+
+                    # Re-query backends with expanded name
+                    retry_results = await self._query_backends(
+                        enabled_backends, expanded_query
+                    )
+
+                    # Calculate new assessment
+                    retry_assessment = self._calculate_assessment(
+                        expanded_query, retry_results, time.time() - start_time
+                    )
+
+                    # If retry gave better results, use it and mark acronym expansion
+                    if retry_assessment.confidence > assessment_result.confidence:
+                        retry_assessment.acronym_expanded_from = query_input.raw_input
+                        retry_assessment.acronym_expansion_used = True
+                        return retry_assessment
+
         return assessment_result
 
     def _get_enabled_backends(self) -> list[Backend]:
@@ -256,6 +298,46 @@ async def _query_backend_with_timing(
         result_dict["evidence_type"] = backend.get_evidence_type().value
         return BackendResult(**result_dict)
 
+    def _should_try_acronym_fallback(
+        self, assessment_result: AssessmentResult, query_input: QueryInput
+    ) -> bool:
+        """Determine if we should try acronym expansion fallback.
+
+        Acronym fallback is attempted when:
+        - Initial assessment is UNKNOWN or has low confidence
+        - No backends returned FOUND status
+        - Input hasn't already been expanded from an acronym
+
+        Args:
+            assessment_result: The initial assessment result
+            query_input: The original query input
+
+        Returns:
+            True if acronym fallback should be attempted
+        """
+        # Don't retry if we already used acronym expansion
+        if query_input.acronym_expanded_from:
+            return False
+
+        # Retry if assessment is UNKNOWN
+        if assessment_result.assessment == AssessmentType.UNKNOWN:
+            return True
+
+        # Retry if confidence is very low (< 0.3)
+        if assessment_result.confidence < 0.3:
+            return True
+
+        # Retry if no backends found anything
+        found_count = sum(
+            1
+            for r in assessment_result.backend_results
+            if r.status == BackendStatus.FOUND
+        )
+        if found_count == 0:
+            return True
+
+        return False
+
     def _calculate_assessment(
         self,
         query_input: QueryInput,
@@ -395,6 +477,8 @@ def _handle_no_results(
             metadata=None,
             reasoning=reasoning,
             processing_time=processing_time,
+            acronym_expanded_from=query_input.acronym_expanded_from,
+            acronym_expansion_used=bool(query_input.acronym_expanded_from),
         )
 
     def _calculate_backend_scores(
@@ -611,6 +695,8 @@ def _make_final_assessment(
             metadata=None,
             reasoning=reasoning,
             processing_time=processing_time,
+            acronym_expanded_from=query_input.acronym_expanded_from,
+            acronym_expansion_used=bool(query_input.acronym_expanded_from),
         )
 
 
diff --git a/src/aletheia_probe/models.py b/src/aletheia_probe/models.py
@@ -28,6 +28,9 @@ class QueryInput(BaseModel):
         default_factory=dict, description="ISSN, DOI, etc."
     )
     aliases: list[str] = Field(default_factory=list, description="Alternative names")
+    acronym_expanded_from: str | None = Field(
+        None, description="Original acronym if expansion was applied"
+    )
 
 
 class BackendResult(BaseModel):
@@ -102,6 +105,12 @@ class AssessmentResult(BaseModel):
         default_factory=datetime.now, description="Assessment timestamp"
     )
     processing_time: float = Field(..., description="Total processing time in seconds")
+    acronym_expanded_from: str | None = Field(
+        None, description="Original acronym if expansion was applied during assessment"
+    )
+    acronym_expansion_used: bool = Field(
+        False, description="Whether acronym expansion was used to get results"
+    )
 
 
 class ConfigBackend(BaseModel):
diff --git a/src/aletheia_probe/normalizer.py b/src/aletheia_probe/normalizer.py
