Maintenance Tasks

from hyrex import HyrexRegistry
from datetime import datetime, timedelta
import requests
import os

hy = HyrexRegistry()

@hy.task
def send_daily_report_email(recipient_email: str, report_data: dict):
    """Send daily analytics report via email"""
    date_str = datetime.now().strftime('%Y-%m-%d')
    email_content = f"Daily Analytics Report - {date_str}\\n\\n"
    email_content += f"Active Users: {report_data['active_users']}\\n"
    email_content += f"Revenue: ${report_data['revenue']:,.2f}\\n"
    email_content += f"New Signups: {report_data['new_signups']}\\n"
    email_content += f"Support Tickets: {report_data['support_tickets']}\\n\\n"
    email_content += "View full dashboard: https://yourapp.com/analytics"
    
    send_email(
        to=recipient_email,
        subject=f"Daily Report - {datetime.now().strftime('%m/%d/%Y')}",
        body=email_content
    )

@hy.task
def backup_database(database_name: str, s3_bucket: str):
    """Create database backup and upload to S3"""
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_filename = f"{database_name}_{timestamp}.sql"
    
    # Create database dump
    os.system(f"pg_dump {database_name} > /tmp/{backup_filename}")
    
    # Upload to S3
    upload_to_s3(f"/tmp/{backup_filename}", s3_bucket, f"backups/{backup_filename}")
    
    # Clean up local file
    os.remove(f"/tmp/{backup_filename}")
    
    return {"backup_file": backup_filename, "size_mb": get_file_size_mb(backup_filename)}

@hy.task
def cleanup_old_files(directory: str, days_old: int = 30):
    """Remove files older than specified days"""
    cutoff_date = datetime.now() - timedelta(days=days_old)
    removed_files = []
    
    for filename in os.listdir(directory):
        file_path = os.path.join(directory, filename)
        if os.path.isfile(file_path):
            file_modified = datetime.fromtimestamp(os.path.getmtime(file_path))
            if file_modified < cutoff_date:
                os.remove(file_path)
                removed_files.append(filename)
    
    return {"removed_files": removed_files, "count": len(removed_files)}

@hy.task
def process_video_thumbnail(video_url: str, output_bucket: str):
    """Generate video thumbnail and upload to storage"""
    import cv2
    import tempfile
    
    # Download video to temp file
    video_response = requests.get(video_url)
    with tempfile.NamedTemporaryFile(suffix='.mp4') as temp_video:
        temp_video.write(video_response.content)
        temp_video.flush()
        
        # Extract frame at 2 seconds
        cap = cv2.VideoCapture(temp_video.name)
        cap.set(cv2.CAP_PROP_POS_MSEC, 2000)  # 2 seconds
        ret, frame = cap.read()
        
        if ret:
            thumbnail_filename = f"thumb_{hash(video_url)}.jpg"
            with tempfile.NamedTemporaryFile(suffix='.jpg') as temp_thumb:
                cv2.imwrite(temp_thumb.name, frame)
                upload_to_s3(temp_thumb.name, output_bucket, f"thumbnails/{thumbnail_filename}")
                
        cap.release()
        
    return {"thumbnail_url": f"s3://{output_bucket}/thumbnails/{thumbnail_filename}"}

@hy.task
def generate_analytics_report():
    """Generate daily analytics data"""
    report_data = {
        "active_users": get_active_users_count(),
        "revenue": get_daily_revenue(),
        "new_signups": get_new_signups_count(),
        "support_tickets": get_open_tickets_count()
    }
    
    # Store in database
    store_daily_metrics(report_data)
    
    # Send to stakeholders
    stakeholder_emails = ["ceo@company.com", "cto@company.com", "head-of-product@company.com"]
    for email in stakeholder_emails:
        send_daily_report_email.send(email, report_data)
    
    return report_data

# Schedule recurring tasks
@hy.cron("0 9 * * *")  # Every day at 9 AM
def daily_tasks():
    """Run daily maintenance tasks"""
    generate_analytics_report.send()
    backup_database.send("production_db", "company-backups")
    cleanup_old_files.send("/app/temp", 7)  # Clean files older than 7 days

@hy.cron("0 2 * * 0")  # Every Sunday at 2 AM
def weekly_tasks():
    """Run weekly maintenance tasks"""
    cleanup_old_files.send("/app/logs", 30)  # Clean logs older than 30 days
    backup_database.send("analytics_db", "company-backups")

import { HyrexRegistry } from 'hyrex';
import axios from 'axios';
import fs from 'fs';
import path from 'path';

const hy = new HyrexRegistry();

const sendDailyReportEmail = hy.task({
    name: "sendDailyReportEmail",
    func: async (recipientEmail: string, reportData: any) => {
        const dateStr = new Date().toISOString().split('T')[0];
        let emailContent = `Daily Analytics Report - ${dateStr}\\n\\n`;
        emailContent += `Active Users: ${reportData.activeUsers}\\n`;
        emailContent += `Revenue: $${reportData.revenue.toLocaleString()}\\n`;
        emailContent += `New Signups: ${reportData.newSignups}\\n`;
        emailContent += `Support Tickets: ${reportData.supportTickets}\\n\\n`;
        emailContent += 'View full dashboard: https://yourapp.com/analytics';
        
        await sendEmail({
            to: recipientEmail,
            subject: `Daily Report - ${new Date().toLocaleDateString()}`,
            body: emailContent
        });
    }
});

const backupDatabase = hy.task({
    name: "backupDatabase",
    func: async (databaseName: string, s3Bucket: string) => {
        const timestamp = new Date().toISOString().replace(/[:.]/g, '-');
        const backupFilename = `${databaseName}_${timestamp}.sql`;
        const tempPath = `/tmp/${backupFilename}`;
        
        // Create database dump
        await execCommand(`pg_dump ${databaseName} > ${tempPath}`);
        
        // Upload to S3
        await uploadToS3(tempPath, s3Bucket, `backups/${backupFilename}`);
        
        // Clean up local file
        fs.unlinkSync(tempPath);
        
        const stats = fs.statSync(tempPath);
        return { 
            backupFile: backupFilename, 
            sizeMb: Math.round(stats.size / 1024 / 1024) 
        };
    }
});

const cleanupOldFiles = hy.task({
    name: "cleanupOldFiles",
    func: async (directory: string, daysOld: number = 30) => {
        const cutoffDate = new Date();
        cutoffDate.setDate(cutoffDate.getDate() - daysOld);
        
        const removedFiles: string[] = [];
        
        const files = fs.readdirSync(directory);
        for (const filename of files) {
            const filePath = path.join(directory, filename);
            const stats = fs.statSync(filePath);
            
            if (stats.isFile() && stats.mtime < cutoffDate) {
                fs.unlinkSync(filePath);
                removedFiles.push(filename);
            }
        }
        
        return { removedFiles, count: removedFiles.length };
    }
});

const processVideoThumbnail = hy.task({
    name: "processVideoThumbnail",
    func: async (videoUrl: string, outputBucket: string) => {
        const ffmpeg = require('fluent-ffmpeg');
        const crypto = require('crypto');
        
        const hash = crypto.createHash('md5').update(videoUrl).digest('hex');
        const thumbnailFilename = `thumb_${hash}.jpg`;
        const tempPath = `/tmp/${thumbnailFilename}`;
        
        // Extract thumbnail at 2 seconds
        await new Promise((resolve, reject) => {
            ffmpeg(videoUrl)
                .seekInput(2)
                .frames(1)
                .output(tempPath)
                .on('end', resolve)
                .on('error', reject)
                .run();
        });
        
        // Upload to S3
        await uploadToS3(tempPath, outputBucket, `thumbnails/${thumbnailFilename}`);
        
        // Clean up
        fs.unlinkSync(tempPath);
        
        return { 
            thumbnailUrl: `s3://${outputBucket}/thumbnails/${thumbnailFilename}`
        };
    }
});

const generateAnalyticsReport = hy.task({
    name: "generateAnalyticsReport",
    func: async () => {
        const reportData = {
            activeUsers: await getActiveUsersCount(),
            revenue: await getDailyRevenue(),
            newSignups: await getNewSignupsCount(),
            supportTickets: await getOpenTicketsCount()
        };
        
        // Store in database
        await storeDailyMetrics(reportData);
        
        // Send to stakeholders
        const stakeholderEmails = [
            "ceo@company.com", 
            "cto@company.com", 
            "head-of-product@company.com"
        ];
        
        for (const email of stakeholderEmails) {
            await sendDailyReportEmail.send(email, reportData);
        }
        
        return reportData;
    }
});

// Schedule recurring tasks
const dailyTasks = hy.cron({
    schedule: "0 9 * * *", // Every day at 9 AM
    name: "dailyTasks",
    func: async () => {
        await generateAnalyticsReport.send();
        await backupDatabase.send("production_db", "company-backups");
        await cleanupOldFiles.send("/app/temp", 7); // Clean files older than 7 days
    }
});

const weeklyTasks = hy.cron({
    schedule: "0 2 * * 0", // Every Sunday at 2 AM
    name: "weeklyTasks", 
    func: async () => {
        await cleanupOldFiles.send("/app/logs", 30); // Clean logs older than 30 days
        await backupDatabase.send("analytics_db", "company-backups");
    }
});

from fastapi import FastAPI
from pydantic import BaseModel
from .tasks import backup_database, cleanup_old_files, process_video_thumbnail

app = FastAPI()

class BackupRequest(BaseModel):
    database_name: str
    s3_bucket: str
    
class CleanupRequest(BaseModel):
    directory: str
    days_old: int = 30
    
class VideoProcessRequest(BaseModel):
    video_url: str
    output_bucket: str

@app.post("/tasks/backup")
async def trigger_backup(request: BackupRequest):
    """Trigger database backup"""
    task = backup_database.send(request.database_name, request.s3_bucket)
    
    return {
        "message": "Backup task started",
        "task_id": task.task_id,
        "database": request.database_name
    }

@app.post("/tasks/cleanup")
async def trigger_cleanup(request: CleanupRequest):
    """Trigger file cleanup"""
    task = cleanup_old_files.send(request.directory, request.days_old)
    
    return {
        "message": "Cleanup task started",
        "task_id": task.task_id,
        "directory": request.directory
    }

@app.post("/tasks/process-video")
async def process_video(request: VideoProcessRequest):
    """Process video thumbnail generation"""
    task = process_video_thumbnail.send(request.video_url, request.output_bucket)
    
    return {
        "message": "Video processing started",
        "task_id": task.task_id,
        "video_url": request.video_url
    }

@app.get("/tasks/status/{task_id}")
async def get_task_status(task_id: str):
    """Get status of background task"""
    task = hy.get_task(task_id)
    
    return {
        "task_id": task_id,
        "status": task.status,
        "progress": task.progress,
        "result": task.result if task.is_complete else None,
        "created_at": task.created_at,
        "completed_at": task.completed_at
    }

import express from 'express';
import { backupDatabase, cleanupOldFiles, processVideoThumbnail } from './tasks';

const app = express();
app.use(express.json());

interface BackupRequest {
    databaseName: string;
    s3Bucket: string;
}

interface CleanupRequest {
    directory: string;
    daysOld?: number;
}

interface VideoProcessRequest {
    videoUrl: string;
    outputBucket: string;
}

app.post('/tasks/backup', async (req, res) => {
    const { databaseName, s3Bucket }: BackupRequest = req.body;
    
    // Trigger database backup
    const task = await backupDatabase.send(databaseName, s3Bucket);
    
    res.json({
        message: 'Backup task started',
        taskId: task.taskId,
        database: databaseName
    });
});

app.post('/tasks/cleanup', async (req, res) => {
    const { directory, daysOld = 30 }: CleanupRequest = req.body;
    
    // Trigger file cleanup
    const task = await cleanupOldFiles.send(directory, daysOld);
    
    res.json({
        message: 'Cleanup task started',
        taskId: task.taskId,
        directory: directory
    });
});

app.post('/tasks/process-video', async (req, res) => {
    const { videoUrl, outputBucket }: VideoProcessRequest = req.body;
    
    // Process video thumbnail generation
    const task = await processVideoThumbnail.send(videoUrl, outputBucket);
    
    res.json({
        message: 'Video processing started',
        taskId: task.taskId,
        videoUrl: videoUrl
    });
});

app.get('/tasks/status/:taskId', async (req, res) => {
    const { taskId } = req.params;
    
    // Get status of background task
    const task = await hy.getTask(taskId);
    
    res.json({
        taskId: taskId,
        status: task.status,
        progress: task.progress,
        result: task.isComplete ? task.result : null,
        createdAt: task.createdAt,
        completedAt: task.completedAt
    });
});