kunthawat/ALwrity
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

AI Backlinker, Google Ads Generator, Letter Writer - WIP

Browse Source
This commit is contained in:
ajaysi
2025-05-06 22:27:43 +05:30
parent 26b02b9719
commit 5f7d319859
38 changed files with 14572 additions and 302 deletions
Download Patch File Download Diff File Expand all files Collapse all files

208
lib/integrations/wix/README.md Normal file
View File

@@ -0,0 +1,208 @@
# Wix Blog Integration for Alwrity
This integration allows you to publish blog content from Alwrity directly to your Wix site using the Wix REST API.
## Features
- **Blog Post Management**: Create, update, and delete blog posts
- **Media Management**: Upload images and other media files
- **SEO Optimization**: Comprehensive SEO settings and analysis
- **Category Management**: Create and manage blog categories
- **Markdown Support**: Write in markdown and publish as HTML
- **Streamlit UI**: User-friendly interface for publishing
## Prerequisites
Before using this integration, you'll need:
1. A Wix site with the Blog feature enabled
2. Wix API credentials (refresh token and site ID)
3. Python 3.7+ with required dependencies
## Getting Wix API Credentials
To use this integration, you need to obtain a refresh token and site ID from Wix:
1. **Create a Wix Developer Account**:
- Go to [Wix Developers](https://dev.wix.com/) and sign up or log in
- Create a new OAuth app
2. **Configure OAuth App**:
- Set a name and description for your app
- Add redirect URLs (e.g., `https://localhost:3000/oauth/callback`)
- Save the app and note the App ID and App Secret
3. **Get a Refresh Token**:
- Follow the OAuth flow to get an authorization code
- Exchange the code for an access token and refresh token
- Detailed instructions: [Wix OAuth Documentation](https://dev.wix.com/api/rest/getting-started/authentication)
4. **Get Your Site ID**:
- Log in to your Wix account
- Go to your site's dashboard
- The site ID is in the URL: `https://manage.wix.com/dashboard/{SITE_ID}/home`
## Installation
The Wix integration is included with Alwrity. No additional installation is required.
## Usage
### Using the Streamlit UI
1. Navigate to the Wix integration in the Alwrity UI
2. Enter your Wix refresh token and site ID
3. Fill in the blog details and content
4. Click "Publish to Wix"
### Using the Python API
```python
from lib.integrations.wix_integration import WixIntegration
# Initialize the integration
wix = WixIntegration(
refresh_token="YOUR_REFRESH_TOKEN",
site_id="YOUR_SITE_ID"
)
# Publish a blog post
result = wix.publish_blog_post(
title="My Blog Post",
content="# Hello World\n\nThis is my blog post.",
is_markdown=True,
tags=["example", "blog"],
categories=["Technology"],
publish=True
)
# Get the published post URL
post_url = result.get("post", {}).get("url")
print(f"Published at: {post_url}")
```
### Using the Command-Line Interface
```bash
# Set environment variables
export WIX_REFRESH_TOKEN="YOUR_REFRESH_TOKEN"
export WIX_SITE_ID="YOUR_SITE_ID"
# List blog posts
python -m lib.integrations.wix_cli list-posts
# Publish a blog post
python -m lib.integrations.wix_cli publish-post \
--title "My Blog Post" \
--content-file blog.md \
--is-markdown \
--tags "example,blog" \
--categories "Technology"
# Generate an SEO report
python -m lib.integrations.wix_cli seo-report \
--title "My Blog Post" \
--keywords "example,blog,technology"
```
## API Reference
### WixIntegration
The main integration class that provides high-level methods for working with Wix blogs.
#### Methods
- `publish_blog_post(title, content, ...)`: Publish a blog post
- `upload_media(file_path, ...)`: Upload a media file
- `get_seo_report(post_id, target_keywords)`: Generate an SEO report
- `list_blog_posts(limit, offset, ...)`: List blog posts
- `list_categories()`: List blog categories
- `create_category(name, description)`: Create a blog category
- `get_post_by_id(post_id)`: Get a blog post by ID
- `get_post_by_title(title)`: Get a blog post by title
- `delete_post(post_id)`: Delete a blog post
### WixAPIClient
Low-level client for interacting with the Wix API.
### WixBlogManager
Handles blog content management, including markdown processing and image handling.
### WixSEOOptimizer
Provides SEO analysis and optimization for blog posts.
## Error Handling
The integration includes comprehensive error handling:
- API errors are logged with detailed information
- Authentication errors provide clear guidance
- File handling errors include path information
- Network errors include retry logic
## Best Practices
1. **Store credentials securely**:
- Use environment variables or a secure credential store
- Don't hardcode credentials in your code
2. **Optimize images before upload**:
- Compress images to reduce file size
- Use appropriate image formats (JPEG for photos, PNG for graphics)
3. **SEO optimization**:
- Use the SEO report to improve your content
- Include relevant keywords in titles and headings
- Add alt text to all images
4. **Content management**:
- Use categories and tags consistently
- Include featured images for better visual appeal
- Write clear, concise meta descriptions
## Troubleshooting
### Common Issues
1. **Authentication Errors**:
- Ensure your refresh token is valid
- Check that your site ID is correct
- Verify that your app has the necessary permissions
2. **API Rate Limits**:
- The Wix API has rate limits that may affect bulk operations
- Add delays between requests if you're publishing many posts
3. **Image Upload Issues**:
- Check that the image file exists and is readable
- Verify that the image format is supported (JPEG, PNG, GIF)
- Ensure the image file size is within Wix limits
4. **Content Formatting Issues**:
- If using markdown, ensure it's valid
- Check for special characters that might cause issues
- Verify that HTML content is properly formatted
### Getting Help
If you encounter issues not covered here:
1. Check the logs for detailed error messages
2. Consult the [Wix API Documentation](https://dev.wix.com/api/rest/getting-started)
3. Contact Alwrity support for assistance
## License
This integration is part of the Alwrity platform and is subject to the same license terms.
## Acknowledgements
- [Wix REST API](https://dev.wix.com/api/rest) for providing the API endpoints
- [Requests](https://docs.python-requests.org/) for HTTP functionality
- [Markdown](https://python-markdown.github.io/) for markdown processing
- [BeautifulSoup](https://www.crummy.com/software/BeautifulSoup/) for HTML parsing
- [Streamlit](https://streamlit.io/) for the user interface

850
lib/integrations/wix/wix_api_client.py Normal file
View File

@@ -0,0 +1,850 @@
"""
Wix API Client for Blog Management
This module provides a comprehensive client for interacting with the Wix API
to manage blog posts, SEO settings, and media uploads.
Documentation: https://dev.wix.com/api/rest/getting-started
"""
import os
import json
import time
import logging
import requests
from typing import Dict, List, Optional, Union, Any, Tuple
from datetime import datetime
import mimetypes
from pathlib import Path
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('wix_api_client')
class WixAPIClient:
"""
Client for interacting with the Wix API for blog management.
This client handles authentication, blog post creation/updating,
media uploads, and SEO settings.
"""
# Base URLs for different Wix API endpoints
BASE_URL = "https://www.wixapis.com"
OAUTH_URL = "https://www.wix.com/oauth"
# API Endpoints
BLOG_API = "/blog/v3"
MEDIA_API = "/site-media/v1"
SEO_API = "/site-properties/v4/seo"
def __init__(
self,
api_key: Optional[str] = None,
refresh_token: Optional[str] = None,
site_id: Optional[str] = None
):
"""
Initialize the Wix API Client.
Args:
api_key: Wix API key (optional if using refresh token)
refresh_token: Wix refresh token for OAuth authentication
site_id: Wix site ID
"""
self.api_key = api_key or os.environ.get('WIX_API_KEY')
self.refresh_token = refresh_token or os.environ.get('WIX_REFRESH_TOKEN')
self.site_id = site_id or os.environ.get('WIX_SITE_ID')
self.access_token = None
self.token_expiry = 0
if not self.refresh_token:
logger.warning("No refresh token provided. Authentication will fail.")
if not self.site_id:
logger.warning("No site ID provided. API calls will fail.")
def _get_headers(self) -> Dict[str, str]:
"""
Get the headers required for API requests.
Returns:
Dict containing the necessary headers for Wix API requests
"""
# Ensure we have a valid access token
self._ensure_valid_token()
headers = {
"Authorization": f"Bearer {self.access_token}",
"wix-site-id": self.site_id,
"Content-Type": "application/json"
}
return headers
def _ensure_valid_token(self) -> None:
"""
Ensure we have a valid access token, refreshing if necessary.
"""
current_time = time.time()
# If token is expired or doesn't exist, refresh it
if not self.access_token or current_time >= self.token_expiry:
self._refresh_access_token()
def _refresh_access_token(self) -> None:
"""
Refresh the access token using the refresh token.
"""
if not self.refresh_token:
raise ValueError("Refresh token is required for authentication")
url = f"{self.OAUTH_URL}/access"
payload = {
"grant_type": "refresh_token",
"refresh_token": self.refresh_token,
"client_id": self.api_key if self.api_key else ""
}
try:
response = requests.post(url, json=payload)
response.raise_for_status()
data = response.json()
self.access_token = data.get("access_token")
# Set token expiry (subtract 5 minutes for safety margin)
expires_in = data.get("expires_in", 3600) # Default to 1 hour if not specified
self.token_expiry = time.time() + expires_in - 300
logger.info("Successfully refreshed access token")
except requests.exceptions.RequestException as e:
logger.error(f"Failed to refresh access token: {str(e)}")
if response.text:
logger.error(f"Response: {response.text}")
raise
def _make_request(
self,
method: str,
endpoint: str,
data: Optional[Dict] = None,
params: Optional[Dict] = None,
files: Optional[Dict] = None
) -> Dict:
"""
Make a request to the Wix API.
Args:
method: HTTP method (GET, POST, PUT, DELETE)
endpoint: API endpoint
data: Request payload
params: Query parameters
files: Files to upload
Returns:
Response data as dictionary
"""
url = f"{self.BASE_URL}{endpoint}"
headers = self._get_headers()
# If we're uploading files, remove the Content-Type header
if files:
headers.pop("Content-Type", None)
try:
response = requests.request(
method=method,
url=url,
headers=headers,
json=data,
params=params,
files=files
)
# Log request details for debugging
logger.debug(f"Request: {method} {url}")
logger.debug(f"Headers: {headers}")
if data:
logger.debug(f"Data: {json.dumps(data)}")
if params:
logger.debug(f"Params: {params}")
# Handle response
response.raise_for_status()
if response.content:
return response.json()
return {}
except requests.exceptions.HTTPError as e:
logger.error(f"HTTP error: {str(e)}")
if response.text:
logger.error(f"Response: {response.text}")
raise
except requests.exceptions.RequestException as e:
logger.error(f"Request error: {str(e)}")
raise
# Blog Post Management
def list_posts(
self,
limit: int = 50,
offset: int = 0,
sort_field: str = "lastPublishedDate",
sort_order: str = "desc",
filter_by: Optional[Dict] = None
) -> Dict:
"""
List blog posts with pagination and sorting.
Args:
limit: Maximum number of posts to return (default: 50)
offset: Pagination offset (default: 0)
sort_field: Field to sort by (default: lastPublishedDate)
sort_order: Sort order, 'asc' or 'desc' (default: desc)
filter_by: Optional filter criteria
Returns:
Dictionary containing blog posts and pagination info
"""
endpoint = f"{self.BLOG_API}/posts/query"
payload = {
"limit": limit,
"offset": offset,
"sort": [
{
"fieldName": sort_field,
"order": sort_order
}
]
}
if filter_by:
payload["filter"] = filter_by
return self._make_request("POST", endpoint, data=payload)
def get_post(self, post_id: str) -> Dict:
"""
Get a specific blog post by ID.
Args:
post_id: ID of the blog post
Returns:
Blog post data
"""
endpoint = f"{self.BLOG_API}/posts/{post_id}"
return self._make_request("GET", endpoint)
def create_post(
self,
title: str,
content: str,
excerpt: Optional[str] = None,
featured_image_id: Optional[str] = None,
tags: Optional[List[str]] = None,
categories: Optional[List[str]] = None,
seo_data: Optional[Dict] = None,
publish: bool = False
) -> Dict:
"""
Create a new blog post.
Args:
title: Post title
content: Post content (HTML)
excerpt: Post excerpt/summary
featured_image_id: ID of the featured image (from media manager)
tags: List of tags
categories: List of category IDs
seo_data: SEO settings for the post
publish: Whether to publish the post immediately
Returns:
Created blog post data
"""
endpoint = f"{self.BLOG_API}/posts"
# Prepare the post data
post_data = {
"post": {
"title": title,
"content": content,
"excerpt": excerpt or "",
"featured_image_id": featured_image_id,
"tags": tags or [],
"categoryIds": categories or []
}
}
# Add SEO data if provided
if seo_data:
post_data["post"]["seoData"] = seo_data
# Create the post
response = self._make_request("POST", endpoint, data=post_data)
# Publish the post if requested
if publish and response.get("post", {}).get("id"):
post_id = response["post"]["id"]
self.publish_post(post_id)
# Refresh the post data to get the published version
response = self.get_post(post_id)
return response
def update_post(
self,
post_id: str,
title: Optional[str] = None,
content: Optional[str] = None,
excerpt: Optional[str] = None,
featured_image_id: Optional[str] = None,
tags: Optional[List[str]] = None,
categories: Optional[List[str]] = None,
seo_data: Optional[Dict] = None,
publish: bool = False
) -> Dict:
"""
Update an existing blog post.
Args:
post_id: ID of the post to update
title: New post title (optional)
content: New post content (HTML) (optional)
excerpt: New post excerpt/summary (optional)
featured_image_id: New featured image ID (optional)
tags: New list of tags (optional)
categories: New list of category IDs (optional)
seo_data: New SEO settings (optional)
publish: Whether to publish the post after updating
Returns:
Updated blog post data
"""
# First, get the current post data
current_post = self.get_post(post_id)
if "post" not in current_post:
raise ValueError(f"Post with ID {post_id} not found")
current_post_data = current_post["post"]
# Update only the fields that were provided
update_data = {
"post": {
"id": post_id,
"title": title if title is not None else current_post_data.get("title", ""),
"content": content if content is not None else current_post_data.get("content", ""),
"excerpt": excerpt if excerpt is not None else current_post_data.get("excerpt", ""),
"featured_image_id": featured_image_id if featured_image_id is not None else current_post_data.get("featuredImageId"),
"tags": tags if tags is not None else current_post_data.get("tags", []),
"categoryIds": categories if categories is not None else current_post_data.get("categoryIds", [])
}
}
# Add SEO data if provided
if seo_data:
update_data["post"]["seoData"] = seo_data
elif "seoData" in current_post_data:
update_data["post"]["seoData"] = current_post_data["seoData"]
# Update the post
endpoint = f"{self.BLOG_API}/posts/{post_id}"
response = self._make_request("PATCH", endpoint, data=update_data)
# Publish the post if requested
if publish:
self.publish_post(post_id)
# Refresh the post data to get the published version
response = self.get_post(post_id)
return response
def delete_post(self, post_id: str) -> Dict:
"""
Delete a blog post.
Args:
post_id: ID of the post to delete
Returns:
Response data
"""
endpoint = f"{self.BLOG_API}/posts/{post_id}"
return self._make_request("DELETE", endpoint)
def publish_post(self, post_id: str) -> Dict:
"""
Publish a draft blog post.
Args:
post_id: ID of the post to publish
Returns:
Published post data
"""
endpoint = f"{self.BLOG_API}/posts/{post_id}/publish"
return self._make_request("POST", endpoint)
def unpublish_post(self, post_id: str) -> Dict:
"""
Unpublish a published blog post (revert to draft).
Args:
post_id: ID of the post to unpublish
Returns:
Unpublished post data
"""
endpoint = f"{self.BLOG_API}/posts/{post_id}/unpublish"
return self._make_request("POST", endpoint)
# Category Management
def list_categories(self) -> Dict:
"""
List all blog categories.
Returns:
Dictionary containing blog categories
"""
endpoint = f"{self.BLOG_API}/categories"
return self._make_request("GET", endpoint)
def create_category(self, label: str, description: Optional[str] = None) -> Dict:
"""
Create a new blog category.
Args:
label: Category name
description: Category description (optional)
Returns:
Created category data
"""
endpoint = f"{self.BLOG_API}/categories"
payload = {
"category": {
"label": label,
"description": description or ""
}
}
return self._make_request("POST", endpoint, data=payload)
def update_category(
self,
category_id: str,
label: Optional[str] = None,
description: Optional[str] = None
) -> Dict:
"""
Update an existing blog category.
Args:
category_id: ID of the category to update
label: New category name (optional)
description: New category description (optional)
Returns:
Updated category data
"""
# First, get the current category data
current_categories = self.list_categories()
current_category = None
for category in current_categories.get("categories", []):
if category.get("id") == category_id:
current_category = category
break
if not current_category:
raise ValueError(f"Category with ID {category_id} not found")
# Update only the fields that were provided
update_data = {
"category": {
"id": category_id,
"label": label if label is not None else current_category.get("label", ""),
"description": description if description is not None else current_category.get("description", "")
}
}
endpoint = f"{self.BLOG_API}/categories/{category_id}"
return self._make_request("PATCH", endpoint, data=update_data)
def delete_category(self, category_id: str) -> Dict:
"""
Delete a blog category.
Args:
category_id: ID of the category to delete
Returns:
Response data
"""
endpoint = f"{self.BLOG_API}/categories/{category_id}"
return self._make_request("DELETE", endpoint)
# Media Management
def upload_image(
self,
file_path: str,
title: Optional[str] = None,
alt_text: Optional[str] = None,
description: Optional[str] = None
) -> Dict:
"""
Upload an image to the Wix media manager.
Args:
file_path: Path to the image file
title: Image title (optional)
alt_text: Image alt text for accessibility (optional)
description: Image description (optional)
Returns:
Uploaded image data
"""
# Check if file exists
if not os.path.isfile(file_path):
raise FileNotFoundError(f"File not found: {file_path}")
# Get file name and mime type
file_name = os.path.basename(file_path)
mime_type, _ = mimetypes.guess_type(file_path)
if not mime_type or not mime_type.startswith('image/'):
raise ValueError(f"File does not appear to be an image: {file_path}")
# Prepare metadata
metadata = {
"title": title or file_name,
"altText": alt_text or "",
"description": description or ""
}
# First, get an upload URL
endpoint = f"{self.MEDIA_API}/files/upload/url"
upload_url_response = self._make_request("POST", endpoint, data={
"mimeType": mime_type,
"fileName": file_name
})
if "uploadUrl" not in upload_url_response:
raise ValueError("Failed to get upload URL")
upload_url = upload_url_response["uploadUrl"]
# Upload the file to the provided URL
with open(file_path, 'rb') as file:
upload_response = requests.post(
upload_url,
files={'file': (file_name, file, mime_type)},
headers={"Content-Type": mime_type}
)
upload_response.raise_for_status()
# Complete the upload with metadata
endpoint = f"{self.MEDIA_API}/files"
complete_data = {
"uploadToken": upload_url_response.get("uploadToken"),
"mediaOptions": {
"mimeType": mime_type,
"fileName": file_name,
"mediaType": "IMAGE",
"title": metadata["title"],
"description": metadata["description"],
"alt": metadata["altText"]
}
}
return self._make_request("POST", endpoint, data=complete_data)
def get_media_item(self, media_id: str) -> Dict:
"""
Get details of a specific media item.
Args:
media_id: ID of the media item
Returns:
Media item data
"""
endpoint = f"{self.MEDIA_API}/files/{media_id}"
return self._make_request("GET", endpoint)
def list_media_items(
self,
media_type: str = "IMAGE",
limit: int = 50,
offset: int = 0
) -> Dict:
"""
List media items with pagination.
Args:
media_type: Type of media to list (IMAGE, VIDEO, AUDIO, DOCUMENT)
limit: Maximum number of items to return
offset: Pagination offset
Returns:
Dictionary containing media items and pagination info
"""
endpoint = f"{self.MEDIA_API}/files/query"
payload = {
"query": {
"paging": {
"limit": limit,
"offset": offset
},
"filter": {
"mediaType": media_type
}
}
}
return self._make_request("POST", endpoint, data=payload)
def delete_media_item(self, media_id: str) -> Dict:
"""
Delete a media item.
Args:
media_id: ID of the media item to delete
Returns:
Response data
"""
endpoint = f"{self.MEDIA_API}/files/{media_id}"
return self._make_request("DELETE", endpoint)
# SEO Management
def get_seo_settings(self, page_url: str) -> Dict:
"""
Get SEO settings for a specific page.
Args:
page_url: URL path of the page (e.g., "/blog/my-post")
Returns:
SEO settings data
"""
endpoint = f"{self.SEO_API}/sites/{self.site_id}/url/{page_url}"
return self._make_request("GET", endpoint)
def update_seo_settings(
self,
page_url: str,
title: Optional[str] = None,
description: Optional[str] = None,
keywords: Optional[List[str]] = None,
og_image_url: Optional[str] = None,
structured_data: Optional[Dict] = None,
no_index: Optional[bool] = None
) -> Dict:
"""
Update SEO settings for a specific page.
Args:
page_url: URL path of the page (e.g., "/blog/my-post")
title: SEO title
description: SEO description
keywords: SEO keywords
og_image_url: Open Graph image URL
structured_data: Structured data (JSON-LD)
no_index: Whether to prevent indexing by search engines
Returns:
Updated SEO settings data
"""
# First, get current SEO settings
try:
current_settings = self.get_seo_settings(page_url)
except:
# If the page doesn't exist yet, start with empty settings
current_settings = {"tags": {}}
# Prepare the update data
seo_data = {
"tags": {}
}
# Update only the fields that were provided
if title is not None:
seo_data["tags"]["title"] = title
elif "title" in current_settings.get("tags", {}):
seo_data["tags"]["title"] = current_settings["tags"]["title"]
if description is not None:
seo_data["tags"]["description"] = description
elif "description" in current_settings.get("tags", {}):
seo_data["tags"]["description"] = current_settings["tags"]["description"]
if keywords is not None:
seo_data["tags"]["keywords"] = ", ".join(keywords)
elif "keywords" in current_settings.get("tags", {}):
seo_data["tags"]["keywords"] = current_settings["tags"]["keywords"]
if og_image_url is not None:
seo_data["tags"]["og:image"] = og_image_url
elif "og:image" in current_settings.get("tags", {}):
seo_data["tags"]["og:image"] = current_settings["tags"]["og:image"]
if structured_data is not None:
seo_data["tags"]["jsonld"] = json.dumps(structured_data)
elif "jsonld" in current_settings.get("tags", {}):
seo_data["tags"]["jsonld"] = current_settings["tags"]["jsonld"]
if no_index is not None:
seo_data["tags"]["robots"] = "noindex" if no_index else "index"
elif "robots" in current_settings.get("tags", {}):
seo_data["tags"]["robots"] = current_settings["tags"]["robots"]
endpoint = f"{self.SEO_API}/sites/{self.site_id}/url/{page_url}"
return self._make_request("PUT", endpoint, data=seo_data)
# Helper Methods
def create_blog_post_with_image(
self,
title: str,
content: str,
image_path: Optional[str] = None,
excerpt: Optional[str] = None,
tags: Optional[List[str]] = None,
categories: Optional[List[str]] = None,
seo_title: Optional[str] = None,
seo_description: Optional[str] = None,
seo_keywords: Optional[List[str]] = None,
publish: bool = False
) -> Dict:
"""
Create a blog post with an optional featured image in one operation.
Args:
title: Post title
content: Post content (HTML)
image_path: Path to featured image (optional)
excerpt: Post excerpt/summary (optional)
tags: List of tags (optional)
categories: List of category IDs (optional)
seo_title: SEO title (optional)
seo_description: SEO description (optional)
seo_keywords: SEO keywords (optional)
publish: Whether to publish the post immediately (optional)
Returns:
Created blog post data
"""
# Upload image if provided
featured_image_id = None
if image_path and os.path.isfile(image_path):
try:
image_response = self.upload_image(
file_path=image_path,
title=title,
alt_text=title
)
featured_image_id = image_response.get("file", {}).get("id")
logger.info(f"Uploaded image with ID: {featured_image_id}")
except Exception as e:
logger.error(f"Failed to upload image: {str(e)}")
# Prepare SEO data
seo_data = None
if seo_title or seo_description or seo_keywords:
seo_data = {
"title": seo_title or title,
"description": seo_description or excerpt or "",
"keywords": seo_keywords or tags or []
}
# Create the blog post
return self.create_post(
title=title,
content=content,
excerpt=excerpt,
featured_image_id=featured_image_id,
tags=tags,
categories=categories,
seo_data=seo_data,
publish=publish
)
def get_or_create_category(self, category_name: str) -> str:
"""
Get a category ID by name, creating it if it doesn't exist.
Args:
category_name: Name of the category
Returns:
Category ID
"""
# List all categories
categories_response = self.list_categories()
categories = categories_response.get("categories", [])
# Check if category exists
for category in categories:
if category.get("label", "").lower() == category_name.lower():
return category.get("id")
# Create category if it doesn't exist
create_response = self.create_category(label=category_name)
return create_response.get("category", {}).get("id")
def get_post_by_slug(self, slug: str) -> Optional[Dict]:
"""
Find a post by its slug.
Args:
slug: Post slug
Returns:
Post data or None if not found
"""
# List posts with a filter for the slug
filter_by = {
"slug": {
"$eq": slug
}
}
response = self.list_posts(limit=1, filter_by=filter_by)
posts = response.get("posts", [])
if posts:
return posts[0]
return None
def get_post_url(self, post_id: str) -> str:
"""
Get the full URL for a blog post.
Args:
post_id: ID of the blog post
Returns:
Full URL to the blog post
"""
post_data = self.get_post(post_id)
slug = post_data.get("post", {}).get("slug", "")
# Get the blog URL prefix
# This is a simplification - in reality, you might need to get this from site settings
return f"/blog/{slug}"

720
lib/integrations/wix/wix_blog_manager.py Normal file
View File

@@ -0,0 +1,720 @@
"""
Wix Blog Manager
This module provides high-level functions for managing blog content on Wix,
including content creation, SEO optimization, and media management.
"""
import os
import re
import logging
import tempfile
import requests
from typing import Dict, List, Optional, Union, Any, Tuple
from datetime import datetime
from pathlib import Path
import markdown
import html2text
from bs4 import BeautifulSoup
from .wix_api_client import WixAPIClient
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('wix_blog_manager')
class WixBlogManager:
"""
High-level manager for Wix blog content.
This class provides convenient methods for common blog management tasks,
building on the lower-level WixAPIClient.
"""
def __init__(
self,
api_key: Optional[str] = None,
refresh_token: Optional[str] = None,
site_id: Optional[str] = None
):
"""
Initialize the Wix Blog Manager.
Args:
api_key: Wix API key (optional if using refresh token)
refresh_token: Wix refresh token for OAuth authentication
site_id: Wix site ID
"""
self.client = WixAPIClient(api_key, refresh_token, site_id)
def publish_markdown_post(
self,
title: str,
markdown_content: str,
featured_image_path: Optional[str] = None,
featured_image_url: Optional[str] = None,
excerpt: Optional[str] = None,
tags: Optional[List[str]] = None,
categories: Optional[List[str]] = None,
seo_title: Optional[str] = None,
seo_description: Optional[str] = None,
seo_keywords: Optional[List[str]] = None,
publish: bool = False
) -> Dict:
"""
Publish a blog post from markdown content.
Args:
title: Post title
markdown_content: Post content in markdown format
featured_image_path: Local path to featured image (optional)
featured_image_url: URL of featured image to download (optional)
excerpt: Post excerpt/summary (optional)
tags: List of tags (optional)
categories: List of category names (optional)
seo_title: SEO title (optional)
seo_description: SEO description (optional)
seo_keywords: SEO keywords (optional)
publish: Whether to publish the post immediately (optional)
Returns:
Published blog post data
"""
# Convert markdown to HTML
html_content = self._markdown_to_html(markdown_content)
# Process images in the content
html_content, embedded_images = self._process_content_images(html_content)
# Handle featured image
featured_image_id = None
temp_image_path = None
if featured_image_url and not featured_image_path:
# Download the image from URL
try:
temp_image_path = self._download_image(featured_image_url)
featured_image_path = temp_image_path
except Exception as e:
logger.error(f"Failed to download featured image: {str(e)}")
if featured_image_path:
try:
image_response = self.client.upload_image(
file_path=featured_image_path,
title=title,
alt_text=title
)
featured_image_id = image_response.get("file", {}).get("id")
logger.info(f"Uploaded featured image with ID: {featured_image_id}")
except Exception as e:
logger.error(f"Failed to upload featured image: {str(e)}")
# Clean up temporary file if created
if temp_image_path and os.path.exists(temp_image_path):
try:
os.remove(temp_image_path)
except:
pass
# Process categories - convert names to IDs
category_ids = []
if categories:
for category_name in categories:
try:
category_id = self.client.get_or_create_category(category_name)
if category_id:
category_ids.append(category_id)
except Exception as e:
logger.error(f"Failed to process category '{category_name}': {str(e)}")
# Generate excerpt if not provided
if not excerpt:
excerpt = self._generate_excerpt(markdown_content)
# Prepare SEO data
seo_data = None
if seo_title or seo_description or seo_keywords:
seo_data = {
"title": seo_title or title,
"description": seo_description or excerpt or "",
"keywords": seo_keywords or tags or []
}
# Create the blog post
response = self.client.create_post(
title=title,
content=html_content,
excerpt=excerpt,
featured_image_id=featured_image_id,
tags=tags,
categories=category_ids,
seo_data=seo_data,
publish=publish
)
# Update SEO settings if the post was published
if publish and response.get("post", {}).get("id"):
post_id = response["post"]["id"]
post_url = self.client.get_post_url(post_id)
try:
self.client.update_seo_settings(
page_url=post_url,
title=seo_title or title,
description=seo_description or excerpt or "",
keywords=seo_keywords or tags,
og_image_url=featured_image_url
)
except Exception as e:
logger.error(f"Failed to update SEO settings: {str(e)}")
return response
def update_markdown_post(
self,
post_id: str,
title: Optional[str] = None,
markdown_content: Optional[str] = None,
featured_image_path: Optional[str] = None,
featured_image_url: Optional[str] = None,
excerpt: Optional[str] = None,
tags: Optional[List[str]] = None,
categories: Optional[List[str]] = None,
seo_title: Optional[str] = None,
seo_description: Optional[str] = None,
seo_keywords: Optional[List[str]] = None,
publish: bool = False
) -> Dict:
"""
Update an existing blog post with markdown content.
Args:
post_id: ID of the post to update
title: New post title (optional)
markdown_content: New post content in markdown format (optional)
featured_image_path: Local path to new featured image (optional)
featured_image_url: URL of new featured image to download (optional)
excerpt: New post excerpt/summary (optional)
tags: New list of tags (optional)
categories: New list of category names (optional)
seo_title: New SEO title (optional)
seo_description: New SEO description (optional)
seo_keywords: New SEO keywords (optional)
publish: Whether to publish the post after updating (optional)
Returns:
Updated blog post data
"""
# Get current post data
current_post = self.client.get_post(post_id)
if "post" not in current_post:
raise ValueError(f"Post with ID {post_id} not found")
# Convert markdown to HTML if provided
html_content = None
if markdown_content:
html_content = self._markdown_to_html(markdown_content)
# Process images in the content
html_content, embedded_images = self._process_content_images(html_content)
# Handle featured image
featured_image_id = None
temp_image_path = None
if featured_image_url and not featured_image_path:
# Download the image from URL
try:
temp_image_path = self._download_image(featured_image_url)
featured_image_path = temp_image_path
except Exception as e:
logger.error(f"Failed to download featured image: {str(e)}")
if featured_image_path:
try:
image_response = self.client.upload_image(
file_path=featured_image_path,
title=title or current_post["post"].get("title", ""),
alt_text=title or current_post["post"].get("title", "")
)
featured_image_id = image_response.get("file", {}).get("id")
logger.info(f"Uploaded featured image with ID: {featured_image_id}")
except Exception as e:
logger.error(f"Failed to upload featured image: {str(e)}")
# Clean up temporary file if created
if temp_image_path and os.path.exists(temp_image_path):
try:
os.remove(temp_image_path)
except:
pass
# Process categories - convert names to IDs
category_ids = None
if categories:
category_ids = []
for category_name in categories:
try:
category_id = self.client.get_or_create_category(category_name)
if category_id:
category_ids.append(category_id)
except Exception as e:
logger.error(f"Failed to process category '{category_name}': {str(e)}")
# Generate excerpt if not provided but markdown is
if not excerpt and markdown_content:
excerpt = self._generate_excerpt(markdown_content)
# Prepare SEO data
seo_data = None
if seo_title or seo_description or seo_keywords:
seo_data = {
"title": seo_title or title or current_post["post"].get("title", ""),
"description": seo_description or excerpt or current_post["post"].get("excerpt", ""),
"keywords": seo_keywords or tags or current_post["post"].get("tags", [])
}
# Update the blog post
response = self.client.update_post(
post_id=post_id,
title=title,
content=html_content,
excerpt=excerpt,
featured_image_id=featured_image_id,
tags=tags,
categories=category_ids,
seo_data=seo_data,
publish=publish
)
# Update SEO settings if needed
if (seo_title or seo_description or seo_keywords or featured_image_url):
post_url = self.client.get_post_url(post_id)
try:
self.client.update_seo_settings(
page_url=post_url,
title=seo_title or title,
description=seo_description or excerpt,
keywords=seo_keywords or tags,
og_image_url=featured_image_url
)
except Exception as e:
logger.error(f"Failed to update SEO settings: {str(e)}")
return response
def find_post_by_title(self, title: str) -> Optional[Dict]:
"""
Find a post by its title (exact match).
Args:
title: Post title to search for
Returns:
Post data or None if not found
"""
# List all posts (this is inefficient but Wix API doesn't support filtering by title)
# In a production environment, you might want to implement pagination
response = self.client.list_posts(limit=100)
posts = response.get("posts", [])
for post in posts:
if post.get("title") == title:
return post
return None
def publish_or_update_markdown_post(
self,
title: str,
markdown_content: str,
featured_image_path: Optional[str] = None,
featured_image_url: Optional[str] = None,
excerpt: Optional[str] = None,
tags: Optional[List[str]] = None,
categories: Optional[List[str]] = None,
seo_title: Optional[str] = None,
seo_description: Optional[str] = None,
seo_keywords: Optional[List[str]] = None,
publish: bool = False,
update_if_exists: bool = True
) -> Dict:
"""
Publish a new post or update an existing one with the same title.
Args:
title: Post title
markdown_content: Post content in markdown format
featured_image_path: Local path to featured image (optional)
featured_image_url: URL of featured image to download (optional)
excerpt: Post excerpt/summary (optional)
tags: List of tags (optional)
categories: List of category names (optional)
seo_title: SEO title (optional)
seo_description: SEO description (optional)
seo_keywords: SEO keywords (optional)
publish: Whether to publish the post immediately (optional)
update_if_exists: Whether to update an existing post with the same title (optional)
Returns:
Published or updated blog post data
"""
# Check if a post with this title already exists
existing_post = self.find_post_by_title(title)
if existing_post and update_if_exists:
# Update existing post
logger.info(f"Updating existing post with title: {title}")
return self.update_markdown_post(
post_id=existing_post["id"],
title=title,
markdown_content=markdown_content,
featured_image_path=featured_image_path,
featured_image_url=featured_image_url,
excerpt=excerpt,
tags=tags,
categories=categories,
seo_title=seo_title,
seo_description=seo_description,
seo_keywords=seo_keywords,
publish=publish
)
else:
# Create new post
logger.info(f"Creating new post with title: {title}")
return self.publish_markdown_post(
title=title,
markdown_content=markdown_content,
featured_image_path=featured_image_path,
featured_image_url=featured_image_url,
excerpt=excerpt,
tags=tags,
categories=categories,
seo_title=seo_title,
seo_description=seo_description,
seo_keywords=seo_keywords,
publish=publish
)
def optimize_seo_for_post(
self,
post_id: str,
seo_title: Optional[str] = None,
seo_description: Optional[str] = None,
seo_keywords: Optional[List[str]] = None,
og_image_url: Optional[str] = None,
structured_data: Optional[Dict] = None
) -> Dict:
"""
Optimize SEO settings for an existing blog post.
Args:
post_id: ID of the blog post
seo_title: SEO title (optional)
seo_description: SEO description (optional)
seo_keywords: SEO keywords (optional)
og_image_url: Open Graph image URL (optional)
structured_data: Structured data (JSON-LD) (optional)
Returns:
Updated SEO settings data
"""
# Get the post URL
post_url = self.client.get_post_url(post_id)
# Update SEO settings
return self.client.update_seo_settings(
page_url=post_url,
title=seo_title,
description=seo_description,
keywords=seo_keywords,
og_image_url=og_image_url,
structured_data=structured_data
)
def generate_structured_data(
self,
post_id: str,
author_name: str,
publisher_name: str,
publisher_logo_url: str
) -> Dict:
"""
Generate structured data (JSON-LD) for a blog post.
Args:
post_id: ID of the blog post
author_name: Name of the author
publisher_name: Name of the publisher
publisher_logo_url: URL of the publisher's logo
Returns:
Structured data as a dictionary
"""
# Get post data
post_data = self.client.get_post(post_id)
post = post_data.get("post", {})
# Get post URL
post_url = self.client.get_post_url(post_id)
# Create structured data
structured_data = {
"@context": "https://schema.org",
"@type": "BlogPosting",
"headline": post.get("title", ""),
"description": post.get("excerpt", ""),
"author": {
"@type": "Person",
"name": author_name
},
"publisher": {
"@type": "Organization",
"name": publisher_name,
"logo": {
"@type": "ImageObject",
"url": publisher_logo_url
}
},
"datePublished": post.get("publishedDate", ""),
"dateModified": post.get("lastPublishedDate", "")
}
# Add featured image if available
if post.get("featuredImageId"):
try:
media_item = self.client.get_media_item(post["featuredImageId"])
image_url = media_item.get("file", {}).get("url", "")
if image_url:
structured_data["image"] = image_url
except:
pass
return structured_data
def apply_structured_data_to_post(
self,
post_id: str,
author_name: str,
publisher_name: str,
publisher_logo_url: str
) -> Dict:
"""
Generate and apply structured data to a blog post.
Args:
post_id: ID of the blog post
author_name: Name of the author
publisher_name: Name of the publisher
publisher_logo_url: URL of the publisher's logo
Returns:
Updated SEO settings data
"""
# Generate structured data
structured_data = self.generate_structured_data(
post_id=post_id,
author_name=author_name,
publisher_name=publisher_name,
publisher_logo_url=publisher_logo_url
)
# Get the post URL
post_url = self.client.get_post_url(post_id)
# Update SEO settings with structured data
return self.client.update_seo_settings(
page_url=post_url,
structured_data=structured_data
)
# Helper methods
def _markdown_to_html(self, markdown_content: str) -> str:
"""
Convert markdown content to HTML.
Args:
markdown_content: Content in markdown format
Returns:
HTML content
"""
# Use the markdown library to convert to HTML
html = markdown.markdown(
markdown_content,
extensions=['extra', 'codehilite', 'tables', 'toc']
)
return html
def _html_to_markdown(self, html_content: str) -> str:
"""
Convert HTML content to markdown.
Args:
html_content: Content in HTML format
Returns:
Markdown content
"""
# Use html2text to convert HTML to markdown
h = html2text.HTML2Text()
h.ignore_links = False
h.ignore_images = False
h.ignore_tables = False
h.ignore_emphasis = False
return h.handle(html_content)
def _process_content_images(self, html_content: str) -> Tuple[str, List[Dict]]:
"""
Process images in HTML content, uploading them to Wix and replacing URLs.
Args:
html_content: HTML content with image tags
Returns:
Tuple of (updated HTML content, list of uploaded image data)
"""
soup = BeautifulSoup(html_content, 'html.parser')
img_tags = soup.find_all('img')
uploaded_images = []
for img in img_tags:
src = img.get('src', '')
alt = img.get('alt', '')
# Skip images that are already hosted on Wix
if 'wixstatic.com' in src:
continue
# Handle images with data URLs
if src.startswith('data:image'):
logger.info("Skipping data URL image - not supported in this implementation")
continue
# Handle remote images
if src.startswith('http://') or src.startswith('https://'):
try:
# Download the image
temp_path = self._download_image(src)
# Upload to Wix
image_response = self.client.upload_image(
file_path=temp_path,
title=alt or "Blog image",
alt_text=alt or "Blog image"
)
# Get the new URL
new_url = image_response.get("file", {}).get("url", "")
if new_url:
# Replace the src attribute
img['src'] = new_url
uploaded_images.append({
'original_url': src,
'wix_url': new_url,
'wix_id': image_response.get("file", {}).get("id", "")
})
# Clean up temp file
if os.path.exists(temp_path):
os.remove(temp_path)
except Exception as e:
logger.error(f"Failed to process image {src}: {str(e)}")
# Handle local images (not implemented in this version)
else:
logger.info(f"Skipping local image {src} - not supported in this implementation")
# Return the updated HTML
return str(soup), uploaded_images
def _download_image(self, url: str) -> str:
"""
Download an image from a URL to a temporary file.
Args:
url: URL of the image
Returns:
Path to the downloaded temporary file
"""
response = requests.get(url, stream=True)
response.raise_for_status()
# Determine file extension
content_type = response.headers.get('content-type', '')
extension = '.jpg' # Default
if 'image/jpeg' in content_type:
extension = '.jpg'
elif 'image/png' in content_type:
extension = '.png'
elif 'image/gif' in content_type:
extension = '.gif'
elif 'image/webp' in content_type:
extension = '.webp'
# Create a temporary file
fd, temp_path = tempfile.mkstemp(suffix=extension)
os.close(fd)
# Write the image data to the file
with open(temp_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
return temp_path
def _generate_excerpt(self, markdown_content: str, max_length: int = 160) -> str:
"""
Generate an excerpt from markdown content.
Args:
markdown_content: Content in markdown format
max_length: Maximum length of the excerpt
Returns:
Generated excerpt
"""
# Convert markdown to plain text
h = html2text.HTML2Text()
h.ignore_links = True
h.ignore_images = True
h.ignore_tables = True
h.ignore_emphasis = True
# First convert markdown to HTML, then HTML to plain text
html = markdown.markdown(markdown_content)
plain_text = h.handle(html)
# Clean up the text
plain_text = re.sub(r'\s+', ' ', plain_text).strip()
# Truncate to max_length
if len(plain_text) <= max_length:
return plain_text
# Try to truncate at a sentence boundary
sentences = re.split(r'(?<=[.!?])\s+', plain_text)
excerpt = ""
for sentence in sentences:
if len(excerpt + sentence) <= max_length:
excerpt += sentence + " "
else:
break
# If we couldn't get a full sentence, just truncate
if not excerpt:
excerpt = plain_text[:max_length-3] + "..."
return excerpt.strip()

350
lib/integrations/wix/wix_blog_publisher.py Normal file
View File

@@ -0,0 +1,350 @@
"""
Wix Blog Publisher for Alwrity
This module integrates the Wix API with the Alwrity AI Writer platform,
allowing users to publish generated blog content directly to their Wix site.
"""
import os
import logging
import tempfile
import streamlit as st
from typing import Dict, List, Optional, Union, Any, Tuple
from pathlib import Path
from .wix_integration import WixIntegration
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('wix_blog_publisher')
def publish_to_wix(
title: str,
content: str,
is_markdown: bool = True,
featured_image_path: Optional[str] = None,
featured_image_url: Optional[str] = None,
excerpt: Optional[str] = None,
tags: Optional[List[str]] = None,
categories: Optional[List[str]] = None,
seo_title: Optional[str] = None,
seo_description: Optional[str] = None,
seo_keywords: Optional[List[str]] = None,
author_name: Optional[str] = None,
publisher_name: Optional[str] = None,
publisher_logo_url: Optional[str] = None,
publish: bool = True,
update_if_exists: bool = True,
api_key: Optional[str] = None,
refresh_token: Optional[str] = None,
site_id: Optional[str] = None
) -> Dict:
"""
Publish a blog post to Wix.
Args:
title: Post title
content: Post content (markdown or HTML)
is_markdown: Whether the content is in markdown format
featured_image_path: Local path to featured image (optional)
featured_image_url: URL of featured image to download (optional)
excerpt: Post excerpt/summary (optional)
tags: List of tags (optional)
categories: List of category names (optional)
seo_title: SEO title (optional)
seo_description: SEO description (optional)
seo_keywords: SEO keywords (optional)
author_name: Name of the author (optional)
publisher_name: Name of the publisher (optional)
publisher_logo_url: URL of the publisher's logo (optional)
publish: Whether to publish the post immediately (optional)
update_if_exists: Whether to update an existing post with the same title (optional)
api_key: Wix API key (optional if using refresh token)
refresh_token: Wix refresh token for OAuth authentication
site_id: Wix site ID
Returns:
Published blog post data
"""
# Initialize Wix integration
wix = WixIntegration(api_key, refresh_token, site_id)
# Publish the blog post
return wix.publish_blog_post(
title=title,
content=content,
is_markdown=is_markdown,
featured_image_path=featured_image_path,
featured_image_url=featured_image_url,
excerpt=excerpt,
tags=tags,
categories=categories,
seo_title=seo_title,
seo_description=seo_description,
seo_keywords=seo_keywords,
author_name=author_name,
publisher_name=publisher_name,
publisher_logo_url=publisher_logo_url,
publish=publish,
update_if_exists=update_if_exists
)
def wix_blog_publisher_ui():
"""
Streamlit UI for publishing blog posts to Wix.
"""
st.title("Publish to Wix")
st.write("Publish your blog content directly to your Wix site.")
# Authentication settings
st.header("Wix Authentication")
# Check for saved credentials
if "wix_refresh_token" in st.session_state and "wix_site_id" in st.session_state:
st.success("✅ Wix credentials are saved in this session.")
show_saved = st.checkbox("Show saved credentials")
if show_saved:
st.text_input("Refresh Token", value=st.session_state.wix_refresh_token, type="password", disabled=True)
st.text_input("Site ID", value=st.session_state.wix_site_id, disabled=True)
clear_creds = st.button("Clear saved credentials")
if clear_creds:
if "wix_refresh_token" in st.session_state:
del st.session_state.wix_refresh_token
if "wix_site_id" in st.session_state:
del st.session_state.wix_site_id
st.rerun()
else:
col1, col2 = st.columns(2)
with col1:
refresh_token = st.text_input("Wix Refresh Token", type="password", help="Your Wix refresh token for API authentication")
with col2:
site_id = st.text_input("Wix Site ID", help="Your Wix site ID")
save_creds = st.checkbox("Save credentials for this session", value=True)
if st.button("Validate Credentials"):
if not refresh_token:
st.error("Refresh token is required.")
return
if not site_id:
st.error("Site ID is required.")
return
# Try to initialize Wix integration to validate credentials
try:
wix = WixIntegration(refresh_token=refresh_token, site_id=site_id)
# Test API call
site_info = wix.get_site_info()
if site_info.get("status") == "connected":
st.success(f"✅ Credentials validated successfully! Found {site_info.get('post_count', 0)} posts and {site_info.get('category_count', 0)} categories.")
# Save credentials if requested
if save_creds:
st.session_state.wix_refresh_token = refresh_token
st.session_state.wix_site_id = site_id
st.rerun()
else:
st.error(f"❌ Failed to validate credentials: {site_info.get('error', 'Unknown error')}")
except Exception as e:
st.error(f"❌ Failed to validate credentials: {str(e)}")
return
# Blog content section
st.header("Blog Content")
# Check if we have content in session state (from other parts of the app)
blog_title = st.text_input(
"Blog Title",
value=st.session_state.get("blog_title", ""),
help="The title of your blog post"
)
content_type = st.radio(
"Content Format",
["Markdown", "HTML"],
horizontal=True,
help="The format of your blog content"
)
is_markdown = content_type == "Markdown"
blog_content = st.text_area(
"Blog Content",
value=st.session_state.get("blog_content", ""),
height=300,
help="The content of your blog post"
)
# Featured image
st.subheader("Featured Image")
image_source = st.radio(
"Image Source",
["None", "Upload", "URL"],
horizontal=True,
help="How to provide the featured image"
)
featured_image_path = None
featured_image_url = None
if image_source == "Upload":
uploaded_file = st.file_uploader("Upload Featured Image", type=["jpg", "jpeg", "png", "gif"])
if uploaded_file:
# Save the uploaded file to a temporary location
with tempfile.NamedTemporaryFile(delete=False, suffix=f".{uploaded_file.name.split('.')[-1]}") as tmp:
tmp.write(uploaded_file.getvalue())
featured_image_path = tmp.name
elif image_source == "URL":
featured_image_url = st.text_input("Featured Image URL", help="URL of the featured image")
# Blog metadata
st.header("Blog Metadata")
col1, col2 = st.columns(2)
with col1:
excerpt = st.text_area(
"Excerpt",
value=st.session_state.get("blog_excerpt", ""),
help="A short summary of your blog post"
)
tags_input = st.text_input(
"Tags (comma-separated)",
value=", ".join(st.session_state.get("blog_tags", [])) if isinstance(st.session_state.get("blog_tags", []), list) else st.session_state.get("blog_tags", ""),
help="Tags for your blog post, separated by commas"
)
tags = [tag.strip() for tag in tags_input.split(",")] if tags_input else None
categories_input = st.text_input(
"Categories (comma-separated)",
value=", ".join(st.session_state.get("blog_categories", [])) if isinstance(st.session_state.get("blog_categories", []), list) else st.session_state.get("blog_categories", ""),
help="Categories for your blog post, separated by commas"
)
categories = [cat.strip() for cat in categories_input.split(",")] if categories_input else None
with col2:
author_name = st.text_input("Author Name", help="Name of the blog post author")
publisher_name = st.text_input("Publisher Name", help="Name of the blog publisher (usually your site name)")
publisher_logo_url = st.text_input("Publisher Logo URL", help="URL of the publisher's logo")
# SEO settings
with st.expander("SEO Settings"):
seo_title = st.text_input("SEO Title", value=blog_title, help="Title for search engines (defaults to blog title)")
seo_description = st.text_area("SEO Description", value=excerpt, help="Description for search engines (defaults to excerpt)")
seo_keywords_input = st.text_input("SEO Keywords (comma-separated)", value=tags_input, help="Keywords for search engines (defaults to tags)")
seo_keywords = [kw.strip() for kw in seo_keywords_input.split(",")] if seo_keywords_input else None
# Publishing options
st.header("Publishing Options")
col1, col2 = st.columns(2)
with col1:
publish = not st.checkbox("Save as draft", help="If checked, the post will be saved as a draft instead of being published")
with col2:
update_if_exists = st.checkbox("Update if exists", value=True, help="If checked, an existing post with the same title will be updated")
# Publish button
if st.button("Publish to Wix", type="primary"):
if not blog_title:
st.error("Blog title is required.")
return
if not blog_content:
st.error("Blog content is required.")
return
# Get credentials
refresh_token = st.session_state.get("wix_refresh_token")
site_id = st.session_state.get("wix_site_id")
if not refresh_token or not site_id:
st.error("Wix credentials are required. Please enter them in the authentication section.")
return
# Show progress
with st.spinner("Publishing to Wix..."):
try:
# Publish to Wix
result = publish_to_wix(
title=blog_title,
content=blog_content,
is_markdown=is_markdown,
featured_image_path=featured_image_path,
featured_image_url=featured_image_url,
excerpt=excerpt,
tags=tags,
categories=categories,
seo_title=seo_title,
seo_description=seo_description,
seo_keywords=seo_keywords,
author_name=author_name,
publisher_name=publisher_name,
publisher_logo_url=publisher_logo_url,
publish=publish,
update_if_exists=update_if_exists,
refresh_token=refresh_token,
site_id=site_id
)
# Clean up temporary file if created
if featured_image_path and os.path.exists(featured_image_path) and featured_image_path.startswith(tempfile.gettempdir()):
try:
os.remove(featured_image_path)
except:
pass
# Show success message
st.success("✅ Blog post published successfully!")
# Show post details
post = result.get("post", {})
st.subheader("Published Post Details")
col1, col2 = st.columns(2)
with col1:
st.write(f"**Title:** {post.get('title', 'N/A')}")
st.write(f"**Status:** {post.get('status', 'N/A')}")
st.write(f"**ID:** {post.get('id', 'N/A')}")
with col2:
st.write(f"**Published Date:** {post.get('publishedDate', 'N/A')}")
st.write(f"**URL:** {post.get('url', 'N/A')}")
st.write(f"**Tags:** {', '.join(post.get('tags', []))}")
# Add a view button if URL is available
if post.get("url"):
st.markdown(f"[View Post]({post.get('url')})")
# Add SEO report button
if st.button("Generate SEO Report"):
with st.spinner("Generating SEO report..."):
try:
wix = WixIntegration(refresh_token=refresh_token, site_id=site_id)
seo_report = wix.get_seo_report(post.get("id"), seo_keywords or tags or [])
st.subheader("SEO Report")
st.write(f"**SEO Score:** {seo_report.get('seo_score', 0):.1f}/100")
st.write("**Recommendations:**")
for i, rec in enumerate(seo_report.get("recommendations", [])):
st.write(f"{i+1}. {rec}")
except Exception as e:
st.error(f"Failed to generate SEO report: {str(e)}")
except Exception as e:
st.error(f"❌ Failed to publish blog post: {str(e)}")
logger.error(f"Failed to publish blog post: {str(e)}")
# For testing the UI directly
if __name__ == "__main__":
wix_blog_publisher_ui()

388
lib/integrations/wix/wix_integration.py Normal file
View File

@@ -0,0 +1,388 @@
"""
Wix Integration for Alwrity
This module provides a high-level interface for integrating Wix blog functionality
with the Alwrity AI Writer platform.
"""
import os
import logging
import json
from typing import Dict, List, Optional, Union, Any, Tuple
from pathlib import Path
from .wix_api_client import WixAPIClient
from .wix_blog_manager import WixBlogManager
from .wix_seo_optimizer import WixSEOOptimizer
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('wix_integration')
class WixIntegration:
"""
Main integration class for Wix blog functionality.
This class provides a simplified interface for common operations,
combining the functionality of the API client, blog manager, and SEO optimizer.
"""
def __init__(
self,
api_key: Optional[str] = None,
refresh_token: Optional[str] = None,
site_id: Optional[str] = None
):
"""
Initialize the Wix Integration.
Args:
api_key: Wix API key (optional if using refresh token)
refresh_token: Wix refresh token for OAuth authentication
site_id: Wix site ID
"""
self.api_client = WixAPIClient(api_key, refresh_token, site_id)
self.blog_manager = WixBlogManager(api_key, refresh_token, site_id)
self.seo_optimizer = WixSEOOptimizer(api_key, refresh_token, site_id)
def publish_blog_post(
self,
title: str,
content: str,
is_markdown: bool = True,
featured_image_path: Optional[str] = None,
featured_image_url: Optional[str] = None,
excerpt: Optional[str] = None,
tags: Optional[List[str]] = None,
categories: Optional[List[str]] = None,
seo_title: Optional[str] = None,
seo_description: Optional[str] = None,
seo_keywords: Optional[List[str]] = None,
author_name: Optional[str] = None,
publisher_name: Optional[str] = None,
publisher_logo_url: Optional[str] = None,
publish: bool = True,
update_if_exists: bool = True
) -> Dict:
"""
Publish a blog post with comprehensive SEO optimization.
Args:
title: Post title
content: Post content (markdown or HTML)
is_markdown: Whether the content is in markdown format
featured_image_path: Local path to featured image (optional)
featured_image_url: URL of featured image to download (optional)
excerpt: Post excerpt/summary (optional)
tags: List of tags (optional)
categories: List of category names (optional)
seo_title: SEO title (optional)
seo_description: SEO description (optional)
seo_keywords: SEO keywords (optional)
author_name: Name of the author (optional)
publisher_name: Name of the publisher (optional)
publisher_logo_url: URL of the publisher's logo (optional)
publish: Whether to publish the post immediately (optional)
update_if_exists: Whether to update an existing post with the same title (optional)
Returns:
Published blog post data
"""
# Generate SEO data if not provided
if not seo_keywords and tags:
seo_keywords = tags
if not seo_title:
seo_title = title
if not seo_description and not excerpt:
if is_markdown:
# Generate description from markdown content
seo_description = self.blog_manager._generate_excerpt(content)
else:
# Generate description from HTML content
seo_description = self.seo_optimizer.generate_meta_description(content)
elif not seo_description:
seo_description = excerpt
# Publish or update the post
if is_markdown:
response = self.blog_manager.publish_or_update_markdown_post(
title=title,
markdown_content=content,
featured_image_path=featured_image_path,
featured_image_url=featured_image_url,
excerpt=excerpt,
tags=tags,
categories=categories,
seo_title=seo_title,
seo_description=seo_description,
seo_keywords=seo_keywords,
publish=publish,
update_if_exists=update_if_exists
)
else:
# Find existing post or create new one
existing_post = self.blog_manager.find_post_by_title(title)
if existing_post and update_if_exists:
# Update existing post
response = self.api_client.update_post(
post_id=existing_post["id"],
title=title,
content=content,
excerpt=excerpt,
tags=tags,
categories=[self.api_client.get_or_create_category(cat) for cat in categories] if categories else None,
seo_data={
"title": seo_title,
"description": seo_description,
"keywords": seo_keywords or []
},
publish=publish
)
else:
# Create new post
response = self.api_client.create_post(
title=title,
content=content,
excerpt=excerpt,
tags=tags,
categories=[self.api_client.get_or_create_category(cat) for cat in categories] if categories else None,
seo_data={
"title": seo_title,
"description": seo_description,
"keywords": seo_keywords or []
},
publish=publish
)
# Apply additional SEO optimization if the post was published
if publish and response.get("post", {}).get("id"):
post_id = response["post"]["id"]
# Apply structured data if author and publisher info is provided
if author_name and publisher_name and publisher_logo_url:
try:
self.seo_optimizer.apply_structured_data_to_post(
post_id=post_id,
author_name=author_name,
publisher_name=publisher_name,
publisher_logo_url=publisher_logo_url
)
except Exception as e:
logger.error(f"Failed to apply structured data: {str(e)}")
# Apply comprehensive SEO optimization
try:
self.seo_optimizer.apply_seo_optimization(
post_id=post_id,
title=seo_title,
description=seo_description,
keywords=seo_keywords,
author_name=author_name,
publisher_name=publisher_name,
publisher_logo_url=publisher_logo_url,
og_image_url=featured_image_url
)
except Exception as e:
logger.error(f"Failed to apply SEO optimization: {str(e)}")
return response
def upload_media(
self,
file_path: str,
title: Optional[str] = None,
alt_text: Optional[str] = None,
description: Optional[str] = None
) -> Dict:
"""
Upload a media file to Wix.
Args:
file_path: Path to the media file
title: Media title (optional)
alt_text: Media alt text (optional)
description: Media description (optional)
Returns:
Uploaded media data
"""
return self.api_client.upload_image(
file_path=file_path,
title=title,
alt_text=alt_text,
description=description
)
def get_seo_report(self, post_id: str, target_keywords: List[str]) -> Dict:
"""
Generate a comprehensive SEO report for a blog post.
Args:
post_id: ID of the blog post
target_keywords: List of target keywords
Returns:
Dictionary with SEO report data
"""
return self.seo_optimizer.generate_seo_report(post_id, target_keywords)
def list_blog_posts(
self,
limit: int = 50,
offset: int = 0,
sort_field: str = "lastPublishedDate",
sort_order: str = "desc"
) -> Dict:
"""
List blog posts with pagination and sorting.
Args:
limit: Maximum number of posts to return (default: 50)
offset: Pagination offset (default: 0)
sort_field: Field to sort by (default: lastPublishedDate)
sort_order: Sort order, 'asc' or 'desc' (default: desc)
Returns:
Dictionary containing blog posts and pagination info
"""
return self.api_client.list_posts(
limit=limit,
offset=offset,
sort_field=sort_field,
sort_order=sort_order
)
def list_categories(self) -> Dict:
"""
List all blog categories.
Returns:
Dictionary containing blog categories
"""
return self.api_client.list_categories()
def create_category(self, name: str, description: Optional[str] = None) -> str:
"""
Create a new blog category.
Args:
name: Category name
description: Category description (optional)
Returns:
ID of the created category
"""
response = self.api_client.create_category(
label=name,
description=description
)
return response.get("category", {}).get("id", "")
def get_post_by_id(self, post_id: str) -> Dict:
"""
Get a blog post by ID.
Args:
post_id: ID of the blog post
Returns:
Blog post data
"""
return self.api_client.get_post(post_id)
def get_post_by_title(self, title: str) -> Optional[Dict]:
"""
Get a blog post by title.
Args:
title: Title of the blog post
Returns:
Blog post data or None if not found
"""
return self.blog_manager.find_post_by_title(title)
def delete_post(self, post_id: str) -> Dict:
"""
Delete a blog post.
Args:
post_id: ID of the blog post
Returns:
Response data
"""
return self.api_client.delete_post(post_id)
def update_post_status(self, post_id: str, publish: bool = True) -> Dict:
"""
Update the publication status of a blog post.
Args:
post_id: ID of the blog post
publish: Whether to publish (True) or unpublish (False) the post
Returns:
Updated blog post data
"""
if publish:
return self.api_client.publish_post(post_id)
else:
return self.api_client.unpublish_post(post_id)
def search_posts(self, query: str, limit: int = 10) -> List[Dict]:
"""
Search for blog posts by content or title.
Args:
query: Search query
limit: Maximum number of results to return
Returns:
List of matching blog posts
"""
# First try to find by title
title_matches = []
try:
all_posts = self.list_blog_posts(limit=100)["posts"]
for post in all_posts:
if query.lower() in post.get("title", "").lower():
title_matches.append(post)
if len(title_matches) >= limit:
break
except Exception as e:
logger.error(f"Error searching posts by title: {str(e)}")
return title_matches[:limit]
def get_site_info(self) -> Dict:
"""
Get information about the Wix site.
Returns:
Dictionary with site information
"""
try:
# Make a simple API call to verify credentials and get site info
posts = self.list_blog_posts(limit=1)
categories = self.list_categories()
return {
"site_id": self.api_client.site_id,
"post_count": posts.get("totalCount", 0),
"category_count": len(categories.get("categories", [])),
"status": "connected"
}
except Exception as e:
logger.error(f"Error getting site info: {str(e)}")
return {
"site_id": self.api_client.site_id,
"status": "error",
"error": str(e)
}

0
lib/integrations/wordpress_blog_uploader.py → lib/integrations/wordpress/wordpress_blog_uploader.py
View File