File size: 6,628 Bytes
1d6d1e6 c7d5529 1d6d1e6 25f22bf 1d6d1e6 25f22bf 1d6d1e6 25f22bf a25d1db 25f22bf 1cf04ef 25f22bf a25d1db 25f22bf a25d1db |
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 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 |
from celery import current_task
from backend.services.content_service import ContentService
from backend.services.linkedin_service import LinkedInService
from backend.utils.database import init_supabase
from backend.celery_config import celery_app
# Configure logging
import logging
logger = logging.getLogger(__name__)
@celery_app.task(bind=True)
def generate_content_task(self, user_id: str, schedule_id: str, supabase_client_config: dict):
"""
Celery task to generate content for a scheduled post.
Args:
user_id (str): User ID
schedule_id (str): Schedule ID
supabase_client_config (dict): Supabase client configuration
Returns:
dict: Result of content generation
"""
try:
logger.info(f"Starting content generation for schedule {schedule_id}")
# Update task state
self.update_state(state='PROGRESS', meta={'status': 'Generating content...'})
# Initialize content service
content_service = ContentService()
# Generate content using content service
generated_content = content_service.generate_post_content(user_id)
# Initialize Supabase client from config
from backend.utils.database import init_supabase
supabase_client = init_supabase(
supabase_client_config['SUPABASE_URL'],
supabase_client_config['SUPABASE_KEY']
)
# Store generated content in database
# We need to get the social account ID from the schedule
schedule_response = (
supabase_client
.table("Scheduling")
.select("id_social")
.eq("id", schedule_id)
.execute()
)
if not schedule_response.data:
raise Exception(f"Schedule {schedule_id} not found")
social_account_id = schedule_response.data[0]['id_social']
# Store the generated content
response = (
supabase_client
.table("Post_content")
.insert({
"social_account_id": social_account_id,
"Text_content": generated_content,
"is_published": False,
"sched": schedule_id
})
.execute()
)
if response.data:
logger.info(f"Content generated and stored for schedule {schedule_id}")
return {
'status': 'success',
'message': f'Content generated for schedule {schedule_id}',
'post_id': response.data[0]['id']
}
else:
logger.error(f"Failed to store generated content for schedule {schedule_id}")
return {
'status': 'failure',
'message': f'Failed to store generated content for schedule {schedule_id}'
}
except Exception as e:
logger.error(f"Error in content generation task for schedule {schedule_id}: {str(e)}")
return {
'status': 'failure',
'message': f'Error in content generation: {str(e)}'
}
@celery_app.task(bind=True)
def publish_post_task(self, schedule_id: str, supabase_client_config: dict):
"""
Celery task to publish a scheduled post.
Args:
schedule_id (str): Schedule ID
supabase_client_config (dict): Supabase client configuration
Returns:
dict: Result of post publishing
"""
try:
logger.info(f"Starting post publishing for schedule {schedule_id}")
# Update task state
self.update_state(state='PROGRESS', meta={'status': 'Publishing post...'})
# Initialize Supabase client from config
from backend.utils.database import init_supabase
supabase_client = init_supabase(
supabase_client_config['SUPABASE_URL'],
supabase_client_config['SUPABASE_KEY']
)
# Fetch the post to publish
response = (
supabase_client
.table("Post_content")
.select("*")
.eq("sched", schedule_id)
.eq("is_published", False)
.order("created_at", desc=True)
.limit(1)
.execute()
)
if not response.data:
logger.info(f"No unpublished posts found for schedule {schedule_id}")
return {
'status': 'info',
'message': f'No unpublished posts found for schedule {schedule_id}'
}
post = response.data[0]
post_id = post.get('id')
text_content = post.get('Text_content')
image_url = post.get('image_content_url')
# Get social network credentials
schedule_response = (
supabase_client
.table("Scheduling")
.select("Social_network(token, sub)")
.eq("id", schedule_id)
.execute()
)
if not schedule_response.data:
raise Exception(f"Schedule {schedule_id} not found")
social_network = schedule_response.data[0].get('Social_network', {})
access_token = social_network.get('token')
user_sub = social_network.get('sub')
if not access_token or not user_sub:
logger.error(f"Missing social network credentials for schedule {schedule_id}")
return {
'status': 'failure',
'message': f'Missing social network credentials for schedule {schedule_id}'
}
# Publish to LinkedIn
linkedin_service = LinkedInService()
publish_response = linkedin_service.publish_post(
access_token, user_sub, text_content, image_url
)
# Update post status in database
update_response = (
supabase_client
.table("Post_content")
.update({"is_published": True})
.eq("id", post_id)
.execute()
)
logger.info(f"Post published successfully for schedule {schedule_id}")
return {
'status': 'success',
'message': f'Post published successfully for schedule {schedule_id}',
'linkedin_response': publish_response
}
except Exception as e:
logger.error(f"Error in publishing task for schedule {schedule_id}: {str(e)}")
return {
'status': 'failure',
'message': f'Error in publishing post: {str(e)}'
}
|