Spaces:
Runtime error
Runtime error
| import requests | |
| from bs4 import BeautifulSoup | |
| import feedparser | |
| from newspaper import Article, Config | |
| from keybert import KeyBERT | |
| from transformers import pipeline, logging as transformers_logging | |
| import streamlit as st | |
| import nltk | |
| import os | |
| # Reduce library noise | |
| transformers_logging.set_verbosity_error() | |
| def download_nltk_data(): | |
| """Download required nltk data once.""" | |
| try: | |
| import nltk | |
| nltk.download('punkt', quiet=True) | |
| nltk.download('punkt_tab', quiet=True) | |
| except: | |
| pass | |
| # Initialize NLTK | |
| download_nltk_data() | |
| class ContentExtractor: | |
| def from_url(url): | |
| try: | |
| user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36' | |
| config = Config() | |
| config.browser_user_agent = user_agent | |
| config.request_timeout = 15 | |
| article = Article(url, config=config) | |
| article.download() | |
| article.parse() | |
| if not article.text: | |
| return {"error": "Fetched successfully but no text content was found. The site might be blocking extraction or uses a complex JS-only layout."} | |
| return { | |
| "title": article.title, | |
| "text": article.text, | |
| "images": [article.top_image] if article.top_image else [], | |
| "meta": article.meta_description, | |
| "keywords": article.keywords | |
| } | |
| except Exception as e: | |
| return {"error": f"Extraction failed: {str(e)}"} | |
| def from_rss(rss_url): | |
| try: | |
| feed = feedparser.parse(rss_url) | |
| entries = [] | |
| for entry in feed.entries[:5]: # Limit to latest 5 | |
| entries.append({ | |
| "title": entry.title, | |
| "link": entry.link, | |
| "summary": entry.get("summary", ""), | |
| "published": entry.get("published", "") | |
| }) | |
| return entries | |
| except Exception as e: | |
| return {"error": str(e)} | |
| def from_sitemap(sitemap_url): | |
| try: | |
| resp = requests.get(sitemap_url) | |
| soup = BeautifulSoup(resp.content, "xml") | |
| urls = [loc.text for loc in soup.find_all("loc")][:10] # Limit to 10 | |
| return urls | |
| except Exception as e: | |
| return {"error": str(e)} | |
| class ForgeAI: | |
| def __init__(self): | |
| # Initializing pipelines (these will be cached by streamlit in app.py) | |
| pass | |
| def get_summarizer(_self): | |
| """Loads a lightweight summarization model once and caches it.""" | |
| try: | |
| # Using distilbart-cnn-6-6 (~460MB) for production stability on CPU | |
| return pipeline("summarization", model="sshleifer/distilbart-cnn-6-6") | |
| except Exception as e: | |
| # Fallback for extreme cases | |
| try: | |
| return pipeline("text-generation", model="gpt2") | |
| except: | |
| return None | |
| def get_kw_model(): | |
| return KeyBERT() | |
| def generate_master_variations(self, content): | |
| """Generates 3 distinct master content variations based on the input.""" | |
| summarizer = self.get_summarizer() | |
| chunk_size = 1000 | |
| input_text = content[:chunk_size] | |
| variations = [] | |
| # Variation 1: Executive Summary (Professional & Concise) | |
| summary_v1 = summarizer(input_text, max_length=150, min_length=50, do_sample=False)[0] | |
| v1_text = summary_v1.get('summary_text', summary_v1.get('generated_text', str(summary_v1))) | |
| variations.append({ | |
| "id": "v1", | |
| "type": "Executive Summary", | |
| "text": v1_text | |
| }) | |
| # Variation 2: Engaging Hook (Curiosity-driven) | |
| # For simplicity in free tier, we re-parse or slightly modify the tone. | |
| # In a full version, we'd use specific prompts. | |
| v2_text = f"Did you know? {v1_text[:100]}... Here's why this matters for you." | |
| variations.append({ | |
| "id": "v2", | |
| "type": "Engaging Hook", | |
| "text": v2_text | |
| }) | |
| # Variation 3: Actionable Insight (Bullet points / Direct) | |
| v3_text = f"Key Takeaways:\n- {v1_text[:80]}\n- Deep dive into the details\n- Final conclusion included." | |
| variations.append({ | |
| "id": "v3", | |
| "type": "Actionable Insight", | |
| "text": v3_text | |
| }) | |
| return variations | |
| def extract_hashtags(self, content): | |
| kw_model = self.get_kw_model() | |
| keywords = kw_model.extract_keywords(content, keyphrase_ngram_range=(1, 1), stop_words='english', top_n=5) | |
| return " ".join([f"#{kw[0].capitalize()}" for kw in keywords]) | |
| def adapt_for_platform(self, master_text, platform, hashtags): | |
| """Adapts a master variation to a specific platform's format.""" | |
| if platform == "LinkedIn": | |
| return f"🎨 Professional Update:\n\n{master_text}\n\n#Professional #Insights {hashtags}" | |
| elif platform == "Twitter/X": | |
| return f"🚀 {master_text[:240]}\n\n{hashtags}" | |
| elif platform == "Instagram": | |
| return f"{master_text}\n.\n.\n.\n📸 Link in bio!\n{hashtags}" | |
| elif platform == "Facebook": | |
| return f"Check this out! 📢\n\n{master_text}\n\nRead more below 👇\n{hashtags}" | |
| elif platform == "YouTube": | |
| return { | |
| "title": f"New Content: {master_text[:50]}...", | |
| "description": f"{master_text}\n\n{hashtags}" | |
| } | |
| elif platform == "Bluesky": | |
| return f"🦋 {master_text[:250]}\n\n{hashtags}" | |
| return master_text | |
| def generate_captions(self, content, platforms): | |
| # Legacy method for backward compatibility if needed, | |
| # but the new flow will use individual methods above. | |
| variations = self.generate_master_variations(content) | |
| hashtags = self.extract_hashtags(content) | |
| results = {} | |
| # By default, use V1 for backward compatibility | |
| master_text = variations[0]['text'] | |
| for platform in platforms: | |
| results[platform] = self.adapt_for_platform(master_text, platform, hashtags) | |
| return results, hashtags | |