turash/eu_funding_api.py

#!/usr/bin/env python3
"""
EU Funding & Tenders Portal API Client

This script provides a Python client for accessing the EU Funding & Tenders Portal APIs.
It supports all major services including Grants & Tenders, Topic Details, Grant Updates,
FAQs, Organization data, Partner Search, and Project Results.

Requirements:
- requests
- json

Install dependencies:
pip install requests

Usage:
python eu_funding_api.py

Author: Generated for City Resource Graph project
Date: November 2025
"""

import requests
import json
import sys
from typing import Dict, List, Optional, Any
from datetime import datetime


class EUFundingAPI:
    """EU Funding & Tenders Portal API Client"""

    BASE_URL = "https://api.tech.ec.europa.eu/search-api/prod/rest"
    SEARCH_API_KEY = "SEDIA"
    FAQ_API_KEY = "SEDIA_FAQ"
    PERSON_API_KEY = "SEDIA_PERSON"

    def __init__(self, timeout: int = 30):
        """Initialize the API client"""
        self.session = requests.Session()
        self.session.headers.update(
            {
                "User-Agent": "EU-Funding-API-Client/1.0",
                "Accept": "application/json",
                "Content-Type": "application/json",
            }
        )
        self.timeout = timeout

    def _make_request(
        self, endpoint: str, method: str = "GET", data: Optional[Dict] = None
    ) -> Dict:
        """Make HTTP request to API"""
        try:
            if method == "POST" and data:
                response = self.session.post(endpoint, json=data, timeout=self.timeout)
            else:
                response = self.session.get(endpoint, timeout=self.timeout)

            response.raise_for_status()
            return response.json()

        except requests.exceptions.RequestException as e:
            print(f"API request failed: {e}")
            return {}

    def _extract_metadata_field(
        self, result: Dict, field_name: str, default: Any = "N/A"
    ) -> Any:
        """
        Extract field from metadata array

        Args:
            result: Result dictionary from API
            field_name: Name of the field to extract
            default: Default value if field not found

        Returns:
            Extracted field value or default
        """
        metadata = result.get("metadata", {})
        field_data = metadata.get(field_name, [])

        if isinstance(field_data, list) and len(field_data) > 0:
            return field_data[0]  # Return first element of array
        elif isinstance(field_data, str):
            return field_data  # Sometimes it's already a string
        else:
            return default

    def _extract_result_data(self, result: Dict) -> Dict:
        """
        Extract common fields from a result

        Args:
            result: Raw result from API

        Returns:
            Cleaned result data
        """
        return {
            "title": self._extract_metadata_field(result, "title"),
            "identifier": self._extract_metadata_field(result, "identifier"),
            "callIdentifier": self._extract_metadata_field(result, "callIdentifier"),
            "status": self._extract_metadata_field(result, "status"),
            "deadline": self._extract_metadata_field(result, "deadlineDate"),
            "description": self._extract_metadata_field(result, "description"),
            "type": self._extract_metadata_field(result, "type"),
            "frameworkProgramme": self._extract_metadata_field(
                result, "frameworkProgramme"
            ),
            "raw_data": result,  # Keep original data for debugging
        }

    def search_grants_tenders(self, query: Optional[Dict] = None) -> Dict:
        """
        Search for grants and tenders

        Args:
            query: Optional search query in Elasticsearch format

        Returns:
            Search results with processed data
        """
        endpoint = f"{self.BASE_URL}/search?apiKey={self.SEARCH_API_KEY}&text=***"

        if query is None:
            # Default query to get all grants and tenders
            query = {
                "bool": {
                    "must": [
                        {"terms": {"type": ["0", "1", "2", "8"]}},  # All types
                        {
                            "terms": {
                                "status": [
                                    "31094501",
                                    "31094502",
                                    "31094503",
                                ]  # All statuses
                            }
                        },
                    ]
                }
            }

        raw_results = self._make_request(endpoint, method="POST", data=query)

        # Process results to extract metadata fields
        if "results" in raw_results:
            processed_results = []
            for result in raw_results["results"]:
                processed_results.append(self._extract_result_data(result))
            raw_results["processed_results"] = processed_results

        return raw_results

    def get_topic_details(self, topic_identifier: str) -> Dict:
        """
        Get detailed information about a specific topic

        Args:
            topic_identifier: Topic identifier (e.g., "HORIZON-EIC-2026-ACCELERATOR-01")

        Returns:
            Topic details with processed data
        """
        query = {"bool": {"must": [{"term": {"identifier": topic_identifier}}]}}
        endpoint = f"{self.BASE_URL}/search?apiKey={self.SEARCH_API_KEY}&text=***"
        raw_results = self._make_request(endpoint, method="POST", data=query)

        # Process results to extract metadata fields
        if "results" in raw_results and raw_results["results"]:
            processed_results = []
            for result in raw_results["results"]:
                processed_results.append(self._extract_result_data(result))
            raw_results["processed_results"] = processed_results

        return raw_results

    def search_grant_updates(self, framework_programme: Optional[str] = None) -> Dict:
        """
        Search for grant updates

        Args:
            framework_programme: Optional framework programme code

        Returns:
            Grant updates
        """
        query = {"bool": {"must": [{"terms": {"type": ["6"]}}]}}  # Grant updates

        if framework_programme:
            query["bool"]["must"].append(
                {"terms": {"frameworkProgramme": [framework_programme]}}
            )

        endpoint = f"{self.BASE_URL}/search?apiKey={self.SEARCH_API_KEY}&text=***"
        return self._make_request(endpoint, method="POST", data=query)

    def search_faqs(self, programme: Optional[str] = None) -> Dict:
        """
        Search FAQs

        Args:
            programme: Optional programme code

        Returns:
            FAQ results
        """
        query = {
            "bool": {
                "must": [
                    {"terms": {"type": ["0", "1"]}},  # FAQ types
                    {"terms": {"status": ["0", "1"]}},  # All statuses
                ]
            }
        }

        if programme:
            query["bool"]["must"].append({"term": {"programme": programme}})

        endpoint = f"{self.BASE_URL}/search?apiKey={self.FAQ_API_KEY}&text=***"
        return self._make_request(endpoint, method="POST", data=query)

    def get_organization_data(self, pic_code: str) -> Dict:
        """
        Get organization public data

        Args:
            pic_code: 9-digit PIC code of the organization

        Returns:
            Organization data
        """
        endpoint = f"{self.BASE_URL}/document/{pic_code}?apiKey={self.PERSON_API_KEY}"
        return self._make_request(endpoint)

    def search_partners(self, topic: str) -> Dict:
        """
        Search for partners in a specific topic

        Args:
            topic: Topic identifier

        Returns:
            Partner search results
        """
        query = {
            "bool": {
                "must": [
                    {"terms": {"topics": [topic]}},
                    {"terms": {"type": ["ORGANISATION", "PERSON"]}},
                ]
            }
        }

        endpoint = f"{self.BASE_URL}/search?apiKey={self.SEARCH_API_KEY}&text=***"
        return self._make_request(endpoint, method="POST", data=query)

    def search_projects(
        self, programme_id: Optional[str] = None, mission_group: Optional[str] = None
    ) -> Dict:
        """
        Search for EU funded projects

        Args:
            programme_id: Programme identifier
            mission_group: Mission group identifier

        Returns:
            Project results
        """
        query = {"bool": {"must": []}}

        if programme_id:
            query["bool"]["must"].append({"terms": {"programId": [programme_id]}})

        if mission_group:
            query["bool"]["must"].append({"terms": {"missionGroup": [mission_group]}})

        endpoint = f"{self.BASE_URL}/search?apiKey={self.SEARCH_API_KEY}&text=***"
        return self._make_request(endpoint, method="POST", data=query)

    def get_facet_data(self, facet_query: Dict) -> Dict:
        """
        Get facet/reference data descriptions

        Args:
            facet_query: Facet query

        Returns:
            Facet data
        """
        endpoint = f"{self.BASE_URL}/facet?apiKey={self.SEARCH_API_KEY}&text=***"
        return self._make_request(endpoint, method="POST", data=facet_query)


def main():
    """Main function demonstrating API usage"""
    print("EU Funding & Tenders Portal API Client")
    print("=" * 50)

    api = EUFundingAPI()

    # Example 1: Search for EIC Accelerator opportunities
    print("\n1. Searching for EIC Accelerator opportunities...")
    eic_query = {
        "bool": {
            "must": [
                {"terms": {"type": ["1", "2", "8"]}},  # Grants
                {
                    "terms": {
                        "status": ["31094501", "31094502", "31094503"]  # All statuses
                    }
                },
                {"term": {"callIdentifier": "HORIZON-EIC-2026-ACCELERATOR-01"}},
            ]
        }
    }

    results = api.search_grants_tenders(eic_query)
    if results:
        processed_results = results.get("processed_results", [])
        print(f"Found {len(processed_results)} EIC Accelerator opportunities")
        for result in processed_results[:3]:  # Show first 3
            title = result.get("title", "N/A")
            print(f"- {title}")
    else:
        print("No results found")

    # Example 2: Get topic details for EIC Accelerator
    print("\n2. Getting EIC Accelerator topic details...")
    topic_details = api.get_topic_details("HORIZON-EIC-2026-ACCELERATOR-01")
    if topic_details:
        print("Topic details retrieved successfully")
        processed_results = topic_details.get("processed_results", [])
        if processed_results:
            topic = processed_results[0]
            print(f"Title: {topic.get('title', 'N/A')}")
            print(f"Status: {topic.get('status', 'N/A')}")
            print(f"Deadline: {topic.get('deadline', 'N/A')}")
    else:
        print("Failed to get topic details")

    # Example 3: Search for grant updates
    print("\n3. Searching for recent grant updates...")
    updates = api.search_grant_updates("43108390")  # Horizon Europe
    if updates:
        print(f"Found {len(updates.get('results', []))} grant updates")
    else:
        print("No grant updates found")

    # Example 4: Search FAQs
    print("\n4. Searching for FAQs...")
    faqs = api.search_faqs()
    if faqs:
        print(f"Found {len(faqs.get('results', []))} FAQs")
    else:
        print("No FAQs found")

    print("\nAPI client demonstration completed!")


def create_eic_accelerator_monitor():
    """
    Create a monitoring script for EIC Accelerator opportunities
    This can be run periodically to check for new opportunities
    """
    api = EUFundingAPI()

    print("EIC Accelerator Monitor")
    print("=" * 30)

    # Query for EIC Accelerator opportunities
    query = {
        "bool": {
            "must": [
                {"terms": {"type": ["1", "2", "8"]}},
                {"terms": {"status": ["31094501", "31094502"]}},  # Open and forthcoming
                {"term": {"frameworkProgramme": "43108390"}},  # Horizon Europe
                {
                    "query_string": {
                        "query": "EIC Accelerator",
                        "default_field": "title",
                    }
                },
            ]
        }
    }

    results = api.search_grants_tenders(query)

    if results and "processed_results" in results:
        opportunities = results["processed_results"]
        print(f"Found {len(opportunities)} EIC Accelerator opportunities:")

        for opp in opportunities:
            title = opp.get("title", "N/A")
            identifier = opp.get("identifier", "N/A")
            status = opp.get("status", "N/A")
            deadline = opp.get("deadline", "N/A")

            print(f"\n📋 {title}")
            print(f"   ID: {identifier}")
            print(f"   Status: {status}")
            print(f"   Deadline: {deadline}")
    else:
        print("No EIC Accelerator opportunities found")


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "monitor":
        create_eic_accelerator_monitor()
    else:
        main()