Markdown Dynamic Content Generation and Automation Workflows: Complete Guide for Automated Documentation, Template Systems, and Content Pipeline Management
Advanced Markdown dynamic content generation and automation workflows enable systematic creation of data-driven documentation, template-based content systems, and intelligent content pipelines that automatically generate, update, and maintain documentation at scale. By implementing comprehensive automation strategies, template engines, and content generation systems, technical writers can build documentation ecosystems that adapt dynamically to changing data sources while maintaining consistency, accuracy, and efficiency across large content repositories.
Why Master Dynamic Content Generation?
Professional content automation provides essential benefits for scalable documentation systems:
- Scalability: Generate thousands of documentation pages automatically from data sources
- Consistency: Maintain uniform formatting and structure across all generated content
- Real-time Updates: Automatically refresh documentation when underlying data changes
- Efficiency: Reduce manual content creation time by 80-90% for repetitive documentation
- Data Integration: Connect documentation directly to APIs, databases, and configuration systems
- Quality Assurance: Implement automated validation and error checking in content pipelines
Foundation Automation Principles
Template-Driven Content Architecture
Understanding the core components of automated content generation:
# Dynamic Content Generation Architecture
## Template Engine Foundation
```yaml
# content-config.yml - Content generation configuration
project:
name: "API Documentation Generator"
version: "2.1.0"
base_url: "https://api.example.com"
data_sources:
- name: "api_endpoints"
type: "openapi"
source: "https://api.example.com/openapi.json"
refresh_interval: "hourly"
- name: "user_metrics"
type: "database"
connection: "postgresql://metrics:5432/analytics"
query: "SELECT * FROM daily_metrics WHERE date >= NOW() - INTERVAL '30 days'"
refresh_interval: "daily"
- name: "feature_flags"
type: "json"
source: "config/features.json"
refresh_interval: "on_change"
templates:
- name: "api_endpoint_doc"
source: "templates/api-endpoint.md.j2"
output_pattern: "docs/api/endpoints/{{endpoint.path|slug}}.md"
data_source: "api_endpoints"
- name: "metrics_report"
source: "templates/metrics-dashboard.md.j2"
output: "docs/analytics/daily-report.md"
data_source: "user_metrics"
- name: "feature_documentation"
source: "templates/feature-guide.md.j2"
output_pattern: "docs/features/{{feature.name|slug}}.md"
data_source: "feature_flags"
automation:
triggers:
- type: "schedule"
cron: "0 */6 * * *" # Every 6 hours
templates: ["api_endpoint_doc", "metrics_report"]
- type: "webhook"
endpoint: "/regenerate"
secret: "${WEBHOOK_SECRET}"
templates: ["feature_documentation"]
- type: "file_watch"
paths: ["config/**/*.json"]
templates: ["feature_documentation"]
validation:
enabled: true
rules:
- check: "required_frontmatter"
fields: ["title", "description", "category"]
- check: "link_validation"
external_links: false
- check: "content_length"
min_words: 50
```
## Basic Template Example
```jinja2
<!-- templates/api-endpoint.md.j2 -->
---
title: "{{endpoint.summary}} - {{endpoint.method|upper}} {{endpoint.path}}"
description: "{{endpoint.description|truncate(160)}}"
category: "API Reference"
tags:
- "{{endpoint.tags|join('", "')}}"
method: "{{endpoint.method|upper}}"
path: "{{endpoint.path}}"
---
# {{endpoint.summary}}
{{endpoint.description}}
## Request Details
**Method:** `{{endpoint.method|upper}}`
**Path:** `{{endpoint.path}}`
**Content-Type:** `{{endpoint.consumes|default(['application/json'])|join(', ')}}`
{% if endpoint.parameters %}
## Parameters
| Name | Type | Required | Description |
|:-----|:-----|:---------|:------------|
{% for param in endpoint.parameters %}
| **{{param.name}}** | {{param.type}} | {{param.required|yesno}} | {{param.description|default('No description provided')}} |
{% endfor %}
{% endif %}
{% if endpoint.requestBody %}
## Request Body
{{endpoint.requestBody.description}}
### Example Request
```json
{{endpoint.requestBody.examples.default|tojsonpretty}}
```
{% endif %}
## Response
{% for status_code, response in endpoint.responses.items() %}
### {{status_code}} - {{response.description}}
```json
{{response.examples.default|tojsonpretty}}
```
{% endfor %}
## Code Examples
### cURL
```bash
curl -X {{endpoint.method|upper}} \
"{{project.base_url}}{{endpoint.path}}" \
{% if endpoint.security %}
-H "Authorization: Bearer YOUR_API_KEY" \
{% endif %}
-H "Content-Type: application/json"
{% if endpoint.requestBody %}
-d '{{endpoint.requestBody.examples.default|tojson}}'
{% endif %}
```
### JavaScript
```javascript
const response = await fetch('{{project.base_url}}{{endpoint.path}}', {
method: '{{endpoint.method|upper}}',
headers: {
'Content-Type': 'application/json',
{% if endpoint.security %}
'Authorization': 'Bearer YOUR_API_KEY',
{% endif %}
},
{% if endpoint.requestBody %}
body: JSON.stringify({{endpoint.requestBody.examples.default|tojsonpretty|indent(4)}})
{% endif %}
});
const data = await response.json();
console.log(data);
```
### Python
```python
import requests
url = "{{project.base_url}}{{endpoint.path}}"
headers = {
"Content-Type": "application/json",
{% if endpoint.security %}
"Authorization": "Bearer YOUR_API_KEY"
{% endif %}
}
{% if endpoint.requestBody %}
data = {{endpoint.requestBody.examples.default|tojsonpretty|indent(4)}}
response = requests.{{endpoint.method}}(url, headers=headers, json=data)
{% else %}
response = requests.{{endpoint.method}}(url, headers=headers)
{% endif %}
print(response.json())
```
---
*Generated automatically from OpenAPI specification on {{generation_timestamp}}*
```
Advanced Template Engine Implementation
Building comprehensive template processing systems with multiple data sources:
# Advanced Template Engine System
## Python-Based Generation Engine
```python
#!/usr/bin/env python3
# content-generator.py - Advanced dynamic content generation system
import os
import json
import yaml
import sqlite3
import asyncio
import aiohttp
import jinja2
from jinja2 import Environment, FileSystemLoader, StrictUndefined
from pathlib import Path
from typing import Dict, Any, List, Optional, Union
from datetime import datetime, timedelta
import hashlib
import logging
from dataclasses import dataclass, field
import schedule
import time
import threading
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import psycopg2
from psycopg2.extras import RealDictCursor
import requests
from urllib.parse import urljoin
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class DataSource:
"""Configuration for data source"""
name: str
type: str
source: str
refresh_interval: str
connection_params: Dict[str, Any] = field(default_factory=dict)
query: Optional[str] = None
transform_function: Optional[str] = None
cache_duration: int = 3600 # seconds
@dataclass
class Template:
"""Template configuration"""
name: str
source: str
output_pattern: Optional[str] = None
output: Optional[str] = None
data_source: str = ""
context_processors: List[str] = field(default_factory=list)
conditions: Dict[str, Any] = field(default_factory=dict)
@dataclass
class GenerationResult:
"""Result of content generation"""
template_name: str
output_path: str
success: bool
error_message: Optional[str] = None
generated_at: datetime = field(default_factory=datetime.now)
data_hash: Optional[str] = None
class ContentGenerationEngine:
"""Advanced content generation engine with multiple data sources"""
def __init__(self, config_path: str):
self.config_path = Path(config_path)
self.config = self._load_config()
self.data_cache = {}
self.cache_timestamps = {}
# Setup Jinja2 environment
self.jinja_env = Environment(
loader=FileSystemLoader(['templates', '.']),
undefined=StrictUndefined,
trim_blocks=True,
lstrip_blocks=True
)
# Add custom filters
self._setup_custom_filters()
# Initialize data sources
self.data_sources = {
source_config['name']: DataSource(**source_config)
for source_config in self.config.get('data_sources', [])
}
# Initialize templates
self.templates = {
template_config['name']: Template(**template_config)
for template_config in self.config.get('templates', [])
}
# Setup automation
self._setup_automation()
def _load_config(self) -> Dict[str, Any]:
"""Load configuration from YAML file"""
try:
with open(self.config_path, 'r') as f:
return yaml.safe_load(f)
except Exception as e:
logger.error(f"Error loading config: {e}")
raise
def _setup_custom_filters(self):
"""Setup custom Jinja2 filters"""
def slug_filter(text: str) -> str:
"""Convert text to URL-friendly slug"""
import re
text = re.sub(r'[^\w\s-]', '', str(text)).strip().lower()
return re.sub(r'[\s_-]+', '-', text)
def yesno_filter(value: bool) -> str:
"""Convert boolean to Yes/No"""
return "Yes" if value else "No"
def tojsonpretty_filter(obj: Any) -> str:
"""Convert object to pretty JSON"""
return json.dumps(obj, indent=2, ensure_ascii=False)
def truncate_words_filter(text: str, length: int = 50) -> str:
"""Truncate text to specified word count"""
words = str(text).split()
if len(words) <= length:
return text
return ' '.join(words[:length]) + '...'
def relative_time_filter(dt: datetime) -> str:
"""Convert datetime to relative time string"""
now = datetime.now()
diff = now - dt
if diff.days > 0:
return f"{diff.days} day{'s' if diff.days > 1 else ''} ago"
elif diff.seconds > 3600:
hours = diff.seconds // 3600
return f"{hours} hour{'s' if hours > 1 else ''} ago"
elif diff.seconds > 60:
minutes = diff.seconds // 60
return f"{minutes} minute{'s' if minutes > 1 else ''} ago"
else:
return "just now"
def format_number_filter(value: Union[int, float], format_type: str = 'default') -> str:
"""Format numbers with various styles"""
if format_type == 'currency':
return f"${value:,.2f}"
elif format_type == 'percentage':
return f"{value:.1f}%"
elif format_type == 'abbreviated':
if value >= 1_000_000:
return f"{value/1_000_000:.1f}M"
elif value >= 1_000:
return f"{value/1_000:.1f}K"
else:
return str(value)
else:
return f"{value:,}"
# Register filters
self.jinja_env.filters['slug'] = slug_filter
self.jinja_env.filters['yesno'] = yesno_filter
self.jinja_env.filters['tojsonpretty'] = tojsonpretty_filter
self.jinja_env.filters['truncate_words'] = truncate_words_filter
self.jinja_env.filters['relative_time'] = relative_time_filter
self.jinja_env.filters['format_number'] = format_number_filter
# Add global functions
self.jinja_env.globals['now'] = datetime.now
self.jinja_env.globals['generation_timestamp'] = datetime.now().isoformat()
async def fetch_data_source(self, source: DataSource) -> Dict[str, Any]:
"""Fetch data from configured source"""
cache_key = source.name
now = time.time()
# Check cache validity
if (cache_key in self.cache_timestamps and
now - self.cache_timestamps[cache_key] < source.cache_duration):
logger.info(f"Using cached data for {source.name}")
return self.data_cache[cache_key]
logger.info(f"Fetching fresh data for {source.name}")
try:
if source.type == 'openapi':
data = await self._fetch_openapi_data(source)
elif source.type == 'database':
data = await self._fetch_database_data(source)
elif source.type == 'json':
data = await self._fetch_json_data(source)
elif source.type == 'api':
data = await self._fetch_api_data(source)
elif source.type == 'csv':
data = await self._fetch_csv_data(source)
else:
raise ValueError(f"Unknown data source type: {source.type}")
# Apply transformation if specified
if source.transform_function:
data = await self._apply_transformation(data, source.transform_function)
# Cache the data
self.data_cache[cache_key] = data
self.cache_timestamps[cache_key] = now
return data
except Exception as e:
logger.error(f"Error fetching data from {source.name}: {e}")
# Return cached data if available, otherwise raise
if cache_key in self.data_cache:
logger.warning(f"Using stale cached data for {source.name}")
return self.data_cache[cache_key]
raise
async def _fetch_openapi_data(self, source: DataSource) -> Dict[str, Any]:
"""Fetch and process OpenAPI specification"""
async with aiohttp.ClientSession() as session:
async with session.get(source.source) as response:
response.raise_for_status()
spec = await response.json()
# Process OpenAPI spec into template-friendly format
endpoints = []
for path, path_methods in spec.get('paths', {}).items():
for method, operation in path_methods.items():
if method.lower() in ['get', 'post', 'put', 'patch', 'delete']:
endpoint = {
'path': path,
'method': method.lower(),
'summary': operation.get('summary', ''),
'description': operation.get('description', ''),
'tags': operation.get('tags', []),
'parameters': operation.get('parameters', []),
'requestBody': operation.get('requestBody'),
'responses': operation.get('responses', {}),
'security': operation.get('security', [])
}
# Process examples
if endpoint['requestBody']:
content = endpoint['requestBody'].get('content', {})
for media_type, schema in content.items():
if 'examples' not in schema and 'example' in schema:
schema['examples'] = {'default': schema['example']}
endpoints.append(endpoint)
return {
'spec': spec,
'endpoints': endpoints,
'info': spec.get('info', {}),
'servers': spec.get('servers', [])
}
async def _fetch_database_data(self, source: DataSource) -> Dict[str, Any]:
"""Fetch data from database"""
connection_string = source.source
if connection_string.startswith('postgresql://'):
import psycopg2.pool
# Create connection pool for better performance
pool = psycopg2.pool.SimpleConnectionPool(
1, 5, connection_string,
cursor_factory=RealDictCursor
)
try:
conn = pool.getconn()
with conn.cursor() as cursor:
cursor.execute(source.query)
results = cursor.fetchall()
# Convert to list of dictionaries
data = [dict(row) for row in results]
return {'rows': data, 'count': len(data)}
finally:
pool.putconn(conn)
pool.closeall()
elif connection_string.startswith('sqlite://'):
db_path = connection_string.replace('sqlite://', '')
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
try:
cursor = conn.cursor()
cursor.execute(source.query)
results = cursor.fetchall()
data = [dict(row) for row in results]
return {'rows': data, 'count': len(data)}
finally:
conn.close()
else:
raise ValueError(f"Unsupported database type in: {connection_string}")
async def _fetch_json_data(self, source: DataSource) -> Dict[str, Any]:
"""Fetch JSON data from file or URL"""
if source.source.startswith('http'):
async with aiohttp.ClientSession() as session:
async with session.get(source.source) as response:
response.raise_for_status()
return await response.json()
else:
with open(source.source, 'r') as f:
return json.load(f)
async def _fetch_api_data(self, source: DataSource) -> Dict[str, Any]:
"""Fetch data from REST API"""
headers = source.connection_params.get('headers', {})
params = source.connection_params.get('params', {})
async with aiohttp.ClientSession() as session:
async with session.get(
source.source,
headers=headers,
params=params
) as response:
response.raise_for_status()
return await response.json()
async def _fetch_csv_data(self, source: DataSource) -> Dict[str, Any]:
"""Fetch and process CSV data"""
import pandas as pd
if source.source.startswith('http'):
df = pd.read_csv(source.source)
else:
df = pd.read_csv(source.source)
# Convert to records format
records = df.to_dict('records')
return {
'records': records,
'columns': df.columns.tolist(),
'count': len(records),
'summary': {
'total_rows': len(records),
'columns': len(df.columns),
'numeric_columns': len(df.select_dtypes(include=['number']).columns),
'text_columns': len(df.select_dtypes(include=['object']).columns)
}
}
async def _apply_transformation(self, data: Dict[str, Any], transform_function: str) -> Dict[str, Any]:
"""Apply custom transformation function to data"""
# This would load and execute custom transformation functions
# For security, this should be sandboxed in production
# Example: Simple transformation functions
if transform_function == 'sort_by_name':
if 'endpoints' in data:
data['endpoints'] = sorted(data['endpoints'], key=lambda x: x.get('summary', ''))
elif transform_function == 'group_by_tag':
if 'endpoints' in data:
grouped = {}
for endpoint in data['endpoints']:
for tag in endpoint.get('tags', ['default']):
if tag not in grouped:
grouped[tag] = []
grouped[tag].append(endpoint)
data['grouped_endpoints'] = grouped
return data
async def generate_content(self, template_name: str, force: bool = False) -> List[GenerationResult]:
"""Generate content for specified template"""
if template_name not in self.templates:
raise ValueError(f"Template {template_name} not found")
template = self.templates[template_name]
results = []
try:
# Load template
jinja_template = self.jinja_env.get_template(template.source)
# Fetch data
data = {}
if template.data_source:
source = self.data_sources[template.data_source]
data = await self.fetch_data_source(source)
# Apply context processors
context = await self._build_context(data, template)
# Check conditions
if not self._check_conditions(context, template):
logger.info(f"Conditions not met for template {template_name}")
return results
# Generate content
if template.output_pattern:
# Generate multiple files based on pattern
results.extend(await self._generate_multiple_files(jinja_template, context, template))
else:
# Generate single file
result = await self._generate_single_file(jinja_template, context, template)
results.append(result)
except Exception as e:
logger.error(f"Error generating content for {template_name}: {e}")
results.append(GenerationResult(
template_name=template_name,
output_path="",
success=False,
error_message=str(e)
))
return results
async def _build_context(self, data: Dict[str, Any], template: Template) -> Dict[str, Any]:
"""Build template context with all required data"""
context = {
'data': data,
'config': self.config,
'generation_time': datetime.now(),
'template_name': template.name
}
# Apply context processors
for processor_name in template.context_processors:
context = await self._apply_context_processor(context, processor_name)
return context
async def _apply_context_processor(self, context: Dict[str, Any], processor_name: str) -> Dict[str, Any]:
"""Apply context processor to enhance template data"""
# Example context processors
if processor_name == 'add_navigation':
context['navigation'] = self._generate_navigation_data(context)
elif processor_name == 'add_metadata':
context['metadata'] = self._generate_metadata(context)
elif processor_name == 'add_related_content':
context['related'] = await self._find_related_content(context)
return context
def _generate_navigation_data(self, context: Dict[str, Any]) -> Dict[str, Any]:
"""Generate navigation data for templates"""
return {
'sections': [
{'name': 'API Reference', 'url': '/api/'},
{'name': 'Guides', 'url': '/guides/'},
{'name': 'Examples', 'url': '/examples/'}
],
'breadcrumbs': self._generate_breadcrumbs(context)
}
def _generate_breadcrumbs(self, context: Dict[str, Any]) -> List[Dict[str, str]]:
"""Generate breadcrumb navigation"""
# Implementation would depend on the specific content structure
return [
{'name': 'Home', 'url': '/'},
{'name': 'Documentation', 'url': '/docs/'}
]
def _generate_metadata(self, context: Dict[str, Any]) -> Dict[str, Any]:
"""Generate metadata for content"""
return {
'generated_at': datetime.now().isoformat(),
'generator_version': '2.1.0',
'data_sources': list(self.data_sources.keys()),
'last_updated': context.get('generation_time', datetime.now()).isoformat()
}
async def _find_related_content(self, context: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Find related content based on context"""
# This would implement content similarity algorithms
return []
def _check_conditions(self, context: Dict[str, Any], template: Template) -> bool:
"""Check if template conditions are met"""
if not template.conditions:
return True
for condition_key, condition_value in template.conditions.items():
if condition_key == 'min_data_count':
data_count = len(context.get('data', {}).get('rows', []))
if data_count < condition_value:
return False
elif condition_key == 'required_fields':
data = context.get('data', {})
for field in condition_value:
if field not in data:
return False
return True
async def _generate_single_file(self, jinja_template, context: Dict[str, Any], template: Template) -> GenerationResult:
"""Generate single output file"""
try:
content = jinja_template.render(**context)
# Create output directory if needed
output_path = Path(template.output)
output_path.parent.mkdir(parents=True, exist_ok=True)
# Write file
with open(output_path, 'w', encoding='utf-8') as f:
f.write(content)
# Calculate content hash for change detection
content_hash = hashlib.sha256(content.encode()).hexdigest()
logger.info(f"Generated {output_path}")
return GenerationResult(
template_name=template.name,
output_path=str(output_path),
success=True,
data_hash=content_hash
)
except Exception as e:
return GenerationResult(
template_name=template.name,
output_path=template.output or "",
success=False,
error_message=str(e)
)
async def _generate_multiple_files(self, jinja_template, context: Dict[str, Any], template: Template) -> List[GenerationResult]:
"""Generate multiple files based on output pattern"""
results = []
try:
# Extract iterable data
data = context.get('data', {})
if 'endpoints' in data:
items = data['endpoints']
item_key = 'endpoint'
elif 'records' in data:
items = data['records']
item_key = 'record'
elif 'rows' in data:
items = data['rows']
item_key = 'row'
else:
items = [data] # Single item
item_key = 'item'
# Generate file for each item
for item in items:
item_context = context.copy()
item_context[item_key] = item
# Render output path pattern
output_pattern_template = self.jinja_env.from_string(template.output_pattern)
output_path = output_pattern_template.render(**item_context)
# Generate content
content = jinja_template.render(**item_context)
# Create output directory
output_path_obj = Path(output_path)
output_path_obj.parent.mkdir(parents=True, exist_ok=True)
# Write file
with open(output_path_obj, 'w', encoding='utf-8') as f:
f.write(content)
content_hash = hashlib.sha256(content.encode()).hexdigest()
results.append(GenerationResult(
template_name=template.name,
output_path=str(output_path_obj),
success=True,
data_hash=content_hash
))
logger.info(f"Generated {output_path_obj}")
except Exception as e:
logger.error(f"Error in multiple file generation: {e}")
results.append(GenerationResult(
template_name=template.name,
output_path="",
success=False,
error_message=str(e)
))
return results
def _setup_automation(self):
"""Setup automation triggers"""
automation_config = self.config.get('automation', {})
triggers = automation_config.get('triggers', [])
for trigger in triggers:
if trigger['type'] == 'schedule':
self._setup_scheduled_trigger(trigger)
elif trigger['type'] == 'file_watch':
self._setup_file_watch_trigger(trigger)
elif trigger['type'] == 'webhook':
self._setup_webhook_trigger(trigger)
def _setup_scheduled_trigger(self, trigger_config: Dict[str, Any]):
"""Setup scheduled content generation"""
cron_expression = trigger_config['cron']
templates = trigger_config['templates']
def job():
asyncio.run(self._run_scheduled_job(templates))
# Simple cron parsing (in production, use croniter or similar)
if cron_expression == "0 */6 * * *": # Every 6 hours
schedule.every(6).hours.do(job)
elif cron_expression == "0 0 * * *": # Daily at midnight
schedule.every().day.at("00:00").do(job)
# Start scheduler in separate thread
def run_scheduler():
while True:
schedule.run_pending()
time.sleep(60)
scheduler_thread = threading.Thread(target=run_scheduler, daemon=True)
scheduler_thread.start()
async def _run_scheduled_job(self, templates: List[str]):
"""Run scheduled content generation job"""
logger.info(f"Running scheduled job for templates: {templates}")
for template_name in templates:
try:
results = await self.generate_content(template_name)
for result in results:
if result.success:
logger.info(f"Successfully generated {result.output_path}")
else:
logger.error(f"Failed to generate content: {result.error_message}")
except Exception as e:
logger.error(f"Error in scheduled job for {template_name}: {e}")
def _setup_file_watch_trigger(self, trigger_config: Dict[str, Any]):
"""Setup file system watcher for automatic regeneration"""
paths = trigger_config['paths']
templates = trigger_config['templates']
class FileChangeHandler(FileSystemEventHandler):
def __init__(self, generator):
self.generator = generator
self.templates = templates
def on_modified(self, event):
if not event.is_directory:
logger.info(f"File changed: {event.src_path}")
asyncio.run(self.generator._run_scheduled_job(self.templates))
event_handler = FileChangeHandler(self)
observer = Observer()
for path in paths:
observer.schedule(event_handler, path, recursive=True)
observer.start()
logger.info(f"Started file watcher for paths: {paths}")
def _setup_webhook_trigger(self, trigger_config: Dict[str, Any]):
"""Setup webhook endpoint for content regeneration"""
# This would typically integrate with a web framework like Flask/FastAPI
# For simplicity, this is just a placeholder
logger.info(f"Webhook trigger configured for endpoint: {trigger_config['endpoint']}")
async def generate_all_content(self, force: bool = False) -> Dict[str, List[GenerationResult]]:
"""Generate content for all configured templates"""
all_results = {}
for template_name in self.templates:
try:
results = await self.generate_content(template_name, force=force)
all_results[template_name] = results
except Exception as e:
logger.error(f"Error generating {template_name}: {e}")
all_results[template_name] = [GenerationResult(
template_name=template_name,
output_path="",
success=False,
error_message=str(e)
)]
return all_results
def get_generation_stats(self) -> Dict[str, Any]:
"""Get statistics about content generation"""
total_templates = len(self.templates)
active_data_sources = len([
source for source in self.data_sources.values()
if source.name in self.cache_timestamps
])
cache_info = {}
for source_name, timestamp in self.cache_timestamps.items():
age_seconds = time.time() - timestamp
cache_info[source_name] = {
'age_seconds': age_seconds,
'age_minutes': age_seconds / 60,
'is_fresh': age_seconds < self.data_sources[source_name].cache_duration
}
return {
'total_templates': total_templates,
'active_data_sources': active_data_sources,
'cache_info': cache_info,
'last_generation': datetime.now().isoformat()
}
# CLI Interface
async def main():
import argparse
parser = argparse.ArgumentParser(description='Dynamic Content Generator')
parser.add_argument('--config', required=True, help='Configuration file path')
parser.add_argument('--template', help='Generate specific template')
parser.add_argument('--force', action='store_true', help='Force regeneration ignoring cache')
parser.add_argument('--stats', action='store_true', help='Show generation statistics')
args = parser.parse_args()
generator = ContentGenerationEngine(args.config)
if args.stats:
stats = generator.get_generation_stats()
print(json.dumps(stats, indent=2))
return
if args.template:
results = await generator.generate_content(args.template, force=args.force)
else:
all_results = await generator.generate_all_content(force=args.force)
results = []
for template_results in all_results.values():
results.extend(template_results)
# Print summary
successful = len([r for r in results if r.success])
failed = len([r for r in results if not r.success])
print(f"Generation complete: {successful} successful, {failed} failed")
for result in results:
if result.success:
print(f"✅ {result.output_path}")
else:
print(f"❌ {result.template_name}: {result.error_message}")
if __name__ == "__main__":
asyncio.run(main())
```
Integration with CI/CD Systems
GitHub Actions Workflow Integration
Implementing automated content generation in continuous integration pipelines:
# CI/CD Integration for Dynamic Content
## GitHub Actions Workflow
```yaml
# .github/workflows/content-generation.yml
name: Dynamic Content Generation
on:
push:
branches: [ main, develop ]
paths:
- 'data/**'
- 'templates/**'
- 'content-config.yml'
schedule:
- cron: '0 6 * * *' # Daily at 6 AM
workflow_dispatch:
inputs:
force_regeneration:
description: 'Force full regeneration'
required: false
default: 'false'
specific_template:
description: 'Generate specific template only'
required: false
jobs:
generate-content:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
with:
token: ${{ secrets.GITHUB_TOKEN }}
fetch-depth: 0
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install jinja2 aiohttp psycopg2-binary pandas pyyaml watchdog schedule
- name: Configure data source credentials
run: |
echo "DATABASE_URL=${{ secrets.DATABASE_URL }}" >> $GITHUB_ENV
echo "API_KEY=${{ secrets.API_KEY }}" >> $GITHUB_ENV
- name: Generate content
run: |
if [ "${{ github.event.inputs.specific_template }}" != "" ]; then
python content-generator.py \
--config content-config.yml \
--template "${{ github.event.inputs.specific_template }}" \
${{ github.event.inputs.force_regeneration == 'true' && '--force' || '' }}
else
python content-generator.py \
--config content-config.yml \
${{ github.event.inputs.force_regeneration == 'true' && '--force' || '' }}
fi
- name: Validate generated content
run: |
python scripts/validate-generated-content.py \
--content-dir docs \
--check-links \
--check-frontmatter \
--output validation-report.json
- name: Build site for testing
run: |
bundle install
bundle exec jekyll build
- name: Test generated content
run: |
python scripts/test-content-quality.py \
--site-dir _site \
--check-performance \
--check-accessibility \
--output test-results.json
- name: Commit generated content
if: success()
run: |
git config --global user.name 'Content Generator Bot'
git config --global user.email '[email protected]'
git add docs/
if ! git diff --staged --quiet; then
git commit -m "🤖 Automated content generation
Generated content for:
- API documentation
- Metrics reports
- Feature guides
Triggered by: ${{ github.event_name }}
${{ github.event.inputs.specific_template && format('Template: {0}', github.event.inputs.specific_template) || '' }}
🤖 Generated with Content Pipeline
"
git push
else
echo "No changes to commit"
fi
- name: Upload generation artifacts
if: always()
uses: actions/upload-artifact@v3
with:
name: content-generation-report
path: |
validation-report.json
test-results.json
generation-stats.json
deploy-content:
runs-on: ubuntu-latest
needs: generate-content
if: github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v4
with:
ref: main # Get the latest content
- name: Deploy to production
run: |
# Deploy generated content to production environment
echo "Deploying generated content to production"
# Example: Deploy to GitHub Pages
npm install -g @11ty/eleventy
eleventy
- name: Notify stakeholders
if: success()
uses: 8398a7/action-slack@v3
with:
status: success
text: "📚 Documentation updated successfully with latest content generation"
env:
SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
- name: Update content metrics
run: |
python scripts/update-content-metrics.py \
--generated-files docs/ \
--analytics-endpoint "${{ secrets.ANALYTICS_ENDPOINT }}" \
--api-key "${{ secrets.ANALYTICS_API_KEY }}"
content-quality-check:
runs-on: ubuntu-latest
needs: generate-content
steps:
- uses: actions/checkout@v4
- name: Quality assurance checks
run: |
# Check for content quality issues
python scripts/content-qa.py \
--content-dir docs \
--check-duplicates \
--check-broken-references \
--check-consistency \
--output qa-report.html
- name: Performance benchmark
run: |
# Benchmark content generation performance
python scripts/benchmark-generation.py \
--config content-config.yml \
--iterations 5 \
--output benchmark-results.json
- name: Update documentation metrics
run: |
# Update internal metrics about documentation
python scripts/update-docs-metrics.py \
--benchmark-data benchmark-results.json \
--qa-data qa-report.html \
--target-endpoint "${{ secrets.METRICS_ENDPOINT }}"
```
## Advanced Content Validation System
```python
#!/usr/bin/env python3
# validate-generated-content.py - Content validation for generated files
import os
import json
import yaml
import requests
from pathlib import Path
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
import frontmatter
from urllib.parse import urljoin, urlparse
import re
from datetime import datetime
import concurrent.futures
import threading
import time
@dataclass
class ValidationResult:
"""Result of content validation"""
file_path: str
validation_type: str
status: str # 'passed', 'failed', 'warning'
message: str
details: Optional[Dict[str, Any]] = None
class ContentValidator:
"""Comprehensive validation for generated content"""
def __init__(self, content_dir: str, base_url: str = ""):
self.content_dir = Path(content_dir)
self.base_url = base_url
self.results = []
self.link_cache = {}
self.validation_stats = {
'total_files': 0,
'passed': 0,
'failed': 0,
'warnings': 0
}
def validate_all_content(self,
check_links: bool = True,
check_frontmatter: bool = True,
check_content_quality: bool = True,
check_consistency: bool = True) -> List[ValidationResult]:
"""Run all validation checks on generated content"""
markdown_files = list(self.content_dir.glob('**/*.md'))
self.validation_stats['total_files'] = len(markdown_files)
print(f"Validating {len(markdown_files)} files...")
# Validate each file
for file_path in markdown_files:
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
# Parse frontmatter and content
post = frontmatter.loads(content)
if check_frontmatter:
self._validate_frontmatter(file_path, post)
if check_content_quality:
self._validate_content_quality(file_path, post)
if check_consistency:
self._validate_consistency(file_path, post)
if check_links:
self._validate_links(file_path, post)
except Exception as e:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="file_processing",
status="failed",
message=f"Error processing file: {str(e)}"
))
# Additional cross-file validations
if check_consistency:
self._validate_cross_file_consistency(markdown_files)
# Update statistics
self._calculate_stats()
return self.results
def _validate_frontmatter(self, file_path: Path, post) -> None:
"""Validate frontmatter completeness and format"""
metadata = post.metadata
# Required fields
required_fields = ['title', 'description', 'category', 'date']
missing_fields = [field for field in required_fields if field not in metadata]
if missing_fields:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="frontmatter",
status="failed",
message=f"Missing required frontmatter fields: {', '.join(missing_fields)}"
))
# Validate field formats
if 'date' in metadata:
try:
if isinstance(metadata['date'], str):
datetime.strptime(metadata['date'], '%Y-%m-%d')
except ValueError:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="frontmatter",
status="failed",
message="Invalid date format in frontmatter (expected YYYY-MM-DD)"
))
# Title length check
if 'title' in metadata and len(metadata['title']) > 200:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="frontmatter",
status="warning",
message="Title is very long (over 200 characters)"
))
# Description length check
if 'description' in metadata:
desc_length = len(metadata['description'])
if desc_length < 50:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="frontmatter",
status="warning",
message="Description is too short (under 50 characters)"
))
elif desc_length > 300:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="frontmatter",
status="warning",
message="Description is too long (over 300 characters)"
))
def _validate_content_quality(self, file_path: Path, post) -> None:
"""Validate content quality metrics"""
content = post.content
# Word count check
word_count = len(content.split())
if word_count < 100:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="content_quality",
status="warning",
message=f"Content is very short ({word_count} words)"
))
# Check for placeholder content
placeholder_patterns = [
r'\[placeholder\]',
r'\[todo\]',
r'\[tbd\]',
r'lorem ipsum',
r'placeholder text'
]
for pattern in placeholder_patterns:
if re.search(pattern, content, re.IGNORECASE):
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="content_quality",
status="failed",
message=f"Placeholder content detected: {pattern}"
))
# Check for proper heading structure
headings = re.findall(r'^(#+)\s+(.+)$', content, re.MULTILINE)
if headings:
# Check if starts with h1
first_heading_level = len(headings[0][0])
if first_heading_level != 1:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="content_quality",
status="warning",
message="Document should start with h1 heading"
))
# Check heading hierarchy
prev_level = 0
for i, (hashes, text) in enumerate(headings):
level = len(hashes)
if level > prev_level + 1 and prev_level > 0:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="content_quality",
status="warning",
message=f"Heading hierarchy skip detected at line with '{text}'"
))
prev_level = level
# Check for code block formatting
code_blocks = re.findall(r'```(\w+)?\n(.*?)\n```', content, re.DOTALL)
for lang, code in code_blocks:
if not lang:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="content_quality",
status="warning",
message="Code block without language specification"
))
# Check for unterminated code blocks
if '```' in code:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="content_quality",
status="failed",
message="Potentially unterminated code block detected"
))
def _validate_consistency(self, file_path: Path, post) -> None:
"""Validate content consistency"""
metadata = post.metadata
content = post.content
# Check if title matches content
title = metadata.get('title', '')
first_heading = re.search(r'^#\s+(.+)$', content, re.MULTILINE)
if first_heading and title:
content_title = first_heading.group(1).strip()
if title.lower() != content_title.lower():
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="consistency",
status="warning",
message="Frontmatter title doesn't match first heading"
))
# Check filename consistency
expected_filename_parts = []
if 'date' in metadata:
date_str = metadata['date']
if isinstance(date_str, str):
expected_filename_parts.append(date_str)
if title:
# Convert title to slug
slug = re.sub(r'[^\w\s-]', '', title.lower())
slug = re.sub(r'[\s_-]+', '-', slug)
expected_filename_parts.append(slug[:50]) # Limit length
if expected_filename_parts:
expected_filename = '-'.join(expected_filename_parts) + '.md'
actual_filename = file_path.name
if actual_filename != expected_filename:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="consistency",
status="warning",
message=f"Filename '{actual_filename}' doesn't follow expected pattern '{expected_filename}'"
))
def _validate_links(self, file_path: Path, post) -> None:
"""Validate internal and external links"""
content = post.content
# Find all markdown links
markdown_links = re.findall(r'\[([^\]]+)\]\(([^)]+)\)', content)
for link_text, link_url in markdown_links:
self._check_single_link(file_path, link_text, link_url)
# Find all HTML links
html_links = re.findall(r'<a[^>]+href=["\']([^"\']+)["\'][^>]*>([^<]*)</a>', content, re.IGNORECASE)
for link_url, link_text in html_links:
self._check_single_link(file_path, link_text, link_url)
def _check_single_link(self, file_path: Path, link_text: str, link_url: str) -> None:
"""Check a single link for validity"""
# Skip anchor links and mailto links
if link_url.startswith('#') or link_url.startswith('mailto:'):
return
# Check internal links
if link_url.startswith('/') or not urlparse(link_url).netloc:
# Internal link
if link_url.startswith('/'):
# Absolute internal link
target_path = self.content_dir / link_url.lstrip('/')
else:
# Relative link
target_path = file_path.parent / link_url
# Check if file exists
if not target_path.exists():
# Try with .md extension
if not str(target_path).endswith('.md'):
target_path_md = target_path.with_suffix('.md')
if not target_path_md.exists():
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="links",
status="failed",
message=f"Broken internal link: '{link_url}' -> '{link_text}'"
))
else:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="links",
status="failed",
message=f"Broken internal link: '{link_url}' -> '{link_text}'"
))
else:
# External link - check if we should validate
if self._should_check_external_link(link_url):
self._check_external_link(file_path, link_text, link_url)
def _should_check_external_link(self, url: str) -> bool:
"""Determine if external link should be checked"""
# Skip certain domains that might block automated requests
skip_domains = [
'localhost',
'127.0.0.1',
'example.com',
'test.com'
]
parsed = urlparse(url)
return parsed.netloc not in skip_domains
def _check_external_link(self, file_path: Path, link_text: str, link_url: str) -> None:
"""Check external link availability"""
if link_url in self.link_cache:
# Use cached result
status = self.link_cache[link_url]
else:
try:
# Make HEAD request first (faster)
response = requests.head(link_url, timeout=10, allow_redirects=True)
status = response.status_code
# If HEAD fails, try GET
if status >= 400:
response = requests.get(link_url, timeout=10, allow_redirects=True)
status = response.status_code
self.link_cache[link_url] = status
except requests.RequestException as e:
status = f"error: {str(e)}"
self.link_cache[link_url] = status
if isinstance(status, int) and status >= 400:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="links",
status="failed",
message=f"Broken external link (HTTP {status}): '{link_url}' -> '{link_text}'"
))
elif isinstance(status, str) and status.startswith("error:"):
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="links",
status="warning",
message=f"External link check failed: '{link_url}' -> '{link_text}'"
))
def _validate_cross_file_consistency(self, files: List[Path]) -> None:
"""Validate consistency across multiple files"""
# Check for duplicate titles
titles = {}
categories = set()
for file_path in files:
try:
with open(file_path, 'r', encoding='utf-8') as f:
post = frontmatter.load(f)
title = post.metadata.get('title', '')
category = post.metadata.get('category', '')
if title:
if title in titles:
self.results.append(ValidationResult(
file_path=str(file_path),
validation_type="cross_file_consistency",
status="warning",
message=f"Duplicate title found: '{title}' (also in {titles[title]})"
))
else:
titles[title] = str(file_path)
if category:
categories.add(category)
except Exception as e:
continue
# Report category statistics
print(f"Found {len(categories)} unique categories: {', '.join(sorted(categories))}")
def _calculate_stats(self) -> None:
"""Calculate validation statistics"""
for result in self.results:
if result.status == "passed":
self.validation_stats['passed'] += 1
elif result.status == "failed":
self.validation_stats['failed'] += 1
elif result.status == "warning":
self.validation_stats['warnings'] += 1
def generate_report(self, output_path: Optional[str] = None) -> Dict[str, Any]:
"""Generate comprehensive validation report"""
report = {
'validation_summary': self.validation_stats,
'validation_timestamp': datetime.now().isoformat(),
'results_by_type': {},
'results_by_status': {},
'detailed_results': []
}
# Group results by type and status
for result in self.results:
# By type
if result.validation_type not in report['results_by_type']:
report['results_by_type'][result.validation_type] = []
report['results_by_type'][result.validation_type].append({
'file': result.file_path,
'status': result.status,
'message': result.message
})
# By status
if result.status not in report['results_by_status']:
report['results_by_status'][result.status] = []
report['results_by_status'][result.status].append({
'file': result.file_path,
'type': result.validation_type,
'message': result.message
})
# Detailed results
report['detailed_results'].append({
'file_path': result.file_path,
'validation_type': result.validation_type,
'status': result.status,
'message': result.message,
'details': result.details
})
if output_path:
with open(output_path, 'w') as f:
json.dump(report, f, indent=2)
return report
def print_summary(self) -> None:
"""Print validation summary to console"""
stats = self.validation_stats
total_issues = stats['failed'] + stats['warnings']
print(f"\n=== Validation Summary ===")
print(f"Total files: {stats['total_files']}")
print(f"Issues found: {total_issues}")
print(f" - Failed: {stats['failed']}")
print(f" - Warnings: {stats['warnings']}")
if total_issues == 0:
print("✅ All validations passed!")
else:
print(f"⚠️ {total_issues} issues need attention")
# Show top issues by type
issue_types = {}
for result in self.results:
if result.status in ['failed', 'warning']:
issue_types[result.validation_type] = issue_types.get(result.validation_type, 0) + 1
print("\nIssues by type:")
for issue_type, count in sorted(issue_types.items(), key=lambda x: x[1], reverse=True):
print(f" - {issue_type}: {count}")
# CLI Interface
def main():
import argparse
parser = argparse.ArgumentParser(description='Validate generated content')
parser.add_argument('--content-dir', required=True, help='Directory containing content to validate')
parser.add_argument('--base-url', help='Base URL for link validation')
parser.add_argument('--check-links', action='store_true', help='Validate internal and external links')
parser.add_argument('--check-frontmatter', action='store_true', help='Validate frontmatter')
parser.add_argument('--check-content-quality', action='store_true', help='Check content quality')
parser.add_argument('--check-consistency', action='store_true', help='Check consistency')
parser.add_argument('--output', help='Output file for validation report')
parser.add_argument('--fail-on-error', action='store_true', help='Exit with error code if validation fails')
args = parser.parse_args()
validator = ContentValidator(args.content_dir, args.base_url or "")
results = validator.validate_all_content(
check_links=args.check_links,
check_frontmatter=args.check_frontmatter,
check_content_quality=args.check_content_quality,
check_consistency=args.check_consistency
)
# Generate and save report
report = validator.generate_report(args.output)
# Print summary
validator.print_summary()
# Exit with error code if requested and validation failed
if args.fail_on_error and validator.validation_stats['failed'] > 0:
exit(1)
if __name__ == "__main__":
main()
```
Integration with Documentation Systems
Dynamic content generation integrates seamlessly with comprehensive documentation workflows. When combined with accessibility and WCAG compliance systems, automated content generation ensures generated documentation maintains accessibility standards while producing consistent, high-quality content that serves users with diverse abilities and technology requirements.
For comprehensive content management, automation systems work effectively with table data validation and quality assurance frameworks to ensure that dynamically generated tabular content maintains data integrity and formatting standards while preventing common table errors through automated validation and correction processes.
When building sophisticated documentation platforms, content generation complements advanced code syntax highlighting systems by enabling automated generation of code documentation with proper syntax highlighting, language detection, and interactive examples that enhance developer experience while maintaining consistency across large codebases.
Advanced Automation Patterns
Multi-Source Content Aggregation
Implementing sophisticated content aggregation from multiple data sources:
# Advanced Content Aggregation Patterns
## Multi-API Content Fusion
```python
#!/usr/bin/env python3
# multi-source-aggregator.py - Advanced content aggregation system
import asyncio
import aiohttp
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
import json
import yaml
from datetime import datetime, timedelta
import logging
from jinja2 import Environment, DictLoader
import hashlib
@dataclass
class DataSourceConfig:
"""Configuration for individual data source"""
name: str
type: str
endpoint: str
auth_method: str = "none"
auth_config: Dict[str, Any] = field(default_factory=dict)
transform_pipeline: List[str] = field(default_factory=list)
cache_ttl: int = 3600
retry_config: Dict[str, Any] = field(default_factory=lambda: {"max_retries": 3, "delay": 1})
@dataclass
class ContentTemplate:
"""Template for generating aggregated content"""
name: str
template_string: str
output_path: str
data_sources: List[str]
aggregation_rules: Dict[str, Any] = field(default_factory=dict)
update_frequency: str = "hourly"
class MultiSourceContentAggregator:
"""Advanced content aggregator supporting multiple data sources"""
def __init__(self, config_path: str):
with open(config_path, 'r') as f:
self.config = yaml.safe_load(f)
self.data_sources = {
ds['name']: DataSourceConfig(**ds)
for ds in self.config['data_sources']
}
self.templates = {
tmpl['name']: ContentTemplate(**tmpl)
for tmpl in self.config['templates']
}
self.data_cache = {}
self.cache_timestamps = {}
self.session = None
# Initialize Jinja2 environment
self.jinja_env = Environment(loader=DictLoader({}))
self._setup_custom_filters()
logging.basicConfig(level=logging.INFO)
self.logger = logging.getLogger(__name__)
async def __aenter__(self):
"""Async context manager entry"""
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Async context manager exit"""
if self.session:
await self.session.close()
def _setup_custom_filters(self):
"""Setup custom Jinja2 filters for content aggregation"""
def merge_lists_filter(list1: List[Any], list2: List[Any], key: str = 'id') -> List[Any]:
"""Merge two lists based on a key"""
result = list1.copy()
existing_keys = {item.get(key) for item in list1}
for item in list2:
if item.get(key) not in existing_keys:
result.append(item)
return result
def group_by_filter(items: List[Dict], key: str) -> Dict[str, List[Dict]]:
"""Group list items by a key"""
grouped = {}
for item in items:
group_key = item.get(key, 'unknown')
if group_key not in grouped:
grouped[group_key] = []
grouped[group_key].append(item)
return grouped
def sort_by_priority_filter(items: List[Dict], priority_key: str = 'priority') -> List[Dict]:
"""Sort items by priority (higher numbers first)"""
return sorted(items, key=lambda x: x.get(priority_key, 0), reverse=True)
def calculate_metrics_filter(items: List[Dict], metric_key: str) -> Dict[str, Any]:
"""Calculate aggregate metrics for a list of items"""
values = [item.get(metric_key, 0) for item in items if isinstance(item.get(metric_key), (int, float))]
if not values:
return {'count': 0, 'sum': 0, 'avg': 0, 'min': 0, 'max': 0}
return {
'count': len(values),
'sum': sum(values),
'avg': sum(values) / len(values),
'min': min(values),
'max': max(values)
}
def filter_recent_filter(items: List[Dict], date_key: str, days: int = 7) -> List[Dict]:
"""Filter items to only include recent entries"""
cutoff_date = datetime.now() - timedelta(days=days)
recent_items = []
for item in items:
item_date = item.get(date_key)
if item_date:
try:
if isinstance(item_date, str):
item_date = datetime.fromisoformat(item_date.replace('Z', '+00:00'))
if item_date >= cutoff_date:
recent_items.append(item)
except ValueError:
continue
return recent_items
# Register filters
self.jinja_env.filters['merge_lists'] = merge_lists_filter
self.jinja_env.filters['group_by'] = group_by_filter
self.jinja_env.filters['sort_by_priority'] = sort_by_priority_filter
self.jinja_env.filters['calculate_metrics'] = calculate_metrics_filter
self.jinja_env.filters['filter_recent'] = filter_recent_filter
async def fetch_all_data(self, force_refresh: bool = False) -> Dict[str, Any]:
"""Fetch data from all configured sources"""
tasks = []
for source_name, source_config in self.data_sources.items():
if force_refresh or self._is_cache_stale(source_name, source_config.cache_ttl):
task = asyncio.create_task(
self._fetch_source_data(source_name, source_config)
)
tasks.append(task)
if tasks:
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(results):
if isinstance(result, Exception):
self.logger.error(f"Error fetching data: {result}")
return self.data_cache
def _is_cache_stale(self, source_name: str, ttl: int) -> bool:
"""Check if cached data is stale"""
if source_name not in self.cache_timestamps:
return True
age = datetime.now().timestamp() - self.cache_timestamps[source_name]
return age > ttl
async def _fetch_source_data(self, source_name: str, config: DataSourceConfig) -> None:
"""Fetch data from a single source with retry logic"""
for attempt in range(config.retry_config.get('max_retries', 3)):
try:
if config.type == 'rest_api':
data = await self._fetch_rest_api_data(config)
elif config.type == 'graphql':
data = await self._fetch_graphql_data(config)
elif config.type == 'database':
data = await self._fetch_database_data(config)
elif config.type == 'file':
data = await self._fetch_file_data(config)
else:
raise ValueError(f"Unknown data source type: {config.type}")
# Apply transformation pipeline
for transform in config.transform_pipeline:
data = await self._apply_transformation(data, transform)
self.data_cache[source_name] = data
self.cache_timestamps[source_name] = datetime.now().timestamp()
self.logger.info(f"Successfully fetched data from {source_name}")
return
except Exception as e:
self.logger.warning(f"Attempt {attempt + 1} failed for {source_name}: {e}")
if attempt < config.retry_config.get('max_retries', 3) - 1:
await asyncio.sleep(config.retry_config.get('delay', 1) * (2 ** attempt))
else:
self.logger.error(f"Failed to fetch data from {source_name} after all retries")
raise
async def _fetch_rest_api_data(self, config: DataSourceConfig) -> Dict[str, Any]:
"""Fetch data from REST API"""
headers = {}
# Handle authentication
if config.auth_method == 'bearer_token':
headers['Authorization'] = f"Bearer {config.auth_config['token']}"
elif config.auth_method == 'api_key':
if config.auth_config.get('header_name'):
headers[config.auth_config['header_name']] = config.auth_config['api_key']
else:
headers['X-API-Key'] = config.auth_config['api_key']
elif config.auth_method == 'basic_auth':
import base64
credentials = f"{config.auth_config['username']}:{config.auth_config['password']}"
encoded = base64.b64encode(credentials.encode()).decode()
headers['Authorization'] = f"Basic {encoded}"
async with self.session.get(config.endpoint, headers=headers) as response:
response.raise_for_status()
return await response.json()
async def _fetch_graphql_data(self, config: DataSourceConfig) -> Dict[str, Any]:
"""Fetch data from GraphQL endpoint"""
headers = {'Content-Type': 'application/json'}
# Handle authentication
if config.auth_method == 'bearer_token':
headers['Authorization'] = f"Bearer {config.auth_config['token']}"
query = config.auth_config.get('query', '')
variables = config.auth_config.get('variables', {})
payload = {
'query': query,
'variables': variables
}
async with self.session.post(config.endpoint, json=payload, headers=headers) as response:
response.raise_for_status()
result = await response.json()
if 'errors' in result:
raise Exception(f"GraphQL errors: {result['errors']}")
return result.get('data', {})
async def _fetch_database_data(self, config: DataSourceConfig) -> Dict[str, Any]:
"""Fetch data from database"""
# This would implement database connectivity
# For this example, we'll simulate with a placeholder
return {
'message': 'Database connectivity would be implemented here',
'config': config.endpoint
}
async def _fetch_file_data(self, config: DataSourceConfig) -> Dict[str, Any]:
"""Fetch data from file"""
if config.endpoint.startswith('http'):
async with self.session.get(config.endpoint) as response:
response.raise_for_status()
if config.endpoint.endswith('.json'):
return await response.json()
elif config.endpoint.endswith('.yaml') or config.endpoint.endswith('.yml'):
text = await response.text()
return yaml.safe_load(text)
else:
return {'content': await response.text()}
else:
# Local file
with open(config.endpoint, 'r') as f:
if config.endpoint.endswith('.json'):
return json.load(f)
elif config.endpoint.endswith('.yaml') or config.endpoint.endswith('.yml'):
return yaml.safe_load(f)
else:
return {'content': f.read()}
async def _apply_transformation(self, data: Dict[str, Any], transform: str) -> Dict[str, Any]:
"""Apply data transformation"""
# Simple transformations - in production, this would be more sophisticated
if transform == 'normalize_dates':
return self._normalize_dates(data)
elif transform == 'extract_summary':
return self._extract_summary(data)
elif transform == 'flatten_nested':
return self._flatten_nested_data(data)
elif transform == 'calculate_derived_metrics':
return self._calculate_derived_metrics(data)
return data
def _normalize_dates(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Normalize date formats in data"""
def normalize_date_value(value):
if isinstance(value, str):
# Try to parse and normalize date strings
for fmt in ['%Y-%m-%dT%H:%M:%SZ', '%Y-%m-%d %H:%M:%S', '%Y-%m-%d']:
try:
dt = datetime.strptime(value, fmt)
return dt.isoformat()
except ValueError:
continue
return value
def normalize_recursive(obj):
if isinstance(obj, dict):
return {k: normalize_recursive(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [normalize_recursive(item) for item in obj]
else:
return normalize_date_value(obj)
return normalize_recursive(data)
def _extract_summary(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Extract summary information from data"""
summary = {
'total_items': 0,
'categories': set(),
'date_range': {'earliest': None, 'latest': None},
'data_types': set()
}
def analyze_item(item, path=""):
if isinstance(item, dict):
for key, value in item.items():
new_path = f"{path}.{key}" if path else key
analyze_item(value, new_path)
if 'category' in key.lower():
summary['categories'].add(str(value))
elif isinstance(item, list):
summary['total_items'] += len(item)
for subitem in item:
analyze_item(subitem, path)
else:
summary['data_types'].add(type(item).__name__)
analyze_item(data)
# Convert sets to lists for JSON serialization
data['_summary'] = {
'total_items': summary['total_items'],
'categories': list(summary['categories']),
'data_types': list(summary['data_types'])
}
return data
def _flatten_nested_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Flatten deeply nested data structures"""
def flatten_dict(d, parent_key='', sep='_'):
items = []
for k, v in d.items():
new_key = f"{parent_key}{sep}{k}" if parent_key else k
if isinstance(v, dict):
items.extend(flatten_dict(v, new_key, sep=sep).items())
else:
items.append((new_key, v))
return dict(items)
if isinstance(data, dict):
data['_flattened'] = flatten_dict(data)
return data
def _calculate_derived_metrics(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Calculate derived metrics from raw data"""
metrics = {}
# Look for numeric fields to calculate aggregations
def find_numeric_fields(obj, path=""):
if isinstance(obj, dict):
for key, value in obj.items():
new_path = f"{path}.{key}" if path else key
if isinstance(value, (int, float)):
if 'metrics' not in data:
data['metrics'] = {}
if path not in data['metrics']:
data['metrics'][path] = []
data['metrics'][path].append(value)
elif isinstance(value, (dict, list)):
find_numeric_fields(value, new_path)
elif isinstance(obj, list):
for item in obj:
find_numeric_fields(item, path)
find_numeric_fields(data)
# Calculate aggregations
if 'metrics' in data:
for field_path, values in data['metrics'].items():
if values:
metrics[field_path] = {
'count': len(values),
'sum': sum(values),
'avg': sum(values) / len(values),
'min': min(values),
'max': max(values)
}
data['_derived_metrics'] = metrics
return data
async def generate_aggregated_content(self, template_name: str) -> str:
"""Generate content using aggregated data from multiple sources"""
if template_name not in self.templates:
raise ValueError(f"Template '{template_name}' not found")
template_config = self.templates[template_name]
# Fetch required data sources
required_data = {}
for source_name in template_config.data_sources:
if source_name not in self.data_cache or self._is_cache_stale(
source_name, self.data_sources[source_name].cache_ttl
):
await self._fetch_source_data(source_name, self.data_sources[source_name])
required_data[source_name] = self.data_cache[source_name]
# Apply aggregation rules
aggregated_data = await self._apply_aggregation_rules(required_data, template_config.aggregation_rules)
# Create template context
context = {
'data': aggregated_data,
'sources': required_data,
'config': self.config,
'generation_time': datetime.now(),
'template_name': template_name
}
# Render template
template = self.jinja_env.from_string(template_config.template_string)
content = template.render(**context)
# Write output file
output_path = template_config.output_path
os.makedirs(os.path.dirname(output_path), exist_ok=True)
with open(output_path, 'w', encoding='utf-8') as f:
f.write(content)
self.logger.info(f"Generated aggregated content: {output_path}")
return content
async def _apply_aggregation_rules(self, data: Dict[str, Any], rules: Dict[str, Any]) -> Dict[str, Any]:
"""Apply aggregation rules to combine data from multiple sources"""
result = {}
for rule_name, rule_config in rules.items():
rule_type = rule_config.get('type', 'merge')
if rule_type == 'merge':
# Merge data from specified sources
sources = rule_config.get('sources', [])
merged_data = {}
for source in sources:
if source in data:
merged_data.update(data[source])
result[rule_name] = merged_data
elif rule_type == 'combine_lists':
# Combine lists from multiple sources
sources = rule_config.get('sources', [])
list_key = rule_config.get('list_key', 'items')
combined_list = []
for source in sources:
if source in data and list_key in data[source]:
source_items = data[source][list_key]
if isinstance(source_items, list):
# Add source identifier to each item
for item in source_items:
if isinstance(item, dict):
item['_source'] = source
combined_list.extend(source_items)
result[rule_name] = combined_list
elif rule_type == 'cross_reference':
# Cross-reference data between sources
primary_source = rule_config.get('primary_source')
reference_source = rule_config.get('reference_source')
join_key = rule_config.get('join_key', 'id')
if primary_source in data and reference_source in data:
primary_data = data[primary_source]
reference_data = data[reference_source]
# Create lookup table for reference data
ref_lookup = {}
if isinstance(reference_data, dict) and 'items' in reference_data:
for item in reference_data['items']:
if isinstance(item, dict) and join_key in item:
ref_lookup[item[join_key]] = item
# Enhance primary data with reference information
enhanced_data = primary_data.copy()
if isinstance(enhanced_data, dict) and 'items' in enhanced_data:
for item in enhanced_data['items']:
if isinstance(item, dict) and join_key in item:
ref_item = ref_lookup.get(item[join_key])
if ref_item:
item['_reference'] = ref_item
result[rule_name] = enhanced_data
return result
# Example usage configuration
example_config = """
data_sources:
- name: "github_issues"
type: "rest_api"
endpoint: "https://api.github.com/repos/owner/repo/issues"
auth_method: "bearer_token"
auth_config:
token: "${GITHUB_TOKEN}"
transform_pipeline:
- "normalize_dates"
- "extract_summary"
cache_ttl: 1800
- name: "user_metrics"
type: "rest_api"
endpoint: "https://analytics.example.com/api/users/metrics"
auth_method: "api_key"
auth_config:
api_key: "${ANALYTICS_API_KEY}"
header_name: "X-Analytics-Key"
transform_pipeline:
- "calculate_derived_metrics"
cache_ttl: 3600
- name: "feature_flags"
type: "file"
endpoint: "config/features.json"
transform_pipeline:
- "extract_summary"
cache_ttl: 600
templates:
- name: "status_dashboard"
output_path: "docs/status/dashboard.md"
data_sources: ["github_issues", "user_metrics", "feature_flags"]
aggregation_rules:
recent_activity:
type: "combine_lists"
sources: ["github_issues"]
list_key: "items"
metrics_summary:
type: "merge"
sources: ["user_metrics"]
feature_status:
type: "cross_reference"
primary_source: "feature_flags"
reference_source: "user_metrics"
join_key: "feature_id"
template_string: |
---
title: "System Status Dashboard"
description: "Real-time system status and metrics overview"
category: "Status"
date: {{ generation_time.strftime('%Y-%m-%d') }}
---
# System Status Dashboard
*Last updated: {{ generation_time.strftime('%Y-%m-%d %H:%M:%S') }} UTC*
## Recent Activity
{% set recent_issues = data.recent_activity | filter_recent('created_at', 7) | sort_by_priority('priority') %}
### Open Issues (Last 7 Days)
| Issue | Priority | Status | Created |
|:------|:---------|:-------|:--------|
{% for issue in recent_issues[:10] %}
| [#{{ issue.number }}]({{ issue.html_url }}) {{ issue.title }} | {{ issue.labels | selectattr('name', 'match', 'priority:.*') | map(attribute='name') | join(', ') | default('normal') }} | {{ issue.state }} | {{ issue.created_at | relative_time }} |
{% endfor %}
## System Metrics
{% set metrics = data.metrics_summary %}
### Performance Overview
- **Active Users**: {{ metrics.active_users | format_number('abbreviated') }}
- **Response Time**: {{ metrics.avg_response_time }}ms
- **Error Rate**: {{ metrics.error_rate }}%
- **Uptime**: {{ metrics.uptime_percentage }}%
### Usage Statistics
{% for category, stats in metrics.usage_by_category.items() %}
- **{{ category }}**: {{ stats.requests | format_number('abbreviated') }} requests
{% endfor %}
## Feature Status
{% set features = data.feature_status %}
### Feature Rollout Status
{% for feature in features.items | group_by('status').items() %}
#### {{ feature[0] | title }} Features
{% for item in feature[1] %}
- **{{ item.name }}**: {{ item.rollout_percentage }}% rollout
{% if item._reference %}
- Usage: {{ item._reference.usage_count | format_number('abbreviated') }}
{% endif %}
{% endfor %}
{% endfor %}
---
*This dashboard is automatically generated from multiple data sources and updated every hour.*
update_frequency: "hourly"
"""
# CLI interface for the aggregator
async def main():
import argparse
import os
parser = argparse.ArgumentParser(description='Multi-source content aggregator')
parser.add_argument('--config', required=True, help='Configuration file path')
parser.add_argument('--template', help='Specific template to generate')
parser.add_argument('--force-refresh', action='store_true', help='Force refresh of all data sources')
args = parser.parse_args()
async with MultiSourceContentAggregator(args.config) as aggregator:
# Fetch all data
await aggregator.fetch_all_data(force_refresh=args.force_refresh)
if args.template:
# Generate specific template
await aggregator.generate_aggregated_content(args.template)
print(f"Generated content for template: {args.template}")
else:
# Generate all templates
for template_name in aggregator.templates:
try:
await aggregator.generate_aggregated_content(template_name)
print(f"✅ Generated: {template_name}")
except Exception as e:
print(f"❌ Failed to generate {template_name}: {e}")
if __name__ == "__main__":
asyncio.run(main())
```
Performance Optimization and Scaling
Content Generation Performance Monitoring
Implementing comprehensive performance monitoring for large-scale content generation:
#!/usr/bin/env python3
# performance-monitor.py - Performance monitoring for content generation
import time
import psutil
import threading
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import statistics
import json
import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import multiprocessing as mp
@dataclass
class PerformanceMetrics:
"""Performance metrics for content generation operations"""
operation_id: str
operation_type: str
start_time: datetime
end_time: Optional[datetime] = None
duration_seconds: Optional[float] = None
memory_usage_mb: Optional[float] = None
cpu_usage_percent: Optional[float] = None
files_processed: Optional[int] = None
output_size_kb: Optional[float] = None
cache_hits: Optional[int] = None
cache_misses: Optional[int] = None
class ContentGenerationPerformanceMonitor:
"""Performance monitoring system for content generation"""
def __init__(self, enable_detailed_monitoring: bool = True):
self.enable_detailed_monitoring = enable_detailed_monitoring
self.metrics_history: List[PerformanceMetrics] = []
self.active_operations: Dict[str, PerformanceMetrics] = {}
self.system_metrics = []
# Start system monitoring thread
if enable_detailed_monitoring:
self._start_system_monitoring()
def _start_system_monitoring(self):
"""Start background system monitoring"""
def monitor_system():
while True:
try:
cpu_percent = psutil.cpu_percent(interval=1)
memory = psutil.virtual_memory()
disk = psutil.disk_usage('/')
self.system_metrics.append({
'timestamp': datetime.now(),
'cpu_percent': cpu_percent,
'memory_percent': memory.percent,
'memory_available_gb': memory.available / (1024**3),
'disk_free_gb': disk.free / (1024**3)
})
# Keep only last hour of system metrics
cutoff = datetime.now() - timedelta(hours=1)
self.system_metrics = [
m for m in self.system_metrics
if m['timestamp'] > cutoff
]
time.sleep(30) # Sample every 30 seconds
except Exception as e:
print(f"System monitoring error: {e}")
time.sleep(60)
monitor_thread = threading.Thread(target=monitor_system, daemon=True)
monitor_thread.start()
def start_operation(self, operation_id: str, operation_type: str) -> PerformanceMetrics:
"""Start monitoring an operation"""
metrics = PerformanceMetrics(
operation_id=operation_id,
operation_type=operation_type,
start_time=datetime.now()
)
self.active_operations[operation_id] = metrics
return metrics
def end_operation(self, operation_id: str,
files_processed: int = 0,
output_size_kb: float = 0,
cache_hits: int = 0,
cache_misses: int = 0) -> PerformanceMetrics:
"""End monitoring and record final metrics"""
if operation_id not in self.active_operations:
return None
metrics = self.active_operations[operation_id]
metrics.end_time = datetime.now()
metrics.duration_seconds = (metrics.end_time - metrics.start_time).total_seconds()
metrics.files_processed = files_processed
metrics.output_size_kb = output_size_kb
metrics.cache_hits = cache_hits
metrics.cache_misses = cache_misses
# Capture resource usage
try:
process = psutil.Process()
memory_info = process.memory_info()
metrics.memory_usage_mb = memory_info.rss / (1024 * 1024)
metrics.cpu_usage_percent = process.cpu_percent()
except Exception:
pass
# Move to history
self.metrics_history.append(metrics)
del self.active_operations[operation_id]
return metrics
def get_performance_summary(self,
operation_type: Optional[str] = None,
time_window: Optional[timedelta] = None) -> Dict[str, Any]:
"""Generate performance summary report"""
# Filter metrics
filtered_metrics = self.metrics_history
if operation_type:
filtered_metrics = [m for m in filtered_metrics if m.operation_type == operation_type]
if time_window:
cutoff = datetime.now() - time_window
filtered_metrics = [m for m in filtered_metrics if m.start_time >= cutoff]
if not filtered_metrics:
return {'error': 'No metrics found for criteria'}
# Calculate statistics
durations = [m.duration_seconds for m in filtered_metrics if m.duration_seconds is not None]
memory_usage = [m.memory_usage_mb for m in filtered_metrics if m.memory_usage_mb is not None]
files_processed = [m.files_processed for m in filtered_metrics if m.files_processed is not None]
summary = {
'operation_count': len(filtered_metrics),
'time_period': str(time_window) if time_window else 'All time',
'operation_type': operation_type or 'All types'
}
if durations:
summary['timing'] = {
'avg_duration': statistics.mean(durations),
'min_duration': min(durations),
'max_duration': max(durations),
'median_duration': statistics.median(durations),
'p95_duration': self._percentile(durations, 95),
'total_time': sum(durations)
}
if memory_usage:
summary['memory'] = {
'avg_memory_mb': statistics.mean(memory_usage),
'peak_memory_mb': max(memory_usage),
'min_memory_mb': min(memory_usage)
}
if files_processed:
total_files = sum(files_processed)
total_time = sum(durations) if durations else 1
summary['throughput'] = {
'total_files': total_files,
'avg_files_per_operation': statistics.mean(files_processed),
'files_per_second': total_files / total_time if total_time > 0 else 0
}
# Cache efficiency
total_hits = sum(m.cache_hits or 0 for m in filtered_metrics)
total_misses = sum(m.cache_misses or 0 for m in filtered_metrics)
total_cache_ops = total_hits + total_misses
if total_cache_ops > 0:
summary['cache'] = {
'hit_rate': (total_hits / total_cache_ops) * 100,
'total_hits': total_hits,
'total_misses': total_misses
}
return summary
def _percentile(self, data: List[float], percentile: int) -> float:
"""Calculate percentile value"""
sorted_data = sorted(data)
index = (percentile / 100) * (len(sorted_data) - 1)
if index.is_integer():
return sorted_data[int(index)]
else:
lower = sorted_data[int(index)]
upper = sorted_data[int(index) + 1]
return lower + (upper - lower) * (index - int(index))
def detect_performance_bottlenecks(self) -> List[Dict[str, Any]]:
"""Detect performance bottlenecks and anomalies"""
bottlenecks = []
# Analyze recent operations
recent_window = timedelta(minutes=30)
recent_metrics = [
m for m in self.metrics_history
if m.start_time >= datetime.now() - recent_window
]
if len(recent_metrics) < 5:
return bottlenecks
# Check for slow operations
durations = [m.duration_seconds for m in recent_metrics if m.duration_seconds]
if durations:
avg_duration = statistics.mean(durations)
threshold = avg_duration * 2
slow_ops = [m for m in recent_metrics if m.duration_seconds and m.duration_seconds > threshold]
if slow_ops:
bottlenecks.append({
'type': 'slow_operations',
'count': len(slow_ops),
'threshold_seconds': threshold,
'details': [
{
'operation_id': op.operation_id,
'duration': op.duration_seconds,
'type': op.operation_type
}
for op in slow_ops[:5] # Show top 5
]
})
# Check for memory issues
memory_usage = [m.memory_usage_mb for m in recent_metrics if m.memory_usage_mb]
if memory_usage:
avg_memory = statistics.mean(memory_usage)
high_memory_threshold = avg_memory * 1.5
high_memory_ops = [m for m in recent_metrics if m.memory_usage_mb and m.memory_usage_mb > high_memory_threshold]
if high_memory_ops:
bottlenecks.append({
'type': 'high_memory_usage',
'count': len(high_memory_ops),
'threshold_mb': high_memory_threshold,
'peak_usage_mb': max(m.memory_usage_mb for m in high_memory_ops)
})
# Check cache efficiency
total_hits = sum(m.cache_hits or 0 for m in recent_metrics)
total_misses = sum(m.cache_misses or 0 for m in recent_metrics)
total_cache_ops = total_hits + total_misses
if total_cache_ops > 0:
hit_rate = (total_hits / total_cache_ops) * 100
if hit_rate < 70: # Less than 70% hit rate
bottlenecks.append({
'type': 'low_cache_efficiency',
'hit_rate': hit_rate,
'total_operations': total_cache_ops,
'recommendation': 'Consider increasing cache size or TTL'
})
# Check system resource usage
if self.system_metrics:
recent_system = [
m for m in self.system_metrics
if m['timestamp'] >= datetime.now() - recent_window
]
if recent_system:
avg_cpu = statistics.mean(m['cpu_percent'] for m in recent_system)
avg_memory = statistics.mean(m['memory_percent'] for m in recent_system)
if avg_cpu > 80:
bottlenecks.append({
'type': 'high_cpu_usage',
'avg_cpu_percent': avg_cpu,
'recommendation': 'Consider optimizing algorithms or scaling horizontally'
})
if avg_memory > 85:
bottlenecks.append({
'type': 'high_memory_pressure',
'avg_memory_percent': avg_memory,
'recommendation': 'Consider increasing available memory or optimizing memory usage'
})
return bottlenecks
def optimize_generation_strategy(self) -> Dict[str, Any]:
"""Generate optimization recommendations based on performance data"""
summary = self.get_performance_summary(time_window=timedelta(hours=24))
bottlenecks = self.detect_performance_bottlenecks()
recommendations = []
# Analyze timing performance
if 'timing' in summary:
avg_duration = summary['timing']['avg_duration']
p95_duration = summary['timing']['p95_duration']
if avg_duration > 30: # 30 seconds
recommendations.append({
'category': 'performance',
'priority': 'high',
'recommendation': 'Consider implementing parallel processing for content generation',
'rationale': f'Average generation time is {avg_duration:.1f} seconds'
})
if p95_duration > avg_duration * 3:
recommendations.append({
'category': 'consistency',
'priority': 'medium',
'recommendation': 'Investigate high variance in processing times',
'rationale': f'95th percentile is {p95_duration/avg_duration:.1f}x average'
})
# Analyze throughput
if 'throughput' in summary:
files_per_second = summary['throughput']['files_per_second']
if files_per_second < 1:
recommendations.append({
'category': 'throughput',
'priority': 'high',
'recommendation': 'Implement batch processing and template caching',
'rationale': f'Current throughput is only {files_per_second:.2f} files/second'
})
# Cache optimization
if 'cache' in summary and summary['cache']['hit_rate'] < 80:
recommendations.append({
'category': 'caching',
'priority': 'medium',
'recommendation': 'Optimize cache configuration and increase cache TTL',
'rationale': f"Cache hit rate is only {summary['cache']['hit_rate']:.1f}%"
})
# Memory optimization
if 'memory' in summary:
peak_memory = summary['memory']['peak_memory_mb']
if peak_memory > 1000: # 1GB
recommendations.append({
'category': 'memory',
'priority': 'medium',
'recommendation': 'Implement streaming processing for large datasets',
'rationale': f'Peak memory usage is {peak_memory:.0f}MB'
})
# Bottleneck-specific recommendations
for bottleneck in bottlenecks:
if bottleneck['type'] == 'slow_operations':
recommendations.append({
'category': 'performance',
'priority': 'high',
'recommendation': 'Profile and optimize slow operations',
'rationale': f"{bottleneck['count']} operations exceeded normal duration"
})
elif bottleneck['type'] == 'low_cache_efficiency':
recommendations.append({
'category': 'caching',
'priority': 'medium',
'recommendation': bottleneck['recommendation'],
'rationale': f"Cache hit rate is {bottleneck['hit_rate']:.1f}%"
})
return {
'analysis_timestamp': datetime.now().isoformat(),
'performance_summary': summary,
'bottlenecks': bottlenecks,
'recommendations': recommendations,
'optimization_score': self._calculate_optimization_score(summary, bottlenecks)
}
def _calculate_optimization_score(self, summary: Dict[str, Any], bottlenecks: List[Dict[str, Any]]) -> int:
"""Calculate optimization score (0-100)"""
score = 100
# Deduct points for performance issues
if 'timing' in summary:
avg_duration = summary['timing']['avg_duration']
if avg_duration > 60:
score -= 20
elif avg_duration > 30:
score -= 10
if 'cache' in summary:
hit_rate = summary['cache']['hit_rate']
if hit_rate < 50:
score -= 20
elif hit_rate < 80:
score -= 10
# Deduct points for bottlenecks
for bottleneck in bottlenecks:
if bottleneck['type'] in ['slow_operations', 'high_memory_usage']:
score -= 15
else:
score -= 5
return max(0, score)
def export_metrics(self, filename: str):
"""Export metrics to JSON file"""
export_data = {
'export_timestamp': datetime.now().isoformat(),
'metrics_history': [
{
'operation_id': m.operation_id,
'operation_type': m.operation_type,
'start_time': m.start_time.isoformat(),
'end_time': m.end_time.isoformat() if m.end_time else None,
'duration_seconds': m.duration_seconds,
'memory_usage_mb': m.memory_usage_mb,
'cpu_usage_percent': m.cpu_usage_percent,
'files_processed': m.files_processed,
'output_size_kb': m.output_size_kb,
'cache_hits': m.cache_hits,
'cache_misses': m.cache_misses
}
for m in self.metrics_history
],
'system_metrics': self.system_metrics[-100:], # Last 100 system samples
'summary': self.get_performance_summary(),
'optimization_analysis': self.optimize_generation_strategy()
}
with open(filename, 'w') as f:
json.dump(export_data, f, indent=2, default=str)
# Global performance monitor instance
performance_monitor = ContentGenerationPerformanceMonitor()
Conclusion
Advanced Markdown dynamic content generation and automation workflows represent transformative approaches to scalable technical documentation that enable systematic creation of data-driven content while maintaining consistency, accuracy, and efficiency across large documentation ecosystems. By implementing comprehensive automation strategies, template engines, and intelligent content pipelines, technical writers can build documentation systems that adapt dynamically to changing requirements while reducing manual effort and ensuring quality standards.
The key to successful content automation lies in understanding the relationship between data sources, template design, and output requirements, enabling the creation of flexible systems that can accommodate diverse content types while maintaining performance and reliability at scale. Whether you’re generating API documentation, metrics reports, or feature guides, the automation techniques and infrastructure patterns covered in this guide provide the foundation for building robust content generation systems.
Remember to implement automation incrementally, starting with simple template-based generation before adding sophisticated data integration and processing capabilities, and ensure that generated content maintains the same quality standards as manually created content through comprehensive validation and testing processes. With proper implementation of dynamic content generation workflows, your documentation system can achieve unprecedented scalability while maintaining the accuracy and consistency essential for effective technical communication.