#!/usr/bin/env python3 """ Auto-Publish to Astro Content Collections Publishes blog posts to Astro content collections, commits to git, and triggers auto-deploy. """ import os import sys import subprocess import argparse import re from pathlib import Path from datetime import datetime from typing import Dict, Optional class AstroPublisher: """Publish blog posts to Astro content collections""" def __init__(self, website_repo: str): """ Initialize Astro publisher Args: website_repo: Path to Astro website repository """ self.website_repo = website_repo self.content_dir = os.path.join(website_repo, 'src/content/blog') self.images_dir = os.path.join(website_repo, 'public/images/blog') def detect_language(self, content: str) -> str: """Detect if content is Thai or English""" thai_chars = sum(1 for c in content if '\u0E00' <= c <= '\u0E7F') total_chars = len(content) thai_ratio = thai_chars / total_chars if total_chars > 0 else 0 return 'th' if thai_ratio > 0.3 else 'en' def generate_slug(self, title: str, lang: str = 'en') -> str: """Generate URL-friendly slug""" # Remove special characters slug = re.sub(r'[^\w\s-]', '', title.lower()) # Replace whitespace with hyphens slug = re.sub(r'[-\s]+', '-', slug) # Remove leading/trailing hyphens slug = slug.strip('-_') # Limit length return slug[:100] def parse_frontmatter(self, content: str) -> Dict: """Parse frontmatter from markdown content""" import yaml if not content.startswith('---'): return {} try: # Extract frontmatter parts = content.split('---', 2) if len(parts) >= 2: frontmatter = yaml.safe_load(parts[1]) return frontmatter or {} except: pass return {} def publish(self, markdown_content: str, images: list = None, use_git: bool = False) -> Dict: """ Publish blog post to Astro content collections Args: markdown_content: Full markdown with frontmatter images: List of image paths to copy use_git: Whether to git commit and push (default: False - direct write only) Returns: Publication result """ try: # Parse frontmatter frontmatter = self.parse_frontmatter(markdown_content) # Get required fields title = frontmatter.get('title', 'Untitled') slug = frontmatter.get('slug') or self.generate_slug(title) lang = frontmatter.get('lang') or self.detect_language(markdown_content) # Determine output path lang_folder = f'({lang})' output_dir = os.path.join(self.content_dir, lang_folder) os.makedirs(output_dir, exist_ok=True) output_path = os.path.join(output_dir, f'{slug}.md') # Write markdown file (ALWAYS do this) with open(output_path, 'w', encoding='utf-8') as f: f.write(markdown_content) print(f"\n✓ Saved: {output_path}") # Copy images if provided if images: images_output = os.path.join(self.images_dir, slug) os.makedirs(images_output, exist_ok=True) for img_path in images: if os.path.exists(img_path): import shutil shutil.copy(img_path, images_output) print(f" ✓ Copied image: {os.path.basename(img_path)}") # Git commit and push (OPTIONAL - only if requested and Gitea configured) git_result = None if use_git: git_result = self.git_commit_and_push(slug, lang) else: print(f" ✓ Direct write complete (no git)") return { 'success': True, 'slug': slug, 'language': lang, 'path': output_path, 'git_result': git_result, 'method': 'direct_write' if not use_git else 'git_push' } except Exception as e: return { 'success': False, 'error': str(e) } def git_commit_and_push(self, slug: str, lang: str) -> Dict: """Commit and push changes to git""" try: # Check if git repo if not os.path.exists(os.path.join(self.website_repo, '.git')): return {'success': False, 'error': 'Not a git repository'} # Git add subprocess.run(['git', 'add', '.'], cwd=self.website_repo, check=True, capture_output=True) # Git commit message = f"Add blog post: {slug} ({lang})" subprocess.run(['git', 'commit', '-m', message], cwd=self.website_repo, check=True, capture_output=True) # Git push subprocess.run(['git', 'push'], cwd=self.website_repo, check=True, capture_output=True) print(f"✓ Committed: {message}") print(f"✓ Pushed to remote") return { 'success': True, 'commit_message': message, 'triggered_deploy': True } except subprocess.CalledProcessError as e: print(f"✗ Git error: {e.stderr.decode() if e.stderr else str(e)}") return {'success': False, 'error': 'Git operation failed'} except Exception as e: print(f"✗ Error: {e}") return {'success': False, 'error': str(e)} def main(): """Test Astro publisher""" parser = argparse.ArgumentParser(description='Publish to Astro') parser.add_argument('--file', required=True, help='Markdown file to publish') parser.add_argument('--website-repo', required=True, help='Path to website repo') parser.add_argument('--image', action='append', help='Image files to copy') parser.add_argument('--use-git', action='store_true', help='Use git commit/push (default: direct write only)') args = parser.parse_args() print(f"\n📝 Publishing to Astro\n") # Read markdown file with open(args.file, 'r', encoding='utf-8') as f: content = f.read() # Publish (default: direct write, no git) publisher = AstroPublisher(args.website_repo) result = publisher.publish(content, args.image, use_git=args.use_git) if result['success']: print(f"\n✅ Published successfully!") print(f" Slug: {result['slug']}") print(f" Language: {result['language']}") print(f" Path: {result['path']}") print(f" Method: {result['method']}") if result.get('git_result') and result['git_result'].get('success'): print(f" ✓ Committed and pushed to Gitea") print(f" ✓ Deployment triggered") else: print(f"\n❌ Publication failed: {result.get('error')}") if __name__ == '__main__': main()