""" Main frontmatter processor """ import sys from pathlib import Path from datetime import datetime from core.database import Database from core.repository import Repository from core.frontmatter import FrontmatterParser from core.rules import RulesEngine from core.template import TemplateManager from core.llm import LLMClient from core.changelog import ChangeLog from core.config import ConfigManager class FrontmatterProcessor: def __init__(self, repo: Repository, template_name: str, whatif: bool = False, no_confirm: bool = False, force: bool = False, add_only: bool = False): self.repo = repo self.template_name = template_name self.whatif = whatif self.no_confirm = no_confirm self.force = force self.add_only = add_only # Initialize configuration self.config = ConfigManager() self.config.ensure_config_exists() # Get API configuration api_config = self.config.get_api_config() self.db = Database(repo.path / 'madomeda.db') self.rules = RulesEngine(self.config.rules_dir) self.templates = TemplateManager(self.config.templates_dir) self.llm = LLMClient(self.config.prompts_dir, api_config) self.changelog = ChangeLog(repo.path) self.parser = FrontmatterParser() def process(self): """Main processing loop""" # Print configuration info if running in interactive shell if sys.stdin.isatty(): self.config.print_config_info() print(f"Processing repository: {self.repo.path}") print(f"Template: {self.template_name}") print(f"Mode: {'DRY RUN' if self.whatif else 'LIVE'}") if self.add_only: print(f"Add-only mode: Preserving existing keys") print() # Get template template = self.templates.load_template(self.template_name) schema = self.templates.get_structured_output_schema(template) # Get all markdown files files = self.repo.get_tracked_files() print(f"Found {len(files)} markdown files") print() # Start changelog session self.changelog.start_session() # Process each file for file_path in files: self._process_file(file_path, template, schema) # Collect all tags and store in database self._collect_all_tags() # Write changelog if not self.whatif: self.changelog.write() self.db.close() def _process_file(self, file_path: Path, template: dict, schema: dict): """Process a single markdown file""" relative_path = file_path.relative_to(self.repo.path) print(f"Processing: {relative_path}") # Add file to database file_id = self.db.add_file(str(relative_path)) # Read file content with open(file_path, 'r', encoding='utf-8') as f: content = f.read() # Parse frontmatter original_fm, body = self.parser.parse(content) if original_fm: print(f" Found existing frontmatter") else: print(f" No frontmatter found") original_fm = {} # Store original frontmatter in database if original_fm: self.db.add_frontmatter(file_id, original_fm, conformant=False) # Get git history git_info = self.repo.get_file_history(file_path) self.db.add_commit_info( file_id, git_info['author_name'], git_info['author_email'], git_info['commit_hash'], git_info['commit_tag'], git_info['latest_hash'] ) # Apply rules normalized_fm, is_conformant = self.rules.apply_rules(dict(original_fm)) # Check if we need to update needs_update = self.force or not is_conformant or not self._matches_template(normalized_fm, template) if not needs_update: print(f" OK Already conformant") if original_fm: # Mark as conformant in database fm_id = self.db.add_frontmatter(file_id, original_fm, conformant=True) print() return # Build new frontmatter from template new_fm = self._build_frontmatter(template, schema, original_fm, normalized_fm, body, file_path, git_info) # In add-only mode, merge with normalized original if self.add_only: # Start with normalized original merged_fm = dict(normalized_fm) # Add missing template fields for key, value in new_fm.items(): if key not in merged_fm: merged_fm[key] = value new_fm = merged_fm # Store tags in database if 'tags' in new_fm and isinstance(new_fm['tags'], list): for tag in new_fm['tags']: self.db.add_tag(tag) # Check for metadata loss discarded = self._check_discarded_metadata(original_fm, new_fm) if discarded and not self.add_only: # Only warn in non-add-only mode print(f" WARNING Metadata will be discarded: {', '.join(discarded)}") if not self.no_confirm and not self.whatif: response = input(" Continue? (y/n): ") if response.lower() != 'y': print(f" Skipped") print() return # Show what would change if self.whatif: print(f" Would update frontmatter:") print(f" Changes: {self._describe_changes(original_fm, new_fm)}") else: # Write new frontmatter new_content = self.parser.serialize(new_fm, body) with open(file_path, 'w', encoding='utf-8') as f: f.write(new_content) # Update database self.db.add_frontmatter(file_id, new_fm, conformant=True) # Add to changelog self.changelog.add_entry(str(relative_path)) print(f" OK Updated frontmatter") print() def _matches_template(self, frontmatter: dict, template: dict) -> bool: """Check if frontmatter has all required template fields""" for field in template.keys(): if field not in frontmatter: return False # Allow empty lists/strings as valid values return True def _build_frontmatter(self, template: dict, schema: dict, original_fm: dict, normalized_fm: dict, body: str, file_path: Path, git_info: dict) -> dict: """Build new frontmatter from template""" new_fm = {} all_tags = self.db.get_all_tags() for field, strategy in template.items(): options = strategy.split('|') value = None for option in options: option = option.strip() if option == 'heur': # Use heuristics value = self._get_heuristic_value(field, normalized_fm, body, file_path, git_info) elif option == 'orig': # Use original value value = normalized_fm.get(field) elif option.startswith('ai'): # Use LLM parts = option.split(None, 1) prompt_name = parts[1] if len(parts) > 1 else 'default' value = self.llm.infer_metadata(field, prompt_name, original_fm, new_fm, body, all_tags, schema) # Accept value if not None (empty lists/strings are valid) if value is not None: break if value is not None: new_fm[field] = value return new_fm def _get_heuristic_value(self, field: str, normalized_fm: dict, body: str, file_path: Path, git_info: dict): """Get field value using heuristics""" if field == 'title': # Try to extract from document title = self.parser.extract_title(body) if not title: # Use filename without extension title = file_path.stem return title elif field == 'created': return git_info.get('created') elif field == 'changed': return git_info.get('changed') elif field == 'authors': return git_info.get('authors', []) elif field == 'version': # Prefer latest tag, fallback to short hash latest_tag = git_info.get('latest_tag', '') if latest_tag: return latest_tag latest_hash = git_info.get('latest_hash', '') return latest_hash[:7] if latest_hash else '' elif field == 'tags': tags = normalized_fm.get('tags', []) # Return None if empty so AI inference is tried return tags if tags else None return None def _check_discarded_metadata(self, original: dict, new: dict) -> list: """Check for metadata that will be discarded""" discarded = [] for key in original.keys(): if key not in new: discarded.append(key) return discarded def _describe_changes(self, original: dict, new: dict) -> str: """Describe changes between frontmatter versions""" changes = [] # New fields for key in new.keys(): if key not in original: changes.append(f"+{key}") # Modified fields for key in new.keys(): if key in original and original[key] != new[key]: changes.append(f"~{key}") # Removed fields for key in original.keys(): if key not in new: changes.append(f"-{key}") return ', '.join(changes) if changes else 'none' def _collect_all_tags(self): """Collect all unique tags from processed files and store in database""" # Tags are collected during file processing all_tags = self.db.get_all_tags() print(f"Total unique tags in repository: {len(all_tags)}")