solventum-image-metadata/backend/app/processors/template_manager.py
SamoilenkoVadym 563d476a94 feat(backend): migrate from Flask to FastAPI with Redis sessions
- Create FastAPI application with async I/O
- Implement Redis session storage (fixes session loss on restart)
- Add JWT authentication with refresh tokens
- Add Microsoft SSO support via MSAL
- Copy all processors from src/ (100% reused, no changes)
- Create file upload/download endpoints
- Create metadata update endpoints
- Create template CRUD endpoints
- Add SQLAlchemy async database models
- Add Docker Compose configuration with Redis

Solves critical issues:
- Session management: Redis replaces in-memory dicts
- Scalability: Async FastAPI + microservices architecture
- File handling: Persistent storage with auto-cleanup

Key files:
- backend/app/main.py - FastAPI entry point
- backend/app/core/redis_client.py - Session store
- backend/app/core/auth.py - JWT authentication
- backend/app/api/* - All REST endpoints
- backend/app/processors/ - Reused from src/

Co-Authored-By: Claude Sonnet 4.5 (1M context) <noreply@anthropic.com>
2026-02-09 13:14:37 +00:00

410 lines
12 KiB
Python

"""Metadata template manager with variable substitution."""
import json
from pathlib import Path
from typing import Dict, List, Optional
from datetime import datetime
from .utils import get_logger
logger = get_logger(__name__)
class TemplateManager:
"""Manage metadata templates with variable substitution."""
# Available variables for substitution
AVAILABLE_VARIABLES = {
'{filename}': 'Original filename without extension',
'{date}': 'Current date (YYYY-MM-DD)',
'{datetime}': 'Current date and time',
'{user}': 'Current username',
'{year}': 'Current year',
'{month}': 'Current month',
'{day}': 'Current day'
}
def __init__(self, templates_path: Optional[str] = None):
"""
Initialize template manager.
Args:
templates_path: Path to JSON file for storing templates
"""
self.templates_path = templates_path or 'metadata_templates.json'
def create_template(
self,
name: str,
title_template: str,
subject_template: str,
keywords_template: str,
description: str = ''
) -> Dict:
"""
Create a new metadata template.
Args:
name: Template name
title_template: Title template with variables (e.g., "{filename} - Product Guide")
subject_template: Subject template with variables
keywords_template: Keywords template with variables
description: Optional description of template usage
Returns:
Template dictionary
"""
template = {
'name': name,
'description': description,
'title': title_template,
'subject': subject_template,
'keywords': keywords_template,
'created_at': self._get_timestamp(),
'updated_at': self._get_timestamp()
}
# Validate template
validation = self.validate_template(template)
if validation['invalid']:
logger.warning(f"Template '{name}' has invalid variables: {validation['invalid']}")
return template
def save_template(self, template: Dict) -> bool:
"""
Save template to storage.
Args:
template: Template dictionary
Returns:
True if successful
"""
try:
templates = self._load_templates()
template['updated_at'] = self._get_timestamp()
templates[template['name']] = template
with open(self.templates_path, 'w', encoding='utf-8') as f:
json.dump(templates, f, indent=2, ensure_ascii=False)
logger.info(f"Saved template: {template['name']}")
return True
except Exception as e:
logger.error(f"Failed to save template '{template['name']}': {e}")
return False
def load_template(self, name: str) -> Optional[Dict]:
"""
Load template by name.
Args:
name: Template name
Returns:
Template dictionary or None if not found
"""
templates = self._load_templates()
template = templates.get(name)
if template:
logger.info(f"Loaded template: {name}")
else:
logger.warning(f"Template not found: {name}")
return template
def list_templates(self) -> List[Dict]:
"""
List all available templates.
Returns:
List of template summaries
"""
templates = self._load_templates()
return [
{
'name': name,
'description': data.get('description', ''),
'created_at': data.get('created_at', ''),
'updated_at': data.get('updated_at', ''),
'variables_used': self._extract_variables(data)
}
for name, data in templates.items()
]
def delete_template(self, name: str) -> bool:
"""
Delete a template.
Args:
name: Template name
Returns:
True if deleted, False if not found
"""
templates = self._load_templates()
if name in templates:
del templates[name]
try:
with open(self.templates_path, 'w', encoding='utf-8') as f:
json.dump(templates, f, indent=2, ensure_ascii=False)
logger.info(f"Deleted template: {name}")
return True
except Exception as e:
logger.error(f"Failed to delete template '{name}': {e}")
return False
logger.warning(f"Template not found: {name}")
return False
def apply_template(
self,
template: Dict,
filename: str,
user: str = 'Unknown',
custom_vars: Optional[Dict[str, str]] = None
) -> Dict[str, str]:
"""
Apply template to generate metadata for a file.
Args:
template: Template dictionary
filename: Filename to process
user: Username for {user} variable
custom_vars: Additional custom variables (e.g., {'product_line': 'Dental'})
Returns:
Dictionary with title, subject, keywords
"""
# Build variable substitution map
variables = self._build_variable_map(filename, user, custom_vars)
# Apply substitutions
metadata = {
'title': self._substitute_variables(template.get('title', ''), variables),
'subject': self._substitute_variables(template.get('subject', ''), variables),
'keywords': self._substitute_variables(template.get('keywords', ''), variables)
}
logger.info(f"Applied template '{template['name']}' to {filename}")
return metadata
def validate_template(self, template: Dict) -> Dict[str, List[str]]:
"""
Validate template for correct variable usage.
Args:
template: Template dictionary
Returns:
Dictionary with 'valid' and 'invalid' variable lists
"""
result = {
'valid': [],
'invalid': []
}
# Extract all variables from template
all_text = (
template.get('title', '') +
template.get('subject', '') +
template.get('keywords', '')
)
# Find all {variable} patterns
import re
variables = re.findall(r'\{[^}]+\}', all_text)
for var in variables:
if var in self.AVAILABLE_VARIABLES:
if var not in result['valid']:
result['valid'].append(var)
else:
if var not in result['invalid']:
result['invalid'].append(var)
return result
def _load_templates(self) -> Dict:
"""Load all templates from file."""
if Path(self.templates_path).exists():
try:
with open(self.templates_path, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
logger.error(f"Failed to load templates: {e}")
return {}
return {}
def _get_timestamp(self) -> str:
"""Get current timestamp as ISO format string."""
return datetime.now().isoformat()
def _build_variable_map(
self,
filename: str,
user: str,
custom_vars: Optional[Dict[str, str]]
) -> Dict[str, str]:
"""
Build variable substitution map.
Args:
filename: Filename (with or without extension)
user: Username
custom_vars: Custom variables
Returns:
Dictionary mapping variable names to values
"""
# Get filename without extension
filename_stem = Path(filename).stem
# Current date/time
now = datetime.now()
variables = {
'{filename}': filename_stem,
'{date}': now.strftime('%Y-%m-%d'),
'{datetime}': now.strftime('%Y-%m-%d %H:%M:%S'),
'{user}': user,
'{year}': str(now.year),
'{month}': now.strftime('%m'),
'{day}': now.strftime('%d')
}
# Add custom variables
if custom_vars:
for key, value in custom_vars.items():
# Ensure custom variables are wrapped in {}
var_key = f'{{{key}}}' if not key.startswith('{') else key
variables[var_key] = value
return variables
def _substitute_variables(self, template_text: str, variables: Dict[str, str]) -> str:
"""
Substitute variables in template text.
Args:
template_text: Text with {variable} placeholders
variables: Variable substitution map
Returns:
Text with variables replaced
"""
result = template_text
for var, value in variables.items():
result = result.replace(var, value)
return result
def _extract_variables(self, template: Dict) -> List[str]:
"""
Extract all variables used in a template.
Args:
template: Template dictionary
Returns:
List of variable names (e.g., ['{filename}', '{date}'])
"""
import re
all_text = (
template.get('title', '') +
template.get('subject', '') +
template.get('keywords', '')
)
variables = re.findall(r'\{[^}]+\}', all_text)
return list(set(variables))
def get_available_variables(self) -> Dict[str, str]:
"""
Get list of available variables with descriptions.
Returns:
Dictionary mapping variable names to descriptions
"""
return self.AVAILABLE_VARIABLES.copy()
def preview_template(
self,
template: Dict,
sample_filename: str = 'example.pdf',
user: str = 'User',
custom_vars: Optional[Dict[str, str]] = None
) -> Dict[str, str]:
"""
Preview template output with sample data.
Args:
template: Template dictionary
sample_filename: Sample filename for preview
user: Sample username
custom_vars: Sample custom variables
Returns:
Preview metadata
"""
return self.apply_template(template, sample_filename, user, custom_vars)
def export_template(self, name: str, export_path: str) -> bool:
"""
Export single template to JSON file.
Args:
name: Template name
export_path: Path to save template
Returns:
True if successful
"""
template = self.load_template(name)
if not template:
return False
try:
with open(export_path, 'w', encoding='utf-8') as f:
json.dump(template, f, indent=2, ensure_ascii=False)
logger.info(f"Exported template '{name}' to {export_path}")
return True
except Exception as e:
logger.error(f"Failed to export template '{name}': {e}")
return False
def import_template(self, import_path: str) -> Optional[Dict]:
"""
Import template from JSON file.
Args:
import_path: Path to template JSON file
Returns:
Imported template dictionary or None
"""
try:
with open(import_path, 'r', encoding='utf-8') as f:
template = json.load(f)
# Validate required fields
required_fields = ['name', 'title', 'subject', 'keywords']
if not all(field in template for field in required_fields):
logger.error(f"Invalid template file: missing required fields")
return None
logger.info(f"Imported template from {import_path}")
return template
except Exception as e:
logger.error(f"Failed to import template: {e}")
return None