064ae085b5
Infrastruktur: - CT 113 auf pve-hetzner erstellt (Docker, Tailscale) - Forgejo-Repo redax-wp angelegt Code (Sprint 2): - docker-compose.yml: wordpress + db + redax-web - .env.example mit allen Variablen - database.py: articles, feeds, feed_items, prompts, settings - wordpress.py: WP REST API Client (create/update post, media upload, Yoast SEO) - rss_fetcher.py: Feed-Import, Blacklist, Teaser-Modus, KI-Rewrite - app.py: Flask Dashboard, Scheduler (publish/rss/briefing), alle API-Routen - templates: base, login, index (Zwei-Spalten-Editor), feeds, history, prompts, settings, hilfe - README.md + .gitignore Made-with: Cursor
121 lines
4.4 KiB
Python
121 lines
4.4 KiB
Python
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
|
|
|
|
# Yoast SEO meta fields
|
|
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}')
|