feat: add acronym export and import commands [AI-assisted] (#1010)

Implemented 'export' and 'import' commands for the acronym database.
This allows users to create datasets of collected acronyms and share them
between installations.

- Added export_all_variants and import_variants to AcronymCache
- Added export and import commands to 'acronym' CLI group
- Renamed 'list' function to 'list_acronyms' to avoid shadowing built-in 'list'
- Fixed Ruff B904 issues in CLI exception handling

Co-authored-by: florath-ai-assistant[bot] <Andreas.Florath@telekom.de>

Author: coding-ai-assistant[bot]

src/aletheia_probe/cache/acronym_cache.py

@@ -338,6 +338,81 @@ def get_acronym_stats(self, entity_type: str | None = None) -> dict[str, int]:
 
             return {"total_count": count}
 
+    def export_all_variants(self) -> list[dict[str, Any]]:
+        """Export all acronym variants from the database.
+
+        Returns:
+            List of dictionaries containing all variant fields.
+        """
+        detail_logger.debug("Exporting all acronym variants")
+
+        with self.get_connection_with_row_factory() as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                """
+                SELECT acronym, entity_type, variant_name, normalized_name,
+                       usage_count, is_canonical, is_ambiguous, source
+                FROM venue_acronym_variants
+                ORDER BY acronym, entity_type, usage_count DESC
+                """
+            )
+            return [dict(row) for row in cursor.fetchall()]
+
+    def import_variants(
+        self, variants: list[dict[str, Any]], merge: bool = True
+    ) -> int:
+        """Import acronym variants into the database.
+
+        Args:
+            variants: List of variant dictionaries (as exported by export_all_variants).
+            merge: If True, merges with existing data (increments counts).
+                   If False, fails on conflict or requires clear first (currently only merge supported).
+
+        Returns:
+            Number of variants imported/updated.
+        """
+        if not variants:
+            return 0
+
+        detail_logger.debug(f"Importing {len(variants)} acronym variants")
+        count = 0
+
+        for variant in variants:
+            # Validate required fields
+            if not all(
+                k in variant for k in ["acronym", "entity_type", "normalized_name"]
+            ):
+                detail_logger.warning(f"Skipping invalid variant: {variant}")
+                continue
+
+            # Default values for missing fields
+            variant_name = variant.get("variant_name", variant["normalized_name"])
+            usage_count = variant.get("usage_count", 1)
+            source = variant.get("source", "import")
+
+            # Use store_variant to handle merge logic
+            self.store_variant(
+                acronym=variant["acronym"],
+                entity_type=variant["entity_type"],
+                variant_name=variant_name,
+                normalized_name=variant["normalized_name"],
+                usage_count=usage_count,
+                source=source,
+            )
+
+            # Restore is_ambiguous status if present
+            if variant.get("is_ambiguous"):
+                self.mark_acronym_as_ambiguous(
+                    variant["acronym"], variant["entity_type"]
+                )
+
+            # Update canonical status
+            self.update_canonical_variant(variant["acronym"], variant["entity_type"])
+
+            count += 1
+
+        return count
+
     def list_all_acronyms(
         self, entity_type: str | None = None, limit: int | None = None, offset: int = 0
     ) -> list[dict[str, str]]:

src/aletheia_probe/cli.py

@@ -412,11 +412,11 @@ def stats() -> None:
     status_logger.info(f"Total acronyms: {total:,}")
 
 
-@acronym.command()
+@acronym.command(name="list")
 @click.option("--limit", type=int, help="Maximum number of entries to display")
 @click.option("--offset", type=int, default=0, help="Number of entries to skip")
 @handle_cli_errors
-def list(limit: int | None, offset: int) -> None:
+def list_acronyms(limit: int | None, offset: int) -> None:
     """List all acronym mappings in the database.
 
     Args:
@@ -446,6 +446,75 @@ def list(limit: int | None, offset: int) -> None:
         status_logger.info(f"\nShowing {shown} of {total_count:,} total acronyms")
 
 
+@acronym.command()
+@click.argument("output_file", type=click.Path())
+@handle_cli_errors
+def export(output_file: str) -> None:
+    """Export the entire acronym database to a JSON file.
+
+    Args:
+        output_file: Path to the output JSON file.
+    """
+    status_logger = get_status_logger()
+    acronym_cache = AcronymCache()
+
+    variants = acronym_cache.export_all_variants()
+
+    try:
+        with open(output_file, "w", encoding="utf-8") as f:
+            json.dump(variants, f, indent=2, ensure_ascii=False)
+
+        status_logger.info(
+            f"Successfully exported {len(variants)} acronym variants to {output_file}"
+        )
+    except Exception as e:
+        status_logger.error(f"Failed to export acronyms: {e}")
+        raise click.ClickException(str(e)) from e
+
+
+@acronym.command(name="import")
+@click.argument("input_file", type=click.Path(exists=True))
+@click.option(
+    "--merge/--no-merge",
+    default=True,
+    help="Merge with existing data (default) or replace",
+)
+@handle_cli_errors
+def import_acronyms(input_file: str, merge: bool) -> None:
+    """Import acronyms from a JSON file.
+
+    Args:
+        input_file: Path to the input JSON file.
+        merge: Whether to merge with existing data.
+    """
+    status_logger = get_status_logger()
+    acronym_cache = AcronymCache()
+
+    try:
+        with open(input_file, encoding="utf-8") as f:
+            variants = json.load(f)
+
+        if not isinstance(variants, list):
+            raise ValueError("Input file must contain a JSON list of variants")
+
+        status_logger.info(f"Read {len(variants)} variants from {input_file}")
+
+        if not merge:
+            if click.confirm(
+                "This will clear existing acronyms before importing. Continue?",
+                abort=True,
+            ):
+                acronym_cache.clear_acronym_database()
+
+        count = acronym_cache.import_variants(variants, merge=True)
+
+        status_logger.info(f"Successfully imported {count} acronym variants")
+
+    except Exception as e:
+        status_logger.error(f"Failed to import acronyms: {e}")
+        raise click.ClickException(str(e)) from e
+
+
 @acronym.command()
 @click.option("--confirm", is_flag=True, help="Skip confirmation prompt")
 @handle_cli_errors