#!/usr/bin/env python3 """ AI Content Monetization Engine - CONFIGURED AND READY Fully autonomous content creation and publishing system """ import os import json import requests from datetime import datetime from typing import Dict, List, Any from dotenv import load_dotenv import anthropic # Load your API keys from .env file load_dotenv() class AIContentEngine: """Autonomous AI-driven content monetization system""" def __init__(self): # Load API keys self.anthropic_key = os.getenv("ANTHROPIC_API_KEY") self.devto_key = os.getenv("DEVTO_API_KEY") self.target_niches = os.getenv("TARGET_NICHES", "technology and AI, business, productivity") # Initialize AI client self.client = anthropic.Anthropic(api_key=self.anthropic_key) self.model = "claude-sonnet-4-5-20250514" print(f"[INIT] Content Engine initialized!") print(f"[INIT] Target niches: {self.target_niches}") print(f"[INIT] Anthropic API: {'✓ Connected' if self.anthropic_key else '✗ Missing'}") print(f"[INIT] Dev.to API: {'✓ Connected' if self.devto_key else '✗ Missing'}") def research_trending_topics(self) -> List[Dict[str, Any]]: """Use AI to find profitable trending topics in your niches""" research_prompt = f"""You are a content strategist. Generate 5 trending article topics for these niches: {self.target_niches} Requirements: - Topics should be currently trending in 2026 - High potential for engagement and revenue - Specific and actionable (not generic) - Perfect for Dev.to, Medium, or tech blogs For each topic, provide: 1. title: Compelling, click-worthy title 2. keywords: 3-5 SEO keywords 3. niche: Which niche it fits 4. why_profitable: Brief explanation Return AS VALID JSON ARRAY ONLY (no markdown, no extra text): [ {{"title": "...", "keywords": ["...", "..."], "niche": "...", "why_profitable": "..."}}, ... ] """ response = self.client.messages.create( model=self.model, max_tokens=2000, messages=[{"role": "user", "content": research_prompt}] ) topics_text = response.content[0].text.strip() # Try to parse JSON try: # Remove markdown code blocks if present if "```json" in topics_text: topics_text = topics_text.split("```json")[1].split("```")[0].strip() elif "```" in topics_text: topics_text = topics_text.split("```")[1].split("```")[0].strip() topics = json.loads(topics_text) print(f"[RESEARCH] Generated {len(topics)} trending topics") return topics except: # Fallback topics if JSON parsing fails print("[RESEARCH] Using fallback topics") return [ { "title": "10 AI Tools That Will Replace Your Job in 2026 (And How to Adapt)", "keywords": ["AI tools", "automation", "future of work"], "niche": "technology and AI", "why_profitable": "High search volume, timely, actionable" } ] def generate_seo_article(self, topic: Dict[str, Any]) -> Dict[str, str]: """Generate a complete, publication-ready article""" content_prompt = f"""Write a complete, SEO-optimized blog article: TITLE: {topic['title']} NICHE: {topic.get('niche', 'technology')} KEYWORDS: {', '.join(topic.get('keywords', []))} REQUIREMENTS: - 1500-2000 words (substantial but not too long) - Engaging introduction with a hook - Use ## for H2 headings and ### for H3 headings - Include specific examples, tools, or actionable tips - Conversational but authoritative tone - Strong conclusion with key takeaways - SEO-optimized but natural (not keyword stuffing) - Format in MARKDOWN Write the complete article NOW (just the article content, no JSON): """ response = self.client.messages.create( model=self.model, max_tokens=8000, messages=[{"role": "user", "content": content_prompt}] ) article_content = response.content[0].text.strip() # Create article object article = { "title": topic['title'], "content": article_content, "tags": topic.get('keywords', [])[:4], # Dev.to max 4 tags "niche": topic.get('niche', 'technology'), "created_at": datetime.now().isoformat() } print(f"[GENERATE] Created {len(article_content)} character article") return article def publish_to_devto(self, article: Dict[str, str]) -> Dict[str, Any]: """Actually publish article to Dev.to using API""" if not self.devto_key: print("[PUBLISH] ✗ Dev.to API key not found") return {"status": "error", "message": "No API key"} # Dev.to API endpoint url = "https://dev.to/api/articles" headers = { "api-key": self.devto_key, "Content-Type": "application/json" } # Prepare article data post_data = { "article": { "title": article['title'], "published": False, # Start as draft for safety "body_markdown": article['content'], "tags": article['tags'], "series": None } } try: print(f"[PUBLISH] Publishing to Dev.to...") response = requests.post(url, headers=headers, json=post_data) if response.status_code == 201: result = response.json() print(f"[PUBLISH] ✓ Successfully published to Dev.to!") print(f"[PUBLISH] URL: {result.get('url', 'N/A')}") print(f"[PUBLISH] Status: Draft (set to published=true to make public)") return { "platform": "dev.to", "status": "success", "url": result.get("url"), "id": result.get("id"), "published_at": datetime.now().isoformat() } else: print(f"[PUBLISH] ✗ Error: {response.status_code}") print(f"[PUBLISH] Response: {response.text}") return { "platform": "dev.to", "status": "error", "error": response.text } except Exception as e: print(f"[PUBLISH] ✗ Exception: {str(e)}") return { "platform": "dev.to", "status": "error", "error": str(e) } def run_test_article(self): """Generate and publish ONE test article""" print("\n" + "="*60) print("AI CONTENT MONETIZATION ENGINE - TEST RUN") print("="*60 + "\n") # 1. Research topics print("[STEP 1] Researching trending topics...") topics = self.research_trending_topics() # 2. Select best topic selected_topic = topics[0] print(f"\n[STEP 2] Selected topic: {selected_topic['title']}") # 3. Generate article print(f"\n[STEP 3] Generating article...") article = self.generate_seo_article(selected_topic) # 4. Save locally timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f"article_{timestamp}.md" filepath = os.path.join(os.path.dirname(__file__), filename) with open(filepath, 'w', encoding='utf-8') as f: f.write(f"# {article['title']}\n\n") f.write(f"**Tags**: {', '.join(article['tags'])}\n\n") f.write(f"---\n\n") f.write(article['content']) print(f"[STEP 4] ✓ Saved article to: {filename}") # 5. Publish to Dev.to print(f"\n[STEP 5] Publishing to Dev.to...") devto_result = self.publish_to_devto(article) # 6. Results print("\n" + "="*60) print("TEST RUN COMPLETE!") print("="*60) results = { "test_date": datetime.now().isoformat(), "topic": selected_topic, "article": { "title": article['title'], "word_count": len(article['content'].split()), "tags": article['tags'], "local_file": filename }, "publication_results": { "devto": devto_result } } # Save results results_file = f"test_results_{timestamp}.json" with open(results_file, 'w', encoding='utf-8') as f: json.dump(results, f, indent=2) print(f"\n✓ Test results saved to: {results_file}") if devto_result.get('status') == 'success': print(f"\n🎉 SUCCESS! Your article is on Dev.to:") print(f" {devto_result.get('url')}") print(f"\n Note: Article is in DRAFT mode for safety.") print(f" Go to Dev.to to review and publish it publicly!") return results def main(): """Run a test article generation""" print("Initializing AI Content Monetization Engine...\n") engine = AIContentEngine() # Run test results = engine.run_test_article() print("\n" + "="*60) print("NEXT STEPS:") print("="*60) print("1. Check the generated article file (.md)") print("2. Check Dev.to dashboard to see your draft article") print("3. Review and publish the article on Dev.to") print("4. Once satisfied, set up daily automation!") print("="*60 + "\n") if __name__ == "__main__": main()