wxr_to_csv.py

#!/usr/bin/env python3
"""
WordPress eXtended RSS (WXR) to CSV Converter

This script converts WordPress export files (WXR format) to CSV files.
It extracts posts, pages, and their metadata into a structured CSV format.
"""

import defusedxml.ElementTree as ET
import csv
import argparse
import html
import re
from datetime import datetime
from pathlib import Path


class WXRToCSVConverter:
    """Convert WordPress WXR files to CSV format."""
    
    def __init__(self):
        self.namespaces = {
            'content': 'http://purl.org/rss/1.0/modules/content/',
            'wfw': 'http://wellformedweb.org/CommentAPI/',
            'dc': 'http://purl.org/dc/elements/1.1/',
            'wp': 'http://wordpress.org/export/1.2/'
        }
    
    def clean_text(self, text):
        """Clean and normalize text content."""
        if not text:
            return ""
        
        # Decode HTML entities
        text = html.unescape(text)
        
        # Remove excessive whitespace
        text = re.sub(r'\s+', ' ', text)
        
        # Strip leading/trailing whitespace
        text = text.strip()
        
        return text
    
    def extract_post_data(self, item):
        """Extract data from a WordPress post/page item."""
        data = {}
        
        # Basic post information
        data['title'] = self.clean_text(item.find('title').text if item.find('title') is not None else "")
        data['link'] = item.find('link').text if item.find('link') is not None else ""
        data['pub_date'] = item.find('pubDate').text if item.find('pubDate') is not None else ""
        data['creator'] = item.find('dc:creator', self.namespaces).text if item.find('dc:creator', self.namespaces) is not None else ""
        data['description'] = self.clean_text(item.find('description').text if item.find('description') is not None else "")
        
        # WordPress-specific fields
        data['post_id'] = item.find('wp:post_id', self.namespaces).text if item.find('wp:post_id', self.namespaces) is not None else ""
        data['post_date'] = item.find('wp:post_date', self.namespaces).text if item.find('wp:post_date', self.namespaces) is not None else ""
        data['post_date_gmt'] = item.find('wp:post_date_gmt', self.namespaces).text if item.find('wp:post_date_gmt', self.namespaces) is not None else ""
        data['post_modified'] = item.find('wp:post_modified', self.namespaces).text if item.find('wp:post_modified', self.namespaces) is not None else ""
        data['post_modified_gmt'] = item.find('wp:post_modified_gmt', self.namespaces).text if item.find('wp:post_modified_gmt', self.namespaces) is not None else ""
        data['comment_status'] = item.find('wp:comment_status', self.namespaces).text if item.find('wp:comment_status', self.namespaces) is not None else ""
        data['ping_status'] = item.find('wp:ping_status', self.namespaces).text if item.find('wp:ping_status', self.namespaces) is not None else ""
        data['post_name'] = item.find('wp:post_name', self.namespaces).text if item.find('wp:post_name', self.namespaces) is not None else ""
        data['status'] = item.find('wp:status', self.namespaces).text if item.find('wp:status', self.namespaces) is not None else ""
        data['post_parent'] = item.find('wp:post_parent', self.namespaces).text if item.find('wp:post_parent', self.namespaces) is not None else ""
        data['menu_order'] = item.find('wp:menu_order', self.namespaces).text if item.find('wp:menu_order', self.namespaces) is not None else ""
        data['post_type'] = item.find('wp:post_type', self.namespaces).text if item.find('wp:post_type', self.namespaces) is not None else ""
        data['post_password'] = item.find('wp:post_password', self.namespaces).text if item.find('wp:post_password', self.namespaces) is not None else ""
        data['is_sticky'] = item.find('wp:is_sticky', self.namespaces).text if item.find('wp:is_sticky', self.namespaces) is not None else ""
        
        # Content
        content_elem = item.find('content:encoded', self.namespaces)
        data['content'] = self.clean_text(content_elem.text if content_elem is not None and content_elem.text else "")
        
        # Excerpt - try different possible elements
        excerpt_elem = None
        # Try wp:post_excerpt first (most common)
        excerpt_elem = item.find('wp:post_excerpt', self.namespaces)
        if excerpt_elem is None:
            # Try excerpt without namespace
            excerpt_elem = item.find('excerpt')
        
        data['excerpt'] = self.clean_text(excerpt_elem.text if excerpt_elem is not None and excerpt_elem.text else "")
        
        # Categories and tags
        categories = []
        tags = []
        
        for category in item.findall('category'):
            if category.get('domain') == 'category':
                categories.append(category.text or "")
            elif category.get('domain') == 'post_tag':
                tags.append(category.text or "")
        
        data['categories'] = "; ".join(categories)
        data['tags'] = "; ".join(tags)
        
        # Custom fields (post meta)
        custom_fields = {}
        for postmeta in item.findall('wp:postmeta', self.namespaces):
            meta_key = postmeta.find('wp:meta_key', self.namespaces)
            meta_value = postmeta.find('wp:meta_value', self.namespaces)
            
            if meta_key is not None and meta_value is not None:
                key = meta_key.text or ""
                value = meta_value.text or ""
                if key and not key.startswith('_'):  # Skip private meta fields
                    custom_fields[key] = value
        
        # Convert custom fields to a readable format
        custom_fields_str = "; ".join([f"{k}: {v}" for k, v in custom_fields.items()])
        data['custom_fields'] = custom_fields_str
        
        return data
    
    def convert_to_csv(self, wxr_file_path, csv_file_path=None, post_types=None):
        """Convert WXR file to CSV."""
        if post_types is None:
            post_types = ['post', 'page']
        
        # Parse the WXR file
        try:
            tree = ET.parse(wxr_file_path)
            root = tree.getroot()
        except ET.ParseError as e:
            raise ValueError(f"Error parsing WXR file: {e}")
        
        # Ensure root was obtained
        if root is None:
            raise ValueError("Error: parsed XML has no root element (empty or invalid WXR file).")
        
        # Find all items (posts, pages, etc.)
        items = root.findall('.//item')
        
        posts_data = []
        
        for item in items:
            # Check if this item is a post type we want to include
            post_type_elem = item.find('wp:post_type', self.namespaces)
            if post_type_elem is not None:
                post_type = post_type_elem.text
                if post_type in post_types:
                    post_data = self.extract_post_data(item)
                    posts_data.append(post_data)
        
        if not posts_data:
            print(f"No posts found with post types: {post_types}")
            return
        
        # Determine output file path
        if csv_file_path is None:
            wxr_path = Path(wxr_file_path)
            csv_file_path = wxr_path.with_suffix('.csv')
        
        # Define CSV columns
        columns = [
            'post_id', 'title', 'post_type', 'status', 'post_date', 'post_modified',
            'creator', 'link', 'post_name', 'description', 'content', 'excerpt',
            'categories', 'tags', 'comment_status', 'ping_status', 'post_parent',
            'menu_order', 'is_sticky', 'post_password', 'custom_fields',
            'pub_date', 'post_date_gmt', 'post_modified_gmt'
        ]
        
        # Write to CSV
        with open(csv_file_path, 'w', newline='', encoding='utf-8') as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=columns)
            writer.writeheader()
            
            for post in posts_data:
                # Ensure all columns are present
                row = {col: post.get(col, '') for col in columns}
                writer.writerow(row)
        
        print(f"Successfully converted {len(posts_data)} items to {csv_file_path}")
        return csv_file_path


def main():
    """Main function for command-line usage."""
    parser = argparse.ArgumentParser(
        description='Convert WordPress eXtended RSS (WXR) files to CSV format'
    )
    parser.add_argument(
        'input_file',
        help='Path to the WXR file to convert'
    )
    parser.add_argument(
        '-o', '--output',
        help='Output CSV file path (default: same name as input with .csv extension)'
    )
    parser.add_argument(
        '-t', '--types',
        nargs='+',
        default=['post', 'page'],
        help='Post types to include (default: post page)'
    )
    
    args = parser.parse_args()
    
    # Check if input file exists
    if not Path(args.input_file).exists():
        print(f"Error: Input file '{args.input_file}' not found.")
        return 1
    
    try:
        converter = WXRToCSVConverter()
        converter.convert_to_csv(args.input_file, args.output, args.types)
        return 0
    except Exception as e:
        print(f"Error: {e}")
        return 1


if __name__ == '__main__':
    exit(main())