frontmatter_writer.py•9.2 kB
"""Frontmatter writer for safely updating poem metadata.
Provides atomic write operations with backup and rollback capabilities.
"""
import shutil
import tempfile
from pathlib import Path
from typing import Any, Optional
from datetime import datetime
import yaml
from pydantic import BaseModel
from ..errors import FrontmatterParseError
class FrontmatterUpdateResult(BaseModel):
"""Result of a frontmatter update operation."""
success: bool
file_path: str
backup_path: Optional[str] = None
tags_added: list[str] = []
tags_removed: list[str] = []
error: Optional[str] = None
def extract_frontmatter_and_content(content: str, file_path: Path) -> tuple[dict[str, Any], str]:
"""Extract YAML frontmatter and content from markdown file.
Args:
content: Full markdown file content
file_path: Path to file (for error messages)
Returns:
Tuple of (frontmatter_dict, content_body)
Raises:
FrontmatterParseError: If frontmatter is malformed
"""
lines = content.split('\n')
# Check for frontmatter delimiter
if not lines or lines[0].strip() != '---':
# No frontmatter - return empty dict and full content
return {}, content
# Find closing delimiter
try:
end_idx = lines[1:].index('---') + 1
except ValueError:
raise FrontmatterParseError(
f"Unclosed frontmatter in {file_path}: missing closing '---'"
)
# Extract frontmatter YAML
frontmatter_lines = lines[1:end_idx]
frontmatter_text = '\n'.join(frontmatter_lines)
try:
frontmatter = yaml.safe_load(frontmatter_text) or {}
except yaml.YAMLError as e:
raise FrontmatterParseError(
f"Invalid YAML in frontmatter of {file_path}: {e}"
)
# Extract content (everything after closing ---)
content_body = '\n'.join(lines[end_idx + 1:])
return frontmatter, content_body
def serialize_frontmatter_and_content(frontmatter: dict[str, Any], content: str) -> str:
"""Serialize frontmatter dict and content back to markdown format.
Args:
frontmatter: Dictionary of frontmatter fields
content: Markdown content body
Returns:
Complete markdown file content with frontmatter
"""
if not frontmatter:
# No frontmatter - return content as-is
return content
# Serialize frontmatter to YAML
yaml_text = yaml.dump(
frontmatter,
default_flow_style=False,
allow_unicode=True,
sort_keys=False, # Preserve field order
).strip()
# Combine with delimiters and content
return f"---\n{yaml_text}\n---\n{content}"
def create_backup(file_path: Path) -> Path:
"""Create a backup of the file before modification.
Args:
file_path: Path to file to back up
Returns:
Path to backup file
"""
backup_path = file_path.with_suffix(file_path.suffix + '.bak')
shutil.copy2(file_path, backup_path)
return backup_path
def atomic_write(file_path: Path, content: str) -> None:
"""Write content to file atomically using temp file + rename.
Args:
file_path: Destination file path
content: Content to write
Raises:
OSError: If write fails
"""
# Create temp file in same directory (ensures same filesystem)
temp_fd, temp_path = tempfile.mkstemp(
dir=file_path.parent,
prefix='.tmp_',
suffix=file_path.suffix,
)
try:
# Write to temp file
with open(temp_fd, 'w', encoding='utf-8') as f:
f.write(content)
# Atomic rename
Path(temp_path).replace(file_path)
except Exception:
# Clean up temp file on error
try:
Path(temp_path).unlink()
except Exception:
pass
raise
def update_poem_frontmatter(
file_path: Path,
updates: dict[str, Any],
create_backup_file: bool = True,
) -> FrontmatterUpdateResult:
"""Update arbitrary frontmatter fields in a poem file.
Args:
file_path: Path to poem markdown file
updates: Dictionary of field updates to apply
create_backup_file: Whether to create .bak file before updating
Returns:
FrontmatterUpdateResult with operation details
"""
result = FrontmatterUpdateResult(
success=False,
file_path=str(file_path),
)
try:
# Read current file
if not file_path.exists():
result.error = f"File not found: {file_path}"
return result
content = file_path.read_text(encoding='utf-8')
# Extract frontmatter and content
frontmatter, content_body = extract_frontmatter_and_content(content, file_path)
# Create backup if requested
backup_path = None
if create_backup_file:
backup_path = create_backup(file_path)
result.backup_path = str(backup_path)
# Apply updates
frontmatter.update(updates)
# Update timestamp
frontmatter['updated_at'] = datetime.now().isoformat()
# Serialize back to markdown
new_content = serialize_frontmatter_and_content(frontmatter, content_body)
# Validate YAML before writing
try:
# Re-parse to validate
extract_frontmatter_and_content(new_content, file_path)
except Exception as e:
result.error = f"YAML validation failed: {e}"
return result
# Atomic write
atomic_write(file_path, new_content)
result.success = True
return result
except Exception as e:
result.error = str(e)
return result
def update_poem_tags(
file_path: Path,
tags_to_add: Optional[list[str]] = None,
tags_to_remove: Optional[list[str]] = None,
create_backup_file: bool = True,
) -> FrontmatterUpdateResult:
"""Update tags in a poem's frontmatter.
Safely adds and removes tags while preserving all other frontmatter fields.
Uses atomic writes with optional backup for safety.
Args:
file_path: Path to poem markdown file
tags_to_add: List of tags to add (will be deduplicated)
tags_to_remove: List of tags to remove
create_backup_file: Whether to create .bak file before updating
Returns:
FrontmatterUpdateResult with operation details
Example:
>>> result = update_poem_tags(
... Path("poem.md"),
... tags_to_add=["water", "childhood"],
... tags_to_remove=["draft"],
... )
>>> result.success
True
>>> result.tags_added
['water', 'childhood']
"""
tags_to_add = tags_to_add or []
tags_to_remove = tags_to_remove or []
result = FrontmatterUpdateResult(
success=False,
file_path=str(file_path),
)
try:
# Read current file
if not file_path.exists():
result.error = f"File not found: {file_path}"
return result
content = file_path.read_text(encoding='utf-8')
# Extract frontmatter and content
frontmatter, content_body = extract_frontmatter_and_content(content, file_path)
# Create backup if requested
backup_path = None
if create_backup_file:
backup_path = create_backup(file_path)
result.backup_path = str(backup_path)
# Get current tags (handle missing field)
current_tags = set(frontmatter.get('tags', []))
# Add new tags
added = []
for tag in tags_to_add:
if tag not in current_tags:
current_tags.add(tag)
added.append(tag)
# Remove tags
removed = []
for tag in tags_to_remove:
if tag in current_tags:
current_tags.remove(tag)
removed.append(tag)
# Update frontmatter
frontmatter['tags'] = sorted(current_tags) # Sort for consistency
frontmatter['updated_at'] = datetime.now().isoformat()
result.tags_added = added
result.tags_removed = removed
# Serialize back to markdown
new_content = serialize_frontmatter_and_content(frontmatter, content_body)
# Validate YAML before writing
try:
# Re-parse to validate
extract_frontmatter_and_content(new_content, file_path)
except Exception as e:
result.error = f"YAML validation failed: {e}"
return result
# Atomic write
atomic_write(file_path, new_content)
result.success = True
return result
except Exception as e:
result.error = str(e)
return result
def rollback_from_backup(file_path: Path) -> bool:
"""Restore a file from its .bak backup.
Args:
file_path: Path to file to restore
Returns:
True if rollback succeeded, False otherwise
"""
backup_path = file_path.with_suffix(file_path.suffix + '.bak')
if not backup_path.exists():
return False
try:
shutil.copy2(backup_path, file_path)
return True
except Exception:
return False