app.py

from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel,load_tool,tool
import datetime
import requests
import pytz
import yaml
from typing import List, Dict, Optional
import tkapi
from tkapi.fractie import Fractie, FractieFilter
from tkapi.persoon import PersoonFilter, PersoonNevenfunctie
from tkapi.util import queries
from tkapi.zaak import Zaak, ZaakSoort
from tools.final_answer import FinalAnswerTool

from Gradio_UI import GradioUI

api = tkapi.TKApi()

@tool
def get_kamervraag_context(zaak_nummer: str) -> Dict[str, any]:
    """
    Retrieves detailed context information for a specific parliamentary item by its number.
    
    Args:
        zaak_nummer: The unique identifier of the parliamentary item (e.g., '2024Z00541')
    
    Returns:
        Dict[str, any]: Dictionary containing detailed item information:
            - 'nummer': Unique identifier of the item
            - 'soort': Type of the item
            - 'onderwerp': Subject/topic of the item
            - 'datum': Date the item was started
            - 'afgedaan': Whether the item has been handled/answered
            - 'indieners': List of submitters with their party
            - 'documenten': List of related documents with their details
            - 'besluiten': List of related decisions
            - 'activiteiten': List of related activities
            
    Example return value:
        {
            'nummer': '2024Z00541',
            'soort': 'Motie',
            'onderwerp': 'Motie van de leden over uitingen van de schutterijcultuur',
            'datum': datetime.date(2024, 1, 18),
            'afgedaan': True,
            'indieners': ['Geert Wilders (PVV)', 'Martin Bosma (PVV)'],
            'documenten': [
                {
                    'nummer': '36410-VIII-56',
                    'soort': 'Motie',
                    'titel': 'Motie van de leden...',
                    'datum': datetime.date(2024, 1, 18)
                }
            ],
            'besluiten': ['Aangenomen op 23 januari 2024'],
            'activiteiten': [
                {
                    'soort': 'Stemmingen',
                    'datum': datetime.date(2024, 1, 23),
                    'status': 'Afgerond'
                }
            ]
        }
    """
    try:
        # Initialize API
        api = tkapi.TKApi(verbose=False)
        
        # Create filter for specific zaak
        filter = Zaak.create_filter()
        filter.filter_nummer(zaak_nummer)
        
        # Get the zaak
        zaken = api.get_zaken(filter=filter)
        zaak = next(iter(zaken), None)
        
        if not zaak:
            print(f"No parliamentary item found with number {zaak_nummer}")
            return {}
            
        # Collect basic information
        result = {
            'nummer': zaak.nummer,
            'soort': zaak.soort.value if zaak.soort else None,
            'onderwerp': zaak.onderwerp,
            'datum': zaak.gestart_op,
            'afgedaan': zaak.afgedaan,
            'indieners': [],
            'documenten': [],
            'besluiten': [],
            'activiteiten': []
        }
        
        # Get indieners (actors)
        actors = zaak.actors
        if actors:
            for actor in actors:
                if hasattr(actor, 'persoon') and actor.persoon:
                    persoon = actor.persoon
                    fractie_str = ""
                    if hasattr(persoon, 'fractie') and persoon.fractie:
                        fractie_str = f" ({persoon.fractie.afkorting})"
                    result['indieners'].append(
                        f"{persoon.roepnaam} {persoon.achternaam}{fractie_str}"
                    )
        #a
        # Get related documents
        for doc in zaak.documenten:
            doc_info = {
                'nummer': doc.nummer if hasattr(doc, 'nummer') else None,
                'soort': doc.soort.value if hasattr(doc, 'soort') and doc.soort else None,
                'titel': doc.titel if hasattr(doc, 'titel') else None,
                'datum': doc.datum if hasattr(doc, 'datum') else None
            }
            result['documenten'].append(doc_info)
        
        # Get decisions
        for besluit in zaak.besluiten:
            if hasattr(besluit, 'tekst') and besluit.tekst:
                result['besluiten'].append(besluit.tekst)
        
        # Get activities
        for activiteit in zaak.activiteiten:
            act_info = {
                'soort': activiteit.soort.value if hasattr(activiteit, 'soort') and activiteit.soort else None,
                'datum': activiteit.datum if hasattr(activiteit, 'datum') else None,
                'status': activiteit.status if hasattr(activiteit, 'status') else None
            }
            result['activiteiten'].append(act_info)
        
        return result
        
    except Exception as e:
        print(f"Error fetching context for parliamentary item: {str(e)}")
        return {}

@tool 
def get_kamervragen(
    start_datetime: datetime.datetime,
    end_datetime: datetime.datetime,
    only_answered: Optional[bool] = None,
    zaak_soort: ZaakSoort = ZaakSoort.SCHRIFTELIJKE_VRAGEN
) -> List[Dict[str, any]]:
    """
    Retrieves basic information about parliamentary items (kamervragen, moties, etc.) within a specified date range.
    
    Args:
        start_datetime: Start date as datetime object
        end_datetime: End date as datetime object
        only_answered: Optional filter for answered items (True = only answered, False = only unanswered, None = both)
        zaak_soort: Type of parliamentary item to filter on. Options from ZaakSoort enum:
            - AMENDEMENT: Amendments
            - BRIEF_COMMISSIE: Committee letters
            - BRIEF_REGERING: Government letters
            - BRIEF_LID: Member letters
            - INITIATIEF_NOTA: Initiative notes
            - INITIATIEF_WETGEVING: Initiative legislation
            - INTERPELLATIE: Interpellations
            - LIJST_MET_VRAGEN: List of questions
            - MEDEDELINGEN: Announcements
            - MONDELINGE_VRAGEN: Oral questions
            - MOTIE: Motions
            - OVERIG: Other
            - PARLEMENTAIR_DOCUMENT: Parliamentary documents
            - RAPPORT: Reports
            - SCHRIFTELIJKE_VRAGEN: Written questions (default)
            - VERSLAG_COMMISSIE: Committee reports
            - VOORDRACHT: Nominations
            - WETSVOORSTEL: Bill proposals
    
    Returns:
        List[Dict[str, any]]: List of dictionaries containing basic item information:
            - 'nummer': Unique identifier of the item
            - 'soort': Type of the item
            - 'onderwerp': Subject/topic of the item
            - 'datum': Date the item was started (as datetime.date)
            - 'afgedaan': Boolean indicating if the item has been handled/answered
            
    Example return value:
        [
            {
                'nummer': '2024Z00541',
                'soort': 'Motie',
                'onderwerp': 'Motie van de leden over uitingen van de schutterijcultuur',
                'datum': datetime.date(2024, 1, 18),
                'afgedaan': True
            }
        ]
    """
    try:
        # Initialize API
        api = TKApi(verbose=False)
        
        # Create filter
        filter = Zaak.create_filter()
        filter.filter_soort(zaak_soort)
        filter.filter_date_range(start_datetime, end_datetime)
        
        if only_answered is not None:
            filter.filter_afgedaan(only_answered)
        
        # Get items
        zaken = api.get_zaken(filter=filter)
        
        # Format results
        results = []
        for zaak in zaken:
            zaak_info = {
                'nummer': zaak.nummer,
                'soort': zaak.soort.value if zaak.soort else None,
                'onderwerp': zaak.onderwerp,
                'datum': zaak.gestart_op,
                'afgedaan': zaak.afgedaan
            }
            results.append(zaak_info)
        
        return results
        
    except Exception as e:
        print(f"Error fetching parliamentary items: {str(e)}")
        return []

@tool
def get_kamerlid_reizen(achternaam: str) -> List[Dict[str, str]]:
    """
    Retrieves all registered travels for a specific parliament member by their last name.
    
    Args:
        achternaam: The last name of the parliament member to search for
                   (case-sensitive, partial matches are supported)
    
    Returns:
        List[Dict[str, str]]: A list of dictionaries containing travel information:
            - 'bestemming': Destination of the travel
            - 'doel': Purpose of the travel
            - 'van': Start date of the travel
            - 'tot': End date of the travel
            - 'betaald_door': Who paid for the travel
            
    Example return value:
        [
            {
                'bestemming': 'Brussels',
                'doel': 'EU Parliament meeting',
                'van': '2024-01-15',
                'tot': '2024-01-17',
                'betaald_door': 'European Parliament'
            }
        ]
        
    Note:
        If multiple parliament members match the given last name, 
        travels for all matching members will be returned.
    """
    try:
        # Initialize API
        api = tkapi.TKApi(verbose=False)
        
        # Create person filter
        filter = PersoonFilter()
        filter.filter_achternaam(achternaam)
        
        # Get matching persons
        persons = api.get_personen(filter=filter)
        
        # Collect all travels
        all_travels = []
        for person in persons:
            for reis in person.reizen:
                travel_info = {
                    'bestemming': reis.bestemming,
                    'doel': reis.doel,
                    'van': str(reis.van) if reis.van else None,
                    'tot': str(reis.tot_en_met) if reis.tot_en_met else None,
                    'betaald_door': reis.betaald_door
                }
                all_travels.append(travel_info)
        
        return all_travels
    
    except Exception as e:
        print(f"Error fetching travel data: {str(e)}")
        return []

@tool
def Get_Active_Fractions() -> list[dict[str, str | int]]:
    """
    Retrieves a list of all currently active fractions (political parties) in the Dutch parliament.
    
    Returns:
        List[Dict[str, Union[str, int]]]: A list of dictionaries containing information about each fraction:
            - 'naam': Full name of the political party (string)
            - 'afkorting': Abbreviation/acronym of the party (string)
            - 'zetels_aantal': Number of seats the party holds (integer)
            
    Example return value:
        [
            {
                'naam': 'Volkspartij voor Vrijheid en Democratie',
                'afkorting': 'VVD',
                'zetels_aantal': 34
            },
            {
                'naam': 'Democraten 66',
                'afkorting': 'D66',
                'zetels_aantal': 24
            }
        ]
    """
    try:
        # Initialize API client
        api = tkapi.TKApi(verbose=False)
        
        # Create filter for active fractions
        filter = Fractie.create_filter()
        filter.filter_actief()
        
        # Get all active fractions
        active_fractions = api.get_fracties(filter=filter)
        
        # Format the results
        results = []
        for fraction in active_fractions:
            fraction_info = {
                'naam': fraction.naam,
                'afkorting': fraction.afkorting,
                'zetels_aantal': fraction.zetels_aantal
            }
            results.append(fraction_info)
            
        return results
        
    except Exception as e:
        print(f"Error fetching fraction data: {str(e)}")
        return []

@tool
def get_fraction_members(fractie_naam: str) -> List[Dict[str, str]]:
    """
    Retrieves all current members of a specific fraction (political party) in the Dutch parliament.
    
    Args:
        fractie_naam: The name of the fraction/party to search for (case-sensitive).
                     Can be either the full name (e.g. 'Volkspartij voor Vrijheid en Democratie')
                     or the abbreviation (e.g. 'VVD')
    
    Returns:
        List[Dict[str, str]]: A list of dictionaries containing member information:
            - 'achternaam': Last name of the member
            - 'geboortedatum': Date of birth
            - 'van': Start date of membership
            - 'fractie': Name of the fraction
            - 'fractie_afkorting': Abbreviation of the fraction
            
    Example return value:
        [
            {
                'achternaam': 'van der Berg',
                'geboortedatum': '1975-03-15',
                'van': '2023-12-06',
                'fractie': 'Volkspartij voor Vrijheid en Democratie',
                'fractie_afkorting': 'VVD'
            }
        ]
        
    Note:
        - This only returns currently active members
        - The search works for both full names and abbreviations
        - If no fraction is found with the given name, an empty list is returned
    """
    try:
        # Initialize API
        api = tkapi.TKApi(verbose=False)
        
        # First try to find the fraction
        filter = FractieFilter()
        filter.filter_actief()  # Only get active fractions
        fracties = api.get_fracties(filter=filter)
        
        # Find matching fraction (check both full name and abbreviation)
        target_fractie = None
        for fractie in fracties:
            if fractie_naam.upper() in [fractie.naam.upper(), fractie.afkorting.upper()]:
                target_fractie = fractie
                break
                
        if not target_fractie:
            print(f"No active fraction found with name or abbreviation: {fractie_naam}")
            return []
            
        # Get active members for the fraction
        members = target_fractie.leden_actief
        
        # Format member information
        member_list = []
        for lid in members:
            persoon = lid.persoon
            member_info = {
                'achternaam': persoon.achternaam,
                'geboortedatum': str(persoon.geboortedatum) if persoon.geboortedatum else None,
                'van': str(lid.van) if lid.van else None,
                'fractie': target_fractie.naam,
                'fractie_afkorting': target_fractie.afkorting
            }
            member_list.append(member_info)
            
        return member_list
    
    except Exception as e:
        print(f"Error fetching fraction members: {str(e)}")
        return []

@tool
def get_kamerlid_geschenken(achternaam: str) -> List[Dict[str, str]]:
    """
    Retrieves all registered gifts for a specific parliament member by their last name.
    
    Args:
        achternaam: The last name of the parliament member to search for
                   (case-sensitive, partial matches are supported)
    
    Returns:
        List[Dict[str, str]]: A list of dictionaries containing gift information:
            - 'persoon': Name of the parliament member
            - 'omschrijving': Description of the gift
            - 'datum': Date the gift was received
            - 'waarde': Value of the gift if specified
            
    Example return value:
        [
            {
                'persoon': 'John van der Berg',
                'omschrijving': 'Book "Politics in the Netherlands"',
                'datum': '2024-01-15',
                'waarde': '€45'
            }
        ]
        
    Note:
        If multiple parliament members match the given last name, 
        gifts for all matching members will be returned.
    """
    try:
        # Initialize API
        api = tkapi.TKApi(verbose=False)
        
        # Create person filter
        filter = PersoonFilter()
        filter.filter_achternaam(achternaam)
        
        # Get matching persons
        persons = api.get_personen(filter=filter)
        
        # Collect all gifts
        all_gifts = []
        for person in persons:
            for geschenk in person.geschenken:
                gift_info = {
                    'persoon': f"{person.roepnaam} {person.achternaam}".strip(),
                    'omschrijving': geschenk.omschrijving,
                    'datum': str(geschenk.datum) if geschenk.datum else None,
                    'waarde': geschenk.waarde if hasattr(geschenk, 'waarde') else None
                }
                all_gifts.append(gift_info)
        
        return all_gifts
    
    except Exception as e:
        print(f"Error fetching gift data: {str(e)}")
        return []

@tool
def get_kamerlid_nevenfuncties(achternaam: str) -> List[Dict[str, str]]:
    """
    Retrieves all registered secondary functions (nevenfuncties) for a specific parliament member by their last name.
    
    Args:
        achternaam: The last name of the parliament member to search for
                   (case-sensitive, partial matches are supported)
    
    Returns:
        List[Dict[str, str]]: A list of dictionaries containing secondary function information:
            - 'persoon': Name of the parliament member
            - 'omschrijving': Description of the function
            - 'vergoeding': Compensation details if any
            - 'van': Start date of the function
            - 'tot': End date of the function (if applicable)
            - 'periode': Period specification (if available)
            
    Example return value:
        [
            {
                'persoon': 'John van der Berg',
                'omschrijving': 'Board member of Foundation XYZ',
                'vergoeding': 'Unpaid',
                'van': '2023-01-01',
                'tot': None,
                'periode': 'Current'
            }
        ]
        
    Note:
        If multiple parliament members match the given last name, 
        secondary functions for all matching members will be returned.
    """
    try:
        # Initialize API
        api = tkapi.TKApi(verbose=False)
        
        # Create person filter
        filter = PersoonFilter()
        filter.filter_achternaam(achternaam)
        
        # Get matching persons
        persons = api.get_personen(filter=filter)
        
        # Collect all secondary functions
        all_functions = []
        for person in persons:
            for functie in person.nevenfuncties:
                function_info = {
                    'persoon': f"{person.roepnaam} {person.achternaam}".strip(),
                    'omschrijving': functie.omschrijving,
                    'vergoeding': functie.vergoeding if hasattr(functie, 'vergoeding') else None,
                    'van': str(functie.van) if functie.van else None,
                    'tot': str(functie.tot) if hasattr(functie, 'tot') and functie.tot else None,
                    'periode': functie.periode if hasattr(functie, 'periode') else None
                }
                all_functions.append(function_info)
        
        return all_functions
    
    except Exception as e:
        print(f"Error fetching secondary function data: {str(e)}")
        return []

@tool
def get_current_time_in_timezone(timezone: str) -> str:
    """A tool that fetches the current local time in a specified timezone.
    Args:
        timezone: A string representing a valid timezone (e.g., 'America/New_York').
    """
    try:
        # Create timezone object
        tz = pytz.timezone(timezone)
        # Get current time in that timezone
        local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")
        return f"The current local time in {timezone} is: {local_time}"
    except Exception as e:
        return f"Error fetching time for timezone '{timezone}': {str(e)}"


final_answer = FinalAnswerTool()
model = HfApiModel(
max_tokens=2096,
temperature=0.5,
model_id='Qwen/Qwen2.5-Coder-32B-Instruct',
custom_role_conversions=None,
)


# Import tool from Hub
image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True)

with open("prompts.yaml", 'r') as stream:
    prompt_templates = yaml.safe_load(stream)
    
agent = CodeAgent(
    model=model,
    tools=[final_answer,
           Get_Active_Fractions,
           get_fraction_members,
           get_kamerlid_reizen,
           get_kamerlid_geschenken,
           get_kamerlid_nevenfuncties,
           get_kamervragen,
           get_kamervraag_context,
           get_current_time_in_timezone], ## the given tools
    max_steps=6,
    verbosity_level=1,
    grammar=None,
    planning_interval=None,
    name=None,
    description=None,
    prompt_templates=prompt_templates
)


GradioUI(agent).launch()