import logging import os import json import requests from typing import Dict, Any, Optional from datetime import datetime from models import ConsolidatedHoroscope logger = logging.getLogger(__name__) class WordPressService: """Service for interacting with WordPress REST API""" def __init__(self): """Initialize WordPress service""" # WordPress API settings self.api_url = os.environ.get("WORDPRESS_API_URL", "") self.username = os.environ.get("WORDPRESS_USERNAME", "") self.password = os.environ.get("WORDPRESS_PASSWORD", "") self.auth = None if self.username and self.password: self.auth = (self.username, self.password) # Check if WordPress integration is properly configured self.is_configured = bool(self.api_url and self.auth) if not self.is_configured: logger.warning("WordPress service not properly configured. Missing API URL, username, or password.") def test_connection(self) -> Dict[str, Any]: """Test the WordPress API connection""" if not self.is_configured: return {"success": False, "error": "WordPress API not configured"} try: response = requests.get(f"{self.api_url}/wp/v2", auth=self.auth) if response.status_code == 200: return {"success": True, "message": "WordPress connection successful"} else: return { "success": False, "error": f"WordPress connection failed: {response.status_code}", "details": response.text } except Exception as e: logger.error(f"Error testing WordPress connection: {str(e)}") return {"success": False, "error": str(e)} def publish_horoscope(self, horoscope: ConsolidatedHoroscope) -> Dict[str, Any]: """ Publish a consolidated horoscope to WordPress Args: horoscope: ConsolidatedHoroscope object to publish Returns: Dictionary with status and post details """ if not self.is_configured: return {"success": False, "error": "WordPress API not configured"} try: # Format the horoscope for WordPress post_data = self._format_horoscope_for_wordpress(horoscope) # Create the post via WordPress API response = requests.post( f"{self.api_url}/wp/v2/posts", auth=self.auth, json=post_data ) if response.status_code in (200, 201): post_data = response.json() return { "success": True, "post_id": post_data.get("id"), "url": post_data.get("link"), "status": post_data.get("status") } else: logger.error(f"WordPress API error: {response.status_code}, {response.text}") return { "success": False, "error": f"WordPress API error: {response.status_code}", "details": response.text } except Exception as e: logger.error(f"Error publishing horoscope to WordPress: {str(e)}") return {"success": False, "error": str(e)} def _format_horoscope_for_wordpress(self, horoscope: ConsolidatedHoroscope) -> Dict[str, Any]: """Format horoscope data for WordPress post""" # Capitalize sign sign = horoscope.sign.capitalize() # Format date formatted_date = horoscope.date.strftime("%B %d, %Y") # Construct title title = f"{sign} Horoscope for {formatted_date}" # Construct content content = f"""

{sign} Daily Horoscope - {formatted_date}

{horoscope.consolidated_prediction}

This horoscope is consolidated from multiple trusted astrological sources.

""" # Extract sources try: sources = json.loads(horoscope.sources) if sources: sources_text = "\n

Sources: " + ", ".join(sources) + "

" content += sources_text except Exception: pass # Construct post data post_data = { "title": title, "content": content, "status": "publish", # Could be: publish, draft, pending, future "excerpt": horoscope.consolidated_prediction[:150] + "...", "categories": [self._get_horoscope_category_id()], "tags": [self._get_tag_id(sign.lower())] } # Schedule post if needed # if schedule_time: # post_data["status"] = "future" # post_data["date"] = schedule_time.isoformat() return post_data def _get_horoscope_category_id(self) -> int: """ Get or create the horoscope category ID Returns: Category ID """ # Default to a common category ID if we can't find/create the specific one default_category_id = 1 # Usually "Uncategorized" if not self.is_configured: return default_category_id try: # Check if Horoscopes category exists response = requests.get( f"{self.api_url}/wp/v2/categories", auth=self.auth, params={"search": "Horoscopes"} ) if response.status_code == 200: categories = response.json() if categories: # Return the first matching category return categories[0].get("id", default_category_id) # Create the category if it doesn't exist create_response = requests.post( f"{self.api_url}/wp/v2/categories", auth=self.auth, json={ "name": "Horoscopes", "slug": "horoscopes", "description": "Daily horoscopes for all zodiac signs" } ) if create_response.status_code in (200, 201): category_data = create_response.json() return category_data.get("id", default_category_id) return default_category_id except Exception as e: logger.error(f"Error getting horoscope category ID: {str(e)}") return default_category_id def _get_tag_id(self, tag_name: str) -> int: """ Get or create a tag ID Args: tag_name: Name of the tag Returns: Tag ID """ # Default to 0 which will be ignored by WordPress default_tag_id = 0 if not self.is_configured: return default_tag_id try: # Check if tag exists response = requests.get( f"{self.api_url}/wp/v2/tags", auth=self.auth, params={"search": tag_name} ) if response.status_code == 200: tags = response.json() if tags: # Return the first matching tag return tags[0].get("id", default_tag_id) # Create the tag if it doesn't exist create_response = requests.post( f"{self.api_url}/wp/v2/tags", auth=self.auth, json={ "name": tag_name, "slug": tag_name.lower().replace(" ", "-") } ) if create_response.status_code in (200, 201): tag_data = create_response.json() return tag_data.get("id", default_tag_id) return default_tag_id except Exception as e: logger.error(f"Error getting tag ID: {str(e)}") return default_tag_id # Create a singleton instance wordpress_service = WordPressService()