GrowthOps_OS / app /tasks.py
Corin1998's picture
Update app/tasks.py
10288f0 verified
Raw History Blame Contribute Delete
1.66 kB
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}