""" title: Zotero Library Search author: Matthieu version: 0.2.0 license: MIT """ import json from pyzotero import zotero from pydantic import BaseModel, Field from typing import Callable, Any, Optional class Tools: class Valves(BaseModel): ZOTERO_API_KEY: str = Field(default="", description="Your Zotero API key") ZOTERO_LIBRARY_ID: str = Field( default="", description="Your Zotero library ID (numeric)" ) ZOTERO_LIBRARY_TYPE: str = Field( default="user", description="Your Zotero library type (user or group)" ) MAX_RESULTS: int = Field( default=10, description="Maximum number of results to return" ) def __init__(self): self.valves = self.Valves() async def search_zotero( self, query: str, __event_emitter__: Optional[Callable[[dict], Any]] = None ) -> str: """ Search your Zotero library for items matching the provided query. :params query: Search term to look for in your Zotero library. :return: Information about matching items in your Zotero library. """ # Emit status if event emitter is provided if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "status": "in_progress", "description": f"Searching Zotero for: {query}", "done": False, }, } ) # Validate required fields if not self.valves.ZOTERO_API_KEY or not self.valves.ZOTERO_LIBRARY_ID: return json.dumps({"error": "Zotero API key and Library ID are required"}) try: # Initialize the Zotero client zot = zotero.Zotero( library_id=self.valves.ZOTERO_LIBRARY_ID, library_type=self.valves.ZOTERO_LIBRARY_TYPE, api_key=self.valves.ZOTERO_API_KEY, ) # Search for items items = zot.items(q=query, limit=self.valves.MAX_RESULTS) # Format results for output results = [] for item in items: data = item.get("data", {}) result = { "key": data.get("key", ""), "title": data.get("title", "Untitled"), "authors": [], "year": "", "itemType": data.get("itemType", ""), "url": data.get("url", ""), "abstract": data.get("abstractNote", ""), } # Get authors creators = data.get("creators", []) for creator in creators: if "name" in creator: result["authors"].append(creator["name"]) else: name = f"{creator.get('firstName', '')} {creator.get('lastName', '')}".strip() if name: result["authors"].append(name) # Get year from date date = data.get("date", "") if len(date) >= 4: result["year"] = date[:4] # Citation formatting if result["authors"] and result["year"]: authors_text = result["authors"][0] if len(result["authors"]) > 1: authors_text += " et al." result["citation"] = ( f"{authors_text} ({result['year']}). {result['title']}" ) else: result["citation"] = result["title"] # Add to results results.append(result) # Send completion status if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "status": "complete", "description": f"Found {len(results)} items", "done": True, }, } ) return json.dumps(results) except Exception as e: error_message = str(e) # Specific error handling for common issues if "Invalid Library ID" in error_message: error_message = "Invalid Zotero Library ID. Please check your settings." elif "Invalid type" in error_message: error_message = "Invalid Zotero Library Type. Use 'user' or 'group'." elif "You do not have permission" in error_message: error_message = "Access denied. Check your API key permissions." # Send error status if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "status": "error", "description": f"Error: {error_message}", "done": True, }, } ) return json.dumps({"error": error_message}) async def get_zotero_item( self, item_key: str, __event_emitter__: Optional[Callable[[dict], Any]] = None ) -> str: """ Get detailed information about a specific Zotero item by its key. :params item_key: The unique key of the Zotero item. :return: Detailed information about the specified Zotero item. """ # Emit status if event emitter is provided if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "status": "in_progress", "description": f"Fetching Zotero item: {item_key}", "done": False, }, } ) # Validate required fields if not self.valves.ZOTERO_API_KEY or not self.valves.ZOTERO_LIBRARY_ID: return json.dumps({"error": "Zotero API key and Library ID are required"}) try: # Initialize the Zotero client zot = zotero.Zotero( library_id=self.valves.ZOTERO_LIBRARY_ID, library_type=self.valves.ZOTERO_LIBRARY_TYPE, api_key=self.valves.ZOTERO_API_KEY, ) # Get the item item = zot.item(item_key) # Get item data data = item.get("data", {}) # Format the result result = { "key": data.get("key", ""), "title": data.get("title", "Untitled"), "authors": [], "year": "", "itemType": data.get("itemType", ""), "url": data.get("url", ""), "abstract": data.get("abstractNote", ""), "tags": [tag.get("tag", "") for tag in data.get("tags", [])], "date": data.get("date", ""), "doi": data.get("DOI", ""), "publisher": data.get("publisher", ""), "publicationTitle": data.get("publicationTitle", ""), "volume": data.get("volume", ""), "issue": data.get("issue", ""), "pages": data.get("pages", ""), "collections": data.get("collections", []), } # Get authors creators = data.get("creators", []) for creator in creators: if "name" in creator: result["authors"].append(creator["name"]) else: name = f"{creator.get('firstName', '')} {creator.get('lastName', '')}".strip() if name: result["authors"].append(name) # Get year from date date = data.get("date", "") if len(date) >= 4: result["year"] = date[:4] # Get attachments try: attachments = zot.children(item_key) result["attachments"] = [] for attachment in attachments: attach_data = attachment.get("data", {}) if attach_data.get("itemType") == "attachment": result["attachments"].append( { "title": attach_data.get("title", ""), "contentType": attach_data.get("contentType", ""), "filename": attach_data.get("filename", ""), "key": attach_data.get("key", ""), } ) except: # If attachments can't be retrieved, just proceed without them result["attachments"] = [] # Send completion status if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "status": "complete", "description": "Item details retrieved", "done": True, }, } ) return json.dumps(result) except Exception as e: error_message = str(e) # Specific error handling for common issues if "Invalid Library ID" in error_message: error_message = "Invalid Zotero Library ID. Please check your settings." elif "Invalid type" in error_message: error_message = "Invalid Zotero Library Type. Use 'user' or 'group'." elif "You do not have permission" in error_message: error_message = "Access denied. Check your API key permissions." elif "not found in collection" in error_message: error_message = f"Item with key '{item_key}' not found in your library." # Send error status if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "status": "error", "description": f"Error: {error_message}", "done": True, }, } ) return json.dumps({"error": error_message})