Remove test file
This commit is contained in:
Test User
@@ -0,0 +1 @@
|
||||
"""Core module initialization"""
|
||||
@@ -0,0 +1,45 @@
|
||||
"""
|
||||
Changelog management
|
||||
"""
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
import subprocess
|
||||
|
||||
|
||||
class ChangeLog:
|
||||
def __init__(self, repo_path: Path):
|
||||
self.changelog_path = repo_path / 'madomeda_changelog.txt'
|
||||
self.current_session = None
|
||||
|
||||
def start_session(self):
|
||||
"""Start a new changelog session"""
|
||||
# Get current commit hash
|
||||
result = subprocess.run(
|
||||
['git', 'rev-parse', 'HEAD'],
|
||||
cwd=self.changelog_path.parent,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
commit_hash = result.stdout.strip() if result.returncode == 0 else 'no-commit'
|
||||
|
||||
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||||
self.current_session = f"{timestamp} (commit: {commit_hash})\n"
|
||||
self.entries = []
|
||||
|
||||
def add_entry(self, file_path: str, message: str = "frontmatter updated"):
|
||||
"""Add a changelog entry"""
|
||||
if self.current_session:
|
||||
self.entries.append(f" {file_path} - {message}")
|
||||
|
||||
def write(self):
|
||||
"""Write changelog to file"""
|
||||
if not self.current_session or not self.entries:
|
||||
return
|
||||
|
||||
content = self.current_session
|
||||
content += '\n'.join(self.entries)
|
||||
content += '\n\n'
|
||||
|
||||
# Append to changelog file
|
||||
with open(self.changelog_path, 'a', encoding='utf-8') as f:
|
||||
f.write(content)
|
||||
@@ -0,0 +1,113 @@
|
||||
"""
|
||||
Database management module
|
||||
"""
|
||||
import sqlite3
|
||||
import json
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class Database:
|
||||
def __init__(self, db_path: Path):
|
||||
self.db_path = db_path
|
||||
self.conn = sqlite3.connect(db_path)
|
||||
self.conn.row_factory = sqlite3.Row
|
||||
self._init_schema()
|
||||
|
||||
def _init_schema(self):
|
||||
cursor = self.conn.cursor()
|
||||
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS files (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
path TEXT UNIQUE NOT NULL,
|
||||
discovered_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
""")
|
||||
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS frontmatter (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
file_id INTEGER NOT NULL,
|
||||
content TEXT NOT NULL,
|
||||
read_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
conformant INTEGER DEFAULT 0,
|
||||
FOREIGN KEY (file_id) REFERENCES files(id)
|
||||
)
|
||||
""")
|
||||
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS commits (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
file_id INTEGER NOT NULL,
|
||||
author_name TEXT,
|
||||
author_email TEXT,
|
||||
commit_hash TEXT,
|
||||
commit_tag TEXT,
|
||||
latest_hash TEXT,
|
||||
FOREIGN KEY (file_id) REFERENCES files(id)
|
||||
)
|
||||
""")
|
||||
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS tags (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
tag TEXT UNIQUE NOT NULL
|
||||
)
|
||||
""")
|
||||
|
||||
self.conn.commit()
|
||||
|
||||
def add_file(self, path: str) -> int:
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute("INSERT OR IGNORE INTO files (path) VALUES (?)", (path,))
|
||||
self.conn.commit()
|
||||
cursor.execute("SELECT id FROM files WHERE path = ?", (path,))
|
||||
return cursor.fetchone()[0]
|
||||
|
||||
def add_frontmatter(self, file_id: int, content: dict, conformant: bool = False) -> int:
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute(
|
||||
"INSERT INTO frontmatter (file_id, content, conformant) VALUES (?, ?, ?)",
|
||||
(file_id, json.dumps(content), 1 if conformant else 0)
|
||||
)
|
||||
self.conn.commit()
|
||||
return cursor.lastrowid
|
||||
|
||||
def update_frontmatter_conformance(self, fm_id: int, conformant: bool):
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute(
|
||||
"UPDATE frontmatter SET conformant = ? WHERE id = ?",
|
||||
(1 if conformant else 0, fm_id)
|
||||
)
|
||||
self.conn.commit()
|
||||
|
||||
def add_commit_info(self, file_id: int, author_name: str, author_email: str,
|
||||
commit_hash: str, commit_tag: str, latest_hash: str):
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute(
|
||||
"""INSERT OR REPLACE INTO commits
|
||||
(file_id, author_name, author_email, commit_hash, commit_tag, latest_hash)
|
||||
VALUES (?, ?, ?, ?, ?, ?)""",
|
||||
(file_id, author_name, author_email, commit_hash, commit_tag, latest_hash)
|
||||
)
|
||||
self.conn.commit()
|
||||
|
||||
def add_tag(self, tag: str):
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute("INSERT OR IGNORE INTO tags (tag) VALUES (?)", (tag,))
|
||||
self.conn.commit()
|
||||
|
||||
def get_all_tags(self) -> list:
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute("SELECT tag FROM tags ORDER BY tag")
|
||||
return [row[0] for row in cursor.fetchall()]
|
||||
|
||||
def get_file_id(self, path: str) -> int:
|
||||
cursor = self.conn.cursor()
|
||||
cursor.execute("SELECT id FROM files WHERE path = ?", (path,))
|
||||
row = cursor.fetchone()
|
||||
return row[0] if row else None
|
||||
|
||||
def close(self):
|
||||
self.conn.close()
|
||||
@@ -0,0 +1,39 @@
|
||||
"""
|
||||
Frontmatter parsing and serialization
|
||||
"""
|
||||
import re
|
||||
import yaml
|
||||
from typing import Optional, Tuple
|
||||
|
||||
|
||||
class FrontmatterParser:
|
||||
FRONTMATTER_PATTERN = re.compile(r'^---\s*\n(.*?\n)---\s*\n', re.DOTALL)
|
||||
|
||||
@staticmethod
|
||||
def parse(content: str) -> Tuple[Optional[dict], str]:
|
||||
"""Parse frontmatter from markdown content"""
|
||||
match = FrontmatterParser.FRONTMATTER_PATTERN.match(content)
|
||||
|
||||
if match:
|
||||
yaml_content = match.group(1)
|
||||
body = content[match.end():]
|
||||
|
||||
try:
|
||||
frontmatter = yaml.safe_load(yaml_content) or {}
|
||||
return frontmatter, body
|
||||
except yaml.YAMLError:
|
||||
return None, content
|
||||
|
||||
return None, content
|
||||
|
||||
@staticmethod
|
||||
def serialize(frontmatter: dict, body: str) -> str:
|
||||
"""Serialize frontmatter and body to markdown"""
|
||||
yaml_str = yaml.dump(frontmatter, default_flow_style=False, allow_unicode=True, sort_keys=False)
|
||||
return f"---\n{yaml_str}---\n{body}"
|
||||
|
||||
@staticmethod
|
||||
def extract_title(body: str) -> Optional[str]:
|
||||
"""Extract title from markdown body"""
|
||||
match = re.search(r'^#\s+(.+)$', body, re.MULTILINE)
|
||||
return match.group(1).strip() if match else None
|
||||
+101
@@ -0,0 +1,101 @@
|
||||
"""
|
||||
LLM communication module
|
||||
"""
|
||||
import os
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from dotenv import load_dotenv
|
||||
from openai import OpenAI
|
||||
|
||||
|
||||
class LLMClient:
|
||||
def __init__(self, prompts_dir: Path):
|
||||
load_dotenv()
|
||||
|
||||
self.api_url = os.getenv('OPENAI_API_URL', 'https://api.openai.com/v1')
|
||||
self.model = os.getenv('OPENAI_MODEL', 'gpt-4')
|
||||
self.api_key = os.getenv('OPENAI_API_KEY', '')
|
||||
|
||||
self.client = OpenAI(
|
||||
base_url=self.api_url,
|
||||
api_key=self.api_key
|
||||
) if self.api_key else None
|
||||
|
||||
self.prompts_dir = prompts_dir
|
||||
if not prompts_dir.exists():
|
||||
prompts_dir.mkdir(parents=True)
|
||||
self._create_default_prompt()
|
||||
|
||||
def _create_default_prompt(self):
|
||||
"""Create default system prompt"""
|
||||
default_prompt = """You are a metadata extraction assistant for markdown documents.
|
||||
|
||||
Your task is to analyze the document content and suggest appropriate frontmatter metadata.
|
||||
|
||||
Guidelines:
|
||||
- For tags: prefer selecting from the provided existing tags list when appropriate
|
||||
- You may create new tags if they better fit the content
|
||||
- New tags must follow the rules: lowercase letters, numbers, and underscores only
|
||||
- Be concise and accurate
|
||||
- Preserve important existing metadata when present"""
|
||||
|
||||
with open(self.prompts_dir / 'default.txt', 'w', encoding='utf-8') as f:
|
||||
f.write(default_prompt)
|
||||
|
||||
def load_prompt(self, name: str = 'default') -> str:
|
||||
"""Load prompt template"""
|
||||
prompt_path = self.prompts_dir / f'{name}.txt'
|
||||
|
||||
if not prompt_path.exists():
|
||||
prompt_path = self.prompts_dir / 'default.txt'
|
||||
if not prompt_path.exists():
|
||||
self._create_default_prompt()
|
||||
|
||||
with open(prompt_path, 'r', encoding='utf-8') as f:
|
||||
return f.read()
|
||||
|
||||
def infer_metadata(self, field: str, prompt_name: str, original_fm: dict,
|
||||
current_fm: dict, document_body: str,
|
||||
all_tags: list, schema: dict) -> Optional[str]:
|
||||
"""Use LLM to infer metadata field value"""
|
||||
if not self.client:
|
||||
return None
|
||||
|
||||
system_prompt = self.load_prompt(prompt_name)
|
||||
|
||||
user_message = f"""Document content:
|
||||
{document_body[:2000]}
|
||||
|
||||
Original frontmatter:
|
||||
{json.dumps(original_fm, indent=2)}
|
||||
|
||||
Current frontmatter being built:
|
||||
{json.dumps(current_fm, indent=2)}
|
||||
|
||||
Available tags in repository:
|
||||
{', '.join(all_tags[:100])}
|
||||
|
||||
Please provide appropriate metadata for this document following the schema."""
|
||||
|
||||
try:
|
||||
response = self.client.chat.completions.create(
|
||||
model=self.model,
|
||||
messages=[
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_message}
|
||||
],
|
||||
response_format={"type": "json_schema", "json_schema": {
|
||||
"name": "frontmatter_response",
|
||||
"strict": True,
|
||||
"schema": schema
|
||||
}},
|
||||
temperature=0.3
|
||||
)
|
||||
|
||||
result = json.loads(response.choices[0].message.content)
|
||||
return result.get(field)
|
||||
|
||||
except Exception as e:
|
||||
print(f" LLM error: {e}")
|
||||
return None
|
||||
@@ -0,0 +1,276 @@
|
||||
"""
|
||||
Main frontmatter processor
|
||||
"""
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
|
||||
self.db = Database(repo.path / 'madomeda.db')
|
||||
self.rules = RulesEngine(repo.path / 'rules')
|
||||
self.templates = TemplateManager(repo.path / 'templates')
|
||||
self.llm = LLMClient(repo.path / 'prompts')
|
||||
self.changelog = ChangeLog(repo.path)
|
||||
self.parser = FrontmatterParser()
|
||||
|
||||
def process(self):
|
||||
"""Main processing loop"""
|
||||
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', [])
|
||||
# Always return the list (even if empty) so it's preserved
|
||||
return tags
|
||||
|
||||
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)}")
|
||||
@@ -0,0 +1,137 @@
|
||||
"""
|
||||
Git repository management
|
||||
"""
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import List, Tuple, Optional
|
||||
|
||||
|
||||
class Repository:
|
||||
def __init__(self, path: Path, ignore_paths: List[str] = None):
|
||||
self.path = path
|
||||
self.ignore_paths = ignore_paths or []
|
||||
|
||||
def get_tracked_files(self) -> List[Path]:
|
||||
"""Get all markdown files tracked by git"""
|
||||
result = subprocess.run(
|
||||
['git', 'ls-files', '*.md'],
|
||||
cwd=self.path,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
|
||||
files = []
|
||||
for line in result.stdout.strip().split('\n'):
|
||||
if not line:
|
||||
continue
|
||||
|
||||
file_path = Path(line)
|
||||
|
||||
# Skip hidden files/folders
|
||||
if any(part.startswith('.') for part in file_path.parts):
|
||||
continue
|
||||
|
||||
# Skip ignored paths
|
||||
if any(str(file_path).startswith(ignored) for ignored in self.ignore_paths):
|
||||
continue
|
||||
|
||||
files.append(self.path / file_path)
|
||||
|
||||
return files
|
||||
|
||||
def get_file_history(self, file_path: Path) -> dict:
|
||||
"""Get git history for a file"""
|
||||
relative_path = file_path.relative_to(self.path)
|
||||
|
||||
# Get latest commit hash
|
||||
result = subprocess.run(
|
||||
['git', 'log', '-1', '--format=%H', '--', str(relative_path)],
|
||||
cwd=self.path,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
latest_hash = result.stdout.strip()
|
||||
|
||||
# Get first commit (creation) info
|
||||
result = subprocess.run(
|
||||
['git', 'log', '--reverse', '--format=%H|%an|%ae', '--', str(relative_path)],
|
||||
cwd=self.path,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
|
||||
first_line = result.stdout.strip().split('\n')[0] if result.stdout.strip() else ''
|
||||
if first_line:
|
||||
commit_hash, author_name, author_email = first_line.split('|', 2)
|
||||
else:
|
||||
commit_hash = author_name = author_email = ''
|
||||
|
||||
# Get tag for commit if any
|
||||
commit_tag = ''
|
||||
if commit_hash:
|
||||
result = subprocess.run(
|
||||
['git', 'describe', '--tags', '--exact-match', commit_hash],
|
||||
cwd=self.path,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
commit_tag = result.stdout.strip() if result.returncode == 0 else ''
|
||||
|
||||
# Get tag for latest commit if any
|
||||
latest_tag = ''
|
||||
if latest_hash:
|
||||
result = subprocess.run(
|
||||
['git', 'describe', '--tags', '--exact-match', latest_hash],
|
||||
cwd=self.path,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
latest_tag = result.stdout.strip() if result.returncode == 0 else ''
|
||||
|
||||
# Get creation date
|
||||
created = None
|
||||
if commit_hash:
|
||||
result = subprocess.run(
|
||||
['git', 'log', '-1', '--format=%aI', commit_hash, '--', str(relative_path)],
|
||||
cwd=self.path,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
created = result.stdout.strip()
|
||||
|
||||
# Get last modified date
|
||||
changed = None
|
||||
if latest_hash:
|
||||
result = subprocess.run(
|
||||
['git', 'log', '-1', '--format=%aI', latest_hash, '--', str(relative_path)],
|
||||
cwd=self.path,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
changed = result.stdout.strip()
|
||||
|
||||
# Get all contributors
|
||||
result = subprocess.run(
|
||||
['git', 'log', '--format=%an|%ae', '--', str(relative_path)],
|
||||
cwd=self.path,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
|
||||
authors = set()
|
||||
for line in result.stdout.strip().split('\n'):
|
||||
if line:
|
||||
name, email = line.split('|', 1)
|
||||
authors.add(name)
|
||||
|
||||
return {
|
||||
'commit_hash': commit_hash,
|
||||
'author_name': author_name,
|
||||
'author_email': author_email,
|
||||
'commit_tag': commit_tag,
|
||||
'latest_hash': latest_hash,
|
||||
'latest_tag': latest_tag,
|
||||
'created': created,
|
||||
'changed': changed,
|
||||
'authors': sorted(list(authors))
|
||||
}
|
||||
+127
@@ -0,0 +1,127 @@
|
||||
"""
|
||||
Frontmatter rules engine
|
||||
"""
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import List, Tuple
|
||||
|
||||
|
||||
class RulesEngine:
|
||||
def __init__(self, rules_dir: Path):
|
||||
self.rules_dir = rules_dir
|
||||
self.rules = self._load_rules()
|
||||
|
||||
def _load_rules(self) -> List[dict]:
|
||||
"""Load all rule files"""
|
||||
if not self.rules_dir.exists():
|
||||
self.rules_dir.mkdir(parents=True)
|
||||
self._create_default_rules()
|
||||
|
||||
rules = []
|
||||
for rule_file in self.rules_dir.glob('*.json'):
|
||||
with open(rule_file, 'r', encoding='utf-8') as f:
|
||||
rules.append(json.load(f))
|
||||
|
||||
return rules
|
||||
|
||||
def _create_default_rules(self):
|
||||
"""Create default rule files"""
|
||||
# Tags normalization rule
|
||||
tags_rule = {
|
||||
"name": "tags_normalization",
|
||||
"description": "Normalize tags to lowercase with underscores",
|
||||
"type": "tags"
|
||||
}
|
||||
|
||||
tag_to_tags_rule = {
|
||||
"name": "tag_to_tags",
|
||||
"description": "Rename 'tag' key to 'tags'",
|
||||
"type": "rename"
|
||||
}
|
||||
|
||||
lowercase_keys_rule = {
|
||||
"name": "lowercase_keys",
|
||||
"description": "All frontmatter keys must be lowercase",
|
||||
"type": "keys"
|
||||
}
|
||||
|
||||
summary_to_description_rule = {
|
||||
"name": "summary_to_description",
|
||||
"description": "Rename 'summary' key to 'description'",
|
||||
"type": "rename"
|
||||
}
|
||||
|
||||
with open(self.rules_dir / 'tags_normalization.json', 'w', encoding='utf-8') as f:
|
||||
json.dump(tags_rule, f, indent=2)
|
||||
|
||||
with open(self.rules_dir / 'tag_to_tags.json', 'w', encoding='utf-8') as f:
|
||||
json.dump(tag_to_tags_rule, f, indent=2)
|
||||
|
||||
with open(self.rules_dir / 'lowercase_keys.json', 'w', encoding='utf-8') as f:
|
||||
json.dump(lowercase_keys_rule, f, indent=2)
|
||||
|
||||
with open(self.rules_dir / 'summary_to_description.json', 'w', encoding='utf-8') as f:
|
||||
json.dump(summary_to_description_rule, f, indent=2)
|
||||
|
||||
def apply_rules(self, frontmatter: dict) -> Tuple[dict, bool]:
|
||||
"""Apply all rules to frontmatter, return (modified_fm, is_conformant)"""
|
||||
if not frontmatter:
|
||||
return frontmatter, True
|
||||
|
||||
modified = dict(frontmatter)
|
||||
violations = []
|
||||
|
||||
# Rule: lowercase keys (run first so we can check other rules)
|
||||
key_changes = {}
|
||||
for key in list(modified.keys()):
|
||||
lower_key = key.lower()
|
||||
if lower_key != key:
|
||||
key_changes[key] = lower_key
|
||||
violations.append('lowercase_keys')
|
||||
|
||||
for old_key, new_key in key_changes.items():
|
||||
modified[new_key] = modified.pop(old_key)
|
||||
|
||||
# Rule: tag -> tags (after lowercase conversion)
|
||||
if 'tag' in modified:
|
||||
modified['tags'] = modified.pop('tag')
|
||||
violations.append('tag_to_tags')
|
||||
|
||||
# Rule: summary -> description (after lowercase conversion)
|
||||
if 'summary' in modified:
|
||||
modified['description'] = modified.pop('summary')
|
||||
violations.append('summary_to_description')
|
||||
|
||||
# Rule: normalize tags
|
||||
if 'tags' in modified:
|
||||
original_tags = modified['tags']
|
||||
|
||||
# Convert to list if comma-separated string
|
||||
if isinstance(original_tags, str):
|
||||
tags_list = [t.strip() for t in original_tags.split(',')]
|
||||
violations.append('tags_format')
|
||||
else:
|
||||
tags_list = original_tags if isinstance(original_tags, list) else [str(original_tags)]
|
||||
|
||||
# Normalize each tag
|
||||
normalized_tags = []
|
||||
for tag in tags_list:
|
||||
tag_str = str(tag)
|
||||
# Replace spaces and hyphens with underscores
|
||||
normalized = tag_str.replace(' ', '_').replace('-', '_')
|
||||
# Convert to lowercase
|
||||
normalized = normalized.lower()
|
||||
# Remove invalid characters (keep only alphanumeric and underscore)
|
||||
normalized = re.sub(r'[^a-z0-9_]', '', normalized)
|
||||
|
||||
if normalized and normalized != tag_str:
|
||||
violations.append('tags_normalized')
|
||||
|
||||
if normalized:
|
||||
normalized_tags.append(normalized)
|
||||
|
||||
modified['tags'] = normalized_tags
|
||||
|
||||
is_conformant = len(violations) == 0
|
||||
return modified, is_conformant
|
||||
@@ -0,0 +1,68 @@
|
||||
"""
|
||||
Template management
|
||||
"""
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class TemplateManager:
|
||||
def __init__(self, templates_dir: Path):
|
||||
self.templates_dir = templates_dir
|
||||
|
||||
if not templates_dir.exists():
|
||||
templates_dir.mkdir(parents=True)
|
||||
self._create_default_template()
|
||||
|
||||
def _create_default_template(self):
|
||||
"""Create default template"""
|
||||
default_template = {
|
||||
"title": "heur|orig",
|
||||
"created": "heur|orig",
|
||||
"changed": "heur|orig",
|
||||
"authors": "heur|orig",
|
||||
"version": "heur|orig",
|
||||
"tags": "orig|ai default"
|
||||
}
|
||||
|
||||
with open(self.templates_dir / 'default.json', 'w', encoding='utf-8') as f:
|
||||
json.dump(default_template, f, indent=2)
|
||||
|
||||
def load_template(self, name: str) -> dict:
|
||||
"""Load template by name"""
|
||||
template_path = self.templates_dir / f'{name}.json'
|
||||
|
||||
if not template_path.exists():
|
||||
# Fallback to default
|
||||
template_path = self.templates_dir / 'default.json'
|
||||
if not template_path.exists():
|
||||
self._create_default_template()
|
||||
|
||||
with open(template_path, 'r', encoding='utf-8') as f:
|
||||
return json.load(f)
|
||||
|
||||
def get_structured_output_schema(self, template: dict) -> dict:
|
||||
"""Generate JSON schema for structured output"""
|
||||
properties = {}
|
||||
required = []
|
||||
|
||||
for field in template.keys():
|
||||
if field == 'tags':
|
||||
properties[field] = {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": f"Field: {field}"
|
||||
}
|
||||
else:
|
||||
properties[field] = {
|
||||
"type": "string",
|
||||
"description": f"Field: {field}"
|
||||
}
|
||||
required.append(field)
|
||||
|
||||
return {
|
||||
"type": "object",
|
||||
"properties": properties,
|
||||
"required": required,
|
||||
"additionalProperties": False
|
||||
}
|
||||
Reference in New Issue
Block a user