from apscheduler.schedulers.background import BackgroundScheduler from sqlalchemy.orm import Session from datetime import datetime, timezone import pytz from src.utils.db import get_db_session from src.marketing.apps.Calendar.model import Calendar from src.marketing.apps.post.model import CalendarPost # from pytz import timezone, UTC, UnknownTimeZoneError from pytz.exceptions import AmbiguousTimeError, NonExistentTimeError from .utils import post_to_twitter, post_to_facebook from sqlalchemy import cast, DateTime, func, text import logging def publish_scheduled_posts(): db: Session = get_db_session() try: # current UTC datetime current_utc = datetime.now(timezone.utc) # Query posts that are scheduled to be published at or before the current UTC time posts = db.query(CalendarPost).join(Calendar).filter( CalendarPost.post_status == "submitted", CalendarPost.schedule_date.isnot(None), CalendarPost.schedule_time.isnot(None), CalendarPost.is_deleted == False, Calendar.is_deleted == False, cast(CalendarPost.schedule_date + CalendarPost.schedule_time, DateTime) <= current_utc ).all() logging.info(f"[CRON] Found {len(posts)} posts with 'submitted' status") for post in posts: if not post.schedule_time: continue scheduled_utc = datetime.combine(post.schedule_date, post.schedule_time).replace(tzinfo=pytz.utc) # # current_utc = datetime.utcnow().replace(tzinfo=pytz.utc) logging.info(f"[CRON DEBUG] Post ID: {post.id} | scheduled_utc: {scheduled_utc} | current_utc: {current_utc}") if scheduled_utc <= current_utc: #todo #1. Loop post type # print("Entered loop for processing posts for checking loop post type") # print("post ============== ",post.post_types) # for post_type in post.post_types: # print("post-type-from-cron", post_type.connected_account_id) for post_type in post.post_types: platform_name = post_type.connected_account.master_account.social_media_name # 2. know which type of post it is by master id platform_post_master_id = post_type.connected_account.master_account.id connected_account_id = post_type.connected_account.id print("connected_account_id", connected_account_id) print(f"[DEBUG] PostType ID={post_type.id} | Platform={platform_name}") print("post-master-account-id:", platform_post_master_id) #=====================Checking the platform name x (twitter) ========================== if platform_name.lower() == "x (twitter)": print("🚀 Ready to post to Twitter:", post_type) response = post_to_twitter(post_type) if not response.get("success"): logging.error(f"[CRON] Twitter posting failed for {post_type.id}: {response.get('error')}") continue # If posting to Twitter was successful, save the external post ID post_type.external_post_id = response.get("post_id") logging.info(f"[CRON] Twitter posting successful for {post_type.id} with external ID {post_type.external_post_id}") # Add sentiment analysis for the posted content try: # Import here to avoid circular imports from src.marketing.apps.Analytics.service import analyze_text_sentiment from src.marketing.apps.Analytics.schema import SentimentAnalysisRequest # Create request for sentiment analysis sentiment_request = SentimentAnalysisRequest( text=post_type.content, post_id=post_type.id ) # Analyze sentiment (this will save to database) analyze_text_sentiment(db, sentiment_request) logging.info(f"[CRON] Sentiment analysis completed for post {post_type.id}") except Exception as e: logging.error(f"[CRON] Error analyzing sentiment for post {post_type.id}: {e}") # =======================If platform is Facebook or Facebook Page=========================== if platform_name.lower() in ("facebook","facebook page"): response = post_to_facebook(post_type) if not response.get("success"): logging.error(f"[CRON] Facebook posting failed for {post_type.id}: {response.get('error')}") continue # If posting to Facebook was successful, save the external post ID post_type.external_post_id = response.get("post_id") logging.info(f"[CRON] Facebook posting successful for {post_type.id} with external ID {post_type.external_post_id}") #4. call the function and post on Twitter # if post_to_twitter(post_type): # logging.info(f"[CRON] Twitter posting successful for {post_type.id}") # #5. On successs update post status to # # post_type.post_status = "published" # else: # logging.warning(f"[CRON] Twitter posting failed for {post_type.id}") # post_to_facebook(post_type) # for post_type in post["post_types"]: # print("post-type-from-corn",post_type["content"]) #2. know which type of post it is by master id #3. create seperate function to post in tweeter #4. call the function and post #5. On successs update post status to post.post_status = "published" # post.updated_at = datetime.utcnow() # logging.info(f"[CRON] → Published Post ID {post.id} at {post.updated_at}") # Set updated_at in calendar's timezone current_utc = datetime.utcnow().replace(tzinfo=pytz.utc) post.updated_at = current_utc # logging.info(f"[CRON] → Published Post ID {post.id} at {post.updated_at} ({calendar_tz})") db.commit() except Exception as e: db.rollback() logging.error(f"Error in cron job: {str(e)}") finally: db.close() def start_cron(): scheduler = BackgroundScheduler() # Schedule post publishing job to run every minute scheduler.add_job(publish_scheduled_posts, 'interval', minutes=1) # Import the Twitter metrics update function from src.marketing.apps.Analytics.twitter_analytics import update_twitter_metrics_daily # Schedule Twitter metrics update to run once daily at 2 AM scheduler.add_job( update_twitter_metrics_daily, 'interval', hours=24, # minutes=5 ) # Start the scheduler scheduler.start()