1503 lines
61 KiB
Python
1503 lines
61 KiB
Python
"""
|
|
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 (fallback)
|
|
OLLAMA_SHADOW = "http://10.30.20.128:11434" # RTX 4080 SUPER, qwen3.8:latest (primary)
|
|
|
|
# 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.8:latest", host: str = OLLAMA_SHADOW,
|
|
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,
|
|
"think": False,
|
|
"options": {"temperature": temperature, "num_predict": max_tokens, "num_ctx": 8192}
|
|
}
|
|
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_SHADOW, OLLAMA_GAMINGPC]))
|
|
for attempt in range(retries):
|
|
for h in hosts:
|
|
try:
|
|
r = requests.post(f"{h}/api/chat", json=payload, timeout=600,
|
|
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.
|
|
# WARNING: 'thinking' is chain-of-thought reasoning, NOT article text.
|
|
# Only fall back to it for JSON/short tasks, never long-form prose.
|
|
if not content:
|
|
content = result["message"].get("thinking", "")
|
|
if content:
|
|
log.warning(f"LLM {model} returned empty content — fell back to 'thinking' field ({len(content)} chars). "
|
|
f"Verify this is real output, not chain-of-thought.")
|
|
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.8:latest", host: str = OLLAMA_SHADOW,
|
|
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="qwen3.8:latest", host=OLLAMA_SHADOW,
|
|
system=system, temperature=0.3, max_tokens=4096)
|
|
|
|
def call_qwen():
|
|
return llm_chat(prompt, model="qwen3.8:latest", host=OLLAMA_SHADOW,
|
|
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="qwen3.8:latest", host=OLLAMA_SHADOW, 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="qwen3.8:latest", host=OLLAMA_SHADOW,
|
|
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="qwen3.8:latest", host=OLLAMA_SHADOW,
|
|
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: <reason>".""",
|
|
model="qwen3.8:latest",
|
|
host=OLLAMA_SHADOW,
|
|
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="qwen3.8:latest", host=OLLAMA_SHADOW, 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="qwen3.8:latest", host=OLLAMA_SHADOW,
|
|
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="qwen3.8:latest", host=OLLAMA_SHADOW, 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="qwen3.8:latest", host=OLLAMA_SHADOW, 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="qwen3.8:latest", host=OLLAMA_SHADOW, 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"""<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
|
<title>{seo_title}</title>
|
|
<meta name="description" content="{seo_desc}">
|
|
<meta property="og:title" content="{seo_title}">
|
|
<meta property="og:description" content="{seo_desc}">
|
|
<meta property="og:image" content="{og_image}">
|
|
<meta property="og:type" content="article">
|
|
<meta name="twitter:card" content="summary_large_image">
|
|
<link rel="canonical" href="https://{vertical}.thetempleofdoom.com/articles/{article.get('slug', '')}">
|
|
<script type="application/ld+json">{json.dumps(json_ld)}</script>
|
|
<link rel="stylesheet" href="/assets/style.css">
|
|
<link href="/rss.xml" rel="alternate" type="application/rss+xml" title="{vertical}.thetempleofdoom.com">
|
|
</head>
|
|
<body>
|
|
<header>
|
|
<nav>
|
|
<a href="/" class="logo">{vertical}.thetempleofdoom.com</a>
|
|
<div class="nav-links">
|
|
<a href="/">Home</a>
|
|
<a href="/articles/">Articles</a>
|
|
<a href="/about/">About</a>
|
|
</div>
|
|
</nav>
|
|
</header>
|
|
<main>
|
|
<article>
|
|
<header class="article-header">
|
|
<h1>{article.get("title", "")}</h1>
|
|
<div class="meta">
|
|
<time>{article.get("published_at", "")}</time>
|
|
<span>{article.get("reading_time_minutes", 5)} min read</span>
|
|
<span>{article.get("word_count", 0)} words</span>
|
|
</div>
|
|
</header>
|
|
<div class="article-content">
|
|
{html_body}
|
|
</div>
|
|
<footer class="article-footer">
|
|
<div class="tags">
|
|
{_build_tag_links(article.get("keywords", []), vertical)}
|
|
</div>
|
|
<div class="sources">
|
|
<h3>Sources</h3>
|
|
{_build_sources_html(article)}
|
|
</div>
|
|
</footer>
|
|
</article>
|
|
<aside class="related">
|
|
<h3>Related Articles</h3>
|
|
<!-- Dynamically populated by site builder -->
|
|
</aside>
|
|
</main>
|
|
<footer class="site-footer">
|
|
<p>© {datetime.now().year} {vertical}.thetempleofdoom.com — Built by AI, curated for humans.</p>
|
|
<nav>
|
|
<a href="/rss.xml">RSS</a>
|
|
<a href="/sitemap.xml">Sitemap</a>
|
|
<a href="/privacy/">Privacy</a>
|
|
</nav>
|
|
</footer>
|
|
</body>
|
|
</html>"""
|
|
|
|
|
|
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'<pre><code class="language-{code_lang}">{_escape_html(code)}</code></pre>')
|
|
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"<h3>{_inline_md(line[4:])}</h3>")
|
|
elif line.startswith("## "):
|
|
html.append(f"<h2>{_inline_md(line[3:])}</h2>")
|
|
elif line.startswith("# "):
|
|
html.append(f"<h1>{_inline_md(line[2:])}</h1>")
|
|
|
|
# Blockquotes
|
|
elif line.startswith("> "):
|
|
html.append(f'<blockquote><p>{_inline_md(line[2:])}</p></blockquote>')
|
|
|
|
# Lists
|
|
elif line.strip().startswith("- ") or line.strip().startswith("* "):
|
|
html.append(f"<li>{_inline_md(line.strip()[2:])}</li>")
|
|
elif re.match(r"^\d+\.", line.strip()):
|
|
text = re.sub(r"^\d+\.\s*", "", line.strip())
|
|
html.append(f"<li>{_inline_md(text)}</li>")
|
|
|
|
# Horizontal rule
|
|
elif line.strip() in ("---", "***", "___"):
|
|
html.append("<hr>")
|
|
|
|
# Empty line
|
|
elif not line.strip():
|
|
html.append("")
|
|
|
|
# Paragraph
|
|
else:
|
|
html.append(f"<p>{_inline_md(line)}</p>")
|
|
|
|
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"<strong>\1</strong>", text)
|
|
# Italic
|
|
text = re.sub(r"\*(.+?)\*", r"<em>\1</em>", text)
|
|
# Inline code
|
|
text = re.sub(r"`(.+?)`", r"<code>\1</code>", text)
|
|
# Links
|
|
text = re.sub(r"\[(.+?)\]\((.+?)\)", r'<a href="\2">\1</a>', 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"""
|
|
<article class="card">
|
|
<h2><a href="/articles/{a.get('slug', '')}.html">{a.get('title', '')}</a></h2>
|
|
<p class="meta">{a.get('published_at', '')} · {a.get('reading_time_minutes', 5)} min read</p>
|
|
<p>{a.get('seo_description', '')}</p>
|
|
</article>"""
|
|
|
|
return f"""<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
|
<title>{vertical}.thetempleofdoom.com — {vertical.title()} articles, guides & insights</title>
|
|
<meta name="description" content="Expert {vertical} articles, tutorials, and insights. Updated daily with fresh, useful content.">
|
|
<link rel="stylesheet" href="/assets/style.css">
|
|
<link href="/rss.xml" rel="alternate" type="application/rss+xml" title="{vertical}.thetempleofdoom.com">
|
|
<script type="application/ld+json">{json.dumps(_site_json_ld(vertical))}</script>
|
|
</head>
|
|
<body>
|
|
<header>
|
|
<nav>
|
|
<a href="/" class="logo">{vertical}.thetempleofdoom.com</a>
|
|
<div class="nav-links">
|
|
<a href="/">Home</a>
|
|
<a href="/articles/">Articles</a>
|
|
<a href="/about/">About</a>
|
|
<a href="/rss.xml" class="rss-link">RSS</a>
|
|
</div>
|
|
</nav>
|
|
</header>
|
|
<main>
|
|
<section class="hero">
|
|
<h1>{vertical.title()} Insights & Guides</h1>
|
|
<p>Expert articles, tutorials, and deep dives. Updated daily.</p>
|
|
</section>
|
|
<section class="articles-grid">
|
|
{articles_html}
|
|
</section>
|
|
</main>
|
|
<footer class="site-footer">
|
|
<p>© {datetime.now().year} {vertical}.thetempleofdoom.com</p>
|
|
<nav>
|
|
<a href="/rss.xml">RSS</a>
|
|
<a href="/sitemap.xml">Sitemap</a>
|
|
<a href="/privacy/">Privacy</a>
|
|
</nav>
|
|
</footer>
|
|
</body>
|
|
</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"""
|
|
<item>
|
|
<title>{_escape_xml(a.get('title', ''))}</title>
|
|
<link>https://{domain}/articles/{a.get('slug', '')}.html</link>
|
|
<guid>https://{domain}/articles/{a.get('slug', '')}.html</guid>
|
|
<description>{_escape_xml(a.get('seo_description', ''))}</description>
|
|
<pubDate>{a.get('published_at', '')}</pubDate>
|
|
</item>"""
|
|
|
|
return f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<rss version="2.0" xmlns:atom="http://www.w3.org/2005/Atom">
|
|
<channel>
|
|
<title>{vertical}.thetempleofdoom.com</title>
|
|
<link>https://{domain}</link>
|
|
<description>Expert {vertical} articles and insights</description>
|
|
<language>en-us</language>
|
|
<lastBuildDate>{datetime.now().isoformat()}</lastBuildDate>
|
|
<atom:link href="https://{domain}/rss.xml" rel="self" type="application/rss+xml"/>
|
|
{items}
|
|
</channel>
|
|
</rss>"""
|
|
|
|
|
|
def _build_sitemap(vertical: str, articles: list[dict]) -> str:
|
|
"""Build XML sitemap."""
|
|
urls = ""
|
|
domain = f"{vertical}.thetempleofdoom.com"
|
|
urls += f"""
|
|
<url>
|
|
<loc>https://{domain}/</loc>
|
|
<changefreq>daily</changefreq>
|
|
<priority>1.0</priority>
|
|
</url>"""
|
|
for a in articles:
|
|
urls += f"""
|
|
<url>
|
|
<loc>https://{domain}/articles/{a.get('slug', '')}.html</loc>
|
|
<lastmod>{a.get('published_at', datetime.now().strftime('%Y-%m-%d'))}</lastmod>
|
|
<changefreq>weekly</changefreq>
|
|
<priority>0.8</priority>
|
|
</url>"""
|
|
return f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">{urls}
|
|
</urlset>"""
|
|
|
|
|
|
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'<a href="/tag/{k.lower().replace(" ", "-")}" class="tag">{k}</a>' for k in (keywords or []))
|
|
|
|
|
|
def _build_sources_html(article: dict) -> str:
|
|
# Extract sources from knowledge package if available
|
|
return "<p>Sources available in the original knowledge package.</p>"
|
|
|
|
|
|
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}")
|