Feature/bibliography export

src/app/api/api_v1/api.py

@@ -2,7 +2,7 @@
 
 from fastapi import APIRouter
 
-from src.app.api.api_v1.endpoints import chat, metric, micro_learning
+from src.app.api.api_v1.endpoints import bibliography, chat, metric, micro_learning
 from src.app.search.api import router as search_router
 from src.app.tutor.api import router as tutor_router
 from src.app.user.api import router as user_router
@@ -16,6 +16,9 @@
     micro_learning.router, prefix="/micro_learning", tags=["micro_learning"]
 )
 api_router.include_router(user_router.router, prefix="/user", tags=["user"])
+api_router.include_router(
+    bibliography.router, prefix="/bibliography", tags=["bibliography"]
+)
 
 
 api_tags_metadata = [
@@ -43,4 +46,8 @@
         "name": "metric",
         "description": "Metric information",
     },
+    {
+        "name": "bibliography",
+        "description": "Bibliography exporters for documents collected by welearn",
+    },
 ]

src/app/api/api_v1/endpoints/bibliography.py

@@ -0,0 +1,32 @@
+from fastapi import APIRouter, BackgroundTasks, Request
+from starlette.responses import PlainTextResponse
+
+from src.app.models.bibliography import DocumentIDs
+from src.app.services.helpers import welearn_document_to_ris
+from src.app.services.sql_db.queries import get_documents_by_ids
+from src.app.shared.utils.dependencies import get_settings
+from src.app.utils.logger import logger as utils_logger
+
+logger = utils_logger(__name__)
+
+router = APIRouter()
+
+settings = get_settings()
+
+
+@router.post("/export_bibliography", response_class=PlainTextResponse)
+async def export_bibliography(
+    request: Request,
+    background_tasks: BackgroundTasks,
+    body: DocumentIDs,
+):
+    documents_ids = [str(u) for u in body.documents_ids]
+
+    docs = get_documents_by_ids(documents_ids)
+
+    ret = ""
+    for doc in docs:
+        ret += welearn_document_to_ris(doc)
+        ret += "\n"
+
+    return ret

src/app/models/bibliography.py

@@ -0,0 +1,7 @@
+from uuid import UUID
+
+from pydantic import BaseModel
+
+
+class DocumentIDs(BaseModel):
+    documents_ids: list[UUID]

src/app/services/helpers.py

@@ -1,3 +1,4 @@
+import datetime
 import re
 from functools import cache
 from typing import Any, List, Optional, cast
@@ -8,7 +9,7 @@
 from json_repair import JSONReturnType
 from langdetect import detect_langs
 from qdrant_client.http.models import models
-from welearn_database.data.models import EmbeddingModel
+from welearn_database.data.models import EmbeddingModel, WeLearnDocument
 
 from src.app.models.collections import Collection
 from src.app.models.documents import JourneySectionType
@@ -134,7 +135,7 @@ def stringify_docs_content(docs: List[Any]) -> str:
 
         documents = "\n\n".join(articles)
     except Exception as e:
-        logger.error("Error in stringify_docs_content: %s", e)
+        logger.exception("Error in stringify_docs_content: %s", e)
         return ""
 
     return documents.strip()
@@ -256,3 +257,74 @@ async def collection_and_model_id_according_lang(
             f"Embedding model '{collection_info.model}' not found in the database."
         )
     return collection_info, emb_models
+
+
+def ris_line(tag: str, value: Any) -> Optional[str]:
+    """Formate une ligne RIS, ou None si la valeur est vide."""
+    if value is None or value == "":
+        return None
+    return f"{tag}  - {value}"
+
+
+def compute_ris_doctype(doc_type: Any | None) -> str:
+    match doc_type:
+        case "book":
+            return "BOOK"
+        case "chapter":
+            return "CHAP"
+        case "article":
+            return "JFULL"
+        case _:
+            return "ELEC"
+
+
+def compute_publication_date_for_ris(pub_date: str | int | float) -> str:
+    try:
+        tmp_ret = datetime.date.fromtimestamp(int(pub_date)).isoformat()
+        return tmp_ret.replace("-", "/")
+    except Exception as e:
+        logger.warning(
+            "Exception occurs during publication formatting, field returned empty", e
+        )
+        return ""
+
+
+def compute_authors_for_ris(authors: dict) -> list[str]:
+    ret = []
+    for author in authors:
+        ret.append(ris_line("AU", author.get("name")))
+    return ret
+
+
+@log_time_and_error_sync
+def welearn_document_to_ris(doc: WeLearnDocument) -> str:
+    details = doc.details or {}
+
+    lines: list[str] = []
+
+    doc_type = details.get("type", None)
+    lines.append(ris_line("TY", compute_ris_doctype(doc_type)))
+    lines.append(ris_line("ID", doc.id))
+    lines.append(ris_line("TI", doc.title))
+    lines.append(ris_line("UR", doc.url))
+    lines.append(ris_line("AB", doc.description))
+    lines.append(ris_line("DB", doc.corpus.source_name))
+    lines.append(ris_line("LA", doc.lang))
+    pub_date = details.get("publication_date", None)
+    if pub_date:
+        ris_pub_date = compute_publication_date_for_ris(pub_date=pub_date)
+        if ris_pub_date:
+            lines.append(ris_line("PY", ris_pub_date))
+    if doc.doi:
+        lines.append(ris_line("DO", doc.doi))
+    authors = details.get("authors", None)
+    if authors:
+        lines.extend(compute_authors_for_ris(authors=authors))
+    publisher = details.get("publisher", None)
+    if publisher:
+        lines.append(ris_line("PB", publisher))
+    license_url = details.get("license_url", None)
+    if license_url:
+        lines.append(ris_line("C1", license_url))
+    lines.append("ER  - ")
+    return "\n".join(lines)

src/app/services/sql_db/queries.py

@@ -6,6 +6,7 @@
 
 from qdrant_client.http.models import ScoredPoint
 from sqlalchemy import func, select
+from sqlalchemy.orm import joinedload
 from welearn_database.data.enumeration import Step
 from welearn_database.data.models import (
     Category,
@@ -82,18 +83,23 @@ def get_document_qty_table_info_sync() -> (
         )  # type: ignore
 
 
+def get_documents_by_ids(documents_ids: list[str]) -> list[WeLearnDocument]:
+    with session_maker() as s:
+        documents = (
+            s.query(WeLearnDocument)
+            .where(WeLearnDocument.id.in_(documents_ids))
+            .options(joinedload(WeLearnDocument.corpus))
+            .all()
+        )
+
+    ret = list(documents)
+
+    return ret
+
+
 def get_documents_payload_by_ids_sync(documents_ids: list[str]) -> list[Document]:
     with session_maker() as s:
-        documents = s.execute(
-            select(
-                WeLearnDocument.title,
-                WeLearnDocument.url,
-                WeLearnDocument.corpus_id,
-                WeLearnDocument.id,
-                WeLearnDocument.description,
-                WeLearnDocument.details,
-            ).where(WeLearnDocument.id.in_(documents_ids))
-        ).all()
+        documents = get_documents_by_ids(documents_ids=documents_ids)
 
         # Batch fetch corpora
         corpus_ids = list({doc.corpus_id for doc in documents})

src/app/tests/api/api_v1/test_bibliography.py

@@ -0,0 +1,73 @@
+import unittest
+from unittest import mock
+
+from fastapi.testclient import TestClient
+
+from src.app.core.config import settings
+from src.main import app
+
+client = TestClient(app)
+
+
+@mock.patch(
+    "src.app.shared.infra.security.check_api_key_sync",
+    new=mock.MagicMock(return_value=True),
+)
+class BibliographyApiTests(unittest.IsolatedAsyncioTestCase):
+    @mock.patch("src.app.api.api_v1.endpoints.bibliography.welearn_document_to_ris")
+    @mock.patch("src.app.api.api_v1.endpoints.bibliography.get_documents_by_ids")
+    async def test_export_bibliography_success(
+        self, get_documents_by_ids_mock, welearn_document_to_ris_mock, *mocks
+    ):
+        doc1 = mock.sentinel.doc1
+        doc2 = mock.sentinel.doc2
+        get_documents_by_ids_mock.return_value = [doc1, doc2]
+        welearn_document_to_ris_mock.side_effect = [
+            "TY  - JFULL\nER  - ",
+            "TY  - BOOK\nER  - ",
+        ]
+
+        payload = {
+            "documents_ids": [
+                "12345678-1234-5678-1234-567812345678",
+                "87654321-4321-8765-4321-876543218765",
+            ]
+        }
+        response = client.post(
+            f"{settings.API_V1_STR}/bibliography/export_bibliography",
+            json=payload,
+            headers={"X-API-Key": "test"},
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.text, "TY  - JFULL\nER  - \nTY  - BOOK\nER  - \n")
+        get_documents_by_ids_mock.assert_called_once_with(payload["documents_ids"])
+        welearn_document_to_ris_mock.assert_has_calls(
+            [mock.call(doc1), mock.call(doc2)]
+        )
+
+    @mock.patch("src.app.api.api_v1.endpoints.bibliography.welearn_document_to_ris")
+    @mock.patch("src.app.api.api_v1.endpoints.bibliography.get_documents_by_ids")
+    async def test_export_bibliography_empty_documents(
+        self, get_documents_by_ids_mock, welearn_document_to_ris_mock, *mocks
+    ):
+        get_documents_by_ids_mock.return_value = []
+
+        response = client.post(
+            f"{settings.API_V1_STR}/bibliography/export_bibliography",
+            json={"documents_ids": ["12345678-1234-5678-1234-567812345678"]},
+            headers={"X-API-Key": "test"},
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.text, "")
+        welearn_document_to_ris_mock.assert_not_called()
+
+    def test_export_bibliography_invalid_payload(self, *mocks):
+        response = client.post(
+            f"{settings.API_V1_STR}/bibliography/export_bibliography",
+            json={"documents_ids": ["not-a-uuid"]},
+            headers={"X-API-Key": "test"},
+        )
+
+        self.assertEqual(response.status_code, 422)

src/app/tests/services/test_helpers.py

@@ -1,15 +1,22 @@
+from types import SimpleNamespace
+from typing import Any, cast
 from unittest import TestCase, mock
 
 import numpy
 from langdetect.language import Language
 
 from src.app.models.documents import Document, DocumentPayloadModel
 from src.app.services.helpers import (
+    compute_authors_for_ris,
+    compute_publication_date_for_ris,
+    compute_ris_doctype,
     convert_embedding_bytes,
     detect_language_from_entry,
     extract_json_from_response,
     linkify_missing_citations,
+    ris_line,
     stringify_docs_content,
+    welearn_document_to_ris,
 )
 from src.app.shared.domain.exceptions import LanguageNotSupportedError
 
@@ -155,3 +162,86 @@ def test_convert_embedding_bytes(self):
         ret = convert_embedding_bytes(embeddings_byte=x.tobytes(), dtype=numpy.float64)
 
         self.assertEqual(x.tolist(), ret.tolist())
+
+    def test_ris_line(self):
+        self.assertEqual(ris_line("TI", "My Title"), "TI  - My Title")
+        self.assertEqual(ris_line("ID", 123), "ID  - 123")
+        self.assertIsNone(ris_line("TI", None))
+        self.assertIsNone(ris_line("TI", ""))
+
+    def test_compute_ris_doctype(self):
+        self.assertEqual(compute_ris_doctype("book"), "BOOK")
+        self.assertEqual(compute_ris_doctype("chapter"), "CHAP")
+        self.assertEqual(compute_ris_doctype("article"), "JFULL")
+        self.assertEqual(compute_ris_doctype("report"), "ELEC")
+        self.assertEqual(compute_ris_doctype(None), "ELEC")
+
+    def test_compute_publication_date_for_ris(self):
+        ret = compute_publication_date_for_ris("1704067200")
+        self.assertRegex(ret, r"^\d{4}/\d{2}/\d{2}$")
+        self.assertEqual(compute_publication_date_for_ris("abc"), "")
+
+    def test_compute_authors_for_ris(self):
+        authors = [{"name": "Alice"}, {"name": "Bob"}]
+        self.assertEqual(
+            compute_authors_for_ris(cast(Any, authors)), ["AU  - Alice", "AU  - Bob"]
+        )
+        self.assertEqual(compute_authors_for_ris(cast(Any, [])), [])
+
+    def test_welearn_document_to_ris_complete(self):
+        doc = SimpleNamespace(
+            id="12345678-1234-5678-1234-567812345678",
+            title="SDG education",
+            url="https://example.org/doc",
+            description="Short description",
+            corpus=SimpleNamespace(source_name="test_corpus"),
+            lang="en",
+            doi="10.1000/test",
+            details={
+                "type": "article",
+                "publication_date": "1704067200",
+                "authors": [{"name": "Alice"}, {"name": "Bob"}],
+                "publisher": "UNESCO",
+                "license_url": "https://license.example.org",
+            },
+        )
+
+        ris = welearn_document_to_ris(cast(Any, doc))
+
+        self.assertIn("TY  - JFULL", ris)
+        self.assertIn("ID  - 12345678-1234-5678-1234-567812345678", ris)
+        self.assertIn("TI  - SDG education", ris)
+        self.assertIn("UR  - https://example.org/doc", ris)
+        self.assertIn("AB  - Short description", ris)
+        self.assertIn("DB  - test_corpus", ris)
+        self.assertIn("LA  - en", ris)
+        self.assertIn("DO  - 10.1000/test", ris)
+        self.assertIn("AU  - Alice", ris)
+        self.assertIn("AU  - Bob", ris)
+        self.assertIn("PB  - UNESCO", ris)
+        self.assertIn("C1  - https://license.example.org", ris)
+        self.assertRegex(ris, r"\nPY  - \d{4}/\d{2}/\d{2}\n")
+        self.assertTrue(ris.endswith("ER  - "))
+
+    def test_welearn_document_to_ris_minimal(self):
+        doc = SimpleNamespace(
+            id="12345678-1234-5678-1234-567812345678",
+            title="Minimal doc",
+            url="https://example.org/minimal",
+            description="dfqsdfsqdf",
+            corpus=SimpleNamespace(source_name="test_corpus"),
+            lang="fr",
+            doi=None,
+            details=None,
+        )
+
+        ris = welearn_document_to_ris(cast(Any, doc))
+
+        self.assertIn("TY  - ELEC", ris)
+        self.assertIn("TI  - Minimal doc", ris)
+        self.assertIn("DB  - test_corpus", ris)
+        self.assertIn("LA  - fr", ris)
+        self.assertNotIn("\nDO  - ", ris)
+        self.assertNotIn("\nPB  - ", ris)
+        self.assertNotIn("\nC1  - ", ris)
+        self.assertTrue(ris.endswith("ER  - "))