from pytz import timezone, UTC, AmbiguousTimeError, NonExistentTimeError from pytz.exceptions import UnknownTimeZoneError from datetime import datetime, time,date import json from src.marketing.apps.post import schema, controller from src.utils.db import get_db from src.utils.db import get_db_session from sqlalchemy.orm import Session from src.marketing.apps.post.model import PostTypeConfig from src.marketing.apps.Account.model import MasterAccount from fastapi import HTTPException, Depends, status from sqlalchemy import func from typing import Optional import logging import requests # or Tweepy if using Twitter API v2 from src.marketing.apps.Account.model import ConnectedAccount from bs4 import BeautifulSoup from src.marketing.apps.post.model import CalendarPostType, PostImage import httpx import os APP_ID = os.getenv("FB_APP_ID") APP_SECRET = os.getenv("FB_APP_SECRET") def convert_to_utc(schedule_date: date, schedule_time: Optional[time], tz_str: str) -> datetime: try: user_tz = timezone((tz_str or "UTC").strip()) except UnknownTimeZoneError: raise HTTPException(status_code=400, detail=f"Invalid timezone: {tz_str}") local_dt = datetime.combine(schedule_date, schedule_time or time(0, 0)) try: local_dt_with_tz = user_tz.localize(local_dt, is_dst=None) except AmbiguousTimeError: local_dt_with_tz = user_tz.localize(local_dt, is_dst=False) except NonExistentTimeError: raise HTTPException(status_code=400, detail="Scheduled time does not exist due to DST transition.") return local_dt_with_tz.astimezone(UTC) # Save social media post type configurations to the database def save_social_config_to_db(db: Session, config_data: dict): for platform_info in config_data.get("platforms", []): platform_name = platform_info.get("platform_display_name", "") # Find the corresponding master account master_account = db.query(MasterAccount).filter_by(social_media_name=platform_name).first() if not master_account: print(f"Master account for platform '{platform_name}' not found. Skipping.") continue post_types = platform_info.get("post_types", []) for post_type_spec in post_types: post_type_name = post_type_spec.get("post_type", "") # Check if the record already exists existing = db.query(PostTypeConfig).filter_by( master_account_id=master_account.id, platform=platform_name, post_type=post_type_name ).first() if existing: print(f"Post type '{post_type_name}' for platform '{platform_name}' already exists. Skipping.") continue post_spec = PostTypeConfig( master_account_id=master_account.id, platform=platform_name, post_type=post_type_name, supports_text=post_type_spec.get("supports_text", False), supports_image=post_type_spec.get("supports_image", False), supports_multiple_images=post_type_spec.get("supports_multiple_images", False), supports_video=post_type_spec.get("supports_video", False), supports_link=post_type_spec.get("supports_link", False), text_max_limit=post_type_spec.get("text_max_limit", 0), text_optimum_limit=post_type_spec.get("text_optimum_limit", 0), image_ratio=post_type_spec.get("image_ratio", ""), image_best_resolution_px=post_type_spec.get("image_best_resolution_px", ""), video_max_size_mb=post_type_spec.get("video_max_size_mb", 0), video_max_length_sec=post_type_spec.get("video_max_length_sec", 0), additional_notes=post_type_spec.get("note", "") ) db.add(post_spec) db.commit() #get post type configs by master account id def format_post_type_configs(configs: list[PostTypeConfig]) -> dict: result = {"platforms": []} platform_dict = {} for config in configs: platform = config.platform post_type = config.post_type if platform not in platform_dict: platform_dict[platform] = { "platform_display_name": platform, "post_types": [] } post_type_data = { "post_type": post_type, "supports_text": config.supports_text, "supports_image": config.supports_image, "supports_multiple_images": config.supports_multiple_images, "supports_video": config.supports_video, "supports_link": config.supports_link, "text_max_limit": config.text_max_limit, "text_optimum_limit": config.text_optimum_limit, "image_ratio": config.image_ratio, "image_best_resolution_px": config.image_best_resolution_px, "video_max_size_mb": config.video_max_size_mb, "video_max_length_sec": config.video_max_length_sec, "note": config.additional_notes } platform_dict[platform]["post_types"].append(post_type_data) # Convert dictionary to array result["platforms"] = list(platform_dict.values()) return result ###===============================================Twitter post automation ======================================================================== #Convert HTML to styled Unicode text for Twitte def html_to_unicode_styled(html: str) -> str: """Convert HTML to styled Unicode text for Twitter.""" soup = BeautifulSoup(html, "html.parser") def style_bold(text): # Map ASCII letters to their Mathematical Bold Unicode equivalents bold_chars = { **{chr(i): chr(0x1D400 + (i - ord('A'))) for i in range(ord('A'), ord('Z')+1)}, **{chr(i): chr(0x1D41A + (i - ord('a'))) for i in range(ord('a'), ord('z')+1)} } return "".join(bold_chars.get(c, c) for c in text) #Applying thes tags for bold text tags_to_handle = [ 'strong', 'b', 'em', 'i', 'u', 'p', 'br', 'h1', 'h2', 'h3', 'ol', 'ul', 'li', 'blockquote' ] # Apply transformations for tag in soup.find_all(tags_to_handle): tag.string = style_bold(tag.get_text()) return soup.get_text() REFRESH_URL = "https://ap.bestbrain.ai/x/refresh" def refresh_twitter_token(branch_id: int, connected_account_id: int): try: db: Session = get_db_session() logging.info(f"❌❌❌ {branch_id}" ) REFRESH_BODY = {"app_id": str(branch_id)} # or your fixed app_id value headers = { "Content-Type": "application/json", "User-Agent": "insomnia/11.2.0" } print("REFRESH_BODY", REFRESH_BODY) resp = requests.request("POST",REFRESH_URL, json=REFRESH_BODY, headers=headers) print(f"Status: {resp.status_code}") print(f"Response: {resp.text}") print(resp) if resp.status_code == 200: data = resp.json() new_token = data.get("tokenData", {}).get("access_token") if not new_token: logging.error("❌ Refresh API returned no access token") return None connected_account = ( db.query(ConnectedAccount) .filter(ConnectedAccount.id == connected_account_id) .first() ) if not connected_account: logging.error(f"❌ ConnectedAccount not found for ID {connected_account_id}") return None connected_account.token = new_token db.commit() db.refresh(connected_account) logging.info(f"🔄 Successfully refreshed Twitter token for ConnectedAccount {connected_account_id}") return new_token else: logging.error(f"❌ Failed to refresh token: {resp.status_code} - {resp.text}") return None except Exception as e: logging.error(f"❌ Error calling token refresh API: {e}") return None finally: db.close() def post_to_twitter(post_type, db:Session = None): if db is None: db: Session = get_db_session() try: access_token = post_type.connected_account.token if not access_token: raise ValueError("Missing access token for Twitter account") content = getattr(post_type, "content", None) if not content: raise ValueError("No content to post for Twitter") url = "https://api.twitter.com/2/tweets" headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json" } twitter_text_html_bold = html_to_unicode_styled(content) payload = {"text": twitter_text_html_bold} response = requests.post(url, json=payload, headers=headers) # If token expired or forbidden, refresh and retry if response.status_code in (401, 403): logging.warning(f"⚠ Twitter {response.status_code} for PostType {post_type.id}, refreshing token...") new_token = refresh_twitter_token( branch_id=post_type.branch_id, connected_account_id=post_type.connected_account.id ) if new_token: headers["Authorization"] = f"Bearer {new_token}" retry_resp = requests.post(url, json=payload, headers=headers) if retry_resp.ok: logging.info(f"✅ Posted to Twitter after token refresh for PostType ID {post_type.id}") post_id = retry_resp.json().get("data", {}).get("id", None) post_type.external_post_id = post_id db.commit() db.refresh(post_type) return { "success": True, "post_id": post_id, } else: logging.error(f"❌ Twitter post failed after token refresh: {retry_resp.status_code} - {retry_resp.text}") return { "success": False, "error": f"Twitter post failed with status {response.status_code}", } # Handle non-successful response if not response.ok: logging.error(f"❌ Twitter API error {response.status_code}: {response.text}") return { "success": False, "error": f"Twitter post failed with status {response.status_code}", } logging.info(f"✅ Posted to Twitter for PostType ID {post_type.id}") post_id = response.json().get("data", {}).get("id", None) # post_type.external_post_id = post_id # db.commit() # db.refresh(post_type) return { "success": True, "post_id": post_id, } except requests.exceptions.RequestException as e: logging.error(f"❌ HTTP request failed for PostType {post_type.id}: {e}") return { "success": False, "error": f"HTTP request failed: {str(e)}", } except Exception as e: logging.error(f"❌ Unexpected error posting to Twitter for PostType {post_type.id}: {e}") return { "success": False, "error": f"Unexpected error: {str(e)}", } #================================================Twitter post automation ======================================================================== #================================================ Facebook post automation ======================================================================== GRAPH = "https://graph.facebook.com/v20.0" def html_to_facebook_text(html: str) -> str: """ Reuse your html_to_unicode_styled, but ensure newlines are preserved. Facebook supports Unicode characters; styled bold via math bold works. """ text = html_to_unicode_styled(html) # Normalize multiple newlines and strip trailing spaces return "\n".join(line.rstrip() for line in text.splitlines()).strip() async def refresh_facebook_token(user_token: str) -> dict: """ Refresh a Facebook access token to get a new long-lived token. Args: user_token: The current user access token to refresh Returns: dict: Contains new access_token and expires_in """ async with httpx.AsyncClient(timeout=20) as client: token_url = f"{GRAPH}/oauth/access_token" r = await client.get(token_url, params={ "grant_type": "fb_exchange_token", "client_id": APP_ID, "client_secret": APP_SECRET, "fb_exchange_token": user_token }) r.raise_for_status() return r.json() def _collect_fb_media_ids(page_id: str, page_token: str, image_urls: list[str]) -> list[str]: """ Upload images as unpublished photos to get media IDs. Later attach them via attached_media on /{page_id}/feed. """ media_ids = [] with requests.Session() as s: for url in image_urls: r = s.post( f"{GRAPH}/{page_id}/photos", data={ "url": url, "published": "false", "access_token": page_token }, timeout=30 ) if r.status_code == 200: mid = r.json().get("id") if mid: media_ids.append(mid) else: logging.error(f"❌ FB photo upload failed: {r.status_code} - {r.text}") return media_ids def post_to_facebook(post_type, db:Session = None): """ Mirrors post_to_twitter: - Formats HTML into styled Unicode text - Posts to /{page_id}/feed (text + link) or text + images via attached_media - On 401/403, refresh the Page Access Token using the stored long-lived user token """ if db is None: db: Session = get_db_session() try: ca = getattr(post_type, "connected_account", None) page_id = getattr(ca, "external_account_id", None) page_token = getattr(ca, "token", None) # Page Access Token # user_long_lived_token = getattr(ca, "user_long_lived_token", None) # Needed to refresh page toke if not page_id: raise ValueError("Missing Facebook page_id on ConnectedAccount") if not page_token: raise ValueError("Missing Facebook page Page Access Token on ConnectedAccount") content_html = getattr(post_type, "content", None) if not content_html: raise ValueError("No content to post for Facebook") message = html_to_facebook_text(content_html) # Optional: derive link and images from your model link = getattr(post_type, "link", None) # image_urls = [img.image_url for img in post_type.images if not img.is_deleted] image_urls = [] # post_images = db.query(PostImage).filter( # PostImage.post_type_id == post_type.id, # PostImage.is_deleted == False # ).all() # for img in post_images: # if img.image_url: # image_urls.append(img.image_url) # If your schema has images on post_type: if hasattr(post_type, "images") and post_type.images: # Expecting each image has .url or .image_url for img in post_type.images: u = getattr(img, "url", None) or getattr(img, "image_url", None) if u: image_urls.append(u) def _do_post(current_page_token: str): payload = {"message": message} # If we have images, upload them first as unpublished and then attach if image_urls: media_ids = _collect_fb_media_ids(page_id, current_page_token, image_urls) if not media_ids: logging.warning("⚠ No media IDs created; falling back to text/link only") # When you use attached_media, you cannot also include 'link' if media_ids: for idx, mid in enumerate(media_ids): payload[f"attached_media[{idx}]"] = json.dumps({"media_fbid": mid}) # Add link only if not using media if link and not any(k.startswith("attached_media") for k in payload.keys()): payload["link"] = link r = requests.post( f"{GRAPH}/{page_id}/feed", data={**payload, "access_token": current_page_token}, timeout=30 ) return r # First attempt r = _do_post(page_token) # If token expired/invalid, refresh and retry if r.status_code in (401, 403): logging.warning(f"⚠ FB {r.status_code} for PostType {post_type.id}; attempting token refresh...") refresh_result = refresh_facebook_token( page_token) new_token = refresh_result["access_token"] expires_in = refresh_result.get("expires_in", 60*60*24*60) expires_in = int(time.time()) + int(expires_in) if new_token: # Persist new page token ca.token = new_token ca.expires_at = expires_in db.commit() db.refresh(ca) r2 = _do_post(new_token) if r2.ok: post_id = (r2.json() or {}).get("id") # FB returns id for post return {"success": True, "post_id": post_id} else: logging.error(f"❌ FB post failed after refresh: {r2.status_code} - {r2.text}") return {"success": False, "error": f"Facebook post failed: {r2.status_code}"} else: return {"success": False, "error": "Failed to refresh Facebook Page token"} if not r.ok: logging.error(f"❌ Facebook API error {r.status_code}: {r.text}") return {"success": False, "error": f"Facebook post failed: {r.status_code}"} post_id = (r.json() or {}).get("id") #calendar_post_type save # post_type.external_post_id = post_id # db.commit() # db.refresh(post_type) return {"success": True, "post_id": post_id} except requests.exceptions.RequestException as e: logging.error(f"❌ HTTP request failed for FB PostType {post_type.id}: {e}") return {"success": False, "error": f"HTTP request failed: {str(e)}"} except Exception as e: logging.error(f"❌ Unexpected error posting to Facebook for PostType {post_type.id}: {e}") return {"success": False, "error": f"Unexpected error: {str(e)}"} finally: db.close()