| ο»Ώ |
| """ |
| Article Rewriter Module |
| ======================= |
| Scrapes articles from URLs, rewrites them with Gemma 4 AI, |
| finds related YouTube videos & images for embedding, |
| generates featured images, and publishes to WordPress. |
| |
| Usage (standalone): |
| python article_rewriter.py <article_url> |
| python article_rewriter.py --file articles.txt |
| """ |
|
|
| import json |
| import os |
| import re |
| import textwrap |
| import time |
| import hashlib |
| import base64 |
| from pathlib import Path |
| from io import BytesIO |
| from urllib.parse import urlparse, urljoin, quote_plus |
|
|
| from dotenv import load_dotenv |
| from bs4 import BeautifulSoup |
| import requests |
| import aiohttp |
| import asyncio |
| from PIL import Image, ImageDraw, ImageFont |
| from pydantic import BaseModel, ValidationError |
| from wordpress_transport import ( |
| post_multipart_to_wordpress_origin, |
| pull_relay_enabled, |
| request_wordpress_json, |
| ) |
|
|
| |
| |
| |
|
|
| class ArticleContent(BaseModel): |
| """Validates the LLM's article rewrite output has all required fields.""" |
| title: str |
| paragraphs: list[str] |
| seo_description: str = "" |
| image_alt_texts: list[str] = [] |
| tags: list[str] = [] |
|
|
|
|
| def _validate_title(generated_title: str, original_title: str) -> str: |
| """ |
| Validate the generated title shares meaningful overlap with the original. |
| Falls back to original title if the model went completely off-topic. |
| """ |
| if not generated_title or not generated_title.strip(): |
| return original_title[:70] |
|
|
| title = generated_title.strip() |
|
|
| |
| if len(title) < 3 or len(title) > 80: |
| return original_title[:70] |
|
|
| |
| |
| def _significant_words(t): |
| return {w.lower().strip('.,!?;:"\'') for w in t.split() if len(w) >= 3} |
|
|
| orig_words = _significant_words(original_title) |
| gen_words = _significant_words(title) |
| overlap = orig_words & gen_words |
|
|
| if not overlap: |
| |
| print(f" [WARN] Generated title shares no words with original. Falling back.") |
| return original_title[:70] |
|
|
| return title |
|
|
| |
| |
| |
| load_dotenv() |
|
|
| from gemma_ai import gemini_generate, image_to_part, image_to_base64 |
|
|
| WP_URL = os.getenv("WP_URL", "").rstrip("/") |
| WP_USERNAME = os.getenv("WP_USERNAME", "") |
| WP_APP_PASSWORD = "".join(os.getenv("WP_APP_PASSWORD", "").split()) |
| WP_CATEGORY_ID = int(os.getenv("WP_CATEGORY_NEWS", os.getenv("WP_REWRITE_CATEGORY_ID", os.getenv("WP_CATEGORY_ID", "1")))) |
| WP_POST_STATUS = os.getenv("WP_POST_STATUS", "publish") |
| LLM_MODEL = os.getenv("LLM_MODEL", "gemma-4-31b-it") |
|
|
| |
| _DATA_DIR = Path("/data") if Path("/data").is_dir() or os.getenv("HUGGING_FACE_SPACES") else Path(__file__).parent |
| if _DATA_DIR == Path("/data"): |
| _DATA_DIR.mkdir(parents=True, exist_ok=True) |
| DOWNLOAD_DIR = _DATA_DIR / "downloads" |
| OUTPUT_DIR = _DATA_DIR / "output" |
|
|
| |
| _ARTICLE_CACHE_PATH = _DATA_DIR / "article_rewriter_cache.json" |
|
|
| |
| HEADERS = { |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36", |
| "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8", |
| "Accept-Language": "en-US,en;q=0.9", |
| "Accept-Encoding": "identity", |
| "Connection": "keep-alive", |
| } |
| |
| if WP_URL: |
| HEADERS["Origin"] = WP_URL |
|
|
|
|
| |
| |
| |
|
|
| def _get_article_cache_key(source_url: str, body_text: str) -> str: |
| """Create a content hash from URL + body fingerprint for cache lookup.""" |
| hasher = hashlib.md5() |
| hasher.update(source_url.encode()) |
| |
| hasher.update(body_text[:2000].encode()) |
| return hasher.hexdigest() |
|
|
| def _load_article_cache() -> dict: |
| if _ARTICLE_CACHE_PATH.exists(): |
| try: |
| with open(_ARTICLE_CACHE_PATH, "r") as f: |
| return json.load(f) |
| except (json.JSONDecodeError, IOError): |
| return {} |
| return {} |
|
|
| def _save_article_cache(cache: dict): |
| try: |
| with open(_ARTICLE_CACHE_PATH, "w") as f: |
| json.dump(cache, f) |
| except IOError: |
| pass |
|
|
|
|
| |
| |
| |
|
|
| def _ensure_dirs(): |
| """Create working directories if they don't exist.""" |
| DOWNLOAD_DIR.mkdir(exist_ok=True) |
| OUTPUT_DIR.mkdir(exist_ok=True) |
|
|
|
|
| def _wp_auth() -> tuple[str, str]: |
| """Return WordPress auth tuple.""" |
| return (WP_USERNAME, WP_APP_PASSWORD) |
|
|
|
|
| def _wp_auth_header() -> str: |
| """Return Basic Auth header string for direct header injection.""" |
| creds = f'{WP_USERNAME}:{WP_APP_PASSWORD}' |
| return f"Basic {base64.b64encode(creds.encode()).decode()}" |
|
|
|
|
| def _wp_ssl_verify() -> bool: |
| """Allow deployments with a broken WordPress certificate chain to opt out.""" |
| return os.getenv("WP_SSL_VERIFY", "true").strip().lower() not in ("0", "false", "no") |
|
|
|
|
| def _sanitize_filename(name: str) -> str: |
| """Create a safe filename from a string.""" |
| safe = re.sub(r'[^\w\-]', '_', name) |
| return safe[:80] |
|
|
|
|
| def _get_font(size: int): |
| """Try to load a nice font, fall back to default.""" |
| font_paths = [ |
| "C:/Windows/Fonts/segoeui.ttf", |
| "C:/Windows/Fonts/arial.ttf", |
| "C:/Windows/Fonts/calibri.ttf", |
| "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", |
| "/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf", |
| "/System/Library/Fonts/Helvetica.ttc", |
| ] |
| for fp in font_paths: |
| try: |
| return ImageFont.truetype(fp, size) |
| except (IOError, OSError): |
| continue |
| try: |
| return ImageFont.truetype("arial.ttf", size) |
| except (IOError, OSError): |
| return ImageFont.load_default() |
|
|
|
|
| def _extract_embed_info(url: str) -> dict | None: |
| """Extract media embed info from a URL if it's a supported platform.""" |
| if not url: |
| return None |
|
|
| |
| yt_match = re.search( |
| r'(?:youtube\.com/(?:watch\?v=|embed/|v/)|youtu\.be/)([\w-]{11})', url |
| ) |
| if yt_match: |
| vid_id = yt_match.group(1) |
| return { |
| "url": f"https://www.youtube.com/watch?v={vid_id}", |
| "title": "YouTube Video", |
| "embed_url": f"https://www.youtube.com/embed/{vid_id}", |
| "provider": "youtube" |
| } |
|
|
| |
| st_match = re.search(r'streamable\.com/(?:e/|s/)?(\w+)', url) |
| if st_match: |
| vid_id = st_match.group(1) |
| return { |
| "url": f"https://streamable.com/{vid_id}", |
| "title": "Streamable Video", |
| "embed_url": f"https://streamable.com/e/{vid_id}", |
| "provider": "streamable" |
| } |
|
|
| |
| vim_match = re.search(r'(?:vimeo\.com|player\.vimeo\.com/video)/(\d+)', url) |
| if vim_match: |
| vid_id = vim_match.group(1) |
| return { |
| "url": f"https://vimeo.com/{vid_id}", |
| "title": "Vimeo Video", |
| "embed_url": f"https://player.vimeo.com/video/{vid_id}", |
| "provider": "vimeo" |
| } |
|
|
| |
| dm_match = re.search(r'dailymotion\.com/(?:video|embed/video)/(\w+)', url) |
| if dm_match: |
| vid_id = dm_match.group(1) |
| return { |
| "url": f"https://www.dailymotion.com/video/{vid_id}", |
| "title": "Dailymotion Video", |
| "embed_url": f"https://www.dailymotion.com/embed/video/{vid_id}", |
| "provider": "dailymotion" |
| } |
|
|
| |
| tt_match = re.search(r'tiktok\.com/(@[\w.-]+)/video/(\d+)', url) |
| if tt_match: |
| |
| clean_url = f"https://www.tiktok.com/{tt_match.group(1)}/video/{tt_match.group(2)}" |
| return { |
| "url": clean_url, |
| "title": "TikTok Video", |
| "embed_url": clean_url, |
| "provider": "tiktok" |
| } |
|
|
| |
| ig_match = re.search(r'instagram\.com/(?:p|reel|reels)/([\w-]+)', url) |
| if ig_match: |
| |
| clean_url = f"https://www.instagram.com/reel/{ig_match.group(1)}/" |
| return { |
| "url": clean_url, |
| "title": "Instagram Post", |
| "embed_url": clean_url, |
| "provider": "instagram" |
| } |
|
|
| |
| fb_match = re.search(r'facebook\.com/(?:watch/?\?v=|[\w.-]+/videos/|video\.php\?v=)(\d+)', url) |
| if fb_match: |
| |
| clean_url = f"https://www.facebook.com/watch/?v={fb_match.group(1)}" |
| return { |
| "url": clean_url, |
| "title": "Facebook Video", |
| "embed_url": clean_url, |
| "provider": "facebook" |
| } |
|
|
| |
| x_match = re.search(r'(?:twitter\.com|x\.com)/([\w-]+)/status/(\d+)', url) |
| if x_match: |
| username = x_match.group(1) |
| tweet_id = x_match.group(2) |
| clean_url = f"https://x.com/{username}/status/{tweet_id}" |
| return { |
| "url": clean_url, |
| "title": "X Post by @{username}", |
| "embed_url": clean_url, |
| "provider": "x" |
| } |
|
|
| |
| threads_match = re.search(r'threads\.net/(@[\w.-]+)/post/([\w-]+)', url) |
| if threads_match: |
| clean_url = f"https://www.threads.net/{threads_match.group(1)}/post/{threads_match.group(2)}" |
| return { |
| "url": clean_url, |
| "title": "Threads Post", |
| "embed_url": clean_url, |
| "provider": "threads" |
| } |
|
|
| |
| reddit_match = re.search(r'reddit\.com/r/[\w-]+/comments/(\w+)', url) |
| if reddit_match: |
| |
| clean_url = url.split("?")[0] |
| return { |
| "url": clean_url, |
| "title": "Reddit Post", |
| "embed_url": clean_url, |
| "provider": "reddit" |
| } |
|
|
| |
| spotify_match = re.search(r'open\.spotify\.com/(track|album|playlist|episode|show)/([\w-]+)', url) |
| if spotify_match: |
| clean_url = f"https://open.spotify.com/{spotify_match.group(1)}/{spotify_match.group(2)}" |
| return { |
| "url": clean_url, |
| "title": "Spotify Media", |
| "embed_url": clean_url, |
| "provider": "spotify" |
| } |
|
|
| |
| soundcloud_match = re.search(r'soundcloud\.com/[\w-]+/[\w-]+', url) |
| if soundcloud_match: |
| clean_url = url.split("?")[0] |
| return { |
| "url": clean_url, |
| "title": "SoundCloud Audio", |
| "embed_url": clean_url, |
| "provider": "soundcloud" |
| } |
|
|
| |
| pinterest_match = re.search(r'pinterest\.com/pin/(\d+)', url) |
| if pinterest_match: |
| clean_url = f"https://www.pinterest.com/pin/{pinterest_match.group(1)}/" |
| return { |
| "url": clean_url, |
| "title": "Pinterest Pin", |
| "embed_url": clean_url, |
| "provider": "pinterest" |
| } |
|
|
| |
| bsky_match = re.search(r'bsky\.app/profile/[\w.-]+/post/([\w-]+)', url) |
| if bsky_match: |
| return { |
| "url": url, |
| "title": "Bluesky Post", |
| "embed_url": url, |
| "provider": "bluesky" |
| } |
|
|
| return None |
|
|
|
|
| |
| |
| |
|
|
| def scrape_article(url: str) -> dict: |
| """ |
| Scrape an article from a URL. |
| Returns: { |
| "title": str, |
| "body": str, # Full article text |
| "images": [str], # List of image URLs found in the article body |
| "featured_image": str, # Source site's chosen featured image (JSON-LD/og:image) |
| "videos": [dict], # Embedded media (YouTube, TikTok, etc.) |
| "source_url": str, |
| "source_domain": str |
| } |
| """ |
| print(f"π Scraping article: {url}") |
|
|
| |
| social_domains = ["tiktok.com", "instagram.com", "facebook.com", "x.com", "twitter.com", "threads.net"] |
| if any(d in url for d in social_domains): |
| print(f" β οΈ [WARN] URL appears to be a social media post. Scraping may be limited.") |
|
|
| try: |
| resp = requests.get(url, headers=HEADERS, timeout=30, allow_redirects=True) |
| resp.raise_for_status() |
| except requests.RequestException as e: |
| raise Exception(f"Failed to fetch article: {e}") |
|
|
| soup = BeautifulSoup(resp.text, "html.parser") |
| domain = urlparse(url).netloc |
|
|
| |
| title = "" |
| |
| og_title = soup.find("meta", property="og:title") |
| if og_title and og_title.get("content"): |
| title = og_title["content"].strip() |
| |
| if not title and soup.title: |
| title = soup.title.string.strip() if soup.title.string else "" |
| |
| if not title: |
| h1 = soup.find("h1") |
| if h1: |
| title = h1.get_text(strip=True) |
|
|
| |
| body_parts = [] |
|
|
| |
| article_selectors = [ |
| "article", |
| '[class*="article-body"]', |
| '[class*="article-content"]', |
| '[class*="post-content"]', |
| '[class*="entry-content"]', |
| '[class*="story-body"]', |
| '[class*="content-body"]', |
| '[role="main"]', |
| "main", |
| '[class*="content"]', |
| ] |
|
|
| content_elem = None |
| for selector in article_selectors: |
| found = soup.select_one(selector) |
| if found: |
| |
| text = found.get_text(strip=True) |
| if len(text) > 200: |
| content_elem = found |
| break |
|
|
| if content_elem is None: |
| |
| content_elem = soup.body if soup.body else soup |
|
|
| |
| |
| noise_selectors = [ |
| '[class*="related-post"]', '[class*="related-article"]', |
| '[class*="popular-post"]', '[class*="recommended"]', |
| '[class*="suggested"]', '[class*="must-read"]', |
| '[class*="read-more"]', '[class*="trending"]', |
| '[class*="sidebar"]', '[class*="newsletter"]', |
| '[class*="social-share"]', '[class*="author-box"]', |
| '[class*="comment"]', '[class*="ad-"]', '[id*="ad-"]', |
| '[class*="advert"]', '[id*="advert"]', '[class*="promo"]', |
| '[class*="sponsor"]', '[class*="banner"]', '[class*="widget"]', |
| '[class*="next-post"]', '[class*="prev-post"]', '[class*="pagination"]', |
| '[class*="td-module-thumb"]', '[class*="td-image-wrap"]', '[class*="entry-thumb"]', |
| '[class*="td_module_wrap"]', '[class*="post-thumbnail"]', |
| 'aside', 'footer', 'nav', 'script', 'style', 'noscript', 'iframe' |
| ] |
| for noise_sel in noise_selectors: |
| for noise_node in content_elem.select(noise_sel): |
| noise_node.decompose() |
|
|
| |
| paragraphs = content_elem.find_all("p") |
| for p in paragraphs: |
| |
| if p.find_parents("blockquote"): |
| continue |
| |
| text = p.get_text(strip=True) |
| if len(text) > 30: |
| body_parts.append(text) |
|
|
| |
| if len(body_parts) < 3: |
| for tag in content_elem.find_all(["h2", "h3", "h4", "li"]): |
| text = tag.get_text(strip=True) |
| if len(text) > 20: |
| body_parts.append(text) |
|
|
| body = "\n\n".join(body_parts) |
|
|
| |
| images = [] |
| featured_image_url = None |
|
|
| |
| for script in soup.find_all("script", type="application/ld+json"): |
| |
| |
| raw = script.string or script.get_text(strip=True) |
| if not raw: |
| continue |
| |
| raw = re.sub(r"<!--.*?-->", "", raw, flags=re.DOTALL) |
| raw = re.sub(r"<!\[CDATA\[", "", raw) |
| raw = re.sub(r"\]\]>", "", raw) |
| raw = raw.strip() |
| if not raw: |
| continue |
| try: |
| data = json.loads(raw) |
| |
| if isinstance(data, dict): |
| if "thumbnailUrl" in data: |
| img_url = data["thumbnailUrl"] |
| if not img_url.startswith("http"): |
| img_url = urljoin(url, img_url) |
| featured_image_url = img_url |
| images.append(img_url) |
| print(f" πΌοΈ Found featured image in JSON-LD thumbnailUrl: {img_url}") |
| |
| if "image" in data and isinstance(data["image"], dict) and "url" in data["image"]: |
| img_url = data["image"]["url"] |
| if not img_url.startswith("http"): |
| img_url = urljoin(url, img_url) |
| if img_url not in images: |
| images.append(img_url) |
| print(f" πΌοΈ Found image in JSON-LD image.url: {img_url}") |
| elif isinstance(data, list): |
| for item in data: |
| if isinstance(item, dict): |
| if "thumbnailUrl" in item: |
| img_url = item["thumbnailUrl"] |
| if not img_url.startswith("http"): |
| img_url = urljoin(url, img_url) |
| if not featured_image_url: |
| featured_image_url = img_url |
| if img_url not in images: |
| images.append(img_url) |
| print(f" πΌοΈ Found featured image in JSON-LD graph thumbnailUrl") |
| if "image" in item and isinstance(item["image"], dict) and "url" in item["image"]: |
| img_url = item["image"]["url"] |
| if not img_url.startswith("http"): |
| img_url = urljoin(url, img_url) |
| if img_url not in images: |
| images.append(img_url) |
| print(f" πΌοΈ Found image in JSON-LD graph image.url") |
| except (json.JSONDecodeError, KeyError, TypeError) as e: |
| print(f" [WARN] JSON-LD parse error: {e}") |
|
|
| |
| og_image = soup.find("meta", property="og:image") |
| if og_image and og_image.get("content"): |
| img_url = og_image["content"] |
| if not img_url.startswith("http"): |
| img_url = urljoin(url, img_url) |
| if not featured_image_url: |
| featured_image_url = img_url |
| print(f" πΌοΈ Found featured image in og:image: {img_url}") |
| if img_url not in images: |
| images.append(img_url) |
| print(f" πΌοΈ Found additional og:image: {img_url}") |
|
|
| |
| twitter_image = soup.find("meta", attrs={"name": "twitter:image"}) |
| if twitter_image and twitter_image.get("content"): |
| img_url = twitter_image["content"] |
| if not img_url.startswith("http"): |
| img_url = urljoin(url, img_url) |
| if not featured_image_url: |
| featured_image_url = img_url |
| print(f" πΌοΈ Found featured image in twitter:image: {img_url}") |
| if img_url not in images: |
| images.append(img_url) |
| print(f" πΌοΈ Found additional twitter:image: {img_url}") |
|
|
| |
| if content_elem: |
| for img in content_elem.find_all("img"): |
| |
| src = None |
| for attr in ["src", "data-src", "data-lazy-src", "data-original", "data-srcset"]: |
| val = img.get(attr, "") |
| if attr == "data-srcset": |
| |
| if val: |
| src = val.split()[0] if val.split() else None |
| elif val and not val.startswith("data:image"): |
| src = val |
| break |
|
|
| if not src: |
| continue |
|
|
| if not src.startswith("http"): |
| src = urljoin(url, src) |
|
|
| |
| width = img.get("width", "") |
| height = img.get("height", "") |
| try: |
| if width and int(width) < 300: |
| continue |
| if height and int(height) < 200: |
| continue |
| except ValueError: |
| pass |
|
|
| if src not in images: |
| images.append(src) |
| print(f" πΌοΈ Found image in article body: {src[:80]}...") |
|
|
| |
| embeds = [] |
| if content_elem: |
| |
| for iframe in content_elem.find_all("iframe"): |
| src = iframe.get("src") or iframe.get("data-src") or "" |
| embed_info = _extract_embed_info(src) |
| if embed_info and embed_info["url"] not in [e["url"] for e in embeds]: |
| embeds.append(embed_info) |
| print(f" π¬ Found embedded {embed_info['provider']}: {embed_info['url']}") |
|
|
| |
| for blockquote in content_elem.find_all("blockquote", class_="tiktok-embed"): |
| cite = blockquote.get("cite") |
| if cite: |
| embed_info = _extract_embed_info(cite) |
| if embed_info and embed_info["url"] not in [e["url"] for e in embeds]: |
| embeds.append(embed_info) |
| print(f" π¬ Found TikTok cite: {embed_info['url']}") |
|
|
| |
| for blockquote in content_elem.find_all("blockquote", class_="instagram-media"): |
| link = blockquote.find("a", href=True) |
| if link: |
| embed_info = _extract_embed_info(link["href"]) |
| if embed_info and embed_info["url"] not in [e["url"] for e in embeds]: |
| embeds.append(embed_info) |
| print(f" π¬ Found Instagram embed: {embed_info['url']}") |
|
|
| |
| for blockquote in content_elem.find_all("blockquote", class_="twitter-tweet"): |
| link = blockquote.find("a", href=True) |
| if link: |
| embed_info = _extract_embed_info(link["href"]) |
| if embed_info and embed_info["url"] not in [e["url"] for e in embeds]: |
| embeds.append(embed_info) |
| print(f" π¬ Found X embed: {embed_info['url']}") |
|
|
| |
| for fb_div in content_elem.find_all(["div", "span"], class_=["fb-video", "fb-post"]): |
| href = fb_div.get("data-href") |
| if href: |
| embed_info = _extract_embed_info(href) |
| if embed_info and embed_info["url"] not in [e["url"] for e in embeds]: |
| embeds.append(embed_info) |
| print(f" π¬ Found Facebook embed: {embed_info['url']}") |
|
|
| |
| for a_tag in content_elem.find_all("a", href=True): |
| href = a_tag["href"] |
| embed_info = _extract_embed_info(href) |
| if embed_info and embed_info["url"] not in [e["url"] for e in embeds]: |
| embeds.append(embed_info) |
| print(f" π¬ Found linked {embed_info['provider']}: {embed_info['url']}") |
|
|
| |
| for video_tag in content_elem.find_all("video"): |
| src = video_tag.get("src") or "" |
| if not src: |
| source_tag = video_tag.find("source") |
| if source_tag: |
| src = source_tag.get("src", "") |
| if src and src.startswith("http"): |
| embeds.append({ |
| "url": src, |
| "title": "Embedded Video", |
| "embed_url": src, |
| "provider": "video" |
| }) |
|
|
| |
| embeds = embeds[:5] |
| |
| images = images[:10] |
|
|
| if not body: |
| raise Exception(f"Could not extract article content from {url}") |
|
|
| print(f" β
Scraped: \"{title[:60]}...\"") |
| print(f" π {len(body_parts)} paragraphs, {len(images)} images, {len(embeds)} embeds found") |
|
|
| return { |
| "title": title, |
| "body": body, |
| "images": images, |
| "featured_image": featured_image_url, |
| "videos": embeds, |
| "source_url": url, |
| "source_domain": domain, |
| } |
|
|
|
|
| |
| |
| |
|
|
| def rewrite_article(original_title: str, original_body: str, source_url: str) -> dict: |
| """ |
| Rewrite an article using Gemma 4 AI. |
| Uses a cache to skip the LLM call for previously processed articles. |
| """ |
| |
| body_text = original_body[:10000] if len(original_body) > 10000 else original_body |
| cache_key = _get_article_cache_key(source_url, body_text) |
| cache = _load_article_cache() |
| if cache_key in cache: |
| print(f" [CACHE] Article rewrite found for this source (skip AI call)") |
| result = cache[cache_key] |
| print(f" β
Rewritten (cached): \"{result['title']}\"") |
| print(f" π {len(result['paragraphs'])} paragraphs generated") |
| return result |
|
|
| prompt = textwrap.dedent(f""" |
| Rewrite the article below into a well-written, in-depth blog post. |
| The tone should be professional yet warm β like a knowledgeable journalist explaining a story to an interested reader over coffee. Not academic, not casual. Think BBC News meets a really good Substack writer. |
| |
| ORIGINAL TITLE: {original_title} |
| |
| ORIGINAL ARTICLE: |
| {body_text} |
| |
| SOURCE: {source_url} |
| |
| Respond ONLY with valid JSON in this exact format (no markdown, no code blocks): |
| {{ |
| "title": "A clear, professional headline that captures the essence (max 70 chars)", |
| "paragraphs": [ |
| "Opening paragraph β set the scene. What happened, who's involved, why it matters. 3-4 sentences that pull the reader in.", |
| "Second paragraph β go deeper into the story. What led to this moment? What's the background context? 3-4 sentences.", |
| "Third paragraph β bring in details, quotes, or specific examples from the original. This is where the story gets interesting. 3-4 sentences.", |
| "Fourth paragraph β explore the implications. What does this mean going forward? What are people saying? 3-4 sentences.", |
| "Fifth paragraph (if relevant) β add additional context, reactions, or related developments. 3-4 sentences." |
| ], |
| "seo_description": "A concise, professional summary of the article (150-160 chars).", |
| "image_alt_texts": [ |
| "Clear, descriptive caption for the main image", |
| "Clear, descriptive caption for a second image if relevant" |
| ], |
| "tags": ["topic1", "topic2", "topic3", "topic4"] |
| }} |
| |
| TITLE RULES: |
| - Max 70 characters. |
| - Clear, professional, and accurate to the story. |
| - Keep the same key people, places, events, and angle from the original title. |
| - If the original names a person or place, your title should too. |
| - If the original title is a QUOTE, keep a version of that quote and name the person. |
| Example: Original -> "High dollar helps exporters - Kojo Poku insists" -> New -> "Kojo Poku: The Dollar Is Helping Exporters" |
| - Avoid clickbait. Sound credible and trustworthy. |
| |
| QUOTE RULES (critical): |
| - If someone said something in the original article, KEEP the direct quote. Do NOT paraphrase it. |
| - Write quotes exactly as they appeared, in quotation marks, with the speaker's name. |
| - If the article includes a statement, opinion, or claim from a named person, ALWAYS attribute it to them. |
| - If no direct quotes exist, report facts without adding opinions. |
| |
| DEPTH & STRUCTURE RULES: |
| - Write 4-5 paragraphs (not just 2-3). Go deeper than a surface summary. |
| - Each paragraph should be 3-4 well-crafted sentences. Substantive, not fluffy. |
| - Paragraph 1: The hook β what happened and why readers should care. |
| - Paragraph 2: The context β how we got here, background that makes the story make sense. |
| - Paragraph 3: The details β specific facts, numbers, quotes, or examples from the original. |
| - Paragraph 4: The implications β what this means, reactions from others, what's next. |
| - Paragraph 5 (optional): Related developments, broader context, or additional perspectives. |
| |
| WRITING STYLE RULES: |
| - Professional but accessible β no academic jargon, no slang. |
| - Use clear, everyday words. A curious teenager should understand every sentence. |
| - Vary your sentence structure. Don't start every sentence with "The" or "It." |
| - Use contractions naturally where they fit ("it's," "don't," "they're"). |
| - Write with authority but humility β present facts, don't editorialize. |
| - Banned words: "delve," "tapestry," "testament," "underscore," "in today's world," "in the realm of," "pivotal," "captivating," "compelling," "fascinating." |
| - Never start with "In a world," "Picture this," or "Imagine." |
| - This is a genuine rewrite, not a paraphrase. Restructure sentences, find the real story, tell it fresh. |
| - Do NOT include links or URLs in paragraphs. |
| - Do NOT reference the original source or say "according to the article." |
| - Write in plain, professional English. |
| """).strip() |
|
|
| try: |
| resp, retries = gemini_generate(prompt, temperature=0.6, json_mode=True) |
| raw_text = resp.text.strip() |
|
|
| if retries > 0: |
| print(f" [RETRY] Rewrite succeeded after {retries} retry(ies)") |
|
|
| |
| raw_text = re.sub(r'^```(?:json)?\s*', '', raw_text) |
| raw_text = re.sub(r'\s*```$', '', raw_text) |
|
|
| content = None |
|
|
| |
| try: |
| parsed = json.loads(raw_text) |
| content = ArticleContent(**parsed).model_dump() |
| except (json.JSONDecodeError, ValidationError) as e: |
| print(f" [DEBUG] Full JSON parse failed: {e}") |
|
|
| |
| if not content: |
| json_match = re.search(r'\{[^{}]*"title"[^{}]*\}', raw_text, re.DOTALL) |
| if json_match: |
| try: |
| parsed = json.loads(json_match.group()) |
| content = ArticleContent(**parsed).model_dump() |
| except (json.JSONDecodeError, ValidationError) as e: |
| print(f" [DEBUG] Regex JSON extract failed: {e}") |
|
|
| |
| if not content: |
| content = _fallback_content(raw_text, original_title) |
|
|
| except Exception as e: |
| print(f" [ERROR] Article rewrite failed: {e}") |
| return { |
| "title": f"Rewritten: {original_title}", |
| "paragraphs": [original_body[:1000]], |
| "seo_description": original_title, |
| "image_alt_texts": [original_title], |
| "tags": [] |
| } |
|
|
| |
| if "title" in content: |
| content["title"] = _validate_title(content["title"], original_title) |
| else: |
| content["title"] = original_title[:70] |
|
|
| |
| if "paragraphs" not in content or not content["paragraphs"]: |
| content["paragraphs"] = [raw_text[:500]] |
| if "tags" not in content: |
| content["tags"] = [] |
| if "seo_description" not in content: |
| content["seo_description"] = content["paragraphs"][0][:160] |
| if "image_alt_texts" not in content: |
| content["image_alt_texts"] = [content.get("title", original_title)] |
|
|
| |
| cache = _load_article_cache() |
| cache[cache_key] = content |
| _save_article_cache(cache) |
| print(f" [CACHE] Saved article rewrite to cache") |
|
|
| print(f" β
Rewritten: \"{content['title']}\"") |
| print(f" π {len(content['paragraphs'])} paragraphs generated") |
| return content |
|
|
|
|
| def _fallback_content(raw_text: str, original_title: str) -> dict: |
| """Create structured content from unstructured text.""" |
| print("[WARN] Could not parse JSON, creating fallback content") |
| lines = [l.strip() for l in raw_text.split("\n") if l.strip() and l.strip() not in ('{', '}', '[', ']')] |
| |
| title = original_title |
| if lines: |
| title_match = re.search(r'"title"\s*:\s*"([^"]+)"', lines[0], re.IGNORECASE) |
| if title_match: |
| title = title_match.group(1)[:80] |
| else: |
| title = lines[0].strip('",\\{}[*: ')[:80] |
|
|
| paragraphs = [l for l in lines if len(l) > 100] |
| if not paragraphs: |
| paragraphs = [raw_text[:500]] |
|
|
| return { |
| "title": title, |
| "paragraphs": paragraphs, |
| "seo_description": paragraphs[0][:160], |
| "image_alt_texts": [original_title], |
| "tags": [], |
| } |
|
|
|
|
| |
| |
| |
|
|
| def find_youtube_videos(queries: list[str], max_videos: int = 2) -> list[dict]: |
| """ |
| Find related YouTube videos by scraping YouTube search results. |
| Returns: [{"url": str, "title": str, "embed_url": str}] |
| """ |
| print("π¬ Searching for related YouTube videos...") |
| videos = [] |
|
|
| for query in queries[:3]: |
| try: |
| search_url = f"https://www.youtube.com/results?search_query={quote_plus(query)}" |
| resp = requests.get(search_url, headers={ |
| "User-Agent": HEADERS["User-Agent"], |
| "Accept-Language": "en-US,en;q=0.9", |
| }, timeout=15) |
|
|
| if resp.status_code != 200: |
| continue |
|
|
| |
| video_ids = re.findall(r'"videoId":"([a-zA-Z0-9_-]{11})"', resp.text) |
| |
| seen = set() |
| unique_ids = [] |
| for vid in video_ids: |
| if vid not in seen: |
| seen.add(vid) |
| unique_ids.append(vid) |
|
|
| |
| for vid_id in unique_ids[:2]: |
| |
| title_match = re.search( |
| rf'"videoId":"{vid_id}".*?"title":\s*\{{"runs":\s*\[\{{"text":\s*"([^"]+)"', |
| resp.text |
| ) |
| vid_title = title_match.group(1) if title_match else f"Related: {query}" |
| |
| video_entry = { |
| "url": f"https://www.youtube.com/watch?v={vid_id}", |
| "title": vid_title, |
| "embed_url": f"https://www.youtube.com/embed/{vid_id}", |
| "video_id": vid_id, |
| } |
| # Avoid duplicates |
| if not any(v["video_id"] == vid_id for v in videos): |
| videos.append(video_entry) |
| print(f" πΉ Found: {vid_title[:60]}...") |
| |
| if len(videos) >= max_videos: |
| break |
| |
| except Exception as e: |
| print(f" [WARN] YouTube search failed for '{query}': {e}") |
| continue |
| |
| if len(videos) >= max_videos: |
| break |
| |
| if not videos: |
| print(" β οΈ No YouTube videos found") |
| else: |
| print(f" β
Found {len(videos)} video(s)") |
| |
| return videos |
| |
| |
| # ββββββββββββββββββββββββββββββββββββββββββββββ |
| # 4. Image Handling |
| # ββββββββββββββββββββββββββββββββββββββββββββββ |
| |
| def filter_irrelevant_images(image_urls: list[str], article_title: str, article_body: str) -> list[str]: |
| """ |
| Filter article images using Gemma 4 AI vision as the primary judge. |
| |
| Process: |
| 1. Minimal pre-filter: drop obvious junk (.gif spacers, tracking pixels, tiny thumbnails) |
| 2. Download each image |
| 3. Send ALL images to Gemma 4 vision with article context |
| 4. AI decides relevance β keeps content images, drops ads/logos/stock/decorations |
| 5. Returns top 3 AI-approved images sorted by quality |
| |
| No URL-pattern blocking β AI is the judge. |
| """ |
| if not image_urls: |
| return [] |
| |
| print(f" AI-filtering {len(image_urls)} image(s) for relevance...") |
| |
| # --- Level 1: Minimal junk filter (only obvious non-images) --- |
| junk_patterns = [ |
| r'\.gif$', # GIFs are usually spacers/animated junk |
| r'spacer', r'pixel', # Tiny tracking/spacer images |
| r'blank', # Blank placeholder images |
| r'tracking', # Analytics/tracking pixels |
| r'beacon', # Tracking beacons |
| r'1x1', # 1x1 tracking pixels |
| r'/pixel\.gif', # Classic tracking pixel |
| r'/spacer\.', # Spacer images |
| ] |
| |
| pre_filtered = [] |
| for url in image_urls: |
| url_lower = url.lower() |
| if any(re.search(pattern, url_lower) for pattern in junk_patterns): |
| print(f" π« Junk filter: {url.split('/')[-1][:40]}...") |
| continue |
| pre_filtered.append(url) |
| |
| if not pre_filtered: |
| return [] |
| |
| print(f" πΈ {len(pre_filtered)} images to evaluate with AI...") |
| |
| # --- Level 2: AI Vision Relevance Check (every image gets analyzed) --- |
| scored = [] |
| for i, url in enumerate(pre_filtered): |
| try: |
| # Rate-limit friendly delays |
| if i > 0: |
| time.sleep(0.5) |
| |
| resp = requests.get(url, headers=HEADERS, timeout=15) |
| |
| # Handle rate limiting |
| if resp.status_code == 429: |
| print(f" β³ Rate limited, waiting 2s: {url.split('/')[-1][:40]}...") |
| time.sleep(2) |
| resp = requests.get(url, headers=HEADERS, timeout=15) |
| |
| if resp.status_code != 200: |
| print(f" π« HTTP {resp.status_code}: {url.split('/')[-1][:40]}...") |
| continue |
| |
| # Convert to PIL Image |
| img = Image.open(BytesIO(resp.content)).convert("RGB") |
| |
| # Check minimum size (skip tiny images) |
| if img.width < 100 or img.height < 100: |
| print(f" π« Too small: {url.split('/')[-1][:40]} ({img.width}x{img.height})") |
| continue |
| |
| # Check for SVG/XML that somehow slipped through |
| ct = resp.headers.get("Content-Type", "") |
| if "svg" in ct.lower() or "xml" in ct.lower(): |
| print(f" π« SVG/XML not supported: {url.split('/')[-1][:40]}...") |
| continue |
| |
| prompt = textwrap.dedent(f""" |
| You are evaluating whether an image is relevant and useful for a blog article. |
| |
| ARTICLE TITLE: {article_title} |
| |
| ARTICLE BODY (first 1000 chars): {article_body[:1000]} |
| |
| Look at the image and decide: |
| 1. Is this image relevant to the article topic? (not an ad, logo, placeholder, or random decoration) |
| 2. Does it add value to the article (illustrates a key point, shows a person/place mentioned, etc.)? |
| 3. Is the image quality good enough (not blurry, not a tiny thumbnail)? |
| |
| Return ONLY valid JSON: |
| {{"relevant": true/false, "reason": "brief explanation", "quality_score": 0.0-1.0}} |
| """).strip() |
| |
| image_part = image_to_part(img) |
| resp_ai, retries = gemini_generate(prompt, images=[image_part], temperature=0.2, json_mode=True) |
| result = json.loads(resp_ai.text.strip()) |
| |
| is_relevant = result.get("relevant", False) |
| quality = result.get("quality_score", 0.0) |
| reason = result.get("reason", "") |
| |
| if is_relevant and quality >= 0.3: |
| scored.append({"url": url, "quality": quality, "reason": reason}) |
| print(f" β
KEPT: {url.split('/')[-1][:40]}... (quality={quality}, {reason[:60]})") |
| else: |
| print(f" π« REJECTED: {url.split('/')[-1][:40]}... ({reason[:60]})") |
| |
| except Exception as e: |
| print(f" π« Error evaluating image: {url.split('/')[-1][:40]}... ({e})") |
| continue |
| |
| if not scored: |
| print(f" β οΈ AI found no relevant images β falling back to first 2") |
| return pre_filtered[:2] |
| |
| # Sort by quality, keep top 3 |
| scored.sort(key=lambda x: x["quality"], reverse=True) |
| top = scored[:3] |
| print(f" π€ Kept {len(top)} AI-verified images (best quality: {top[0]['quality']})") |
| return [s["url"] for s in top] |
| |
| |
| |
| |
| def _generate_search_queries(article_title: str, article_body: str) -> list[str]: |
| """Use Gemma 4 to generate 3 smart, specific search queries for finding relevant images.""" |
| prompt = textwrap.dedent(f""" |
| Generate exactly 3 short search queries to find relevant photos for this article. |
| Each query should be 2-5 words, specific to the topic. |
| |
| ARTICLE TITLE: {article_title} |
| ARTICLE BODY (first 500 chars): {article_body[:500]} |
| |
| Return ONLY a JSON array of 3 strings: ["query 1", "query 2", "query 3"] |
| Do NOT include generic terms like "photo" or "image" β just the topic keywords. |
| """).strip() |
| try: |
| resp, _ = gemini_generate(prompt, temperature=0.3, json_mode=True) |
| queries = json.loads(resp.text.strip()) |
| if isinstance(queries, list) and len(queries) >= 1: |
| return [q.strip() for q in queries[:3] if q.strip()] |
| except Exception as e: |
| print(f" [WARN] AI query generation failed: {e}") |
| # Fallback |
| return [article_title, article_title.split(":")[0].strip() if ":" in article_title else article_title] |
| |
| |
| def _validate_image_with_ai(img: Image.Image, article_title: str, article_body: str) -> dict: |
| """Use Gemma 4 vision to validate whether an image is genuinely relevant to the article.""" |
| prompt = textwrap.dedent(f""" |
| Look at this image and the article below. Is this image genuinely relevant and useful for the article? |
| |
| ARTICLE: {article_title} |
| BODY (first 300 chars): {article_body[:300]} |
| |
| Reject if: stock photo, ad/logo, unrelated topic, too blurry, or generic placeholder. |
| Accept if: shows people/places/events mentioned, illustrates the topic directly, or is a real editorial photo. |
| |
| Return ONLY JSON: {{"relevant": true/false, "confidence": 0.0-1.0, "reason": "brief explanation"}} |
| """).strip() |
| try: |
| image_part = image_to_part(img) |
| resp, _ = gemini_generate(prompt, images=[image_part], temperature=0.2, json_mode=True) |
| return json.loads(resp.text.strip()) |
| except Exception as e: |
| return {"relevant": True, "confidence": 0.5, "reason": f"AI validation failed ({e}), keeping image"} |
| |
| |
| def _search_ddg_images(query: str, max_results: int = 20) -> list[str]: |
| """ |
| Search DuckDuckGo for images by parsing regular search results. |
| DuckDuckGo's /i.js API is blocked (403), so we scrape image URLs from |
| the regular web search results page which includes embedded image thumbnails |
| and links to full-size images on source pages. |
| """ |
| import urllib.parse |
| |
| # Use a session to maintain cookies |
| session = requests.Session() |
| headers = { |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36", |
| "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8", |
| "Accept-Language": "en-US,en;q=0.9", |
| "Accept-Encoding": "gzip, deflate", |
| "Connection": "keep-alive", |
| "Upgrade-Insecure-Requests": "1", |
| "Sec-Fetch-Dest": "document", |
| "Sec-Fetch-Mode": "navigate", |
| "Sec-Fetch-Site": "none", |
| "Sec-Fetch-User": "?1", |
| "Cache-Control": "max-age=0", |
| } |
| session.headers.update(headers) |
| |
| # First, visit DDG homepage to get cookies |
| try: |
| session.get("https://duckduckgo.com/", timeout=10) |
| except Exception: |
| pass |
| |
| # Search with images keyword |
| search_url = f"https://html.duckduckgo.com/html/?q={urllib.parse.quote_plus(query + ' photo image')}" |
| try: |
| resp = session.get(search_url, timeout=15) |
| resp.raise_for_status() |
| soup = BeautifulSoup(resp.text, "html.parser") |
| except Exception as e: |
| print(f" β οΈ DDG web search failed: {e}") |
| return [] |
| |
| img_urls = [] |
| |
| # Extract image URLs from DDG result snippets |
| # DDG HTML includes <img> tags with src pointing to external images |
| for img in soup.find_all("img"): |
| src = img.get("src") or img.get("data-src") or "" |
| # DDG proxies images through their own CDN - extract the actual URL |
| if "duckduckgo.com" in src.lower(): |
| # DDG image proxy URL format: https://tseX.mm.bing.net/th?id=... |
| # or direct external URLs |
| if src.startswith("http") and "tse" in src.lower(): |
| # These are Bing-style thumbnails, try to get the original |
| pass |
| elif src.startswith("http"): |
| img_urls.append(src) |
| elif src.startswith("http") and not src.startswith("data:"): |
| img_urls.append(src) |
| |
| # Also extract URLs from result links that point to image files |
| for a in soup.find_all("a", class_="result__snippet"): |
| href = a.get("href", "") |
| # Decode DDG's redirect URLs |
| if "uddg=" in href: |
| from urllib.parse import unquote |
| actual_url = unquote(href.split("uddg=")[1].split("&")[0]) |
| if any(actual_url.lower().endswith(ext) for ext in [".jpg", ".jpeg", ".png", ".webp", ".gif"]): |
| if actual_url not in img_urls: |
| img_urls.append(actual_url) |
| |
| # Also check meta og:image tags in result snippets (DDG includes these) |
| for meta in soup.find_all("meta"): |
| content = meta.get("content", "") |
| if content.startswith("http") and any(ext in content.lower() for ext in [".jpg", ".jpeg", ".png", ".webp"]): |
| if content not in img_urls: |
| img_urls.append(content) |
| |
| return img_urls[:max_results] |
| |
| |
| def _search_wikimedia_images(query: str, max_results: int = 10) -> list[str]: |
| """ |
| Search Wikimedia Commons API for relevant images. |
| Returns full-size image URLs (not thumbnails). |
| """ |
| api_url = ( |
| "https://commons.wikimedia.org/w/api.php" |
| "?action=query" |
| "&format=json" |
| "&generator=search" |
| "&gsrnamespace=6" # File namespace |
| f"&gsrsearch={quote_plus(query)}" |
| "&prop=imageinfo" |
| "&iiprop=url|extmetadata|size" |
| f"&gsrlimit={max_results}" |
| ) |
| headers = { |
| "User-Agent": "ContentAgent/1.0 (contact: admin@afripulsetv.com)", |
| } |
| try: |
| resp = requests.get(api_url, headers=headers, timeout=10) |
| resp.raise_for_status() |
| data = resp.json() |
| pages = data.get("query", {}).get("pages", {}) |
| urls = [] |
| for page_id, page in pages.items(): |
| imageinfo = page.get("imageinfo", []) |
| if imageinfo: |
| info = imageinfo[0] |
| url = info.get("url", "") |
| width = info.get("width", 0) |
| height = info.get("height", 0) |
| # Skip tiny files (e.g. signatures, icons) |
| if url and url.startswith("http") and width >= 400 and height >= 300: |
| if url not in urls: |
| urls.append(url) |
| return urls |
| except Exception as e: |
| print(f" β οΈ Wikimedia search failed: {e}") |
| return [] |
| |
| |
| def _search_related_image(query: str, article_title: str = "", article_body: str = "") -> str: |
| """ |
| Search for relevant images using: |
| 1. Wikimedia Commons API (reliable, free, no key needed) |
| 2. DuckDuckGo web search as backup |
| 3. Gemma 4 AI validation for relevance |
|
|
| If article_title and article_body are provided: |
| - AI generates 3 smart search queries from article content |
| - Each image is validated by Gemma 4 vision for relevance |
| - Rejects stock photos, ads, logos, placeholders |
|
|
| Falls back to size-only validation if AI fails. |
| """ |
| headers = { |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36", |
| } |
| |
| # ββ Step 1: Generate search queries ββ |
| if article_title and article_body: |
| print(f" π§ AI generating search queries for: {article_title[:60]}...") |
| search_queries = _generate_search_queries(article_title, article_body) |
| print(f" π AI queries: {search_queries}") |
| else: |
| search_queries = [query] |
| |
| # ββ Step 2: Search Wikimedia Commons (reliable, always works) ββ |
| all_img_urls = [] |
| print(f" ποΈ Searching Wikimedia Commons...") |
| for sq in search_queries: |
| try: |
| urls = _search_wikimedia_images(sq, max_results=8) |
| for u in urls: |
| if u not in all_img_urls: |
| all_img_urls.append(u) |
| if urls: |
| print(f" πΈ Wikimedia query '{sq[:40]}...' β {len(urls)} images") |
| except Exception as e: |
| print(f" β οΈ Wikimedia search failed for '{sq}': {e}") |
| |
| # ββ Step 3: Try DDG as backup ββ |
| if len(all_img_urls) < 5: |
| print(f" π Trying DuckDuckGo for more results...") |
| for sq in search_queries: |
| try: |
| urls = _search_ddg_images(sq, max_results=10) |
| for u in urls: |
| if u not in all_img_urls: |
| all_img_urls.append(u) |
| except Exception: |
| pass |
| |
| if not all_img_urls: |
| print(" β οΈ No images found from any source") |
| return "" |
| |
| print(f" πΈ Total candidates: {len(all_img_urls)} image URLs") |
| |
| # ββ Step 3: Validate images ββ |
| use_ai_validation = bool(article_title and article_body) |
| rejected_count = 0 |
| error_count = 0 |
| size_count = 0 |
| |
| for i, img_url in enumerate(all_img_urls[:12]): |
| try: |
| # Add delay between requests to avoid rate limiting |
| if i > 0: |
| time.sleep(0.5) |
| |
| resp_img = requests.get(img_url, timeout=15, headers=headers, allow_redirects=True) |
| |
| # Handle rate limiting with retry |
| if resp_img.status_code == 429: |
| print(f" β³ [{i+1}/12] Rate limited (429), waiting 2s and retrying: {img_url[:60]}...") |
| time.sleep(2) |
| resp_img = requests.get(img_url, timeout=15, headers=headers, allow_redirects=True) |
| |
| if resp_img.status_code != 200: |
| print(f" π« [{i+1}/12] HTTP {resp_img.status_code}: {img_url[:70]}...") |
| error_count += 1 |
| continue |
| |
| # Check content type |
| ct = resp_img.headers.get("Content-Type", "") |
| if "svg" in ct.lower() or "xml" in ct.lower(): |
| print(f" π« [{i+1}/12] SVG/XML not supported: {img_url[:70]}...") |
| error_count += 1 |
| continue |
| |
| img_data = BytesIO(resp_img.content) |
| with Image.open(img_data) as img: |
| w, h = img.size |
| if w < 400 or h < 300: |
| print(f" π« [{i+1}/12] Too small ({w}x{h}): {img_url[:70]}...") |
| size_count += 1 |
| continue |
| |
| if use_ai_validation: |
| result = _validate_image_with_ai(img, article_title, article_body) |
| is_relevant = result.get("relevant", False) |
| confidence = result.get("confidence", 0.0) |
| reason = result.get("reason", "")[:60] |
| print(f" π€ [{i+1}/12] {'β
' if is_relevant else 'π«'} ({w}x{h}) conf={confidence:.1f} β {reason}") |
| if is_relevant and confidence >= 0.5: |
| print(f" β
AI-validated related image: {img_url[:80]}...") |
| return img_url |
| else: |
| rejected_count += 1 |
| else: |
| print(f" β
Found related image: {img_url[:80]}... ({w}x{h})") |
| return img_url |
| except Exception as e: |
| print(f" π« [{i+1}/12] Error: {e}") |
| error_count += 1 |
| continue |
| |
| print(f" π Summary: {error_count} errors, {size_count} too small, {rejected_count} AI-rejected out of {min(12, len(all_img_urls))} checked") |
| |
| # ββ Step 4: Fallback β try remaining URLs without AI ββ |
| if use_ai_validation: |
| print(f" β οΈ AI rejected all β trying remaining {len(all_img_urls) - 12} images by size only...") |
| for i, img_url in enumerate(all_img_urls[12:20]): |
| try: |
| if i > 0: |
| time.sleep(0.5) |
| resp_img = requests.get(img_url, timeout=15, headers=headers, allow_redirects=True) |
| if resp_img.status_code == 429: |
| time.sleep(2) |
| resp_img = requests.get(img_url, timeout=15, headers=headers, allow_redirects=True) |
| if resp_img.status_code != 200: |
| continue |
| ct = resp_img.headers.get("Content-Type", "") |
| if "svg" in ct.lower() or "xml" in ct.lower(): |
| continue |
| img_data = BytesIO(resp_img.content) |
| with Image.open(img_data) as img: |
| w, h = img.size |
| if w >= 600 and h >= 400: |
| print(f" β
Fallback related image (size only): {img_url[:80]}... ({w}x{h})") |
| return img_url |
| except Exception: |
| continue |
| |
| print(" β οΈ No suitable related image found") |
| return "" |
| |
| |
| def download_article_images(image_urls: list[str], max_images: int = 5, referer_url: str = None) -> list[Image.Image]: |
| """ |
| Download images from URLs and return as PIL Image objects. |
| Filters out broken images and tiny ones. |
| """ |
| print(f"πΌ Downloading article images ({len(image_urls)} found)...") |
| images = [] |
| |
| # Build image download headers with Referer to bypass hotlink protection |
| img_headers = dict(HEADERS) |
| img_headers["Accept"] = "image/avif,image/webp,image/apng,image/*,*/*;q=0.8" |
| if referer_url: |
| img_headers["Referer"] = referer_url |
| |
| async def download_image(session, img_url): |
| try: |
| async with session.get(img_url, headers=img_headers, timeout=aiohttp.ClientTimeout(total=15)) as resp: |
| resp.raise_for_status() |
| content_type = resp.headers.get("Content-Type", "") |
| if not content_type.startswith("image/"): |
| return None |
| img_data = BytesIO(await resp.read()) |
| img = Image.open(img_data).convert("RGB") |
| w, h = img.size |
| if w < 300 or h < 200: |
| print(f" π« Dropping tiny image: {w}x{h}") |
| return None |
| print(f" β
Downloaded image: {w}x{h}") |
| return img |
| except Exception as e: |
| print(f" [WARN] Failed to download image: {e}") |
| return None |
| |
| async def download_all(): |
| async with aiohttp.ClientSession() as session: |
| tasks = [download_image(session, url) for url in image_urls[:max_images + 3]] |
| results = await asyncio.gather(*tasks) |
| return [img for img in results if img is not None] |
| |
| images = asyncio.run(download_all()) |
| |
| if not images: |
| print(" β οΈ No usable images downloaded, will use placeholders") |
| # Create a simple placeholder |
| placeholder = Image.new("RGB", (1280, 720), color=(30, 30, 60)) |
| draw = ImageDraw.Draw(placeholder) |
| font = _get_font(48) |
| draw.text( |
| (640, 360), "Article Image", |
| fill=(200, 200, 200), font=font, anchor="mm" |
| ) |
| images.append(placeholder) |
| |
| print(f" πΈ {len(images)} usable image(s) ready") |
| return images |
| |
| |
| def create_article_featured_image( |
| images: list[Image.Image], output_name: str |
| ) -> Path: |
| """ |
| Create a 3:1 editorial collage featured image. |
| Always uses the 3-panel layout β repeats images if fewer than 3 available. |
| """ |
| print("πΌ Creating featured image for article...") |
| _ensure_dirs() |
| |
| WIDTH, HEIGHT = 1280, 720 |
| |
| # Pad to 3 images by repeating the first one so we always get the 3:1 layout |
| padded = list(images) |
| while len(padded) < 3: |
| padded.append(images[0]) |
| |
| canvas = _collage_3_to_1(padded, WIDTH, HEIGHT) |
| |
| # Ensure output directory exists (defensive) |
| OUTPUT_DIR.mkdir(parents=True, exist_ok=True) |
| |
| output_path = OUTPUT_DIR / f"featured_article_{output_name}.jpg" |
| canvas.save(str(output_path), "JPEG", quality=92) |
| print(f" β
Featured image saved: {output_path}") |
| return output_path |
| |
| |
| def _collage_3_to_1(images: list[Image.Image], width: int, height: int) -> Image.Image: |
| """3:1 editorial layout β hero 75% left, supporting 25% right stacked.""" |
| canvas = Image.new("RGB", (width, height), (15, 15, 15)) |
| hero_w = int(width * 0.75) # 960px |
| hero_h = height # 720px |
| side_w = width - hero_w # 320px |
| side_h = height // 2 # 360px each |
| |
| # Hero image (largest, left panel) |
| hero = _crop_to_fill(images[0], hero_w, hero_h) |
| canvas.paste(hero, (0, 0)) |
| |
| # Supporting images stacked on right |
| top = _crop_to_fill(images[1], side_w, side_h) |
| canvas.paste(top, (hero_w, 0)) |
| |
| bottom = _crop_to_fill(images[2], side_w, side_h) |
| canvas.paste(bottom, (hero_w, side_h)) |
| |
| # Thin divider lines |
| draw = ImageDraw.Draw(canvas) |
| # Vertical divider |
| draw.line([(hero_w, 0), (hero_w, height)], fill=(40, 40, 40), width=2) |
| # Horizontal divider on right panel |
| draw.line([(hero_w, side_h), (width, side_h)], fill=(40, 40, 40), width=2) |
| |
| return canvas |
| |
| |
| def _crop_to_fill(img: Image.Image, target_w: int, target_h: int) -> Image.Image: |
| """Center-crop an image to the target aspect ratio, then resize.""" |
| img = img.convert("RGB") |
| src_w, src_h = img.size |
| target_ratio = target_w / target_h |
| src_ratio = src_w / src_h |
| |
| if src_ratio > target_ratio: |
| # Image too wide β crop sides |
| new_w = int(src_h * target_ratio) |
| left = (src_w - new_w) // 2 |
| img = img.crop((left, 0, left + new_w, src_h)) |
| else: |
| # Image too tall β crop top/bottom |
| new_h = int(src_w / target_ratio) |
| top = (src_h - new_h) // 2 |
| img = img.crop((0, top, src_w, top + new_h)) |
| |
| return img.resize((target_w, target_h), Image.LANCZOS) |
| |
| |
| # ββββββββββββββββββββββββββββββββββββββββββββββ |
| # 5. Gutenberg Content Builder |
| # ββββββββββββββββββββββββββββββββββββββββββββββ |
| |
| def build_article_gutenberg_content( |
| paragraphs: list[str], |
| article_embeds: list[dict], |
| wp_image_ids: list[dict], |
| source_url: str, |
| ) -> str: |
| """ |
| Build WordPress Gutenberg block content with: |
| - Paragraphs |
| - YouTube video embeds |
| - Image blocks |
| - Source link button |
| - Ad shortcodes |
| """ |
| blocks = [] |
| |
| # Simple strategy: Insert images and embeds evenly |
| image_count = len(wp_image_ids) |
| video_count = len(article_embeds) |
| para_count = len(paragraphs) |
| |
| # Calculate insertion intervals |
| video_insert_points = [] |
| if video_count > 0: |
| interval = max(1, para_count // (video_count + 1)) |
| video_insert_points = [interval * (i + 1) for i in range(video_count)] |
| |
| image_insert_points = [] |
| if image_count > 0: |
| interval = max(1, para_count // (image_count + 1)) |
| image_insert_points = [interval * (i + 1) for i in range(image_count)] |
| |
| video_idx = 0 |
| image_idx = 0 |
| |
| for i, para in enumerate(paragraphs): |
| blocks.append(f'<!-- wp:paragraph -->\n<p>{para}</p>\n<!-- /wp:paragraph -->') |
| |
| # Insert video/media embed if this is the right spot |
| if video_idx < len(article_embeds) and (i + 1) in video_insert_points: |
| vid = article_embeds[video_idx] |
| vid_url = vid["url"] |
| provider = vid.get("provider", "youtube") |
| |
| # Choose type: Most social/media embeds use 'rich' |
| v_type = "video" if provider in ["youtube", "vimeo", "dailymotion", "streamable", "video"] else "rich" |
| |
| # Choose aspect ratio |
| aspect_class = "wp-embed-aspect-16-9" |
| if provider in ["tiktok", "instagram"]: |
| aspect_class = "wp-embed-aspect-9-16" |
| elif provider in ["spotify", "soundcloud", "threads", "reddit", "pinterest", "bluesky"]: |
| aspect_class = "" |
| |
| # SPECIAL CASE: Instagram, Facebook, X need Custom HTML blocks |
| if provider in ["instagram", "facebook"]: |
| if provider == "instagram": |
| # Force clean URL if not already |
| clean_url = vid_url.split("?")[0].rstrip("/") |
| embed_html = ( |
| f'<blockquote class="instagram-media" data-instgrm-captioned data-instgrm-permalink="{clean_url}/?utm_source=ig_embed&utm_campaign=loading" data-instgrm-version="14" style=" background:#FFF; border:0; border-radius:3px; box-shadow:0 0 1px 0 rgba(0,0,0,0.5),0 1px 10px 0 rgba(0,0,0,0.15); margin: 1px; max-width:540px; min-width:326px; padding:0; width:99.375%; width:-webkit-calc(100% - 2px); width:calc(100% - 2px);">' |
| f'<div style="padding:16px;"> <a href="{clean_url}/?utm_source=ig_embed&utm_campaign=loading" style=" background:#FFFFFF; line-height:0; padding:0 0; text-align:center; text-decoration:none; width:100%;" target="_blank">View this post on Instagram</a></div>' |
| f'</blockquote>\n<script async src="//www.instagram.com/embed.js"></script>' |
| ) |
| else: # Facebook |
| clean_url = vid_url.split("?")[0] |
| embed_html = ( |
| f'<div class="fb-post" data-href="{clean_url}" data-width="500" data-show-text="true">' |
| f'<blockquote cite="{clean_url}" class="fb-xfbml-parse-ignore">View post on Facebook</blockquote>' |
| f'</div>\n<script async defer src="https://connect.facebook.net/en_US/sdk.js#xfbml=1&version=v12.0"></script>' |
| ) |
| |
| blocks.append( |
| f'<!-- wp:html -->\n' |
| f'<div class="wp-block-html-embed {provider}-embed">\n{embed_html}\n</div>\n' |
| f'<!-- /wp:html -->' |
| ) |
| else: |
| # X/Twitter: use a simple paragraph with the URL on its own line |
| # WordPress auto-embeds standalone URLs when surrounded by whitespace |
| if provider == "x": |
| clean_url = vid_url.split("?")[0].rstrip("/") |
| blocks.append( |
| f'<!-- wp:paragraph -->\n' |
| f'<p>{clean_url}</p>\n' |
| f'<!-- /wp:paragraph -->' |
| ) |
| else: |
| # Use standard Gutenberg embed block for everything else |
| embed_json = { |
| "url": vid_url, |
| "type": v_type, |
| "providerNameSlug": provider, |
| "responsive": True |
| } |
| if aspect_class: |
| embed_json["className"] = f"{aspect_class} wp-has-aspect-ratio" |
| |
| blocks.append( |
| f'<!-- wp:embed {json.dumps(embed_json)} -->\n' |
| f'<figure class="wp-block-embed is-type-{v_type} is-provider-{provider} wp-block-embed-{provider} {"" if not aspect_class else aspect_class + " wp-has-aspect-ratio"}">' |
| f'<div class="wp-block-embed__wrapper">\n{vid_url}\n</div></figure>\n' |
| f'<!-- /wp:embed -->' |
| ) |
| video_idx += 1 |
| |
| # Insert image block if this is the right spot |
| if image_idx < len(wp_image_ids) and (i + 1) in image_insert_points: |
| img = wp_image_ids[image_idx] |
| img_id = img["id"] |
| img_url = img["url"] |
| alt_text = img.get("alt", "Related image") |
| blocks.append( |
| f'<!-- wp:image {{"id":{img_id},"sizeSlug":"large","linkDestination":"none"}} -->\n' |
| f'<figure class="wp-block-image size-large">' |
| f'<img src="{img_url}" alt="{alt_text}" class="wp-image-{img_id}"/>' |
| f'</figure>\n' |
| f'<!-- /wp:image -->' |
| ) |
| image_idx += 1 |
| |
| # Add any remaining embeds |
| while video_idx < len(article_embeds): |
| vid = article_embeds[video_idx] |
| vid_url = vid["url"] |
| provider = vid.get("provider", "youtube") |
| |
| v_type = "video" if provider in ["youtube", "vimeo", "dailymotion", "streamable", "video"] else "rich" |
| |
| aspect_class = "wp-embed-aspect-16-9" |
| if provider in ["tiktok", "instagram"]: |
| aspect_class = "wp-embed-aspect-9-16" |
| elif provider in ["spotify", "soundcloud", "x", "threads", "reddit", "pinterest", "bluesky"]: |
| aspect_class = "" |
| |
| # SPECIAL CASE: Instagram and Facebook |
| if provider in ["instagram", "facebook"]: |
| if provider == "instagram": |
| clean_url = vid_url.split("?")[0].rstrip("/") |
| embed_html = ( |
| f'<blockquote class="instagram-media" data-instgrm-captioned data-instgrm-permalink="{clean_url}/?utm_source=ig_embed&utm_campaign=loading" data-instgrm-version="14" style=" background:#FFF; border:0; border-radius:3px; box-shadow:0 0 1px 0 rgba(0,0,0,0.5),0 1px 10px 0 rgba(0,0,0,0.15); margin: 1px; max-width:540px; min-width:326px; padding:0; width:99.375%; width:-webkit-calc(100% - 2px); width:calc(100% - 2px);">' |
| f'<div style="padding:16px;"> <a href="{clean_url}/?utm_source=ig_embed&utm_campaign=loading" style=" background:#FFFFFF; line-height:0; padding:0 0; text-align:center; text-decoration:none; width:100%;" target="_blank">View this post on Instagram</a></div>' |
| f'</blockquote>\n<script async src="//www.instagram.com/embed.js"></script>' |
| ) |
| else: |
| clean_url = vid_url.split("?")[0] |
| embed_html = ( |
| f'<div class="fb-post" data-href="{clean_url}" data-width="500" data-show-text="true">' |
| f'<blockquote cite="{clean_url}" class="fb-xfbml-parse-ignore">View post on Facebook</blockquote>' |
| f'</div>\n<script async defer src="https://connect.facebook.net/en_US/sdk.js#xfbml=1&version=v12.0"></script>' |
| ) |
| |
| blocks.append( |
| f'<!-- wp:html -->\n' |
| f'<div class="wp-block-html-embed {provider}-embed">\n{embed_html}\n</div>\n' |
| f'<!-- /wp:html -->' |
| ) |
| else: |
| # X/Twitter: use a simple paragraph with the URL |
| if provider == "x": |
| clean_url = vid_url.split("?")[0].rstrip("/") |
| blocks.append( |
| f'<!-- wp:paragraph -->\n' |
| f'<p>{clean_url}</p>\n' |
| f'<!-- /wp:paragraph -->' |
| ) |
| else: |
| embed_json = { |
| "url": vid_url, |
| "type": v_type, |
| "providerNameSlug": provider, |
| "responsive": True |
| } |
| if aspect_class: |
| embed_json["className"] = f"{aspect_class} wp-has-aspect-ratio" |
| |
| blocks.append( |
| f'<!-- wp:embed {json.dumps(embed_json)} -->\n' |
| f'<figure class="wp-block-embed is-type-{v_type} is-provider-{provider} wp-block-embed-{provider} {"" if not aspect_class else aspect_class + " wp-has-aspect-ratio"}">' |
| f'<div class="wp-block-embed__wrapper">\n{vid_url}\n</div></figure>\n' |
| f'<!-- /wp:embed -->' |
| ) |
| video_idx += 1 |
| |
| # Add any remaining images |
| while image_idx < len(wp_image_ids): |
| img = wp_image_ids[image_idx] |
| img_id = img["id"] |
| img_url = img["url"] |
| alt_text = img.get("alt", "Related image") |
| blocks.append( |
| f'<!-- wp:image {{"id":{img_id},"sizeSlug":"large","linkDestination":"none"}} -->\n' |
| f'<figure class="wp-block-image size-large">' |
| f'<img src="{img_url}" alt="{alt_text}" class="wp-image-{img_id}"/>' |
| f'</figure>\n' |
| f'<!-- /wp:image -->' |
| ) |
| image_idx += 1 |
| |
| # Ad shortcodes |
| blocks.append( |
| '<!-- wp:shortcode -->\n[ads2] [ads1]\n<!-- /wp:shortcode -->' |
| ) |
| |
| return "\n\n".join(blocks) |
| |
| |
| # ββββββββββββββββββββββββββββββββββββββββββββββ |
| # 6. WordPress Publishing |
| # ββββββββββββββββββββββββββββββββββββββββββββββ |
| |
| def upload_media_to_wp(file_path: Path, filename: str = None) -> dict: |
| """ |
| Upload an image to WordPress. |
| Returns: {"id": int, "url": str} |
| """ |
| if filename is None: |
| filename = file_path.name |
| |
| print(f"βοΈ Uploading {filename} to WordPress...") |
| |
| ext = file_path.suffix.lower() |
| mime_map = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png"} |
| mime_type = mime_map.get(ext, "image/jpeg") |
| |
| with open(file_path, "rb") as f: |
| file_data = f.read() |
| |
| upload_method = os.getenv("WP_UPLOAD_METHOD", "auto").strip().lower() |
| if pull_relay_enabled(): |
| return _upload_media_to_wp_bridge(file_data, filename, mime_type) |
| |
| upload_errors = [] |
| if upload_method in ("auto", "rest"): |
| try: |
| return _upload_media_to_wp_rest(file_data, filename, mime_type) |
| except Exception as e: |
| if upload_method == "rest": |
| raise |
| upload_errors.append(f"REST error: {e}") |
| print(f" β οΈ REST media upload failed; trying authenticated upload bridge: {e}") |
| |
| if upload_method in ("auto", "bridge"): |
| try: |
| return _upload_media_to_wp_bridge(file_data, filename, mime_type) |
| except Exception as e: |
| if upload_method == "bridge": |
| raise |
| upload_errors.append(f"Bridge error: {e}") |
| print(f" β οΈ Upload bridge failed; trying XML-RPC fallback: {e}") |
| |
| if upload_method in ("auto", "xmlrpc"): |
| try: |
| return _upload_media_to_wp_xmlrpc(file_data, filename, mime_type) |
| except Exception as e: |
| upload_errors.append(f"XML-RPC error: {e}") |
| raise Exception("WordPress media upload failed. " + "; ".join(upload_errors)) |
| |
| raise ValueError("WP_UPLOAD_METHOD must be auto, rest, bridge, or xmlrpc") |
| |
| |
| def _upload_media_to_wp_rest(file_data: bytes, filename: str, mime_type: str) -> dict: |
| """Upload media through the WordPress REST API.""" |
| url = f"{WP_URL}/wp-json/wp/v2/media" |
| base_headers = { |
| "Authorization": _wp_auth_header(), |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36", |
| "Accept": "application/json, */*", |
| "Origin": WP_URL, |
| "Referer": f"{WP_URL}/wp-admin/upload.php", |
| } |
| |
| response = requests.post( |
| url, headers=base_headers, files={"file": (filename, file_data, mime_type)}, |
| timeout=(15, 60), |
| verify=_wp_ssl_verify(), |
| ) |
| try: |
| return _parse_wp_rest_media_response(response, "REST multipart") |
| except Exception as multipart_error: |
| print(f" β οΈ REST multipart upload failed; trying REST raw upload: {multipart_error}") |
| |
| raw_headers = base_headers | { |
| "Content-Disposition": f'attachment; filename="{filename}"', |
| "Content-Type": mime_type, |
| } |
| response = requests.post( |
| url, |
| headers=raw_headers, |
| data=file_data, |
| timeout=(15, 60), |
| verify=_wp_ssl_verify(), |
| ) |
| return _parse_wp_rest_media_response(response, "REST raw") |
| |
| |
| def _upload_media_to_wp_bridge(file_data: bytes, filename: str, mime_type: str) -> dict: |
| """Upload through an authenticated admin-ajax action when REST is blocked.""" |
| url = os.getenv( |
| "WP_BRIDGE_URL", |
| f"{WP_URL}/wp-admin/admin-ajax.php?action=content_agent_upload", |
| ) |
| fields = { |
| "content_agent_user": WP_USERNAME, |
| "content_agent_password": WP_APP_PASSWORD, |
| "file": (filename, file_data, mime_type), |
| } |
| status_code, content_type, response_text, origin = post_multipart_to_wordpress_origin( |
| url, fields, verify_ssl=_wp_ssl_verify() |
| ) |
| print(f" [DEBUG] Upload bridge origin: {origin}") |
| |
| if status_code != 200 or "json" not in content_type.lower(): |
| raise Exception(f"Upload bridge failed ({status_code}): {response_text[:300]}") |
| |
| payload = json.loads(response_text) |
| if not payload.get("success"): |
| raise Exception(f"Upload bridge rejected the file: {payload.get('data')}") |
| |
| data = payload.get("data") or {} |
| result = {"id": int(data.get("id") or 0), "url": data.get("url", "")} |
| if not result["id"] or not result["url"]: |
| raise Exception(f"Upload bridge returned an incomplete response: {data}") |
| |
| print(f" β
Uploaded via authenticated bridge: {result['url']} (ID: {result['id']})") |
| return result |
| |
| |
| def _parse_wp_rest_media_response(response, method_label: str) -> dict: |
| """Validate and normalize a WordPress REST media response.""" |
| # Check if we got a bot verification page instead of a response |
| content_type = response.headers.get("Content-Type", "") |
| if "text/html" in content_type or "<html" in response.text.lower() or "bot verification" in response.text.lower(): |
| raise Exception( |
| f"WordPress blocked the upload (bot verification/WAF). " |
| f"Check your security plugin settings and whitelist the REST API." |
| ) |
| |
| if response.status_code not in (200, 201): |
| raise Exception( |
| f"WordPress media upload failed ({response.status_code}): " |
| f"{response.text[:300]}" |
| ) |
| |
| try: |
| data = response.json() |
| except Exception as e: |
| print(f" [ERROR] Failed to parse WordPress JSON response: {e}") |
| print(f" [DEBUG] Response Status: {response.status_code}") |
| print(f" [DEBUG] Response Text: {response.text[:500]}") |
| raise Exception(f"Invalid JSON response from WordPress during media upload: {response.text[:200]}") |
| |
| result = { |
| "id": data.get("id"), |
| "url": data.get("source_url", data.get("guid", {}).get("rendered", "")) |
| } |
| if not result["id"]: |
| raise Exception(f"WordPress upload succeeded but no ID returned. Response: {response.text[:200]}") |
| |
| print(f" β
Uploaded via {method_label}: {result['url']} (ID: {result['id']})") |
| return result |
| |
| |
| def _upload_media_to_wp_xmlrpc(file_data: bytes, filename: str, mime_type: str) -> dict: |
| """Upload media through WordPress XML-RPC as a fallback for REST/WAF blocks.""" |
| import xmlrpc.client |
| |
| endpoint = os.getenv("WP_XMLRPC_URL", f"{WP_URL}/xmlrpc.php") |
| xmlrpc_user = os.getenv("WP_XMLRPC_USERNAME", WP_USERNAME) |
| xmlrpc_pass = os.getenv("WP_XMLRPC_PASSWORD", WP_APP_PASSWORD) |
| print(" βοΈ Retrying media upload via WordPress XML-RPC...") |
| payload = { |
| "name": filename, |
| "type": mime_type, |
| "bits": xmlrpc.client.Binary(file_data), |
| "overwrite": False, |
| } |
| request_body = xmlrpc.client.dumps((0, xmlrpc_user, xmlrpc_pass, payload), methodname="wp.uploadFile", allow_none=True) |
| response = requests.post( |
| endpoint, |
| data=request_body.encode("utf-8"), |
| headers={ |
| "Content-Type": "text/xml", |
| "User-Agent": ( |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " |
| "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36" |
| ), |
| }, |
| timeout=(15, 300), |
| verify=_wp_ssl_verify(), |
| ) |
| content_type = response.headers.get("Content-Type", "") |
| if response.status_code != 200 or "text/html" in content_type or "<html" in response.text.lower(): |
| raise Exception(f"XML-RPC upload failed ({response.status_code}): {response.text[:300]}") |
| |
| try: |
| params, _ = xmlrpc.client.loads(response.content) |
| except xmlrpc.client.Fault: |
| raise |
| except Exception as e: |
| raise Exception(f"Invalid XML-RPC response from WordPress: {e}; {response.text[:300]}") |
| |
| data = params[0] if params else {} |
| result = { |
| "id": int(data.get("id") or 0), |
| "url": data.get("url", ""), |
| } |
| if not result["id"] or not result["url"]: |
| raise Exception(f"XML-RPC upload returned an incomplete response: {data}") |
| |
| print(f" β
Uploaded via XML-RPC: {result['url']} (ID: {result['id']})") |
| return result |
| |
| |
| def upload_image_from_url_to_wp(image_url: str, filename: str, alt_text: str = "") -> dict: |
| """ |
| Download an image from URL and upload it to WordPress. |
| Returns: {"id": int, "url": str, "alt": str} |
| """ |
| print(f"βοΈ Uploading remote image to WordPress: {filename}") |
| |
| try: |
| resp = requests.get(image_url, headers=HEADERS, timeout=15) |
| resp.raise_for_status() |
| |
| content_type = resp.headers.get("Content-Type", "image/jpeg") |
| if "png" in content_type: |
| ext = ".png" |
| else: |
| ext = ".jpg" |
| |
| if not filename.endswith(ext): |
| filename = filename + ext |
| |
| upload_type = content_type if content_type.startswith("image/") else "image/jpeg" |
| if pull_relay_enabled(): |
| media = _upload_media_to_wp_bridge(resp.content, filename, upload_type) |
| data = {"id": media["id"], "source_url": media["url"]} |
| else: |
| upload_method = os.getenv("WP_UPLOAD_METHOD", "auto").strip().lower() |
| rest_error = None |
| uploaded_data = None |
| if upload_method in ("auto", "rest"): |
| try: |
| media = _upload_media_to_wp_rest(resp.content, filename, upload_type) |
| except Exception as e: |
| rest_error = e |
| if upload_method == "rest": |
| raise |
| print(f" β οΈ REST remote image upload failed; trying XML-RPC fallback: {e}") |
| else: |
| uploaded_data = {"id": media["id"], "source_url": media["url"]} |
| if upload_method in ("auto", "xmlrpc") and uploaded_data is None: |
| try: |
| media = _upload_media_to_wp_xmlrpc(resp.content, filename, upload_type) |
| uploaded_data = {"id": media["id"], "source_url": media["url"]} |
| except Exception as e: |
| if rest_error: |
| raise Exception(f"WordPress remote image upload failed. REST error: {rest_error}; XML-RPC error: {e}") |
| raise |
| if uploaded_data is None: |
| raise ValueError("WP_UPLOAD_METHOD must be auto, rest, or xmlrpc") |
| data = uploaded_data |
| |
| # Set alt text if provided |
| if alt_text: |
| try: |
| request_wordpress_json( |
| "POST", |
| f"{WP_URL}/wp-json/wp/v2/media/{data['id']}", |
| {"alt_text": alt_text}, |
| WP_USERNAME, |
| WP_APP_PASSWORD, |
| timeout=15, |
| ) |
| except Exception: |
| pass |
| |
| result = { |
| "id": data["id"], |
| "url": data.get("source_url", data.get("guid", {}).get("rendered", "")), |
| "alt": alt_text, |
| } |
| print(f" β
Uploaded: {result['url']} (ID: {result['id']})") |
| return result |
| |
| except Exception as e: |
| print(f" [WARN] Failed to upload image from URL: {e}") |
| return None |
| |
| |
| def publish_article_post( |
| title: str, |
| content_html: str, |
| featured_media_id: int, |
| seo_description: str = "", |
| tags: list[str] = None, |
| ) -> dict: |
| """ |
| Create and publish a WordPress post via REST API. |
| Returns: {"id": int, "url": str} |
| """ |
| print("π Publishing rewritten article to WordPress...") |
| |
| post_data = { |
| "title": title, |
| "content": content_html, |
| "status": WP_POST_STATUS, |
| "categories": [WP_CATEGORY_ID], |
| "featured_media": featured_media_id, |
| } |
| |
| # Add excerpt/SEO description |
| if seo_description: |
| post_data["excerpt"] = seo_description |
| |
| # Handle tags β create if they don't exist |
| if tags: |
| tag_ids = [] |
| for tag_name in tags[:10]: # Max 10 tags |
| try: |
| # Check if tag exists |
| resp = requests.get( |
| f"{WP_URL}/wp-json/wp/v2/tags", |
| params={"search": tag_name}, |
| auth=_wp_auth(), |
| timeout=10 |
| ) |
| existing = resp.json() |
| if existing: |
| tag_ids.append(existing[0]["id"]) |
| else: |
| # Create new tag |
| create_resp = request_wordpress_json( |
| "POST", |
| f"{WP_URL}/wp-json/wp/v2/tags", |
| {"name": tag_name}, |
| WP_USERNAME, |
| WP_APP_PASSWORD, |
| timeout=10, |
| ) |
| if create_resp.status_code in (200, 201): |
| tag_ids.append(create_resp.json()["id"]) |
| except Exception as e: |
| print(f" [WARN] Failed to handle tag '{tag_name}': {e}") |
| if tag_ids: |
| post_data["tags"] = tag_ids |
| |
| url = f"{WP_URL}/wp-json/wp/v2/posts" |
| response = request_wordpress_json( |
| "POST", url, post_data, WP_USERNAME, WP_APP_PASSWORD, timeout=30 |
| ) |
| |
| if response.status_code not in (200, 201): |
| raise Exception( |
| f"WordPress post creation failed ({response.status_code}): " |
| f"{response.text[:300]}" |
| ) |
| |
| try: |
| data = response.json() |
| except Exception as e: |
| print(f" [ERROR] Failed to parse WordPress JSON response: {e}") |
| print(f" [DEBUG] Response Status: {response.status_code}") |
| print(f" [DEBUG] Response Text: {response.text[:500]}") |
| raise Exception(f"Invalid JSON response from WordPress during post creation: {response.text[:200]}") |
| |
| result = { |
| "id": data.get("id"), |
| "url": data.get("link", data.get("guid", {}).get("rendered", "")) |
| } |
| if not result["id"]: |
| raise Exception(f"WordPress post created but no ID returned. Response: {response.text[:200]}") |
| |
| print(f" β
Published: {result['url']}") |
| return result |
| |
| |
| # ββββββββββββββββββββββββββββββββββββββββββββββ |
| # 7. Full Article Rewrite Pipeline |
| # ββββββββββββββββββββββββββββββββββββββββββββββ |
| |
| def process_article(url: str, dry_run: bool = False) -> dict: |
| """ |
| Full pipeline for rewriting an article: |
| scrape β rewrite β find media β create featured image β upload β publish. |
| """ |
| print(f"\n{'='*60}") |
| print(f"π° Rewriting article: {url}") |
| print(f"{'='*60}") |
| |
| _ensure_dirs() |
| |
| # Validate WP credentials before doing expensive work |
| if not all([WP_URL, WP_USERNAME, WP_APP_PASSWORD]): |
| missing = [k for k, v in [("WP_URL", WP_URL), ("WP_USERNAME", WP_USERNAME), ("WP_APP_PASSWORD", WP_APP_PASSWORD)] if not v] |
| msg = f"WordPress credentials not set. Missing: {', '.join(missing)}. Set them in .env or deployment config." |
| print(f" [ERROR] {msg}") |
| raise RuntimeError(msg) |
| |
| # Generate a safe name from the URL |
| url_hash = hashlib.md5(url.encode()).hexdigest()[:8] |
| safe_name = _sanitize_filename( |
| urlparse(url).path.strip("/").split("/")[-1] or url_hash |
| ) |
| if not safe_name: |
| safe_name = url_hash |
| |
| # Step 1: Scrape the article |
| article = scrape_article(url) |
| |
| # Step 2: Rewrite with Gemini AI |
| rewritten = rewrite_article( |
| article["title"], article["body"], article["source_url"] |
| ) |
| |
| # Rate limiting is handled by gemma_ai.py (4s gap between calls) |
| # No additional cooldown needed |
| print(f" [COOLDOWN] Rate limit enforced by gemma_ai wrapper") |
| |
| # Step 3: Use embeds found in the source article |
| article_embeds = article.get("videos", []) # Key returned by scrape_article |
| if article_embeds: |
| print(f"π¬ Found {len(article_embeds)} media embed(s) in source article") |
| else: |
| print("π¬ No media embeds found in source article β skipping") |
| |
| # Step 4: Intelligent Image Filtering |
| filtered_image_urls = filter_irrelevant_images( |
| article["images"], rewritten["title"], article["body"] |
| ) |
| |
| # Step 4b: FALLBACK CHAIN β if AI filter removed everything, try recovery |
| source_featured = article.get("featured_image", "") |
| if not filtered_image_urls and source_featured: |
| print(f" [FALLBACK] AI filter removed all images, using source's featured image") |
| print(f" πΌοΈ Source featured image: {source_featured}") |
| filtered_image_urls = [source_featured] |
| elif not filtered_image_urls: |
| print(f" [FALLBACK] No images after filtering, searching for related image with AI...") |
| related = _search_related_image( |
| query=rewritten["title"], |
| article_title=rewritten["title"], |
| article_body=" ".join(rewritten["paragraphs"][:2]), |
| ) |
| if related: |
| filtered_image_urls = [related] |
| else: |
| print(f" β οΈ No images available at all β featured image will use placeholder") |
| |
| # Step 5: Download & process images from the article |
| article_images = download_article_images(filtered_image_urls, referer_url=article["source_url"]) |
| |
| # Step 6: Create featured image |
| featured_path = create_article_featured_image( |
| article_images, safe_name |
| ) |
| |
| if dry_run: |
| print(f"\nβ
DRY RUN complete for: {url}") |
| print(f" New title: {rewritten['title']}") |
| print(f" Paragraphs: {len(rewritten['paragraphs'])}") |
| print(f" Media embeds: {len(article_embeds)}") |
| print(f" Images: {len(article_images)}") |
| return { |
| "source_url": url, |
| "title": rewritten["title"], |
| "paragraphs": rewritten["paragraphs"], |
| "seo_description": rewritten.get("seo_description", ""), |
| "youtube_videos": article_embeds, # Keep key for dashboard compatibility if needed |
| "image_count": len(article_images), |
| "featured_image": str(featured_path), |
| "status": "dry_run_complete", |
| } |
| |
| # Step 7: Upload featured image to WordPress |
| featured_media = upload_media_to_wp( |
| featured_path, f"featured-article-{safe_name}.jpg" |
| ) |
| |
| # Step 8: Upload article images to WordPress for embedding |
| wp_images = [] |
| alt_texts = rewritten.get("image_alt_texts", []) |
| # Upload up to 3 body images from the FILTERED list |
| for i, img_url in enumerate(filtered_image_urls[:3]): |
| alt_text = alt_texts[i] if i < len(alt_texts) else f"Image for {rewritten['title']}" |
| wp_img = upload_image_from_url_to_wp( |
| img_url, f"article-{safe_name}-{i}.jpg", alt_text |
| ) |
| if wp_img: |
| wp_images.append(wp_img) |
| |
| # Step 9: Build Gutenberg content |
| content_html = build_article_gutenberg_content( |
| paragraphs=rewritten["paragraphs"], |
| article_embeds=article_embeds, |
| wp_image_ids=wp_images, |
| source_url=url, |
| ) |
| |
| # Step 10: Publish to WordPress |
| post = publish_article_post( |
| title=rewritten["title"], |
| content_html=content_html, |
| featured_media_id=featured_media["id"], |
| seo_description=rewritten.get("seo_description", ""), |
| tags=rewritten.get("tags", []), |
| ) |
| |
| result = { |
| "source_url": url, |
| "title": rewritten["title"], |
| "post_url": post["url"], |
| "post_id": post["id"], |
| "featured_image_id": featured_media["id"], |
| "youtube_videos": len(article_embeds), |
| "embedded_images": len(wp_images), |
| } |
| |
| print(f"\nπ SUCCESS! Rewritten article published: {post['url']}") |
| return result |
| |
| |
| # ββββββββββββββββββββββββββββββββββββββββββββββ |
| # 8. Standalone CLI |
| # ββββββββββββββββββββββββββββββββββββββββββββββ |
| |
| if __name__ == "__main__": |
| import argparse |
| |
| parser = argparse.ArgumentParser( |
| description="Article Rewriter β Scrape, rewrite, and publish articles", |
| formatter_class=argparse.RawDescriptionHelpFormatter, |
| epilog=textwrap.dedent(""" |
| Examples: |
| python article_rewriter.py https://example.com/some-article |
| python article_rewriter.py --file articles.txt |
| python article_rewriter.py --dry-run https://example.com/article |
| """) |
| ) |
| parser.add_argument( |
| "urls", nargs="*", |
| help="Article URLs to rewrite and publish" |
| ) |
| parser.add_argument( |
| "--file", "-f", type=str, |
| help="Path to a text file containing article URLs (one per line)" |
| ) |
| parser.add_argument( |
| "--dry-run", action="store_true", |
| help="Scrape and rewrite but don't publish to WordPress" |
| ) |
| |
| args = parser.parse_args() |
| |
| # Collect URLs |
| urls = list(args.urls) if args.urls else [] |
| |
| if args.file: |
| file_path = Path(args.file) |
| if not file_path.exists(): |
| print(f"[ERROR] File not found: {args.file}") |
| import sys |
| sys.exit(1) |
| with open(file_path) as f: |
| for line in f: |
| line = line.strip() |
| if line and not line.startswith("#"): |
| urls.append(line) |
| |
| if not urls: |
| parser.print_help() |
| print("\n[ERROR] No URLs provided.") |
| import sys |
| sys.exit(1) |
| |
| # Validate config |
| if not os.getenv("GOOGLE_API_KEY"): |
| print("[ERROR] GOOGLE_API_KEY not set in .env") |
| import sys |
| sys.exit(1) |
| if not args.dry_run and not all([WP_URL, WP_USERNAME, WP_APP_PASSWORD]): |
| print("[ERROR] WordPress credentials not set in .env") |
| import sys |
| sys.exit(1) |
| |
| print(f"\nπ Found {len(urls)} article URL(s) to process") |
| print(f" WordPress: {WP_URL}") |
| print(f" Post status: {WP_POST_STATUS}") |
| print(f" Dry run: {'YES' if args.dry_run else 'NO'}") |
| |
| results = [] |
| errors = [] |
| |
| for i, url in enumerate(urls, 1): |
| print(f"\n[{i}/{len(urls)}] Processing...") |
| try: |
| result = process_article(url, dry_run=args.dry_run) |
| results.append(result) |
| except Exception as e: |
| print(f"\nβ ERROR processing {url}: {e}") |
| import traceback |
| traceback.print_exc() |
| errors.append({"url": url, "error": str(e)}) |
| |
| # Summary |
| print(f"\n{'='*60}") |
| print("π ARTICLE REWRITER SUMMARY") |
| print(f"{'='*60}") |
| print(f" Total articles: {len(urls)}") |
| print(f" Successful: {len(results)}") |
| print(f" Errors: {len(errors)}") |
| |
| if results: |
| print(f"\nβ
Successful articles:") |
| for r in results: |
| print(f" - {r.get('title', 'N/A')}") |
| if "post_url" in r: |
| print(f" β {r['post_url']}") |
| |
| if errors: |
| print(f"\nβ Failed articles:") |
| for e in errors: |
| print(f" - {e['url']}: {e['error']}") |
| |
| # Save results |
| _ensure_dirs() |
| results_path = OUTPUT_DIR / "rewrite_results.json" |
| with open(results_path, "w", encoding="utf-8") as f: |
| json.dump({"results": results, "errors": errors}, f, indent=2, ensure_ascii=False) |
| print(f"\nπ Results saved to: {results_path}") |
| |
