Spaces:
Sleeping
Sleeping
| from shiny import reactive, render, ui | |
| import os | |
| import requests | |
| from dotenv import load_dotenv | |
| from bs4 import BeautifulSoup | |
| import subprocess | |
| import time | |
| from llm_connect import get_response | |
| load_dotenv() | |
| PAGE_ID = os.getenv("FB_PAGE_ID") | |
| ACCESS_TOKEN = os.getenv("FB_PAGE_ACCESS_TOKEN") | |
| generated_fb_post = reactive.Value("") | |
| def scrape_shopify_blog(url: str) -> str: | |
| try: | |
| resp = requests.get(url, timeout=10) | |
| soup = BeautifulSoup(resp.content, 'html.parser') | |
| section = soup.find( | |
| 'div', | |
| class_='w940 align-center size-content' | |
| ) | |
| return section.get_text(strip=True, separator=' ') if section else "" | |
| except Exception as e: | |
| return f"β Error scraping blog: {e}" | |
| def generate_facebook_post(topic: str = "", url: str = "", min_len: int = 500, max_len: int = 1000) -> str: | |
| # Handle blog scraping | |
| if url: | |
| scraped = scrape_shopify_blog(url) | |
| if scraped.startswith("β") or not scraped.strip(): | |
| return f"β οΈ Failed to extract blog content from URL: {url}" | |
| prompt = ( | |
| f"You are a social media manager for a hobby e-commerce company called 'Ultima Supply'.\n" | |
| f"Write a detailed, engaging Facebook post (min {min_len} characters, max {max_len} characters) summarizing the following Shopify blog:\n\n" | |
| f"{scraped}\n\n" | |
| f"The post MUST include this exact URL: {url}\n" | |
| f"Use a casual, friendly tone with emojis.\n" | |
| f"VERY IMPORTANT: The post MUST include exactly 5 to 10 SEO-relevant hashtags at the end.\n" | |
| f"Do not include hashtags inline β list them as a group at the end of the post.\n" | |
| f"Keep everything under {max_len} characters total." | |
| ) | |
| elif topic: | |
| prompt = ( | |
| f"You are a social media manager for a hobby e-commerce company called 'Ultima Supply'.\n" | |
| f"Write a short, engaging Facebook post (min {min_len} characters, max {max_len} characters) about: '{topic}'.\n" | |
| f"Use a casual, friendly tone with fun emojis.\n" | |
| f"VERY IMPORTANT: Include exactly 5 to 10 SEO-relevant hashtags grouped at the end of the post.\n" | |
| f"Do NOT place hashtags inline β they must be in a clean list at the end.\n" | |
| f"The entire post must be under {max_len} characters total." | |
| ) | |
| else: | |
| return "β οΈ Provide either a topic or a Shopify blog URL." | |
| # Step 1: Generate initial post | |
| post = get_response( | |
| input=prompt, | |
| template=lambda x: x.strip(), | |
| llm="gemini", | |
| md=False, | |
| temperature=0.9, | |
| max_tokens=250 | |
| ).strip() | |
| if len(post) <= max_len: | |
| return post | |
| time.sleep(5) | |
| # Step 2: Shorten if needed | |
| shorten_prompt = ( | |
| f"Shorten this Facebook post to {max_len} characters or fewer.\n" | |
| f"You MUST keep the Shopify blog link ({url}) and all original hashtags:\n\n{post}" | |
| if url else | |
| f"Shorten this Facebook post to {max_len} characters or fewer, keeping the tone, emojis, and all original hashtags:\n\n{post}" | |
| ) | |
| shortened = get_response( | |
| input=shorten_prompt, | |
| template=lambda x: x.strip(), | |
| llm="gemini", | |
| md=False, | |
| temperature=0.7, | |
| max_tokens=200 | |
| ).strip() | |
| return shortened[:max_len] | |
| def post_to_facebook(message: str) -> str: | |
| proxy_url = "https://facebook-proxy.onrender.com/post-to-facebook" # your actual deployed Render URL | |
| try: | |
| res = requests.post(proxy_url, json={"message": message}, timeout=10) | |
| res.raise_for_status() | |
| data = res.json() | |
| # Safely handle both proxy success and raw Facebook responses | |
| if isinstance(data, dict): | |
| if "status" in data and data["status"] == "success": | |
| return f"β Post successful! Facebook Post ID: {data.get('post_id', 'unknown')}" | |
| elif "id" in data or "post_id" in data: | |
| post_id = data.get("post_id") or data.get("id") | |
| return f"β Post successful! Facebook Post ID: {post_id}" | |
| elif "error" in data: | |
| return f"β Facebook error: {data['error']}" | |
| return f"β οΈ Unexpected response: {data}" | |
| except requests.RequestException as e: | |
| return f"β Network or proxy error: {e}" | |
| except Exception as e: | |
| return f"β Unexpected error: {e}" | |
| def server(input, output, session): | |
| post_status = reactive.Value("") | |
| def fb_post_draft(): | |
| topic = input.fb_topic().strip() | |
| url = input.fb_url().strip() | |
| if not topic and not url: | |
| return ui.HTML("<p><strong>β οΈ Enter a topic or a blog URL.</strong></p>") | |
| post = generate_facebook_post(topic=topic, url=url) | |
| if post.startswith("β οΈ") or post.startswith("β"): | |
| return ui.HTML(f"<p><strong>{post}</strong></p>") | |
| generated_fb_post.set(post) | |
| return ui.HTML(f"<p><strong>Generated Facebook Post:</strong><br>{post}</p>") | |
| def fb_post_status(): | |
| return post_status() | |
| def _(): | |
| post = generated_fb_post() | |
| if not post: | |
| post_status.set("β οΈ No post generated yet.") | |
| else: | |
| result = post_to_facebook(post) | |
| post_status.set(result) | |