Why Automate Reporting?
Manual reporting takes hours each week. This tutorial shows you how to configure your AI agent to automatically generate weekly and monthly M&E reports from spreadsheet data, complete with indicator tracking, trend analysis, and actionable recommendations.

# config.py
REPORT_CONFIG = {
    'weekly': {
        'day': 'Friday',
        'time': '09:00',
        'period_days': 7,
        'include': ['performance', 'challenges', 'next_steps']
    },
    'monthly': {
        'day': '1',
        'time': '09:00',
        'period_days': 30,
        'include': ['performance', 'trends', 'recommendations']
    },
    'output_dir': 'reports/generated/'
}

import gspread
import pandas as pd
from google.oauth2.service_account import Credentials
from datetime import datetime, timedelta
import os
def load_data(source='google_sheets', file_path='indicator_data.xlsx'):
    """Load indicator data from Google Sheets or Excel."""
    if source == 'google_sheets':
        SCOPES = ['https://www.googleapis.com/auth/spreadsheets.readonly']
        creds = Credentials.from_service_account_file('credentials.json', scopes=SCOPES)
        client = gspread.authorize(creds)
        sheet = client.open('M&E Indicator Tracker').sheet1
        return sheet.get_all_records()
    else:
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"Excel file not found: {file_path}")
        df = pd.read_excel(file_path)
        return df.to_dict('records')
def filter_recent_data(data, days=7):
    """Filter data from the last N days."""
    cutoff = datetime.now() - timedelta(days=days)
    filtered = []
    for record in data:
        if 'date' in record:
            try:
                record_date = datetime.strptime(record['date'], '%Y-%m-%d')
                if record_date >= cutoff:
                    filtered.append(record)
            except (ValueError, KeyError):
                # Skip records without valid date
                continue
    return filtered
def calculate_metrics(data):
    """Calculate performance metrics from indicator data."""
    metrics = {
        'total': len(data),
        'on_track': 0,
        'off_track': 0,
        'at_risk': 0,
        'avg_achievement': 0,
        'completed': 0,
        'delayed': 0
    }
    
    if not data:
        return metrics
    
    achievement_sum = 0
    achievement_count = 0
    
    for record in data:
        # Calculate achievement
        if 'target' in record and 'actual' in record:
            try:
                target = float(record['target'])
                actual = float(record['actual'])
                if target > 0:
                    achievement = (actual / target) * 100
                    achievement_sum += achievement
                    achievement_count += 1
                    
                    if achievement >= 90:
                        metrics['on_track'] += 1
                    elif achievement >= 70:
                        metrics['at_risk'] += 1
                    else:
                        metrics['off_track'] += 1
            except (ValueError, TypeError):
                continue
        
        # Track status
        if 'status' in record:
            status = str(record['status']).lower()
            if status in ['completed', 'done', 'finished']:
                metrics['completed'] += 1
            elif status in ['delayed', 'behind', 'overdue']:
                metrics['delayed'] += 1
    
    if achievement_count > 0:
        metrics['avg_achievement'] = round(achievement_sum / achievement_count, 1)
    
    return metrics

import anthropic
from dotenv import load_dotenv
load_dotenv()
def generate_report(data, metrics, period='weekly'):
    """Generate AI-powered report from indicator data."""
    
    client = anthropic.Anthropic(api_key=os.getenv('ANTHROPIC_API_KEY'))
    
    # Prepare data summary
    top_indicators = sorted(data, key=lambda x: x.get('actual', 0) / x.get('target', 1) if x.get('target', 0) > 0 else 0, reverse=True)[:3]
    bottom_indicators = sorted(data, key=lambda x: x.get('actual', 0) / x.get('target', 1) if x.get('target', 0) > 0 else 0)[:3]
    
    prompt = f"""
    You are a senior M&E expert. Generate a {period} performance report.
    
    DATA SUMMARY:
    - Total indicators: {metrics['total']}
    - On track: {metrics['on_track']} ({round(metrics['on_track']/metrics['total']*100 if metrics['total']>0 else 0)}%)
    - At risk: {metrics['at_risk']}
    - Off track: {metrics['off_track']}
    - Average achievement: {metrics['avg_achievement']}%
    - Completed activities: {metrics['completed']}
    - Delayed activities: {metrics['delayed']}
    
    TOP PERFORMING INDICATORS:
    {top_indicators}
    
    BOTTOM PERFORMING INDICATORS:
    {bottom_indicators}
    
    Generate a report with:
    1. Executive Summary (3-4 sentences)
    2. Performance Overview (key achievements, main challenges)
    3. Priority Actions (5 specific recommendations)
    4. Risk Assessment (identify high-risk areas)
    5. Next Steps (actions for next period)
    
    Format the report with clear sections and bullet points.
    Mark critical issues with [CRITICAL].
    """
    
    response = client.messages.create(
        model="claude-3-sonnet-20240229",
        max_tokens=2000,
        temperature=0.3,
        messages=[{"role": "user", "content": prompt}]
    )
    
    return response.content[0].text

def save_report(content, period='weekly'):
    """Save generated report to file."""
    os.makedirs('reports', exist_ok=True)
    timestamp = datetime.now().strftime('%Y%m%d_%H%M')
    filename = f"reports/{period}_report_{timestamp}.md"
    
    with open(filename, 'w', encoding='utf-8') as f:
        # Add header
        f.write(f"# {period.capitalize()} Report\n")
        f.write(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n")
        f.write(content)
    
    return filename
def send_report(filename, recipients=None):
    """Send report via email."""
    if not recipients:
        return
    
    # Placeholder for email integration
    # Use smtplib for email, or SendGrid/Mailgun API
    print(f"Report ready for delivery to {len(recipients)} recipients")
    print(f"Report file: {filename}")

import schedule
import time
def run_report(period='weekly', source='google_sheets'):
    """Generate and save report."""
    try:
        data = load_data(source)
        filtered = filter_recent_data(data, days=7 if period=='weekly' else 30)
        metrics = calculate_metrics(filtered)
        report = generate_report(filtered, metrics, period)
        filename = save_report(report, period)
        print(f"[{datetime.now()}] {period.capitalize()} report saved: {filename}")
        return filename
    except Exception as e:
        print(f"[ERROR] Failed to generate {period} report: {e}")
        return None
# Schedule weekly report
schedule.every().friday.at("09:00").do(lambda: run_report('weekly'))
# Schedule monthly report
schedule.every().day.at("09:00").do(lambda: run_report('monthly'))
print("Scheduler started. Running reports will be generated automatically...")
while True:
    schedule.run_pending()
    time.sleep(60)

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
def start_scheduler():
    """Start background scheduler with job configuration."""
    scheduler = BackgroundScheduler()
    
    # Weekly on Fridays at 9:00 AM
    scheduler.add_job(
        lambda: run_report('weekly'),
        trigger=CronTrigger(day_of_week='fri', hour=9, minute=0),
        id='weekly_report'
    )
    
    # Monthly on 1st day at 9:00 AM
    scheduler.add_job(
        lambda: run_report('monthly'),
        trigger=CronTrigger(day=1, hour=9, minute=0),
        id='monthly_report'
    )
    
    scheduler.start()
    print("Background scheduler started...")
    return scheduler

# reporting_system.py
import sys
from datetime import datetime
class ReportSystem:
    def __init__(self, data_source='google_sheets'):
        self.data_source = data_source
        self.config = REPORT_CONFIG
    
    def run_weekly(self):
        """Generate weekly report."""
        return self._run_report('weekly')
    
    def run_monthly(self):
        """Generate monthly report."""
        return self._run_report('monthly')
    
    def _run_report(self, period):
        """Internal report generation method."""
        print(f"[{datetime.now()}] Generating {period} report...")
        
        days = 7 if period == 'weekly' else 30
        data = load_data(self.data_source)
        filtered = filter_recent_data(data, days)
        
        if not filtered:
            print(f"No data found for {period} report")
            return None
        
        metrics = calculate_metrics(filtered)
        report = generate_report(filtered, metrics, period)
        filename = save_report(report, period)
        
        print(f"Report saved to: {filename}")
        return filename
    
    def start(self):
        """Start automated reporting."""
        from apscheduler.schedulers.background import BackgroundScheduler
        from apscheduler.triggers.cron import CronTrigger
        
        scheduler = BackgroundScheduler()
        
        scheduler.add_job(
            self.run_weekly,
            trigger=CronTrigger(day_of_week='fri', hour=9, minute=0),
            id='weekly_report'
        )
        
        scheduler.add_job(
            self.run_monthly,
            trigger=CronTrigger(day=1, hour=9, minute=0),
            id='monthly_report'
        )
        
        scheduler.start()
        print("Reporting system running...")
        return scheduler
if __name__ == "__main__":
    system = ReportSystem()
    
    if len(sys.argv) > 1:
        if sys.argv[1] == '--weekly':
            system.run_weekly()
        elif sys.argv[1] == '--monthly':
            system.run_monthly()
        elif sys.argv[1] == '--test':
            system.run_weekly()
    else:
        system.start()
        try:
            while True:
                import time
                time.sleep(3600)
        except KeyboardInterrupt:
            print("Shutting down...")

def combine_sources(sources):
    """Combine data from multiple sources."""
    combined = []
    for source in sources:
        if source['type'] == 'google_sheets':
            data = load_data('google_sheets')
        elif source['type'] == 'excel':
            data = load_data('excel', source['file'])
        else:
            continue
        combined.extend(data)
    return combined

def convert_to_html(report_content, period):
    """Convert Markdown report to HTML email format."""
    html_template = f"""
    <!DOCTYPE html>
    <html>
    <head><style>
        body {{ font-family: Arial, sans-serif; margin: 20px; }}
        h1 {{ color: #1a1a2e; }}
        h2 {{ color: #4a9eff; }}
        .section {{ margin: 15px 0; }}
        .critical {{ color: #dc3545; font-weight: bold; }}
    </style></head>
    <body>
        <h1>{period.capitalize()} Report</h1>
        <div class="section">
            {report_content.replace('\n', '<br>')}
        </div>
        <p>Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}</p>
    </body>
    </html>
    """
    return html_template

def manage_versions(period, max_versions=10):
    """Keep only the latest N report versions."""
    import glob
    pattern = f"reports/{period}_report_*.md"
    files = sorted(glob.glob(pattern), reverse=True)
    
    if len(files) > max_versions:
        for old_file in files[max_versions:]:
            os.remove(old_file)
            print(f"Removed old report: {old_file}")

Master AI Agent Development for M&E

The AI Agents for Evaluators Certificate teaches you to build practical, reliable AI agents for real M&E workflows—including automated reporting, spreadsheet integration, and validation.

Course Features: 32 lectures · Lifetime access · Certificate included · 241 students enrolled

Enroll Now – $249 Lifetime Access

Bundle with AI in M&E course and save 30%