Initial commit: WordPress News Import System
This commit is contained in:
commit
5f923d8ece
8 changed files with 948 additions and 0 deletions
278
wordpress_api.py
Normal file
278
wordpress_api.py
Normal file
|
|
@ -0,0 +1,278 @@
|
|||
#!/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 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:
|
||||
response = self._get('posts', params={'search': title, 'per_page': 10})
|
||||
posts = response.json()
|
||||
|
||||
# Exakte Übereinstimmung prüfen
|
||||
for post in posts:
|
||||
if post.get('title', {}).get('rendered', '') == title:
|
||||
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,
|
||||
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
|
||||
if check_duplicate:
|
||||
existing_id = self.check_post_exists(title)
|
||||
if existing_id:
|
||||
print(f"Beitrag '{title}' existiert bereits (ID: {existing_id})")
|
||||
return existing_id
|
||||
|
||||
# Post-Daten zusammenstellen
|
||||
post_data = {
|
||||
'title': title,
|
||||
'content': content,
|
||||
'status': status,
|
||||
**kwargs
|
||||
}
|
||||
|
||||
if featured_media:
|
||||
post_data['featured_media'] = featured_media
|
||||
if categories:
|
||||
post_data['categories'] = categories
|
||||
if tags:
|
||||
post_data['tags'] = tags
|
||||
|
||||
# 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})")
|
||||
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 get_categories(self) -> List[Dict[str, Any]]:
|
||||
"""Holt alle verfügbaren Kategorien"""
|
||||
try:
|
||||
response = self._get('categories', params={'per_page': 100})
|
||||
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"""
|
||||
categories = self.get_categories()
|
||||
for cat in categories:
|
||||
if cat['name'].lower() == name.lower():
|
||||
return cat['id']
|
||||
|
||||
# Kategorie erstellen
|
||||
try:
|
||||
response = self._post('categories', data={'name': name})
|
||||
return response.json()['id']
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"Fehler beim Erstellen der Kategorie: {e}")
|
||||
return None
|
||||
|
||||
def get_tags(self) -> List[Dict[str, Any]]:
|
||||
"""Holt alle verfügbaren Tags"""
|
||||
try:
|
||||
response = self._get('tags', params={'per_page': 100})
|
||||
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"""
|
||||
tags = self.get_tags()
|
||||
for tag in tags:
|
||||
if tag['name'].lower() == name.lower():
|
||||
return tag['id']
|
||||
|
||||
# Tag erstellen
|
||||
try:
|
||||
response = self._post('tags', data={'name': name})
|
||||
return response.json()['id']
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"Fehler beim Erstellen des Tags: {e}")
|
||||
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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue