Spaces:
Sleeping
Sleeping
| """ | |
| WordPress API interactions for posting content. | |
| """ | |
| import requests | |
| from logger import logger | |
| from config import WP_URL, WP_USER, WP_APP_PASS, DEFAULT_POST_STATUS | |
| class WordPressClient: | |
| """Handles WordPress API interactions.""" | |
| def __init__(self, wp_url=WP_URL, username=WP_USER, app_password=WP_APP_PASS): | |
| """Initialize WordPress client.""" | |
| if not all([wp_url, username, app_password]): | |
| raise ValueError("WordPress credentials not set in environment") | |
| self.wp_url = wp_url.rstrip("/") | |
| self.auth = (username, app_password) | |
| logger.debug("WordPressClient initialized") | |
| def create_or_get_tags(self, tag_names): | |
| """ | |
| Create tags in WordPress or get existing ones. | |
| Workflow: | |
| 1. Search for existing tag by name | |
| 2. If found, use that tag ID | |
| 3. If not found, create new tag | |
| 4. Log failures only when tag creation actually fails | |
| Args: | |
| tag_names (list): List of tag names | |
| Returns: | |
| list: List of tag IDs | |
| Raises: | |
| Exception: If API call fails | |
| """ | |
| try: | |
| logger.info(f"Processing {len(tag_names)} tags") | |
| tag_ids = [] | |
| for tag_name in tag_names: | |
| try: | |
| # Step 1: Search for existing tag | |
| logger.debug(f"Searching for tag '{tag_name}'") | |
| search_response = requests.get( | |
| f"{self.wp_url}/wp-json/wp/v2/tags", | |
| auth=self.auth, | |
| params={"search": tag_name, "per_page": 100}, | |
| timeout=10, | |
| ) | |
| search_response.raise_for_status() | |
| search_results = search_response.json() | |
| existing_tag = None | |
| # Look for exact or close match | |
| for tag in search_results: | |
| if tag.get("name", "").lower() == tag_name.lower(): | |
| existing_tag = tag | |
| break | |
| if existing_tag: | |
| tag_id = existing_tag.get("id") | |
| tag_ids.append(tag_id) | |
| logger.info(f"Tag '{tag_name}' found (ID {tag_id})") | |
| continue | |
| # Step 2: Tag not found, create it | |
| logger.debug(f"Tag '{tag_name}' not found, creating new tag") | |
| create_response = requests.post( | |
| f"{self.wp_url}/wp-json/wp/v2/tags", | |
| auth=self.auth, | |
| json={"name": tag_name}, | |
| timeout=10, | |
| ) | |
| create_response.raise_for_status() | |
| created_tag = create_response.json() | |
| tag_id = created_tag.get("id") | |
| if tag_id: | |
| tag_ids.append(tag_id) | |
| logger.info(f"Tag '{tag_name}' created (ID {tag_id})") | |
| else: | |
| logger.warning(f"No ID returned for created tag '{tag_name}'") | |
| except requests.RequestException as e: | |
| logger.error(f"Tag creation failed for '{tag_name}': {e}") | |
| logger.info(f"Processed {len(tag_ids)} tags successfully") | |
| return tag_ids | |
| except Exception as e: | |
| logger.error(f"Failed to process tags: {e}") | |
| raise | |
| def create_or_get_categories(self, category_names): | |
| """ | |
| Create categories in WordPress or get existing ones. | |
| Args: | |
| category_names (list): List of category names | |
| Returns: | |
| list: List of category IDs | |
| """ | |
| try: | |
| logger.info(f"Processing {len(category_names)} categories") | |
| category_ids = [] | |
| for cat_name in category_names: | |
| try: | |
| search_response = requests.get( | |
| f"{self.wp_url}/wp-json/wp/v2/categories", | |
| auth=self.auth, | |
| params={"search": cat_name, "per_page": 100}, | |
| timeout=10, | |
| ) | |
| search_response.raise_for_status() | |
| search_results = search_response.json() | |
| existing_cat = None | |
| for cat in search_results: | |
| if cat.get("name", "").lower() == cat_name.lower(): | |
| existing_cat = cat | |
| break | |
| if existing_cat: | |
| cat_id = existing_cat.get("id") | |
| category_ids.append(cat_id) | |
| logger.info(f"Category '{cat_name}' found (ID {cat_id})") | |
| continue | |
| create_response = requests.post( | |
| f"{self.wp_url}/wp-json/wp/v2/categories", | |
| auth=self.auth, | |
| json={"name": cat_name}, | |
| timeout=10, | |
| ) | |
| create_response.raise_for_status() | |
| created_cat = create_response.json() | |
| cat_id = created_cat.get("id") | |
| if cat_id: | |
| category_ids.append(cat_id) | |
| logger.info(f"Category '{cat_name}' created (ID {cat_id})") | |
| else: | |
| logger.warning(f"No ID returned for created category '{cat_name}'") | |
| except requests.RequestException as e: | |
| logger.error(f"Category creation failed for '{cat_name}': {e}") | |
| logger.info(f"Processed {len(category_ids)} categories successfully") | |
| return category_ids | |
| except Exception as e: | |
| logger.error(f"Failed to process categories: {e}") | |
| raise | |
| def upload_media(self, image_data, filename, content_type="image/avif"): | |
| """ | |
| Upload image to WordPress media library. | |
| Args: | |
| image_data (bytes): Image file data | |
| filename (str): Filename for the image | |
| content_type (str): MIME type | |
| Returns: | |
| dict: Contains 'id' and 'url' keys | |
| Raises: | |
| Exception: If upload fails | |
| """ | |
| try: | |
| logger.info(f"Uploading media: {filename}") | |
| response = requests.post( | |
| f"{self.wp_url}/wp-json/wp/v2/media", | |
| auth=self.auth, | |
| headers={ | |
| "Content-Disposition": f'attachment; filename="{filename}"', | |
| "Content-Type": content_type, | |
| }, | |
| data=image_data, | |
| timeout=30, | |
| ) | |
| response.raise_for_status() | |
| data = response.json() | |
| media_id = data.get("id") | |
| media_url = data.get("source_url") | |
| if not media_id or not media_url: | |
| raise ValueError("Invalid media response structure") | |
| logger.info(f"Media uploaded: ID {media_id}") | |
| return {"id": media_id, "url": media_url} | |
| except requests.RequestException as e: | |
| logger.error(f"Failed to upload media: {e}") | |
| raise | |
| def get_post(self, post_id): | |
| """Retrieve an existing WordPress post by ID.""" | |
| try: | |
| response = requests.get( | |
| f"{self.wp_url}/wp-json/wp/v2/posts/{post_id}", | |
| auth=self.auth, | |
| timeout=15, | |
| ) | |
| response.raise_for_status() | |
| return response.json() | |
| except requests.RequestException as e: | |
| logger.error(f"Failed to fetch post {post_id}: {e}") | |
| raise | |
| def get_post_id_by_slug(self, slug): | |
| """Return the post ID for a given slug, or None if not found.""" | |
| try: | |
| response = requests.get( | |
| f"{self.wp_url}/wp-json/wp/v2/posts", | |
| auth=self.auth, | |
| params={"slug": slug, "status": "any", "per_page": 1}, | |
| timeout=15, | |
| ) | |
| response.raise_for_status() | |
| posts = response.json() | |
| if posts: | |
| post_id = posts[0].get("id") | |
| logger.info(f"Found existing WP post by slug '{slug}': ID {post_id}") | |
| return post_id | |
| except requests.RequestException as e: | |
| logger.warning(f"Slug lookup failed for '{slug}': {e}") | |
| return None | |
| def get_media(self, media_id): | |
| """Retrieve an existing WordPress media item by ID.""" | |
| try: | |
| response = requests.get( | |
| f"{self.wp_url}/wp-json/wp/v2/media/{media_id}", | |
| auth=self.auth, | |
| timeout=15, | |
| ) | |
| response.raise_for_status() | |
| return response.json() | |
| except requests.RequestException as e: | |
| logger.error(f"Failed to fetch media {media_id}: {e}") | |
| raise | |
| def update_post(self, post_id, title, content, tag_ids, media_id, slug=None, category_ids=None, seo_data=None): | |
| """ | |
| Update an existing WordPress post. | |
| Args: | |
| post_id (int): ID of the existing post | |
| title (str): Post title | |
| content (str): Post content (HTML) | |
| tag_ids (list): List of tag IDs | |
| media_id (int): Featured image media ID | |
| slug (str, optional): URL slug | |
| category_ids (list, optional): List of category IDs | |
| seo_data (dict, optional): SEO metadata for Rank Math | |
| Returns: | |
| dict: Post data including 'id' and 'edit_url' | |
| Raises: | |
| Exception: If post update fails | |
| """ | |
| try: | |
| logger.info(f"Updating existing post: ID {post_id}") | |
| payload = { | |
| "title": title, | |
| "content": content, | |
| "tags": tag_ids, | |
| "featured_media": media_id, | |
| } | |
| if slug: | |
| payload["slug"] = slug | |
| if category_ids: | |
| payload["categories"] = category_ids | |
| if seo_data: | |
| payload["meta"] = { | |
| "rank_math_focus_keyword": seo_data.get("focus_keyword", ""), | |
| "rank_math_description": seo_data.get("meta_description", ""), | |
| } | |
| logger.debug(f"Sending update payload for post {post_id}: slug={payload.get('slug')}, " | |
| f"categories={payload.get('categories')}, meta={payload.get('meta')}") | |
| response = requests.post( | |
| f"{self.wp_url}/wp-json/wp/v2/posts/{post_id}", | |
| auth=self.auth, | |
| json=payload, | |
| timeout=30, | |
| ) | |
| response.raise_for_status() | |
| post_data = response.json() | |
| edit_url = f"{self.wp_url}/wp-admin/post.php?post={post_id}&action=edit" | |
| returned_meta = post_data.get("meta", {}) | |
| logger.debug(f"WP returned meta for post {post_id}: {returned_meta}") | |
| if seo_data and not returned_meta.get("rank_math_focus_keyword"): | |
| logger.warning( | |
| "rank_math_focus_keyword not found in WP response meta — " | |
| "ensure Rank Math is active and has REST API support enabled" | |
| ) | |
| logger.info(f"Post updated: ID {post_id}") | |
| return { | |
| "id": post_id, | |
| "title": post_data.get("title", {}).get("rendered", title), | |
| "edit_url": edit_url, | |
| } | |
| except requests.RequestException as e: | |
| logger.error(f"Failed to update post {post_id}: {e}") | |
| raise | |
| def create_draft_post(self, title, content, tag_ids, media_id, slug=None, category_ids=None, seo_data=None): | |
| """ | |
| Create a draft post in WordPress. | |
| Args: | |
| title (str): Post title | |
| content (str): Post content (HTML) | |
| tag_ids (list): List of tag IDs | |
| media_id (int): Featured image media ID | |
| slug (str, optional): URL slug | |
| category_ids (list, optional): List of category IDs | |
| seo_data (dict, optional): SEO metadata for Rank Math | |
| Returns: | |
| dict: Post data including 'id' and 'edit_url' | |
| Raises: | |
| Exception: If post creation fails | |
| """ | |
| try: | |
| logger.info(f"Creating draft post: {title}") | |
| payload = { | |
| "title": title, | |
| "content": content, | |
| "status": DEFAULT_POST_STATUS, | |
| "tags": tag_ids, | |
| "featured_media": media_id, | |
| } | |
| if slug: | |
| payload["slug"] = slug | |
| if category_ids: | |
| payload["categories"] = category_ids | |
| if seo_data: | |
| payload["meta"] = { | |
| "rank_math_focus_keyword": seo_data.get("focus_keyword", ""), | |
| "rank_math_description": seo_data.get("meta_description", ""), | |
| } | |
| logger.debug(f"Sending create payload: slug={payload.get('slug')}, " | |
| f"categories={payload.get('categories')}, meta={payload.get('meta')}") | |
| response = requests.post( | |
| f"{self.wp_url}/wp-json/wp/v2/posts", | |
| auth=self.auth, | |
| json=payload, | |
| timeout=30, | |
| ) | |
| response.raise_for_status() | |
| post_data = response.json() | |
| post_id = post_data.get("id") | |
| edit_url = f"{self.wp_url}/wp-admin/post.php?post={post_id}&action=edit" | |
| returned_meta = post_data.get("meta", {}) | |
| logger.debug(f"WP returned meta for new post {post_id}: {returned_meta}") | |
| if seo_data and not returned_meta.get("rank_math_focus_keyword"): | |
| logger.warning( | |
| "rank_math_focus_keyword not found in WP response meta — " | |
| "ensure Rank Math is active and has REST API support enabled" | |
| ) | |
| logger.info(f"Draft post created: ID {post_id}") | |
| return { | |
| "id": post_id, | |
| "title": post_data.get("title", {}).get("rendered", title), | |
| "edit_url": edit_url, | |
| } | |
| except requests.RequestException as e: | |
| logger.error(f"Failed to create draft post: {e}") | |
| raise | |