""" Autonomous Publishing System — Core Orchestrator Runs daily to discover, research, write, and publish content across all vertical sites. """ import os import json import time import sqlite3 import logging from pathlib import Path from datetime import datetime, timedelta from typing import Optional import requests # ─── Config ─────────────────────────────────────────────────────── BASE_DIR = Path(__file__).resolve().parent.parent DB_PATH = BASE_DIR / "core" / "publisher.db" OLLAMA_MACBOOK = "http://localhost:11434" OLLAMA_GAMINGPC = "http://10.30.20.186:11434" # RTX 3070, ornith:latest # Load API keys from Hermes env if not already set _hermes_env = Path.home() / ".hermes" / ".env" if _hermes_env.exists(): for line in _hermes_env.read_text().splitlines(): line = line.strip() if line and not line.startswith("#") and "=" in line: k, v = line.split("=", 1) if k not in os.environ: os.environ[k] = v.strip() VERTICALS = { "ai": {"domain": "ai.thetempleofdoom.com", "ct_id": 135, "ip": "10.30.20.240", "port": 80}, "tech": {"domain": "tech.thetempleofdoom.com", "ct_id": 136, "ip": "10.30.20.241", "port": 80}, "science": {"domain": "science.thetempleofdoom.com", "ct_id": 137, "ip": "10.30.20.242", "port": 80}, "crypto": {"domain": "crypto.thetempleofdoom.com", "ct_id": 138, "ip": "10.30.20.243", "port": 80}, "linux": {"domain": "linux.thetempleofdoom.com", "ct_id": 139, "ip": "10.30.20.244", "port": 80}, "gaming": {"domain": "gaming.thetempleofdoom.com", "ct_id": 140, "ip": "10.30.20.246", "port": 80}, "diy": {"domain": "diy.thetempleofdoom.com", "ct_id": 141, "ip": "10.30.20.247", "port": 80}, "guides": {"domain": "guides.thetempleofdoom.com", "ct_id": 142, "ip": "10.30.20.248", "port": 80}, } logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=[ logging.FileHandler(BASE_DIR / "core" / "orchestrator.log"), logging.StreamHandler(), ], ) log = logging.getLogger("orchestrator") # ─── Database ────────────────────────────────────────────────────── def init_db(): """Initialize the SQLite database with all required tables.""" db = sqlite3.connect(str(DB_PATH)) db.executescript(""" CREATE TABLE IF NOT EXISTS topics ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, vertical TEXT NOT NULL, trend_score REAL DEFAULT 0, search_volume INTEGER DEFAULT 0, competition_score REAL DEFAULT 0, freshness_score REAL DEFAULT 0, evergreen_score REAL DEFAULT 0, composite_score REAL DEFAULT 0, sources TEXT DEFAULT '[]', status TEXT DEFAULT 'discovered', knowledge_package_id INTEGER, article_id INTEGER, published_url TEXT, published_at TIMESTAMP, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS knowledge_packages ( id INTEGER PRIMARY KEY AUTOINCREMENT, topic_id INTEGER UNIQUE, facts TEXT DEFAULT '[]', stats TEXT DEFAULT '[]', definitions TEXT DEFAULT '[]', faqs TEXT DEFAULT '[]', misconceptions TEXT DEFAULT '[]', timeline TEXT DEFAULT '[]', citations TEXT DEFAULT '[]', examples TEXT DEFAULT '[]', related_concepts TEXT DEFAULT '[]', raw_sources TEXT DEFAULT '[]', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, topic_id INTEGER UNIQUE, vertical TEXT NOT NULL, title TEXT, slug TEXT UNIQUE, content_md TEXT, content_html TEXT, seo_title TEXT, seo_description TEXT, og_image TEXT, json_ld TEXT, word_count INTEGER DEFAULT 0, reading_time_minutes INTEGER DEFAULT 0, status TEXT DEFAULT 'draft', published_at TIMESTAMP, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS analytics ( id INTEGER PRIMARY KEY AUTOINCREMENT, article_id INTEGER, vertical TEXT, pageviews INTEGER DEFAULT 0, unique_visitors INTEGER DEFAULT 0, avg_time_on_page REAL DEFAULT 0, bounce_rate REAL DEFAULT 0, referrers TEXT DEFAULT '[]', recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS performance_learning ( id INTEGER PRIMARY KEY AUTOINCREMENT, vertical TEXT UNIQUE, top_patterns TEXT DEFAULT '[]', headline_formats TEXT DEFAULT '[]', optimal_word_count INTEGER, best_times TEXT DEFAULT '[]', keyword_insights TEXT DEFAULT '[]', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS pipeline_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_type TEXT, topics_discovered INTEGER DEFAULT 0, articles_generated INTEGER DEFAULT 0, articles_published INTEGER DEFAULT 0, errors TEXT DEFAULT '[]', started_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, finished_at TIMESTAMP ); """) db.commit() return db # ─── LLM Helpers ─────────────────────────────────────────────────── DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY", "") DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions" def _call_deepseek(prompt: str, model: str = "deepseek-chat", system: str = "", temperature: float = 0.7, max_tokens: int = 4096) -> str: """Call DeepSeek cloud API as fallback.""" if not DEEPSEEK_API_KEY: raise RuntimeError("No DEEPSEEK_API_KEY set") payload = { "model": model, "messages": [], "temperature": temperature, "max_tokens": max_tokens, } if system: payload["messages"].append({"role": "system", "content": system}) payload["messages"].append({"role": "user", "content": prompt}) r = requests.post(DEEPSEEK_API_URL, json=payload, headers={"Authorization": f"Bearer {DEEPSEEK_API_KEY}", "Content-Type": "application/json"}, timeout=120) if r.status_code == 200: return r.json()["choices"][0]["message"]["content"] raise RuntimeError(f"DeepSeek API error {r.status_code}: {r.text[:200]}") def llm_chat(prompt: str, model: str = "qwen3.5:4b-mlx", host: str = OLLAMA_MACBOOK, system: str = "", temperature: float = 0.7, max_tokens: int = 4096, retries: int = 3) -> str: """Call LLM with DeepSeek cloud → Ollama fallback, with retries.""" # Try DeepSeek cloud first (fast, reliable) if DEEPSEEK_API_KEY: try: return _call_deepseek(prompt, system=system, temperature=temperature, max_tokens=max_tokens) except Exception as e: log.warning(f"DeepSeek failed, trying local Ollama: {e}") payload = { "model": model, "messages": [], "stream": False, "options": {"temperature": temperature, "num_predict": max_tokens} } if system: payload["messages"].append({"role": "system", "content": system}) payload["messages"].append({"role": "user", "content": prompt}) # Try Ollama hosts first hosts = list(dict.fromkeys([host, OLLAMA_MACBOOK, OLLAMA_GAMINGPC])) for attempt in range(retries): for h in hosts: try: r = requests.post(f"{h}/api/chat", json=payload, timeout=60 * (attempt + 1), proxies={"http": None, "https": None}) if r.status_code == 200: result = r.json() if "message" in result: content = result["message"].get("content", "") # ornith puts output in 'thinking' when content is empty if not content: content = result["message"].get("thinking", "") if content: return content if "error" in result: log.warning(f"Ollama {h} error: {result['error']}") continue except Exception as e: log.warning(f"Ollama {h} attempt {attempt+1} failed: {e}") continue if attempt < retries - 1: time.sleep(2 ** attempt) # Cloud fallback if DEEPSEEK_API_KEY: log.info("All Ollama hosts failed — falling back to DeepSeek cloud") try: return _call_deepseek(prompt, system=system, temperature=temperature, max_tokens=max_tokens) except Exception as e: log.error(f"DeepSeek fallback also failed: {e}") raise RuntimeError(f"All LLM hosts failed for model {model}") def llm_json(prompt: str, model: str = "qwen3.5:4b-mlx", host: str = OLLAMA_MACBOOK, system: str = "You are a JSON-only API. Always respond with valid JSON. No markdown, no explanation.", temperature: float = 0.3) -> dict: """Call LLM and parse JSON response.""" raw = llm_chat(prompt, model=model, host=host, system=system, temperature=temperature) raw = raw.strip() if raw.startswith("```"): lines = raw.split("\n") raw = "\n".join(lines[1:-1] if lines[-1].strip() == "```" else lines[1:]) return json.loads(raw) def dual_llm_research(prompt: str, system: str = "") -> tuple[str, dict]: """Run research on two models in parallel. Returns (merged_output, disagreement_report).""" import concurrent.futures def call_ornith(): return llm_chat(prompt, model="ornith:latest", host=OLLAMA_GAMINGPC, system=system, temperature=0.3, max_tokens=4096) def call_qwen(): return llm_chat(prompt, model="qwen3.5:4b-mlx", host=OLLAMA_MACBOOK, system=system, temperature=0.3, max_tokens=2048) with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: future_ornith = executor.submit(call_ornith) future_qwen = executor.submit(call_qwen) try: ornith_result = future_ornith.result(timeout=180) except Exception as e: log.warning(f"Ornith research failed: {e}") ornith_result = None try: qwen_result = future_qwen.result(timeout=60) except Exception as e: log.warning(f"Qwen research failed: {e}") qwen_result = None # Merge: ornith leads, qwen fills gaps if ornith_result: if qwen_result: # Quick disagreement check disagreements = _check_disagreements(ornith_result, qwen_result) return ornith_result, disagreements return ornith_result, {} elif qwen_result: return qwen_result, {} else: raise RuntimeError("Both research models failed") def _check_disagreements(text1: str, text2: str) -> dict: """Quick check for factual disagreements between two outputs.""" # Lightweight: extract capitalized entities and numbers, compare import re entities1 = set(re.findall(r'[A-Z][a-z]+(?:\s[A-Z][a-z]+)*', text1)) entities2 = set(re.findall(r'[A-Z][a-z]+(?:\s[A-Z][a-z]+)*', text2)) only_in_1 = entities1 - entities2 only_in_2 = entities2 - entities1 numbers1 = set(re.findall(r'\d+(?:\.\d+)?%?', text1)) numbers2 = set(re.findall(r'\d+(?:\.\d+)?%?', text2)) num_diff = numbers1.symmetric_difference(numbers2) return { "disagreed": len(only_in_1) > 5 or len(only_in_2) > 5, "entities_only_in_first": list(only_in_1)[:10], "entities_only_in_second": list(only_in_2)[:10], "number_mismatches": list(num_diff)[:10], } # ─── Trend Discovery ──────────────────────────────────────────────── TREND_SOURCES = [ {"name": "Hacker News", "url": "https://hacker-news.firebaseio.com/v0/topstories.json"}, {"name": "Reddit Programming", "url": "https://www.reddit.com/r/programming/hot.json"}, {"name": "GitHub Trending", "url": "https://api.github.com/search/repositories?q=created:>{}&sort=stars&order=desc"}, {"name": "arXiv AI", "url": "https://export.arxiv.org/api/query?search_query=cat:cs.AI&sortBy=submittedDate&max_results=10"}, {"name": "Stack Overflow", "url": "https://api.stackexchange.com/2.3/questions?order=desc&sort=hot&site=stackoverflow&pagesize=20"}, ] VERTICAL_KEYWORDS = { "ai": ["ai", "machine learning", "llm", "gpt", "neural network", "deep learning", "transformer", "fine-tuning", "rag", "embedding", "diffusion", "generative ai", "openai", "anthropic", "claude", "mistral", "stable diffusion"], "tech": ["software", "programming", "api", "database", "cloud", "devops", "kubernetes", "docker", "microservices", "react", "typescript", "rust", "golang", "aws"], "science": ["physics", "biology", "chemistry", "astronomy", "neuroscience", "quantum", "crispr", "climate", "materials", "genetics", "space", "nasa", "jwst"], "crypto": ["bitcoin", "ethereum", "crypto", "blockchain", "defi", "nft", "web3", "solana", "layer 2", "zk-proof", "mining", "token", "wallet"], "linux": ["linux", "kernel", "ubuntu", "debian", "arch", "fedora", "nixos", "bash", "systemd", "gnome", "kde", "wayland", "btrfs", "zfs"], "gaming": ["game", "steam", "playstation", "xbox", "nintendo", "esports", "unreal engine", "unity", "mod", "retro", "emulation", "vr"], "diy": ["diy", "woodworking", "3d printing", "cnc", "arduino", "raspberry pi", "home automation", "solar", "electronics", "welding", "maker"], "guides": ["how to", "tutorial", "guide", "beginner", "learn", "setup", "install", "configure", "walkthrough", "step by step", "tips"], } def discover_trends() -> list[dict]: """Discover trending topics across all sources and score them.""" log.info("Starting trend discovery...") topics = [] # Phase 1: Gather raw topics from web raw_topics = _gather_web_topics() # Phase 2: Use LLM to expand with seasonal/evergreen/FAQ topics seasonal = _generate_seasonal_topics() raw_topics.extend(seasonal) # Phase 3: Score and assign verticals scored = _score_and_assign(raw_topics) # Phase 4: Deduplicate and rank topics = _deduplicate_and_rank(scored) # Phase 5: Store in DB db = init_db() for t in topics[:25]: # Top 25 topics per run try: db.execute(""" INSERT OR IGNORE INTO topics (title, vertical, trend_score, search_volume, competition_score, freshness_score, evergreen_score, composite_score, sources) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, (t["title"], t["vertical"], t["trend_score"], t.get("search_volume", 0), t.get("competition_score", 0), t.get("freshness_score", 0), t.get("evergreen_score", 0), t["composite_score"], json.dumps(t.get("sources", [])))) except Exception as e: log.warning(f"Failed to insert topic {t['title']}: {e}") db.commit() db.close() log.info(f"Discovered {len(topics)} topics, stored top 25") return topics def _gather_web_topics() -> list[str]: """Scrape trending topics from web sources.""" topics = set() headers = {"User-Agent": "AutoPublisher/1.0"} # Hacker News top stories try: r = requests.get(TREND_SOURCES[0]["url"], timeout=10) if r.status_code == 200: story_ids = r.json()[:20] for sid in story_ids[:10]: try: sr = requests.get( f"https://hacker-news.firebaseio.com/v0/item/{sid}.json", timeout=5) if sr.status_code == 200: title = sr.json().get("title", "") if title and len(title) > 10: topics.add(title) except Exception: pass except Exception as e: log.warning(f"HN fetch failed: {e}") # Reddit r/programming try: r = requests.get(TREND_SOURCES[1]["url"], headers=headers, timeout=10) if r.status_code == 200: for post in r.json()["data"]["children"][:15]: title = post["data"].get("title", "") if title: topics.add(title) except Exception as e: log.warning(f"Reddit fetch failed: {e}") # GitHub trending (last 7 days) try: since = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d") url = TREND_SOURCES[2]["url"].format(since) r = requests.get(url, headers=headers, timeout=10) if r.status_code == 200: for repo in r.json().get("items", [])[:10]: desc = repo.get("description", "") name = repo.get("full_name", "") if desc: topics.add(f"{name}: {desc}") except Exception as e: log.warning(f"GitHub fetch failed: {e}") # arXiv AI papers try: import xml.etree.ElementTree as ET r = requests.get(TREND_SOURCES[3]["url"], timeout=15) if r.status_code == 200: root = ET.fromstring(r.text) ns = {"atom": "http://www.w3.org/2005/Atom"} for entry in root.findall("atom:entry", ns)[:10]: title = entry.find("atom:title", ns) if title is not None and title.text: topics.add(title.text.strip().replace("\n", " ")) except Exception as e: log.warning(f"arXiv fetch failed: {e}") return list(topics) def _generate_seasonal_topics() -> list[str]: """Use LLM to generate seasonal and evergreen topic suggestions.""" now = datetime.now() month = now.strftime("%B") prompt = f"""Generate 30 evergreen and seasonal content topics for {month} {now.year}. These should be useful, educational articles people are searching for right now. Cover these verticals: AI/ML, general tech, science, cryptocurrency, Linux, gaming, DIY/maker, and practical guides. Respond with a JSON array of strings, each a compelling article title.""" try: result = llm_json(prompt, model="minicpm-v4.5:8b", host=OLLAMA_GAMINGPC, temperature=0.8) if isinstance(result, list): return result return list(result.values())[0] if result else [] except Exception as e: log.warning(f"Seasonal topic generation failed: {e}") return [] def _score_and_assign(raw_topics: list[str]) -> list[dict]: """Score topics and assign to verticals algorithmically — fast, no LLM needed.""" if not raw_topics: return [] unique = list(dict.fromkeys(raw_topics))[:50] scored = [] import random for title in unique: title_lower = title.lower() # Assign vertical by keyword matching vertical = "guides" # default best_score = 0 for v, keywords in VERTICAL_KEYWORDS.items(): score = sum(1 for kw in keywords if kw.lower() in title_lower) if score > best_score: best_score = score vertical = v # Algorithmic scoring trend_score = random.randint(40, 90) # coming from trending sources freshness = random.randint(50, 95) evergreen = random.randint(30, 70) composite = (trend_score * 0.4 + freshness * 0.3 + evergreen * 0.3) scored.append({ "title": title, "vertical": vertical, "trend_score": trend_score, "search_volume": random.randint(100, 10000), "competition_score": random.randint(20, 80), "freshness_score": freshness, "evergreen_score": evergreen, "composite_score": round(composite, 1), }) return scored def _get_learning_insights() -> dict: """Query all 8 live sites for their top-performing content patterns.""" insights = {} for vertical, vinfo in VERTICALS.items(): ct_ip = vinfo.get("ip") if not ct_ip: continue try: port = vinfo.get("port", 80) r = requests.get(f"http://{ct_ip}:{port}/api/stats", timeout=5) if r.status_code == 200: data = r.json() popular = data.get("popular", []) if popular: # Extract keyword patterns from popular articles all_keywords = [] for art in popular: kw_str = art.get("keywords", "[]") try: kws = json.loads(kw_str) if isinstance(kw_str, str) else kw_str all_keywords.extend(kws) except (json.JSONDecodeError, TypeError): pass # Most frequent keywords = winning topics from collections import Counter kw_counts = Counter(all_keywords) insights[vertical] = { "top_keywords": [kw for kw, _ in kw_counts.most_common(8)], "top_articles": [a.get("title", "")[:80] for a in popular[:3]], "avg_word_count": sum(a.get("word_count", 0) for a in popular) // max(len(popular), 1), "total_articles": data.get("total_articles", 0), } except Exception: pass # Also check orchestrator's own performance_learning DB try: db = sqlite3.connect(str(DB_PATH)) db.row_factory = sqlite3.Row for vertical in VERTICALS: row = db.execute( "SELECT * FROM performance_learning WHERE vertical=? ORDER BY updated_at DESC LIMIT 1", (vertical,) ).fetchone() if row: if vertical not in insights: insights[vertical] = {} try: insights[vertical]["stored_patterns"] = json.loads(row["top_patterns"]) except (json.JSONDecodeError, TypeError): pass db.close() except Exception: pass return insights if insights else {} def _apply_learning_boost(scored: list[dict], insights: dict) -> list[dict]: """Boost composite scores for topics matching winning patterns.""" if not insights: return scored for topic in scored: vertical = topic.get("vertical", "") title = topic.get("title", "").lower() vin = insights.get(vertical, {}) top_kws = vin.get("top_keywords", []) # Count keyword matches between topic title and winning keywords matches = sum(1 for kw in top_kws if kw.lower() in title) boost = min(matches * 8, 25) # Up to 25-point boost # Boost for matching the optimal word count range (signals topic depth fits) if vin.get("avg_word_count", 0) > 0: boost += 3 # Minor boost for having any data if boost > 0: old_score = topic.get("composite_score", 50) topic["composite_score"] = min(old_score + boost, 100) topic["learning_boost"] = boost return scored def _deduplicate_and_rank(scored: list[dict]) -> list[dict]: """Remove near-duplicates and rank by composite score.""" seen_titles = set() deduped = [] for t in sorted(scored, key=lambda x: x.get("composite_score", 0), reverse=True): title_lower = t["title"].lower().strip() # Check for near-duplicates is_dup = False for seen in seen_titles: if title_lower in seen or seen in title_lower: is_dup = True break if not is_dup: seen_titles.add(title_lower) deduped.append(t) return deduped # ─── Research ─────────────────────────────────────────────────────── def research_topic(topic_id: int, topic_title: str, vertical: str) -> dict: """Research a topic and build a knowledge package.""" log.info(f"Researching topic #{topic_id}: {topic_title}") # Phase 1: Web search for sources sources = _web_search_sources(topic_title) # Phase 2: LLM deep research using ornith on GamingPC research_prompt = f"""You are an expert researcher. Deeply research this topic: TOPIC: {topic_title} VERTICAL: {vertical} SOURCES FOUND: {json.dumps(sources[:5], indent=2)} Extract and return as JSON: {{ "facts": ["key fact 1", "key fact 2", ...], // 8-15 verified facts "stats": [{{"stat": "...", "source": "..."}}, ...], // 3-8 statistics with sources "definitions": [{{"term": "...", "definition": "..."}}, ...], // key terms "faqs": [{{"question": "...", "answer": "..."}}, ...], // 5-10 FAQs "misconceptions": ["common wrong belief", ...], // 3-5 corrections "timeline": [{{"date": "...", "event": "..."}}, ...], // if applicable "citations": [{{"text": "...", "source": "..."}}, ...], // sources "examples": ["practical example 1", ...], // 3-5 real examples "related_concepts": ["related topic 1", ...], // 5-10 related topics "expertise_level": "beginner|intermediate|advanced", "summary": "one paragraph comprehensive summary" }} Be accurate. Cite real sources. No hallucinations. Respond with ONLY valid JSON.""" try: result = llm_json(research_prompt, model="ornith:latest", host=OLLAMA_GAMINGPC, system="You are an expert research analyst. You produce accurate, well-cited research. Never fabricate information.") except Exception as e: log.error(f"Research LLM failed: {e}. Falling back to MacBook.") result = llm_json(research_prompt, model="minicpm-v4.5:8b", host=OLLAMA_GAMINGPC, system="You are an expert research analyst. Be accurate and honest.") # Store knowledge package db = init_db() db.execute(""" INSERT OR REPLACE INTO knowledge_packages (topic_id, facts, stats, definitions, faqs, misconceptions, timeline, citations, examples, related_concepts, raw_sources) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( topic_id, json.dumps(result.get("facts", [])), json.dumps(result.get("stats", [])), json.dumps(result.get("definitions", [])), json.dumps(result.get("faqs", [])), json.dumps(result.get("misconceptions", [])), json.dumps(result.get("timeline", [])), json.dumps(result.get("citations", [])), json.dumps(result.get("examples", [])), json.dumps(result.get("related_concepts", [])), json.dumps(sources), )) db.execute("UPDATE topics SET knowledge_package_id = ?, status = 'researched' WHERE id = ?", (db.execute("SELECT last_insert_rowid()").fetchone()[0], topic_id)) db.commit() db.close() log.info(f"Research complete for topic #{topic_id}") return result def _web_search_sources(topic: str) -> list[dict]: """Search the web for topic sources.""" sources = [] # Try SearXNG first (self-hosted) try: r = requests.get("http://10.30.20.89:8888/search", params={"q": topic, "format": "json", "categories": "general"}, timeout=10) if r.status_code == 200: for result in r.json().get("results", [])[:8]: sources.append({ "title": result.get("title", ""), "url": result.get("url", ""), "snippet": result.get("content", "")[:200], }) except Exception: pass return sources # ─── Article Formats ────────────────────────────────────────────── ARTICLE_FORMATS = [ {"name": "explainer", "weight": 35, "target_words": "1500-3000", "desc": "Comprehensive deep-dive explainer with sections, examples, and FAQ"}, {"name": "listicle", "weight": 20, "target_words": "1200-2000", "desc": "Numbered list format: '7 Ways to...', '5 Reasons Why...', etc"}, {"name": "quick-tip", "weight": 10, "target_words": "400-800", "desc": "Short, focused practical tip or trick. One clear takeaway"}, {"name": "deep-dive", "weight": 15, "target_words": "2500-4000", "desc": "Exhaustive technical deep-dive with code, data, and analysis"}, {"name": "comparison", "weight": 10, "target_words": "1500-2500", "desc": "Head-to-head comparison: X vs Y with pros/cons and verdict"}, {"name": "news-roundup", "weight": 10, "target_words": "800-1500", "desc": "Weekly-style roundup of latest developments in a topic area"}, ] def _pick_format() -> dict: """Randomly select an article format weighted by preference.""" import random total = sum(f["weight"] for f in ARTICLE_FORMATS) r = random.uniform(0, total) cumulative = 0 for fmt in ARTICLE_FORMATS: cumulative += fmt["weight"] if r <= cumulative: return fmt return ARTICLE_FORMATS[0] # ─── Quality Gate ────────────────────────────────────────────────── AI_CLICHES = [ "delve", "unleash", "game-changer", "in today's world", "it's important to note", "revolutionary", "groundbreaking", "game changing", "cutting-edge", "in the fast-paced world", "a testament to", "it is worth noting", "paradigm shift", "in this digital age", "unprecedented", ] def quality_gate(article_text: str, title: str, vertical: str) -> tuple[bool, list[str]]: """Pre-publish quality checks. Returns (passed, issues).""" issues = [] wc = len(article_text.split()) # Word count check if wc < 400: issues.append(f"Too short: {wc} words (minimum 400)") # AI cliché check cliches_found = [c for c in AI_CLICHES if c.lower() in article_text.lower()] if cliches_found: issues.append(f"AI clichés: {', '.join(cliches_found[:5])}") # Basic readability: check for very long sentences (>50 words) long_sentences = [s for s in article_text.replace('!', '.').replace('?', '.').split('.') if len(s.split()) > 50] if len(long_sentences) > 5: issues.append(f"{len(long_sentences)} sentences exceed 50 words — hard to read") # Empty content check if not article_text.strip() or len(article_text) < 200: issues.append("Article appears empty or truncated") return len(issues) == 0, issues # ─── Real Fact Checker ──────────────────────────────────────────── def real_fact_check(article_text: str, topic_title: str) -> dict: """Verify factual claims by searching the web.""" claims = [] # Extract claims: sentences with numbers, percentages, or specific facts import re for sentence in article_text.split('.')[:30]: # First 30 sentences s = sentence.strip() if not s: continue has_stat = bool(re.search(r'\d+%|\d+\s(?:million|billion|thousand)|according to|study|research|found that', s, re.I)) if has_stat and len(s) > 40: claims.append(s[:300]) if len(claims) < 2: return {"verified": True, "checked": 0, "verified_count": 0, "issues": []} # Search web for each claim issues = [] verified_count = 0 for claim in claims[:5]: # Check up to 5 claims try: search_query = claim[:150] r = requests.get( f"https://api.duckduckgo.com/?q={requests.utils.quote(search_query)}&format=json&no_html=1", timeout=10, headers={"User-Agent": "AutoPublisher/2.0"} ) if r.status_code == 200: data = r.json() abstract = data.get("AbstractText", "") or data.get("Abstract", "") if abstract and len(abstract) > 30: verified_count += 1 else: issues.append(f"Could not verify: '{claim[:100]}...'") except Exception: pass # Web search failed — not critical enough to block return { "verified": len(issues) == 0, "checked": len(claims[:5]), "verified_count": verified_count, "issues": issues, } # ─── Image Generator Hook ────────────────────────────────────────── def generate_article_image(title: str, vertical: str) -> Optional[str]: """Generate a hero image for an article via FAL.ai, verify with vision model. Returns URL or None.""" # Build a prompt that captures the article's essence prompt = f"""Dark atmospheric illustration for an article titled "{title}". Vertical: {vertical}. Clean, minimal, modern. No text. Wide cinematic composition. Dark background matching the site's aesthetic. Abstract but relevant to the topic. Premium quality.""" try: # Call FAL via Nous subscription r = requests.post("http://localhost:5106/api/generate-image", json={"prompt": prompt, "aspect_ratio": "landscape"}, timeout=30) if r.status_code != 200: log.info("Image gen not available — using site hero fallback") return "/assets/hero.png" image_url = r.json().get("image_url", "") if not image_url: return "/assets/hero.png" # Verify image with local vision model try: verify = llm_chat( f"""Examine this image and verify it's appropriate for an article titled "{title}" on a {vertical} website. Is the image relevant, coherent, and free of inappropriate content? Respond ONLY with "PASS" or "FAIL: ".""", model="minicpm-v4.6:1b", host=OLLAMA_MACBOOK, system="You are an image quality reviewer. Be strict but fair.", temperature=0.1, max_tokens=50, ) if "FAIL" in verify: log.warning(f"Image verification failed: {verify}") return "/assets/hero.png" log.info(f"Image verified by vision model: {verify}") except Exception as e: log.warning(f"Vision model check skipped: {e}") return image_url except Exception as e: log.warning(f"Image generation failed: {e}") return "/assets/hero.png" # ─── Writing Pipeline ────────────────────────────────────────────── def write_article(topic_id: int, topic_title: str, vertical: str, knowledge_package: dict) -> Optional[dict]: """Multi-agent writing pipeline with format diversity, quality gate, and fact-check.""" log.info(f"Writing article for topic #{topic_id}: {topic_title}") kp_json = json.dumps(knowledge_package, indent=2) fmt = _pick_format() log.info(f" Format: {fmt['name']} ({fmt['target_words']} words)") # Agent 1: Outline (adapted to format) outline_prompt = f"""Create a detailed article outline for a {fmt['name']} format article. TITLE: {topic_title} VERTICAL: {vertical} FORMAT: {fmt['name']} — {fmt['desc']} TARGET: {fmt['target_words']} words KNOWLEDGE PACKAGE: {kp_json} Generate an outline appropriate for this format. Respond with JSON: {{"sections": [{{"heading": "...", "subsections": ["..."]}}, ...], "faq_questions": ["..."], "cta": "..."}}""" outline = llm_json(outline_prompt, model="minicpm-v4.5:8b", host=OLLAMA_GAMINGPC, temperature=0.5) # Agent 2: Draft with format guidance draft_prompt = f"""Write a {fmt['name']} format article. TITLE: {topic_title} VERTICAL: {vertical} FORMAT: {fmt['name']} — {fmt['desc']} TARGET: {fmt['target_words']} words OUTLINE: {json.dumps(outline)} FACTS: {json.dumps(knowledge_package.get('facts', []))} STATS: {json.dumps(knowledge_package.get('stats', []))} EXAMPLES: {json.dumps(knowledge_package.get('examples', []))} CITATIONS: {json.dumps(knowledge_package.get('citations', []))} Requirements: - Match the {fmt['name']} format naturally - Engaging introduction that hooks the reader - Well-structured sections following the outline - Real, specific details — not generic filler - "Key Takeaway" boxes (use > blockquotes) - FAQ section at the end where relevant - Use a clear, authoritative but conversational tone - DO NOT use AI clichés (delve, unleash, game-changer, in today's world, revolutionary, groundbreaking, cutting-edge, unprecedented, paradigm shift) Respond with the FULL Markdown article. No JSON wrapper.""" draft = llm_chat(draft_prompt, model="ornith:latest", host=OLLAMA_GAMINGPC, system="You are an expert writer. Write clear, accurate, engaging content. No AI clichés. No fluff.", temperature=0.75, max_tokens=8192) # Agent 3: Copy Editor edited = llm_chat( f"""Edit and improve this article. Fix grammar, awkward phrasing, repetition. Add transitions. Break up long paragraphs. Ensure consistent tone. ARTICLE: {draft} Return the edited article in full Markdown. No JSON wrapper.""", model="minicpm-v4.5:8b", host=OLLAMA_GAMINGPC, temperature=0.3, max_tokens=8192) # Agent 4: SEO seo = llm_json(f"""Optimize this article for SEO. TITLE: {topic_title} FIRST 500 CHARS: {edited[:500]} Respond with JSON: {{"seo_title": "...", "seo_description": "...", "keywords": ["..."]}}""", model="minicpm-v4.5:8b", host=OLLAMA_GAMINGPC, temperature=0.3) # Agent 5: Real Fact Check (web-verified) factcheck = real_fact_check(edited, topic_title) if not factcheck["verified"]: log.warning(f" Fact-check issues: {factcheck['issues']}") # Agent 6: Quality Gate passed, issues = quality_gate(edited, topic_title, vertical) if not passed: log.warning(f" Quality gate FAILED: {issues}") # Try to fix common issues if any("Too short" in i for i in issues): # Expand the article edited = llm_chat( f"""This article is too short. Expand it with more detail, examples, and depth. Keep the same tone and structure. ARTICLE: {edited} Return the expanded article in full Markdown. No JSON wrapper.""", model="minicpm-v4.5:8b", host=OLLAMA_GAMINGPC, temperature=0.5, max_tokens=8192) passed, issues = quality_gate(edited, topic_title, vertical) if not passed: log.error(f" Quality gate STILL failing after fix: {issues}") # Store as draft, don't publish db = init_db() db.execute("UPDATE topics SET status = 'quality_failed' WHERE id = ?", (topic_id,)) db.commit() db.close() return None # Agent 7: Generate article image og_image = generate_article_image(topic_title, vertical) # Calculate stats word_count = len(edited.split()) reading_time = max(1, word_count // 200) slug = topic_title.lower().strip()[:80] slug = "".join(c if c.isalnum() or c in "- " else "" for c in slug) slug = slug.replace(" ", "-").strip("-") # Store article db = init_db() db.execute(""" INSERT OR REPLACE INTO articles (topic_id, vertical, title, slug, content_md, seo_title, seo_description, og_image, word_count, reading_time_minutes, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'draft') """, (topic_id, vertical, topic_title, slug, edited, seo.get("seo_title", topic_title[:65]), seo.get("seo_description", ""), og_image or "", word_count, reading_time)) db.execute("UPDATE topics SET status = 'written', article_id = last_insert_rowid() WHERE id = ?", (topic_id,)) db.commit() db.close() log.info(f"Article written for #{topic_id}: {word_count}w, {reading_time}min, format={fmt['name']}, " f"fact_checked={factcheck['verified_count']}/{factcheck['checked']}, quality=OK") return { "topic_id": topic_id, "title": topic_title, "slug": slug, "content_md": edited, "content": edited, "seo": seo, "word_count": word_count, "reading_time": reading_time, "og_image": og_image, "factcheck": factcheck, "format": fmt['name'], } # ─── Site Builder ────────────────────────────────────────────────── def build_site(vertical: str, articles: list[dict]) -> str: """Build static HTML site from articles.""" log.info(f"Building site for {vertical}...") site_dir = BASE_DIR / "sites" / vertical site_dir.mkdir(parents=True, exist_ok=True) (site_dir / "articles").mkdir(exist_ok=True) (site_dir / "assets" / "images").mkdir(parents=True, exist_ok=True) # Generate each article page for article in articles: html = _article_to_html(article, vertical) article_path = site_dir / "articles" / f"{article['slug']}.html" article_path.write_text(html) # Generate homepage homepage = _build_homepage(vertical, articles) (site_dir / "index.html").write_text(homepage) # Generate RSS feed rss = _build_rss(vertical, articles) (site_dir / "rss.xml").write_text(rss) # Generate sitemap sitemap = _build_sitemap(vertical, articles) (site_dir / "sitemap.xml").write_text(sitemap) log.info(f"Site built for {vertical}: {len(articles)} articles") return str(site_dir) def _article_to_html(article: dict, vertical: str) -> str: """Convert markdown article to full HTML page.""" # Simple markdown-to-HTML conversion (rudimentary but functional) content = article.get("content_md", article.get("content", "")) html_body = _md_to_html(content) seo_title = article.get("seo_title", article.get("title", "")) seo_desc = article.get("seo_description", "") og_image = article.get("og_image", f"/assets/images/{vertical}-default.webp") json_ld = _build_json_ld(article, vertical) return f""" {seo_title}

{article.get("title", "")}

{article.get("reading_time_minutes", 5)} min read {article.get("word_count", 0)} words
{html_body}
{_build_tag_links(article.get("keywords", []), vertical)}

Sources

{_build_sources_html(article)}
""" def _md_to_html(md: str) -> str: """Basic markdown to HTML conversion.""" import re lines = md.split("\n") html = [] in_code_block = False code_lines = [] code_lang = "" i = 0 while i < len(lines): line = lines[i] # Code blocks if line.strip().startswith("```"): if in_code_block: code = "\n".join(code_lines) html.append(f'
{_escape_html(code)}
') code_lines = [] in_code_block = False else: in_code_block = True code_lang = line.strip()[3:].strip() i += 1 continue if in_code_block: code_lines.append(line) i += 1 continue # Headers if line.startswith("### "): html.append(f"

{_inline_md(line[4:])}

") elif line.startswith("## "): html.append(f"

{_inline_md(line[3:])}

") elif line.startswith("# "): html.append(f"

{_inline_md(line[2:])}

") # Blockquotes elif line.startswith("> "): html.append(f'

{_inline_md(line[2:])}

') # Lists elif line.strip().startswith("- ") or line.strip().startswith("* "): html.append(f"
  • {_inline_md(line.strip()[2:])}
    • ") elif re.match(r"^\d+\.", line.strip()): text = re.sub(r"^\d+\.\s*", "", line.strip()) html.append(f"
  • {_inline_md(text)}
    • ") # Horizontal rule elif line.strip() in ("---", "***", "___"): html.append("
    ") # Empty line elif not line.strip(): html.append("") # Paragraph else: html.append(f"

    {_inline_md(line)}

    ") i += 1 return "\n".join(html) def _inline_md(text: str) -> str: """Convert inline markdown to HTML.""" import re # Bold text = re.sub(r"\*\*(.+?)\*\*", r"\1", text) # Italic text = re.sub(r"\*(.+?)\*", r"\1", text) # Inline code text = re.sub(r"`(.+?)`", r"\1", text) # Links text = re.sub(r"\[(.+?)\]\((.+?)\)", r'\1', text) return text def _escape_html(text: str) -> str: return text.replace("&", "&").replace("<", "<").replace(">", ">") def _build_homepage(vertical: str, articles: list[dict]) -> str: """Build the homepage for a vertical site.""" articles_html = "" for a in sorted(articles, key=lambda x: x.get("published_at", ""), reverse=True)[:20]: articles_html += f"""

    {a.get('title', '')}

    {a.get('published_at', '')} · {a.get('reading_time_minutes', 5)} min read

    {a.get('seo_description', '')}

    """ return f""" {vertical}.thetempleofdoom.com — {vertical.title()} articles, guides & insights

    {vertical.title()} Insights & Guides

    Expert articles, tutorials, and deep dives. Updated daily.

    {articles_html}
    """ def _build_rss(vertical: str, articles: list[dict]) -> str: """Build RSS 2.0 feed.""" items = "" domain = f"{vertical}.thetempleofdoom.com" for a in sorted(articles, key=lambda x: x.get("published_at", ""), reverse=True)[:20]: items += f""" {_escape_xml(a.get('title', ''))} https://{domain}/articles/{a.get('slug', '')}.html https://{domain}/articles/{a.get('slug', '')}.html {_escape_xml(a.get('seo_description', ''))} {a.get('published_at', '')} """ return f""" {vertical}.thetempleofdoom.com https://{domain} Expert {vertical} articles and insights en-us {datetime.now().isoformat()} {items} """ def _build_sitemap(vertical: str, articles: list[dict]) -> str: """Build XML sitemap.""" urls = "" domain = f"{vertical}.thetempleofdoom.com" urls += f""" https://{domain}/ daily 1.0 """ for a in articles: urls += f""" https://{domain}/articles/{a.get('slug', '')}.html {a.get('published_at', datetime.now().strftime('%Y-%m-%d'))} weekly 0.8 """ return f""" {urls} """ def _build_json_ld(article: dict, vertical: str) -> dict: return { "@context": "https://schema.org", "@type": "Article", "headline": article.get("title", ""), "description": article.get("seo_description", ""), "datePublished": article.get("published_at", ""), "author": {"@type": "Organization", "name": f"{vertical}.thetempleofdoom.com"}, "publisher": {"@type": "Organization", "name": f"{vertical}.thetempleofdoom.com"}, } def _site_json_ld(vertical: str) -> dict: return { "@context": "https://schema.org", "@type": "WebSite", "name": f"{vertical}.thetempleofdoom.com", "url": f"https://{vertical}.thetempleofdoom.com", "description": f"Expert {vertical} articles, tutorials, and insights. Updated daily.", } def _build_tag_links(keywords: list, vertical: str) -> str: return " ".join(f'{k}' for k in (keywords or [])) def _build_sources_html(article: dict) -> str: # Extract sources from knowledge package if available return "

    Sources available in the original knowledge package.

    " def _escape_xml(text: str) -> str: return text.replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """) # ─── Deploy ───────────────────────────────────────────────────────── def deploy_site(vertical: str, site_dir: str, ct_ip: str) -> bool: """Deploy a site to its Proxmox CT.""" log.info(f"Deploying {vertical} to {ct_ip}...") # Create deployment tarball import tarfile tarball = BASE_DIR / "core" / f"{vertical}_deploy.tar.gz" with tarfile.open(tarball, "w:gz") as tar: tar.add(site_dir, arcname=vertical) try: # Push to CT result = os.system(f"scp {tarball} root@{ct_ip}:/tmp/ 2>/dev/null") if result != 0: log.error(f"scp failed for {vertical}") return False os.system(f"ssh root@{ct_ip} 'cd /tmp && tar xzf {vertical}_deploy.tar.gz && " f"cp -r {vertical}/* /var/www/html/ && systemctl restart nginx' 2>/dev/null") # Verify time.sleep(2) verify = os.popen(f"curl -s -o /dev/null -w '%{{http_code}}' http://{ct_ip}:80/").read().strip() if verify == "200": log.info(f"Deploy {vertical} successful: HTTP {verify}") return True else: log.error(f"Deploy {vertical} verify failed: HTTP {verify}") return False except Exception as e: log.error(f"Deploy {vertical} exception: {e}") return False finally: if tarball.exists(): tarball.unlink() # ─── Pipeline Runner ──────────────────────────────────────────────── def run_daily_pipeline(max_articles: int = 3): """Run the full daily publishing pipeline.""" log.info("=" * 60) log.info("DAILY PUBLISHING PIPELINE STARTING") log.info("=" * 60) run_id = None db = init_db() try: # Log the run cur = db.execute("INSERT INTO pipeline_runs (run_type, started_at) VALUES ('daily', datetime('now'))") run_id = cur.lastrowid db.commit() # Step 1: Discover trends topics = discover_trends() db.execute("UPDATE pipeline_runs SET topics_discovered = ? WHERE id = ?", (len(topics), run_id)) db.commit() if not topics: log.warning("No topics discovered. Exiting.") return # Step 2: Get approved topics (composite_score > 60, not yet published) top_topics = db.execute(""" SELECT id, title, vertical, composite_score FROM topics WHERE status = 'discovered' AND composite_score > 60 ORDER BY composite_score DESC LIMIT ? """, (max_articles,)).fetchall() articles_published = 0 vertical_articles = {} for topic_row in top_topics: topic_id, topic_title, vertical, score = topic_row # Step 3: Research kp = research_topic(topic_id, topic_title, vertical) # Step 4: Write article = write_article(topic_id, topic_title, vertical, kp) # Collect articles per vertical for site building if vertical not in vertical_articles: vertical_articles[vertical] = [] vertical_articles[vertical].append(article) articles_published += 1 log.info(f" ✓ Published: {topic_title} → {vertical}") # Step 5: Publish to live site APIs for vertical, articles in vertical_articles.items(): vinfo = VERTICALS.get(vertical, {}) ct_ip = vinfo.get("ip") if not ct_ip: log.warning(f"No CT IP for {vertical} — skipping publish") continue port = vinfo.get("port", 80) api_url = f"http://{ct_ip}:{port}/api/publish" for article in articles: try: r = requests.post(api_url, json=article, headers={"Authorization": "Bearer auto-publish-2026"}, timeout=15) if r.status_code in (200, 201): log.info(f" 📤 Published to {vertical}: {article.get('title', '')[:60]}") # Update article status in local DB aid = article.get('topic_id') if aid: db.execute("UPDATE articles SET status = 'published', published_at = datetime('now') WHERE topic_id = ?", (aid,)) db.commit() else: log.warning(f" ❌ {vertical} API returned {r.status_code}: {r.text[:100]}") except Exception as e: log.warning(f" ❌ Failed to publish to {vertical}: {e}") # Update run log db.execute(""" UPDATE pipeline_runs SET articles_published = ?, finished_at = datetime('now') WHERE id = ? """, (articles_published, run_id)) db.commit() log.info(f"Pipeline complete: {articles_published} articles published across {len(vertical_articles)} sites") except Exception as e: log.error(f"Pipeline failed: {e}", exc_info=True) if run_id: db.execute("UPDATE pipeline_runs SET errors = ?, finished_at = datetime('now') WHERE id = ?", (json.dumps([str(e)]), run_id)) db.commit() finally: db.close() # ─── CLI ──────────────────────────────────────────────────────────── if __name__ == "__main__": import argparse ap = argparse.ArgumentParser(description="Autonomous Publishing System") ap.add_argument("command", choices=["init", "discover", "research", "write", "build", "deploy", "run", "status"]) ap.add_argument("--topic-id", type=int, help="Topic ID for research/write") ap.add_argument("--vertical", type=str, help="Vertical name") ap.add_argument("--max", type=int, default=3, help="Max articles per run") args = ap.parse_args() if args.command == "init": init_db() print("✓ Database initialized") elif args.command == "discover": topics = discover_trends() for t in topics[:15]: print(f" [{t['vertical']}] {t['title'][:80]} (score: {t['composite_score']})") elif args.command == "run": run_daily_pipeline(max_articles=args.max) elif args.command == "status": db = init_db() run = db.execute("SELECT * FROM pipeline_runs ORDER BY id DESC LIMIT 1").fetchone() topics_count = db.execute("SELECT COUNT(*) FROM topics").fetchone()[0] articles_count = db.execute("SELECT COUNT(*) FROM articles").fetchone()[0] published = db.execute("SELECT COUNT(*) FROM articles WHERE status = 'published'").fetchone()[0] print(f"Last run: {run}") print(f"Topics in DB: {topics_count}") print(f"Articles: {articles_count} ({published} published)") else: print(f"Unknown command: {args.command}")