Spaces:
Sleeping
Sleeping
File size: 4,807 Bytes
2197ab7 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 |
"""File processing module for fabric-to-espanso."""
from pathlib import Path
from typing import List, Dict, Any, Optional
from datetime import datetime
import logging
from .markdown_parser import parse_markdown_file
from .exceptions import ProcessingError
logger = logging.getLogger('fabric_to_espanso')
def find_markdown_files(
root_dir: Path,
max_depth: int = 2,
pattern: str = "*.md"
) -> List[Path]:
"""Find markdown files in directory up to specified depth.
Args:
root_dir: Root directory to search in
max_depth: Maximum directory depth to search
pattern: Glob pattern for files to find
Returns:
List of paths to markdown files
Raises:
ValueError: If root_dir doesn't exist or isn't a directory
"""
if not root_dir.exists():
raise ValueError(f"Directory does not exist: {root_dir}")
if not root_dir.is_dir():
raise ValueError(f"Path is not a directory: {root_dir}")
files: List[Path] = []
try:
# Convert depth to parts for comparison
root_parts = len(root_dir.parts)
# Use rglob to find all markdown files
for file_path in root_dir.rglob(pattern):
# Skip if too deep
if len(file_path.parts) - root_parts > max_depth:
continue
# Skip README.md files
if file_path.name.lower() == "readme.md":
continue
if file_path.name.lower() == "user.md":
continue
if file_path.is_file():
files.append(file_path)
logger.debug(f"Found {len(files)} markdown files in {root_dir}")
return files
except Exception as e:
logger.error(f"Error finding markdown files: {str(e)}", exc_info=True)
raise ProcessingError(f"Failed to find markdown files: {str(e)}") from e
def process_markdown_file(
file_path: Path,
trigger_prefix: str
) -> Optional[Dict[str, Any]]:
"""Process a single markdown file.
Args:
file_path: Path to markdown file
trigger_prefix: Prefix for espanso triggers
Returns:
Dictionary with file information or None if processing fails
Raises:
ProcessingError: If file processing fails
"""
try:
content, extracted_sections = parse_markdown_file(str(file_path))
if extracted_sections is None:
logger.warning(f"No sections extracted from {file_path}")
extracted_sections = content
return {
'filename': file_path.parent.name,
'content': content,
'purpose': extracted_sections,
'last_modified': datetime.fromtimestamp(file_path.stat().st_mtime),
'filesize': file_path.stat().st_size,
'trigger': trigger_prefix,
'label': file_path.stem # filename without extension
}
except Exception as e:
logger.error(f"Error processing {file_path}: {str(e)}", exc_info=True)
raise ProcessingError(f"Failed to process {file_path}: {str(e)}") from e
def process_markdown_files(
markdown_folder: Path | str,
# TODO: make 'max_depth' a parameter
max_depth: int = 2,
trigger_prefix: str = ";;fab"
) -> List[Dict[str, Any]]:
"""Process all markdown files in directory.
Args:
markdown_folder: Directory containing markdown files
max_depth: Maximum directory depth to search
trigger_prefix: Prefix for espanso triggers
Returns:
List of processed file information
Raises:
ProcessingError: If processing fails
ValueError: If markdown_folder is invalid
"""
root_dir = Path(markdown_folder)
processed_files: List[Dict[str, Any]] = []
try:
# Find all markdown files
markdown_files = find_markdown_files(root_dir, max_depth)
# Process each file
for file_path in markdown_files:
try:
if result := process_markdown_file(file_path, trigger_prefix):
processed_files.append(result)
logger.info(f"Processed: {file_path.parent.name}")
except ProcessingError as e:
logger.error(str(e))
continue
logger.info(f"Successfully processed {len(processed_files)} files in fabric patterns folder")
return processed_files
except Exception as e:
logger.error(f"Error processing markdown files: {str(e)}", exc_info=True)
if isinstance(e, (ProcessingError, ValueError)):
raise
raise ProcessingError(f"Unexpected error processing files: {str(e)}") from e |