Add fields metadata suggestions

app/activities/__init__.py

@@ -6,11 +6,13 @@
 from .check_funding_relevance import check_funding_relevance
 from .extract_metadata import extract_metadata_with_llm
 from .extract_pdf_content import extract_pdf_text
+from .resolve_metadata import resolve_metadata_suggestions
 from .update_workflow import update_workflow
 
 REGISTERED_ACTIVITIES = [
     extract_pdf_text,
     extract_metadata_with_llm,
+    resolve_metadata_suggestions,
     update_workflow,
     check_funding_relevance,
 ]
@@ -19,6 +21,7 @@
     "REGISTERED_ACTIVITIES",
     "extract_pdf_text",
     "extract_metadata_with_llm",
+    "resolve_metadata_suggestions",
     "update_workflow",
     "check_funding_relevance",
 ]

app/activities/check_funding_relevance.py

@@ -6,7 +6,7 @@
 from pydantic import BaseModel, Field
 from temporalio import activity
 
-from app.activities._llm import build_agent
+from app.agent import build_agent
 from app.config import get_settings
 from app.observability import propagate_langfuse_context
 from app.workflows.specs import WorkflowContext

app/activities/extract_metadata.py

@@ -13,10 +13,10 @@
 from temporalio import activity
 from temporalio.common import RetryPolicy
 
-from app.activities._llm import build_agent
+from app.agent import build_agent
 from app.config import get_settings
 from app.observability import propagate_langfuse_context
-from app.schemas.metadata_suggestions import ExtractedMetadata, MetadataSuggestions
+from app.schemas.metadata_suggestions import ExtractedMetadata
 from app.workflows.specs import WorkflowContext
 
 EXTRACT_METADATA_RETRY_POLICY = RetryPolicy(
@@ -97,15 +97,15 @@ def _clear_absent_fields(output: ExtractedMetadata, text: str) -> None:
 async def extract_metadata_with_llm(
     request: ExtractMetadataRequest,
     context: WorkflowContext,
-) -> MetadataSuggestions:
-    """Generate typed metadata suggestions using an LLM."""
+) -> ExtractedMetadata:
+    """Extract raw metadata suggestions using an LLM."""
     if len(request.text.strip()) < MIN_TEXT_CHARS:
         # No usable text: skip the LLM entirely rather than let it fabricate.
-        return MetadataSuggestions(suggestions=[])
+        return ExtractedMetadata()
 
     agent = build_agent(get_settings().llm, ExtractedMetadata, INSTRUCTIONS)
     with propagate_langfuse_context(context, trace_name="extract_metadata"):
         result = await agent.run(request.text)
 
     _clear_absent_fields(result.output, request.text)
-    return result.output.to_suggestions()
+    return result.output

app/activities/extract_pdf_content.py

@@ -2,15 +2,14 @@
 # SPDX-License-Identifier: MIT
 
 from datetime import timedelta
-from urllib.parse import urlparse
 
 import httpx
 from pydantic import BaseModel
 from temporalio import activity
 from temporalio.common import RetryPolicy
 from temporalio.exceptions import ApplicationError
 
-from app.config import get_settings
+from app.activities.utils import http_verify
 from app.extractors import get_extractor
 from app.extractors.errors import InvalidPageSelectionError
 
@@ -38,34 +37,13 @@ class ExtractPdfContentResponse(BaseModel):
     pages_extracted: list[int]  # List of 1-indexed page numbers that were extracted
 
 
-def _http_verify_for_pdf_url(url: str) -> bool:
-    settings = get_settings()
-
-    parsed_url = urlparse(url)
-    hostname = (parsed_url.hostname or "").lower()
-
-    if hostname in ("localhost", "127.0.0.1", "::1") or hostname.endswith(".localhost"):
-        return False
-
-    allowlist = {
-        host.strip().lower()
-        for host in (settings.pdf_http_allowlist or "").split(",")
-        if host.strip()
-    }
-
-    if hostname not in allowlist:
-        raise ValueError(f"PDF downloads from '{hostname}' are not allowed")
-
-    return True
-
-
 @activity.defn
 async def extract_pdf_text(
     request: ExtractPdfContentRequest,
 ) -> ExtractPdfContentResponse:
     """Download PDF from a URL and extract its content using the specified extractor."""
     try:
-        verify = _http_verify_for_pdf_url(request.url)
+        verify = http_verify(request.url)
     except ValueError as e:
         raise ApplicationError(
             str(e),

app/activities/resolve_metadata.py

@@ -0,0 +1,253 @@
+# SPDX-FileCopyrightText: 2026 CERN.
+# SPDX-License-Identifier: MIT
+
+"""Activity that resolves funders, awards, and licenses via Invenio vocabularies."""
+
+import asyncio
+import json
+import logging
+from datetime import timedelta
+
+import httpx
+from pydantic import BaseModel, Field
+from temporalio import activity
+from temporalio.common import RetryPolicy
+from temporalio.exceptions import ApplicationError
+
+from app.activities.utils import http_verify
+from app.config import get_settings
+from app.schemas.extracted_metadata import ExtractedMetadata, FunderEnum
+from app.schemas.metadata_suggestions import (
+    FundingSuggestion,
+    LicenseSuggestion,
+    MetadataSuggestions,
+)
+from app.schemas.resolved_fields import (
+    ResolvedAward,
+    ResolvedFunder,
+    ResolvedFunding,
+    ResolvedLicense,
+)
+
+logger = logging.getLogger(__name__)
+
+RESOLVE_METADATA_RETRY_POLICY = RetryPolicy(
+    initial_interval=timedelta(seconds=2),
+    backoff_coefficient=2,
+    maximum_interval=timedelta(seconds=30),
+    maximum_attempts=3,
+)
+
+_INVENIO_HEADERS = {"Accept": "application/vnd.inveniordm.v1+json"}
+
+FUNDER_ROR_IDS = {
+    FunderEnum.NIH: "01cwqze88",
+    FunderEnum.NSF: "021nxhr62",
+    FunderEnum.UKRI: "001aqnf71",
+    FunderEnum.FNS: "00yjd3n13",
+    FunderEnum.EC: "00k4n6c32",
+    FunderEnum.FCT: "00snfqn58",
+    FunderEnum.NWO: "04jsz6e67",
+    FunderEnum.NHMRC: "011kf5r70",
+    FunderEnum.ANR: "00rbzpz17",
+    FunderEnum.ARC: "05mmh0f86",
+}
+
+
+class ResolveMetadataRequest(BaseModel):
+    """Request to resolve and generate metadata suggestions from raw metadata."""
+
+    metadata: ExtractedMetadata = Field(description="Raw metadata to resolve")
+
+
+def _log_resolve_error(
+    field: str, instance: str, error: httpx.HTTPError | json.JSONDecodeError
+) -> None:
+    reason = (
+        "request failed"
+        if isinstance(error, httpx.HTTPError)
+        else "invalid JSON response"
+    )
+    logger.error(f"Error resolving {field} {instance} ({reason})", exc_info=error)
+
+
+async def _resolve_license(
+    client: httpx.AsyncClient, license_id: str, base_url: str
+) -> ResolvedLicense:
+    try:
+        response = await client.get(
+            f"{base_url}/api/vocabularies/licenses/{license_id}",
+            headers=_INVENIO_HEADERS,
+        )
+        if response.is_success:
+            data = response.json()
+            return ResolvedLicense(
+                id=data["id"],
+                title=data.get("title_l10n"),
+                description=data.get("description_l10n"),
+                link=data.get("props", {}).get("url"),
+            )
+    except (httpx.HTTPError, json.JSONDecodeError) as e:
+        _log_resolve_error("license", license_id, e)
+    return ResolvedLicense(id=license_id)
+
+
+async def _resolve_award_by_number(
+    client: httpx.AsyncClient,
+    number: str,
+    funder_id: str | None,
+    base_url: str,
+) -> tuple[None, None] | tuple[ResolvedAward, ResolvedFunder]:
+    try:
+        params: dict = {"q": f"number:{number}", "size": 1}
+        if funder_id:
+            params["funders"] = funder_id
+        resp = await client.get(
+            f"{base_url}/api/awards",
+            params=params,
+            headers=_INVENIO_HEADERS,
+        )
+        if not resp.is_success:
+            return None, None
+        hits = resp.json().get("hits", {}).get("hits", [])
+        if not hits:
+            return None, None
+        hit = hits[0]
+        return ResolvedAward(
+            id=hit.get("id"),
+            number=hit.get("number"),
+            title=hit.get("title_l10n"),
+        ), ResolvedFunder(id=hit["funder"]["id"], name=hit["funder"]["name"])
+    except (httpx.HTTPError, json.JSONDecodeError) as e:
+        _log_resolve_error("award by number", number, e)
+    return None, None
+
+
+async def _resolve_award_by_title(
+    client: httpx.AsyncClient,
+    title: str,
+    funder_id: str | None,
+    base_url: str,
+) -> tuple[None, None] | tuple[ResolvedAward, ResolvedFunder]:
+    try:
+        params: dict = {"q": title, "size": 2, "sort": "bestmatch"}
+        if funder_id:
+            params["funders"] = funder_id
+        resp = await client.get(
+            f"{base_url}/api/awards",
+            params=params,
+            headers=_INVENIO_HEADERS,
+        )
+        if not resp.is_success:
+            return None, None
+        hits = resp.json().get("hits", {}).get("hits", [])
+        # Use the result when unambiguous: 1 match or acronym is found in the 1st result
+        if len(hits) != 1:
+            if (
+                not hits
+                or not (acronym := hits[0].get("acronym"))
+                or acronym not in title
+            ):
+                return None, None
+        hit = hits[0]
+        return ResolvedAward(
+            id=hit.get("id"),
+            number=hit.get("number"),
+            title=hit.get("title_l10n"),
+        ), ResolvedFunder(id=hit["funder"]["id"], name=hit["funder"]["name"])
+    except (httpx.HTTPError, json.JSONDecodeError) as e:
+        _log_resolve_error("award by title", title, e)
+    return None, None
+
+
+async def _resolve_funding_entry(
+    client: httpx.AsyncClient,
+    funder_name: FunderEnum | None,
+    title: str | None,
+    number: str | None,
+    base_url: str,
+) -> ResolvedFunding | None:
+    ror_id = FUNDER_ROR_IDS.get(funder_name) if funder_name else None
+
+    resolved_award: ResolvedAward | None = None
+    resolved_funder: ResolvedFunder | None = None
+    if number:
+        resolved_award, resolved_funder = await _resolve_award_by_number(
+            client, number, ror_id, base_url
+        )
+    if resolved_award is None and title:
+        resolved_award, resolved_funder = await _resolve_award_by_title(
+            client, title, ror_id, base_url
+        )
+    # 0 or >1 award matches: pass whatever info we have as a custom award
+    if resolved_award is None and (number or title):
+        resolved_award = ResolvedAward(
+            number=number or None,
+            title=title or None,
+        )
+    if not resolved_funder and funder_name and ror_id:
+        resolved_funder = ResolvedFunder(id=ror_id, name=funder_name)
+
+    if resolved_funder is None and resolved_award is None:
+        return None
+    return ResolvedFunding(funder=resolved_funder, award=resolved_award)
+
+
+@activity.defn
+async def resolve_metadata_suggestions(
+    request: ResolveMetadataRequest,
+) -> MetadataSuggestions:
+    """Resolve funders, awards, and licenses, and return typed metadata suggestions."""
+    extracted = request.metadata
+    result = MetadataSuggestions.from_extracted(extracted)
+
+    base_url = get_settings().invenio_base_url
+    if not base_url:
+        logger.warning("INVENIO_BASE_URL is not configured, skipping resolution")
+        return result
+
+    try:
+        verify = http_verify(base_url)
+    except ValueError as e:
+        raise ApplicationError(
+            str(e),
+            type="HostNotAllowed",
+            non_retryable=True,
+        ) from e
+
+    async with httpx.AsyncClient(verify=verify) as client:
+        # Resolve licenses in parallel, fall back to id if unable to resolve
+        if extracted.license:
+            resolved_licenses = await asyncio.gather(
+                *[
+                    _resolve_license(client, lic.lower(), base_url)
+                    for lic in extracted.license
+                ]
+            )
+            result.suggestions.append(LicenseSuggestion(value=list(resolved_licenses)))
+
+        # Resolve Funding (awards and funders)
+        funders = extracted.funding_funders
+        size = max(
+            len(extracted.funding_titles),
+            len(extracted.funding_numbers),
+            len(funders),
+        )
+        if size:
+            resolved = await asyncio.gather(
+                *[
+                    _resolve_funding_entry(
+                        client,
+                        funders[i] if i < len(funders) else None,
+                        extracted.at(extracted.funding_titles, i),
+                        extracted.at(extracted.funding_numbers, i),
+                        base_url,
+                    )
+                    for i in range(size)
+                ]
+            )
+            entries = [r for r in resolved if r is not None]
+            if entries:
+                result.suggestions.append(FundingSuggestion(value=entries))
+
+    return result