Spaces:
Running
Running
File size: 4,214 Bytes
c84fdae | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | """Temporal worker — runs all workflows and activities."""
import asyncio
from temporalio.client import Client, Schedule, ScheduleActionStartWorkflow, ScheduleSpec, ScheduleIntervalSpec
from temporalio.worker import Worker, UnsandboxedWorkflowRunner
from datetime import timedelta
from src.config import get_settings
from src.temporal_workflows.workflows import TrendToPostPublishWorkflow
from src.temporal_workflows.scheduler_workflows import (
TrendFetchCronWorkflow,
AutoPostSchedulerWorkflow,
fetch_trends_activity,
pick_best_unposted_trend,
get_default_agent_version_id,
trigger_post_workflow,
)
from src.temporal_workflows import activities
from loguru import logger
settings = get_settings()
async def _ensure_schedules(client: Client):
"""Create Temporal schedules if they don't already exist."""
# 1. Trend fetch — every 2 hours
try:
await client.create_schedule(
"trend-fetch-every-2h",
Schedule(
action=ScheduleActionStartWorkflow(
TrendFetchCronWorkflow.run,
id="trend-fetch-cron",
task_queue=settings.temporal_task_queue,
),
spec=ScheduleSpec(intervals=[ScheduleIntervalSpec(every=timedelta(hours=2))]),
),
)
logger.info("Created schedule: trend-fetch-every-2h")
except Exception as e:
if "already exists" in str(e).lower():
logger.info("Schedule trend-fetch-every-2h already exists")
else:
logger.warning(f"Could not create trend-fetch schedule: {e}")
# 2. Auto post — every 4 hours (6 posts/day ceiling)
try:
await client.create_schedule(
"auto-post-every-4h",
Schedule(
action=ScheduleActionStartWorkflow(
AutoPostSchedulerWorkflow.run,
args=[settings.post_platform],
id="auto-post-cron",
task_queue=settings.temporal_task_queue,
),
spec=ScheduleSpec(intervals=[ScheduleIntervalSpec(every=timedelta(hours=4))]),
),
)
logger.info("Created schedule: auto-post-every-4h")
except Exception as e:
if "already exists" in str(e).lower():
logger.info("Schedule auto-post-every-4h already exists")
else:
logger.warning(f"Could not create auto-post schedule: {e}")
async def main():
"""Start the Temporal worker and register cron schedules."""
client = await Client.connect(
f"{settings.temporal_host}:{settings.temporal_port}",
namespace=settings.temporal_namespace,
)
logger.info(f"Connected to Temporal at {settings.temporal_host}:{settings.temporal_port}")
# Register cron schedules on startup
await _ensure_schedules(client)
worker = Worker(
client,
task_queue=settings.temporal_task_queue,
workflows=[
TrendToPostPublishWorkflow,
TrendFetchCronWorkflow,
AutoPostSchedulerWorkflow,
],
activities=[
# Core pipeline
activities.fetch_trend_details,
activities.generate_post_content,
activities.generate_images,
activities.moderate_content,
activities.store_media_to_cdn,
activities.save_post_draft,
activities.publish_to_platform,
activities.record_execution_metrics,
# Scheduler
fetch_trends_activity,
pick_best_unposted_trend,
get_default_agent_version_id,
trigger_post_workflow,
],
workflow_runner=UnsandboxedWorkflowRunner(),
max_concurrent_activities=10,
max_concurrent_workflow_tasks=10,
)
logger.info(f"Worker running on task queue: {settings.temporal_task_queue}")
logger.info(f"Auto-approve: {settings.auto_approve} | Platform: {settings.post_platform}")
try:
await worker.run()
except KeyboardInterrupt:
logger.info("Worker interrupted")
finally:
await client.close()
if __name__ == "__main__":
asyncio.run(main())
|