Create workflows/content_generator.py

workflows/content_generator.py

@@ -0,0 +1,43 @@
+from prefect import task
+from crewai import Crew
+from crews.agents import content_writer_agent, image_prompt_agent, quality_assurance_agent
+from crews.tasks import create_content_tasks
+from config.schemas import InstagramPost
+import json
+import re
+
+@task(retries=2)
+def generate_instagram_content(rss_entry: dict) -> dict:
+    """Use CrewAI to generate Instagram post content"""
+    
+    # Create tasks
+    tasks = create_content_tasks(
+        rss_content=rss_entry["content"],
+        rss_title=rss_entry["title"]
+    )
+    
+    # Create crew
+    crew = Crew(
+        agents=[content_writer_agent, image_prompt_agent, quality_assurance_agent],
+        tasks=tasks,
+        verbose=True
+    )
+    
+    # Execute crew
+    result = crew.kickoff()
+    
+    # Parse result (CrewAI returns string)
+    result_str = str(result)
+    
+    # Extract JSON from result
+    json_match = re.search(r'\{[\s\S]*\}', result_str)
+    if json_match:
+        json_str = json_match.group()
+        post_data = json.loads(json_str)
+    else:
+        raise ValueError("No valid JSON found in CrewAI output")
+    
+    # Validate against schema
+    validated_post = InstagramPost(**post_data)
+    
+    return validated_post.model_dump()