from celery import Celery import os from datetime import datetime from pathlib import Path from pptx import Presentation from docx import Document REDIS_URL = os.getenv("REDIS_URL", "redis://127.0.0.1:6379/0") celery_app = Celery("growthops_tasks", broker=REDIS_URL, backend=REDIS_URL) EXPORT_DIR = Path("/data/exports") EXPORT_DIR.mkdir(parents=True, exist_ok=True) @celery_app.task(bind=True) def generate_pptx(self, title: str = "GrowthOps Report", items: list[str] | None = None): items = items or ["Overview", "Tenants", "Usage", "Jobs"] prs = Presentation() title_slide_layout = prs.slide_layouts[0] slide = prs.slides.add_slide(title_slide_layout) slide.shapes.title.text = title slide.placeholders[1].text = f"Generated at {datetime.utcnow().isoformat()}Z" bullet_layout = prs.slide_layouts[1] s2 = prs.slides.add_slide(bullet_layout) s2.shapes.title.text = "Contents" tf = s2.shapes.placeholders[1].text_frame for it in items: p = tf.add_paragraph() p.text = it p.level = 0 fname = f"report-{datetime.utcnow().strftime('%Y%m%dT%H%M%SZ')}.pptx" out = EXPORT_DIR / fname prs.save(out) return {"file": str(out), "filename": fname} @celery_app.task(bind=True) def generate_docx(self, title: str = "GrowthOps Summary", body: str = "Hello"): doc = Document() doc.add_heading(title, 0) doc.add_paragraph(f"Generated at {datetime.utcnow().isoformat()}Z") doc.add_paragraph(body) fname = f"summary-{datetime.utcnow().strftime('%Y%m%dT%H%M%SZ')}.docx" out = EXPORT_DIR / fname doc.save(out) return {"file": str(out), "filename": fname}