Spaces:
Sleeping
Sleeping
File size: 3,650 Bytes
c2fb256 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | import requests
from requests.auth import HTTPBasicAuth
from typing import Dict, Optional
import os
import mimetypes
from smolagents.tools import Tool
class WordPressMediaTool(Tool):
name = "wordpress_media"
description = "Uploads media files to WordPress and returns the attachment URL"
inputs = {
'file_path': {'type': 'string', 'description': 'Local path to the media file to upload'},
'title': {
'type': 'string',
'description': 'Title for the media attachment',
'nullable': True
},
'alt_text': {
'type': 'string',
'description': 'Alt text for the media attachment',
'nullable': True
}
}
output_type = "object"
def __init__(self, wordpress_credentials):
super().__init__()
self.credentials = wordpress_credentials
def forward(self, file_path: str, title: str = None, alt_text: str = None) -> Dict:
"""Uploads a media file to WordPress
Args:
file_path: Path to the media file
title: Optional title for the media attachment
alt_text: Optional alt text for the media attachment
Returns:
Dict containing the response from WordPress with media details
"""
if not self.credentials:
return {"error": "WordPress credentials not configured"}
if not os.path.exists(file_path):
return {"error": f"File not found: {file_path}"}
api_url = f"{self.credentials['site_url']}/wp-json/wp/v2/media"
auth = HTTPBasicAuth(
self.credentials['username'], self.credentials['app_password'])
# Determine mime type
mime_type, _ = mimetypes.guess_type(file_path)
if not mime_type:
return {"error": "Could not determine file type"}
# Prepare headers
headers = {
'Content-Type': mime_type,
'Content-Disposition': f'attachment; filename="{os.path.basename(file_path)}"'
}
try:
# Upload file
with open(file_path, 'rb') as file:
response = requests.post(
api_url,
headers=headers,
auth=auth,
data=file
)
response.raise_for_status()
response_data = response.json()
# Update media properties if title or alt_text provided
if title or alt_text:
media_id = response_data['id']
update_url = f"{api_url}/{media_id}"
update_data = {}
if title:
update_data['title'] = {'rendered': title}
if alt_text:
update_data['alt_text'] = alt_text
if update_data:
update_response = requests.post(
update_url,
json=update_data,
auth=auth
)
update_response.raise_for_status()
response_data = update_response.json()
return {
"status": "success",
"attachment_id": response_data.get("id"),
"url": response_data.get("source_url"),
"title": response_data.get("title", {}).get("rendered"),
"alt_text": response_data.get("alt_text")
}
except requests.exceptions.RequestException as e:
print(f"Failed to upload media: {str(e)}")
return {"error": f"Failed to upload media: {str(e)}"}
|