#!/usr/bin/env python3 """ WordPress API Helper Bietet Funktionen zum Erstellen von WordPress-Beiträgen und Hochladen von Medien über die WordPress REST-API mit Duplikatsprüfung. """ import os import requests import hashlib from typing import Optional, Dict, Any, List from urllib.parse import urljoin import mimetypes class WordPressAPI: """WordPress REST-API Client mit Duplikatsprüfung""" def __init__(self, url: str, username: str, app_password: str): """ Initialisiert den WordPress API Client Args: url: WordPress URL (z.B. https://news.rpi-virtuell.de) username: WordPress Benutzername app_password: WordPress Anwendungspasswort """ self.url = url.rstrip('/') self.api_base = urljoin(self.url, '/wp-json/wp/v2/') self.auth = (username, app_password.replace(' ', '')) self.session = requests.Session() self.session.auth = self.auth def _get(self, endpoint: str, params: Optional[Dict] = None) -> requests.Response: """GET-Request an WordPress API""" url = urljoin(self.api_base, endpoint) response = self.session.get(url, params=params) response.raise_for_status() return response def _post(self, endpoint: str, data: Optional[Dict] = None, files: Optional[Dict] = None, headers: Optional[Dict] = None) -> requests.Response: """POST-Request an WordPress API""" url = urljoin(self.api_base, endpoint) response = self.session.post(url, json=data, files=files, headers=headers) response.raise_for_status() return response def _put(self, endpoint: str, data: Optional[Dict] = None) -> requests.Response: """PUT-Request an WordPress API (für Updates)""" url = urljoin(self.api_base, endpoint) response = self.session.put(url, json=data) response.raise_for_status() return response def check_post_exists(self, title: str) -> Optional[int]: """ Prüft, ob ein Beitrag mit dem Titel bereits existiert Args: title: Titel des Beitrags Returns: Post-ID wenn gefunden, sonst None """ try: # Suche mit verschiedenen Parametern response = self._get('posts', params={ 'search': title, 'per_page': 100, # Erhöht für bessere Suche 'status': 'any' # Alle Status (draft, publish, etc.) }) posts = response.json() # Normalisiere Titel für Vergleich title_lower = title.lower().strip() # Exakte Übereinstimmung prüfen for post in posts: # Prüfe rendered Titel rendered_title = post.get('title', {}).get('rendered', '').strip() if rendered_title.lower() == title_lower: print(f" → Beitrag gefunden (ID: {post['id']}, Status: {post['status']})") return post['id'] # Prüfe auch raw Titel falls vorhanden raw_title = post.get('title', {}).get('raw', '').strip() if raw_title and raw_title.lower() == title_lower: print(f" → Beitrag gefunden (ID: {post['id']}, Status: {post['status']})") return post['id'] return None except requests.exceptions.RequestException as e: print(f"Fehler bei der Suche nach Beitrag: {e}") return None def check_media_exists(self, filename: str, file_hash: Optional[str] = None) -> Optional[int]: """ Prüft, ob eine Mediendatei bereits existiert Args: filename: Dateiname file_hash: Optional MD5-Hash der Datei für präzisere Prüfung Returns: Media-ID wenn gefunden, sonst None """ try: # Suche nach Dateiname response = self._get('media', params={'search': filename, 'per_page': 10}) media_items = response.json() for item in media_items: source_url = item.get('source_url', '') if filename in source_url: # Wenn Hash gegeben, zusätzlich prüfen if file_hash: # Hash kann nicht direkt verglichen werden, # daher nur Dateiname-Check return item['id'] return item['id'] return None except requests.exceptions.RequestException as e: print(f"Fehler bei der Suche nach Medien: {e}") return None def upload_media(self, file_path: str, title: Optional[str] = None, alt_text: Optional[str] = None, check_duplicate: bool = True) -> Optional[int]: """ Lädt eine Mediendatei zu WordPress hoch Args: file_path: Pfad zur Datei title: Titel der Mediendatei (optional) alt_text: Alt-Text für Bilder (optional) check_duplicate: Prüfung auf Duplikate (Standard: True) Returns: Media-ID der hochgeladenen Datei, oder None bei Fehler """ if not os.path.exists(file_path): print(f"Datei nicht gefunden: {file_path}") return None filename = os.path.basename(file_path) # Duplikatsprüfung if check_duplicate: existing_id = self.check_media_exists(filename) if existing_id: print(f"Medien-Datei '{filename}' existiert bereits (ID: {existing_id})") return existing_id # MIME-Type ermitteln mime_type, _ = mimetypes.guess_type(file_path) if not mime_type: mime_type = 'application/octet-stream' # Datei hochladen try: with open(file_path, 'rb') as f: files = { 'file': (filename, f, mime_type) } headers = { 'Content-Disposition': f'attachment; filename="{filename}"' } if title: headers['Content-Title'] = title if alt_text: headers['Content-Alt-Text'] = alt_text url = urljoin(self.api_base, 'media') response = self.session.post(url, files=files, headers=headers) response.raise_for_status() media_data = response.json() media_id = media_data['id'] print(f"Medien-Datei '{filename}' hochgeladen (ID: {media_id})") return media_id except requests.exceptions.RequestException as e: print(f"Fehler beim Hochladen der Medien-Datei: {e}") return None def create_post(self, title: str, content: str, status: str = 'draft', featured_media: Optional[int] = None, categories: Optional[List[int]] = None, tags: Optional[List[int]] = None, excerpt: Optional[str] = None, check_duplicate: bool = True, **kwargs) -> Optional[int]: """ Erstellt einen neuen WordPress-Beitrag Args: title: Titel des Beitrags content: Inhalt des Beitrags (HTML) status: Status (draft, publish, etc.) featured_media: ID des Beitragsbilds categories: Liste der Kategorie-IDs tags: Liste der Tag-IDs check_duplicate: Prüfung auf Duplikate (Standard: True) **kwargs: Weitere WordPress-Post-Felder Returns: Post-ID des erstellten Beitrags, oder None bei Fehler """ # Duplikatsprüfung - bei Duplikat Update durchführen if check_duplicate: existing_id = self.check_post_exists(title) if existing_id: print(f"Beitrag '{title}' existiert bereits (ID: {existing_id}) - Aktualisiere...") return self.update_post( post_id=existing_id, title=title, content=content, status=status, featured_media=featured_media, categories=categories, tags=tags, excerpt=excerpt, **kwargs ) # Post-Daten zusammenstellen post_data = { 'title': title, 'content': content, 'status': status, **kwargs } if excerpt: post_data['excerpt'] = excerpt if featured_media: post_data['featured_media'] = featured_media if categories: post_data['categories'] = categories if tags: post_data['tags'] = tags # Debug: Zeige was gesendet wird print(f"Erstelle Beitrag mit Daten:") print(f" - Status: {status}") if tags: print(f" - Tags: {tags}") if categories: print(f" - Kategorien: {categories}") if 'date' in post_data: print(f" - Datum: {post_data['date']}") if 'date_gmt' in post_data: print(f" - Datum GMT: {post_data['date_gmt']}") # Beitrag erstellen try: response = self._post('posts', data=post_data) post = response.json() post_id = post['id'] print(f"✅ Beitrag '{title}' erstellt (ID: {post_id}, Status: {status})") # Debug: Zeige was WordPress zurückgibt if 'tags' in post and post['tags']: print(f" WordPress-Tags: {post['tags']}") if 'date' in post: print(f" WordPress-Datum: {post['date']}") return post_id except requests.exceptions.RequestException as e: print(f"Fehler beim Erstellen des Beitrags: {e}") if hasattr(e.response, 'text'): print(f"Details: {e.response.text}") return None def update_post(self, post_id: int, title: Optional[str] = None, content: Optional[str] = None, status: Optional[str] = None, featured_media: Optional[int] = None, categories: Optional[List[int]] = None, tags: Optional[List[int]] = None, excerpt: Optional[str] = None, **kwargs) -> Optional[int]: """ Aktualisiert einen existierenden WordPress-Beitrag Args: post_id: ID des zu aktualisierenden Beitrags title: Neuer Titel (optional) content: Neuer Inhalt (optional) status: Neuer Status (optional) featured_media: ID des Beitragsbilds (optional) categories: Liste der Kategorie-IDs (optional) tags: Liste der Tag-IDs (optional) excerpt: Auszug (optional) **kwargs: Weitere WordPress-Post-Felder Returns: Post-ID des aktualisierten Beitrags, oder None bei Fehler """ # Post-Daten zusammenstellen (nur Felder die gesetzt sind) post_data = {**kwargs} if title is not None: post_data['title'] = title if content is not None: post_data['content'] = content if status is not None: post_data['status'] = status if excerpt is not None: post_data['excerpt'] = excerpt if featured_media is not None: post_data['featured_media'] = featured_media if categories is not None: post_data['categories'] = categories if tags is not None: post_data['tags'] = tags # Debug: Zeige was aktualisiert wird print(f"Aktualisiere Beitrag (ID: {post_id}):") if title: print(f" - Titel: {title}") if status: print(f" - Status: {status}") if tags: print(f" - Tags: {tags}") if categories: print(f" - Kategorien: {categories}") if 'date' in post_data: print(f" - Datum: {post_data['date']}") if 'date_gmt' in post_data: print(f" - Datum GMT: {post_data['date_gmt']}") # Beitrag aktualisieren try: response = self._put(f'posts/{post_id}', data=post_data) post = response.json() print(f"✅ Beitrag aktualisiert (ID: {post_id}, Status: {post.get('status')})") # Debug: Zeige was WordPress zurückgibt if 'tags' in post and post['tags']: print(f" WordPress-Tags: {post['tags']}") if 'date' in post: print(f" WordPress-Datum: {post['date']}") return post_id except requests.exceptions.RequestException as e: print(f"Fehler beim Aktualisieren des Beitrags: {e}") if hasattr(e.response, 'text'): print(f"Details: {e.response.text}") return None def get_categories(self, search: Optional[str] = None) -> List[Dict[str, Any]]: """ Holt alle verfügbaren Kategorien oder sucht nach einer bestimmten Kategorie Args: search: Optionaler Suchbegriff """ try: params = {'per_page': 100} if search: params['search'] = search response = self._get('categories', params=params) return response.json() except requests.exceptions.RequestException as e: print(f"Fehler beim Abrufen der Kategorien: {e}") return [] def get_or_create_category(self, name: str) -> Optional[int]: """Holt oder erstellt eine Kategorie""" # Erst gezielt nach dieser Kategorie suchen categories = self.get_categories(search=name) for cat in categories: if cat['name'].lower() == name.lower(): print(f" → Kategorie '{name}' gefunden (ID: {cat['id']})") return cat['id'] # Kategorie erstellen try: response = self._post('categories', data={'name': name}) return response.json()['id'] except requests.exceptions.RequestException as e: # Prüfe ob Fehler durch bereits existierende Kategorie if e.response is not None and e.response.status_code == 400: try: error_data = e.response.json() # WordPress gibt bei 'term_exists' die term_id zurück! if error_data.get('code') == 'term_exists': term_id = error_data.get('data', {}).get('term_id') if term_id: print(f" → Kategorie '{name}' existiert bereits (ID: {term_id})") return term_id except: pass # Fallback: Erneut suchen categories = self.get_categories() for cat in categories: if cat['name'].lower() == name.lower(): return cat['id'] print(f"Fehler beim Erstellen der Kategorie '{name}': {e}") if hasattr(e, 'response') and e.response is not None: try: error_data = e.response.json() print(f"Details: {error_data}") except: print(f"Response: {e.response.text}") return None def get_tags(self, search: Optional[str] = None) -> List[Dict[str, Any]]: """ Holt alle verfügbaren Tags oder sucht nach einem bestimmten Tag Args: search: Optionaler Suchbegriff """ try: params = {'per_page': 100} if search: params['search'] = search response = self._get('tags', params=params) return response.json() except requests.exceptions.RequestException as e: print(f"Fehler beim Abrufen der Tags: {e}") return [] def get_or_create_tag(self, name: str) -> Optional[int]: """Holt oder erstellt einen Tag""" # Erst gezielt nach diesem Tag suchen tags = self.get_tags(search=name) for tag in tags: if tag['name'].lower() == name.lower(): print(f" → Tag '{name}' gefunden (ID: {tag['id']})") return tag['id'] # Tag erstellen try: response = self._post('tags', data={'name': name}) return response.json()['id'] except requests.exceptions.RequestException as e: # Prüfe ob Fehler durch bereits existierenden Tag if e.response is not None and e.response.status_code == 400: try: error_data = e.response.json() # WordPress gibt bei 'term_exists' die term_id zurück! if error_data.get('code') == 'term_exists': term_id = error_data.get('data', {}).get('term_id') if term_id: print(f" → Tag '{name}' existiert bereits (ID: {term_id})") return term_id except: pass # Fallback: Erneut suchen tags = self.get_tags() for tag in tags: if tag['name'].lower() == name.lower(): return tag['id'] print(f"Fehler beim Erstellen des Tags '{name}': {e}") if hasattr(e, 'response') and e.response is not None: try: error_data = e.response.json() print(f"Details: {error_data}") except: print(f"Response: {e.response.text}") return None def calculate_file_hash(file_path: str) -> str: """Berechnet MD5-Hash einer Datei""" hash_md5 = hashlib.md5() with open(file_path, 'rb') as f: for chunk in iter(lambda: f.read(4096), b""): hash_md5.update(chunk) return hash_md5.hexdigest()