import os import requests from requests.auth import HTTPBasicAuth import logger as flog class WordPressClient: def __init__(self): self.base_url = os.environ.get('WP_URL', 'http://wordpress').rstrip('/') self.api_url = f"{self.base_url}/wp-json/wp/v2" self.username = os.environ.get('WP_USERNAME', 'admin') self.app_password = os.environ.get('WP_APP_PASSWORD', '') self.auth = HTTPBasicAuth(self.username, self.app_password) def _get(self, endpoint, params=None): r = requests.get(f"{self.api_url}/{endpoint}", auth=self.auth, params=params or {}, timeout=15) r.raise_for_status() return r.json() def _post(self, endpoint, data): r = requests.post(f"{self.api_url}/{endpoint}", auth=self.auth, json=data, timeout=30) r.raise_for_status() return r.json() def _put(self, endpoint, data): r = requests.put(f"{self.api_url}/{endpoint}", auth=self.auth, json=data, timeout=30) r.raise_for_status() return r.json() def is_reachable(self) -> bool: try: requests.get(f"{self.base_url}/wp-json/", timeout=5) return True except Exception: return False def get_categories(self) -> list: try: return self._get('categories', {'per_page': 100, 'hide_empty': False}) except Exception as e: flog.error('wp_get_categories_failed', error=str(e)) return [] def get_tags(self) -> list: try: return self._get('tags', {'per_page': 100}) except Exception as e: flog.error('wp_get_tags_failed', error=str(e)) return [] def upload_media(self, image_url: str, filename: str = 'featured.jpg') -> int | None: """Download image from URL and upload to WordPress Media Library.""" try: img_response = requests.get(image_url, timeout=15) img_response.raise_for_status() content_type = img_response.headers.get('Content-Type', 'image/jpeg') r = requests.post( f"{self.api_url}/media", auth=self.auth, headers={ 'Content-Disposition': f'attachment; filename="{filename}"', 'Content-Type': content_type, }, data=img_response.content, timeout=30 ) r.raise_for_status() media_id = r.json()['id'] flog.info('wp_media_uploaded', media_id=media_id, url=image_url) return media_id except Exception as e: flog.error('wp_media_upload_failed', error=str(e), url=image_url) return None def create_post(self, title: str, content: str, status: str = 'publish', scheduled_at: str = None, category_ids: list = None, tag_ids: list = None, featured_media_id: int = None, seo_title: str = None, seo_description: str = None, focus_keyword: str = None) -> dict: """ Create a WordPress post. status: 'publish' | 'draft' | 'future' (requires scheduled_at) Returns: {'id': int, 'url': str} """ data = { 'title': title, 'content': content, 'status': status, } if scheduled_at and status == 'future': data['date'] = scheduled_at if category_ids: data['categories'] = category_ids if tag_ids: data['tags'] = tag_ids if featured_media_id: data['featured_media'] = featured_media_id if any([seo_title, seo_description, focus_keyword]): data['meta'] = {} if seo_title: data['meta']['_yoast_wpseo_title'] = seo_title if seo_description: data['meta']['_yoast_wpseo_metadesc'] = seo_description if focus_keyword: data['meta']['_yoast_wpseo_focuskw'] = focus_keyword result = self._post('posts', data) return {'id': result['id'], 'url': result['link']} def update_post(self, wp_post_id: int, **kwargs) -> dict: result = self._put(f'posts/{wp_post_id}', kwargs) return {'id': result['id'], 'url': result['link']} def get_post(self, wp_post_id: int) -> dict: return self._get(f'posts/{wp_post_id}') def post_exists(self, title: str) -> bool: """Prüft ob ein Artikel mit diesem Titel bereits existiert (Duplikat-Schutz).""" try: results = self._get('posts', {'search': title[:50], 'per_page': 5, 'status': 'any'}) for p in results: if p.get('title', {}).get('rendered', '').strip() == title.strip(): return True return False except Exception: return False class WordPressMirrorClient: """Verwaltet mehrere WordPress-Targets (Primary + N Mirrors).""" def __init__(self): self.targets = [] # Primary (aus WP_URL) primary_url = os.environ.get('WP_URL', '').rstrip('/') if primary_url: self.targets.append({ 'name': 'primary', 'label': primary_url.replace('https://', '').replace('http://', ''), 'url': primary_url, 'username': os.environ.get('WP_USERNAME', 'admin'), 'app_password': os.environ.get('WP_APP_PASSWORD', ''), 'primary': True, 'enabled': True, }) # Mirror 1 (aus WP_MIRROR_URL) mirror_url = os.environ.get('WP_MIRROR_URL', '').rstrip('/') mirror_enabled = os.environ.get('WP_MIRROR_ENABLED', 'false').lower() == 'true' if mirror_url and mirror_enabled: self.targets.append({ 'name': 'mirror_1', 'label': mirror_url.replace('https://', '').replace('http://', ''), 'url': mirror_url, 'username': os.environ.get('WP_MIRROR_USERNAME', 'admin'), 'app_password': os.environ.get('WP_MIRROR_APP_PASSWORD', ''), 'primary': False, 'enabled': True, }) # Erweiterbar: Mirror 2, 3, ... aus WP_MIRROR2_URL usw. for i in range(2, 10): m_url = os.environ.get(f'WP_MIRROR{i}_URL', '').rstrip('/') m_enabled = os.environ.get(f'WP_MIRROR{i}_ENABLED', 'false').lower() == 'true' if m_url and m_enabled: self.targets.append({ 'name': f'mirror_{i}', 'label': m_url.replace('https://', '').replace('http://',''), 'url': m_url, 'username': os.environ.get(f'WP_MIRROR{i}_USERNAME', 'admin'), 'app_password': os.environ.get(f'WP_MIRROR{i}_APP_PASSWORD', ''), 'primary': False, 'enabled': True, }) def _client_for(self, target: dict) -> WordPressClient: """Erstellt einen temporären WordPressClient für ein Target.""" c = WordPressClient.__new__(WordPressClient) c.base_url = target['url'] c.api_url = f"{target['url']}/wp-json/wp/v2" c.username = target['username'] c.app_password = target['app_password'] c.auth = HTTPBasicAuth(c.username, c.app_password) c._get = c.__class__._get.__get__(c) c._post = c.__class__._post.__get__(c) c._put = c.__class__._put.__get__(c) return c def publish_to_all(self, title: str, content: str, status: str = 'publish', scheduled_at: str = None, category_ids: list = None, featured_media_id: int = None, seo_title: str = None, seo_description: str = None, focus_keyword: str = None) -> dict: """ Veröffentlicht auf allen aktiven Targets. Returns: { 'primary': {'id': int, 'url': str}, 'mirrors': [{'name': str, 'label': str, 'id': int, 'url': str, 'error': str|None}] } """ results = {'primary': None, 'mirrors': []} for target in self.targets: if not target['enabled']: continue client = self._client_for(target) try: # Duplikat-Prüfung auf Mirror (nicht auf Primary) if not target['primary']: if client.post_exists(title): flog.info('mirror_skip_duplicate', target=target['name'], title=title[:50]) results['mirrors'].append({ 'name': target['name'], 'label': target['label'], 'id': None, 'url': None, 'error': 'Duplikat übersprungen', }) continue result = client.create_post( title=title, content=content, status=status, scheduled_at=scheduled_at, category_ids=category_ids, featured_media_id=featured_media_id, seo_title=seo_title, seo_description=seo_description, focus_keyword=focus_keyword, ) flog.info('published_to_target', target=target['name'], wp_id=result['id'], url=result['url']) if target['primary']: results['primary'] = result else: results['mirrors'].append({ 'name': target['name'], 'label': target['label'], 'id': result['id'], 'url': result['url'], 'error': None, }) except Exception as e: flog.error('publish_target_failed', target=target['name'], error=str(e)) if target['primary']: raise # Primary-Fehler weitergeben else: results['mirrors'].append({ 'name': target['name'], 'label': target['label'], 'id': None, 'url': None, 'error': str(e), }) return results def get_active_targets(self) -> list: return [t for t in self.targets if t['enabled']]