wordpresspub/wp_api_examples.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
WordPress REST API 使用示例
演示常见的WordPress操作
"""

import os
import requests
import json
from requests.auth import HTTPBasicAuth
from datetime import datetime
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# WordPress 配置信息
WP_URL = os.getenv('WORDPRESS_URL')
WP_USER = os.getenv('WORDPRESS_USERNAME')
WP_APP_PASSWORD = os.getenv('WORDPRESS_APP_PASSWORD')

# 移除密码中的空格
if WP_APP_PASSWORD:
    WP_APP_PASSWORD = WP_APP_PASSWORD.replace(" ", "")

# 检查必需的环境变量
if not all([WP_URL, WP_USER, WP_APP_PASSWORD]):
    print("❌ 错误: 请在 .env 文件中配置 WORDPRESS_URL, WORDPRESS_USERNAME, WORDPRESS_APP_PASSWORD")
    exit(1)

class WordPressAPI:
    def __init__(self, url, username, app_password):
        self.base_url = url.rstrip('/')
        self.api_url = f"{self.base_url}/wp-json/wp/v2"
        self.username = username
        self.app_password = app_password
        self.auth = HTTPBasicAuth(username, app_password)
        
    def get_posts(self, per_page=10, status='publish'):
        """获取文章列表"""
        params = {
            'per_page': per_page,
            'status': status
        }
        response = requests.get(f"{self.api_url}/posts", params=params)
        if response.status_code == 200:
            return response.json()
        else:
            print(f"获取文章失败: {response.status_code}")
            return None
    
    def get_post_by_id(self, post_id):
        """根据ID获取特定文章"""
        response = requests.get(f"{self.api_url}/posts/{post_id}")
        if response.status_code == 200:
            return response.json()
        else:
            print(f"获取文章失败: {response.status_code}")
            return None
    
    def create_post(self, title, content, status='draft', categories=None, tags=None):
        """创建新文章"""
        post_data = {
            'title': title,
            'content': content,
            'status': status
        }
        
        if categories:
            post_data['categories'] = categories
        if tags:
            post_data['tags'] = tags
            
        response = requests.post(
            f"{self.api_url}/posts",
            json=post_data,
            auth=self.auth
        )
        
        if response.status_code == 201:
            return response.json()
        else:
            print(f"创建文章失败: {response.status_code}")
            print(response.text)
            return None
    
    def update_post(self, post_id, title=None, content=None, status=None):
        """更新文章"""
        update_data = {}
        if title:
            update_data['title'] = title
        if content:
            update_data['content'] = content
        if status:
            update_data['status'] = status
            
        response = requests.post(
            f"{self.api_url}/posts/{post_id}",
            json=update_data,
            auth=self.auth
        )
        
        if response.status_code == 200:
            return response.json()
        else:
            print(f"更新文章失败: {response.status_code}")
            return None
    
    def delete_post(self, post_id, force=False):
        """删除文章"""
        params = {'force': 'true'} if force else {}
        response = requests.delete(
            f"{self.api_url}/posts/{post_id}",
            params=params,
            auth=self.auth
        )
        
        if response.status_code == 200:
            return response.json()
        else:
            print(f"删除文章失败: {response.status_code}")
            return None
    
    def get_categories(self):
        """获取分类列表"""
        response = requests.get(f"{self.api_url}/categories")
        if response.status_code == 200:
            return response.json()
        else:
            print(f"获取分类失败: {response.status_code}")
            return None
    
    def create_category(self, name, description="", parent=0):
        """创建新分类"""
        category_data = {
            'name': name,
            'description': description,
            'parent': parent
        }
        
        response = requests.post(
            f"{self.api_url}/categories",
            json=category_data,
            auth=self.auth
        )
        
        if response.status_code == 201:
            return response.json()
        else:
            print(f"创建分类失败: {response.status_code}")
            return None
    
    def get_tags(self):
        """获取标签列表"""
        response = requests.get(f"{self.api_url}/tags")
        if response.status_code == 200:
            return response.json()
        else:
            print(f"获取标签失败: {response.status_code}")
            return None
    
    def create_tag(self, name, description=""):
        """创建新标签"""
        tag_data = {
            'name': name,
            'description': description
        }
        
        response = requests.post(
            f"{self.api_url}/tags",
            json=tag_data,
            auth=self.auth
        )
        
        if response.status_code == 201:
            return response.json()
        else:
            print(f"创建标签失败: {response.status_code}")
            return None
    
    def upload_media(self, file_path, title=None, alt_text=None):
        """上传媒体文件"""
        if not os.path.exists(file_path):
            print(f"文件不存在: {file_path}")
            return None
            
        filename = os.path.basename(file_path)
        
        with open(file_path, 'rb') as f:
            files = {
                'file': (filename, f, 'application/octet-stream')
            }
            
            data = {}
            if title:
                data['title'] = title
            if alt_text:
                data['alt_text'] = alt_text
                
            response = requests.post(
                f"{self.api_url}/media",
                files=files,
                data=data,
                auth=self.auth
            )
            
            if response.status_code == 201:
                return response.json()
            else:
                print(f"上传媒体失败: {response.status_code}")
                return None
    
    def get_media(self, per_page=10):
        """获取媒体列表"""
        params = {'per_page': per_page}
        response = requests.get(f"{self.api_url}/media", params=params)
        if response.status_code == 200:
            return response.json()
        else:
            print(f"获取媒体失败: {response.status_code}")
            return None
    
    def search_posts(self, search_term, per_page=10):
        """搜索文章"""
        params = {
            'search': search_term,
            'per_page': per_page
        }
        response = requests.get(f"{self.api_url}/posts", params=params)
        if response.status_code == 200:
            return response.json()
        else:
            print(f"搜索文章失败: {response.status_code}")
            return None

def demo_basic_operations():
    """演示基本操作"""
    print("🚀 WordPress REST API 基本操作演示\n")
    
    wp = WordPressAPI(WP_URL, WP_USER, WP_APP_PASSWORD)
    
    # 1. 获取现有文章
    print("📝 获取现有文章...")
    posts = wp.get_posts(per_page=5)
    if posts:
        print(f"找到 {len(posts)} 篇文章:")
        for post in posts:
            print(f"  - {post['title']['rendered']} (ID: {post['id']})")
    else:
        print("没有找到文章")
    
    # 2. 获取分类
    print("\n📂 获取分类...")
    categories = wp.get_categories()
    if categories:
        print(f"找到 {len(categories)} 个分类:")
        for cat in categories:
            print(f"  - {cat['name']} (ID: {cat['id']}, 文章数: {cat['count']})")
    
    # 3. 获取标签
    print("\n🏷️ 获取标签...")
    tags = wp.get_tags()
    if tags:
        print(f"找到 {len(tags)} 个标签:")
        for tag in tags[:5]:  # 只显示前5个
            print(f"  - {tag['name']} (ID: {tag['id']}, 使用次数: {tag['count']})")
    else:
        print("没有找到标签")
    
    # 4. 获取媒体文件
    print("\n🖼️ 获取媒体文件...")
    media = wp.get_media(per_page=5)
    if media:
        print(f"找到 {len(media)} 个媒体文件:")
        for item in media:
            print(f"  - {item['title']['rendered']} (类型: {item['media_type']})")
    
    print("\n✅ 基本操作演示完成!")

def demo_create_content():
    """演示内容创建"""
    print("\n🎨 WordPress 内容创建演示\n")
    
    wp = WordPressAPI(WP_URL, WP_USER, WP_APP_PASSWORD)
    
    # 创建测试分类
    print("📂 创建测试分类...")
    test_category = wp.create_category(
        name="API测试分类",
        description="这是通过API创建的测试分类"
    )
    
    if test_category:
        category_id = test_category['id']
        print(f"✅ 分类创建成功 (ID: {category_id})")
        
        # 创建测试标签
        print("\n🏷️ 创建测试标签...")
        test_tag = wp.create_tag(
            name="API测试",
            description="API测试标签"
        )
        
        if test_tag:
            tag_id = test_tag['id']
            print(f"✅ 标签创建成功 (ID: {tag_id})")
            
            # 创建测试文章
            print("\n📝 创建测试文章...")
            current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            test_post = wp.create_post(
                title=f"API测试文章 - {current_time}",
                content=f"""
                <h2>这是通过WordPress REST API创建的测试文章</h2>
                <p>创建时间: {current_time}</p>
                <p>这篇文章演示了如何使用WordPress REST API进行内容管理:</p>
                <ul>
                    <li>创建文章</li>
                    <li>设置分类和标签</li>
                    <li>使用HTML格式的内容</li>
                </ul>
                <p><strong>注意:</strong> 这是一个测试文章，可以安全删除。</p>
                """,
                status='draft',  # 草稿状态
                categories=[category_id],
                tags=[tag_id]
            )
            
            if test_post:
                post_id = test_post['id']
                print(f"✅ 文章创建成功 (ID: {post_id})")
                print(f"   标题: {test_post['title']['rendered']}")
                print(f"   状态: {test_post['status']}")
                print(f"   链接: {test_post['link']}")
                
                # 询问是否删除测试内容
                print("\n🗑️ 清理测试内容...")
                
                # 删除测试文章
                if wp.delete_post(post_id, force=True):
                    print(f"✅ 测试文章已删除 (ID: {post_id})")
                
                # 删除测试分类和标签需要小心，因为可能有其他内容在使用
                print("⚠️ 测试分类和标签保留，请手动删除如果不需要")
                
            else:
                print("❌ 文章创建失败")
        else:
            print("❌ 标签创建失败")
    else:
        print("❌ 分类创建失败")
    
    print("\n✅ 内容创建演示完成!")

def main():
    """主函数"""
    print("WordPress REST API 使用示例")
    print("=" * 40)
    
    # 基本操作演示
    demo_basic_operations()
    
    # 内容创建演示
    demo_create_content()
    
    print("\n🎉 所有演示完成!")
    print("\n💡 提示:")
    print("- 所有创建的测试内容都已清理")
    print("- 你可以修改这个脚本来适应你的具体需求")
    print("- 更多API文档: https://developer.wordpress.org/rest-api/")

if __name__ == "__main__":
    main()