""" Living Site Engine — Dynamic, self-learning authority website. Each vertical gets its own instance with its own database, identity, and content. """ import os import json import sqlite3 import hashlib import time from pathlib import Path from datetime import datetime, timedelta from functools import wraps from flask import Flask, request, jsonify, render_template_string, g, abort, Response # ─── Config ──────────────────────────────────────────────────────── VERTICAL = os.environ.get("PUBLISHER_VERTICAL", "guides") DOMAIN = f"{VERTICAL}.thetempleofdoom.com" DB_PATH = Path(f"/var/lib/publisher/{VERTICAL}.db") SECRET = os.environ.get("PUBLISHER_SECRET", "auto-publish-2026") # Per-vertical identity IDENTITIES = { "ai": { "name": "AI Insights", "tagline": "Exploring machine intelligence, one breakthrough at a time.", "primary": "#7c3aed", "accent": "#a78bfa", "bg": "#0f0720", "card_bg": "#1a1030", "font_heading": "'Space Grotesk', sans-serif", "gradient": "linear-gradient(135deg, #7c3aed, #a78bfa)", "description": "Deep dives into AI, machine learning, LLMs, and the future of intelligence.", }, "tech": { "name": "Tech Frontier", "tagline": "The software, systems, and stacks shaping tomorrow.", "primary": "#2563eb", "accent": "#60a5fa", "bg": "#0a1628", "card_bg": "#112240", "font_heading": "'JetBrains Mono', monospace", "gradient": "linear-gradient(135deg, #2563eb, #06b6d4)", "description": "Programming, cloud infrastructure, DevOps, and the tools that power the internet.", }, "science": { "name": "Science Decoded", "tagline": "The universe, explained clearly.", "primary": "#059669", "accent": "#34d399", "bg": "#061a12", "card_bg": "#0d2b1e", "font_heading": "'Crimson Text', serif", "gradient": "linear-gradient(135deg, #059669, #06b6d4)", "description": "Physics, biology, chemistry, astronomy — the frontiers of human knowledge.", }, "crypto": { "name": "Crypto Compass", "tagline": "Navigate the blockchain wilderness.", "primary": "#f59e0b", "accent": "#fbbf24", "bg": "#1a1400", "card_bg": "#261f00", "font_heading": "'DM Mono', monospace", "gradient": "linear-gradient(135deg, #f59e0b, #ef4444)", "description": "Bitcoin, Ethereum, DeFi, and the technology rebuilding finance.", }, "linux": { "name": "Linux Lab", "tagline": "Open source. Open mind. Open systems.", "primary": "#f97316", "accent": "#fb923c", "bg": "#1a0e00", "card_bg": "#261500", "font_heading": "'Fira Code', monospace", "gradient": "linear-gradient(135deg, #f97316, #ef4444)", "description": "Linux, kernel hacking, system administration, and the FOSS ecosystem.", }, "gaming": { "name": "Game Layer", "tagline": "The art, tech, and culture of play.", "primary": "#ec4899", "accent": "#f472b6", "bg": "#1a0010", "card_bg": "#260018", "font_heading": "'Press Start 2P', cursive", "gradient": "linear-gradient(135deg, #ec4899, #8b5cf6)", "description": "Video games, game development, esports, hardware, and gaming culture.", }, "diy": { "name": "Maker Forge", "tagline": "Build it yourself. Build it better.", "primary": "#ef4444", "accent": "#f87171", "bg": "#1a0808", "card_bg": "#261010", "font_heading": "'Oswald', sans-serif", "gradient": "linear-gradient(135deg, #ef4444, #f97316)", "description": "3D printing, woodworking, electronics, home automation, and the maker movement.", }, "guides": { "name": "Practical Guides", "tagline": "Learn anything. Step by step.", "primary": "#0891b2", "accent": "#22d3ee", "bg": "#061a1e", "card_bg": "#0d2b30", "font_heading": "'Inter', sans-serif", "gradient": "linear-gradient(135deg, #0891b2, #06b6d4)", "description": "Clear, practical tutorials and how-to guides for everything that matters.", }, } IDENTITY = IDENTITIES.get(VERTICAL, IDENTITIES["guides"]) app = Flask(__name__) # ─── Database ────────────────────────────────────────────────────── def get_db(): if "db" not in g: Path(DB_PATH).parent.mkdir(parents=True, exist_ok=True) g.db = sqlite3.connect(str(DB_PATH)) g.db.row_factory = sqlite3.Row g.db.execute("PRAGMA journal_mode=WAL") g.db.execute("PRAGMA foreign_keys=ON") return g.db def init_db(): db = get_db() db.executescript(""" CREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, slug TEXT UNIQUE NOT NULL, content_html TEXT NOT NULL, content_md TEXT NOT NULL DEFAULT '', excerpt TEXT DEFAULT '', seo_title TEXT, seo_description TEXT, og_image TEXT, keywords TEXT DEFAULT '[]', word_count INTEGER DEFAULT 0, reading_time INTEGER DEFAULT 0, status TEXT DEFAULT 'published', published_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS pageviews ( id INTEGER PRIMARY KEY AUTOINCREMENT, path TEXT NOT NULL, article_id INTEGER, referrer TEXT DEFAULT '', ip_hash TEXT DEFAULT '', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS topics ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, score REAL DEFAULT 0, status TEXT DEFAULT 'discovered', article_id INTEGER, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS learning ( id INTEGER PRIMARY KEY AUTOINCREMENT, metric TEXT NOT NULL, value TEXT NOT NULL, recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE VIRTUAL TABLE IF NOT EXISTS articles_fts USING fts5( title, content_md, excerpt, keywords, content='articles', content_rowid='id' ); CREATE TRIGGER IF NOT EXISTS articles_ai AFTER INSERT ON articles BEGIN INSERT INTO articles_fts(rowid, title, content_md, excerpt, keywords) VALUES (new.id, new.title, new.content_md, new.excerpt, new.keywords); END; CREATE TRIGGER IF NOT EXISTS articles_ad AFTER DELETE ON articles BEGIN INSERT INTO articles_fts(articles_fts, rowid, title, content_md, excerpt, keywords) VALUES ('delete', old.id, old.title, old.content_md, old.excerpt, old.keywords); END; CREATE TRIGGER IF NOT EXISTS articles_au AFTER UPDATE ON articles BEGIN INSERT INTO articles_fts(articles_fts, rowid, title, content_md, excerpt, keywords) VALUES ('delete', old.id, old.title, old.content_md, old.excerpt, old.keywords); INSERT INTO articles_fts(rowid, title, content_md, excerpt, keywords) VALUES (new.id, new.title, new.content_md, new.excerpt, new.keywords); END; CREATE INDEX IF NOT EXISTS idx_pageviews_path ON pageviews(path); CREATE INDEX IF NOT EXISTS idx_pageviews_created ON pageviews(created_at); CREATE INDEX IF NOT EXISTS idx_articles_published ON articles(published_at); CREATE INDEX IF NOT EXISTS idx_articles_slug ON articles(slug); """) @app.teardown_appcontext def close_db(exception): db = g.pop("db", None) if db: db.close() # ─── Helpers ─────────────────────────────────────────────────────── def record_view(path, article_id=None): db = get_db() ip = request.remote_addr or "" ip_hash = hashlib.sha256(ip.encode()).hexdigest()[:16] ref = request.headers.get("Referer", "")[:500] db.execute( "INSERT INTO pageviews (path, article_id, referrer, ip_hash) VALUES (?, ?, ?, ?)", (path, article_id, ref, ip_hash) ) db.commit() def get_popular(limit=6): db = get_db() rows = db.execute(""" SELECT a.*, COUNT(p.id) as views FROM articles a LEFT JOIN pageviews p ON a.id = p.article_id WHERE a.status = 'published' GROUP BY a.id ORDER BY views DESC LIMIT ? """, (limit,)).fetchall() return [dict(r) for r in rows] def get_related(article_id, vertical=None, limit=4): db = get_db() # Simple: same vertical, different article, most recent rows = db.execute(""" SELECT * FROM articles WHERE id != ? AND status = 'published' ORDER BY published_at DESC LIMIT ? """, (article_id, limit)).fetchall() return [dict(r) for r in rows] def search_articles(query, limit=20): db = get_db() rows = db.execute(""" SELECT a.* FROM articles a JOIN articles_fts fts ON a.id = fts.rowid WHERE articles_fts MATCH ? AND a.status = 'published' ORDER BY rank LIMIT ? """, (query, limit)).fetchall() return [dict(r) for r in rows] def build_rss(): db = get_db() articles = db.execute( "SELECT * FROM articles WHERE status='published' ORDER BY published_at DESC LIMIT 20" ).fetchall() items = "" for a in articles: a = dict(a) items += f""" {_xml(a['title'])} https://{DOMAIN}/articles/{a['slug']} https://{DOMAIN}/articles/{a['slug']} {_xml(a.get('excerpt', ''))} {a.get('published_at', '')} """ return f""" {IDENTITY['name']} https://{DOMAIN} {IDENTITY['description']} en-us {datetime.now().isoformat()} {items} """ def build_sitemap_xml(): db = get_db() articles = db.execute( "SELECT slug, updated_at FROM articles WHERE status='published' ORDER BY published_at DESC" ).fetchall() urls = f""" https://{DOMAIN}/ daily 1.0 https://{DOMAIN}/search weekly 0.6 """ for a in articles: urls += f""" https://{DOMAIN}/articles/{a['slug']} {a['updated_at'][:10]} weekly 0.8 """ return f""" {urls} """ def build_json_ld(article=None): if article: return { "@context": "https://schema.org", "@type": "Article", "headline": article.get("title", ""), "description": article.get("excerpt", ""), "datePublished": article.get("published_at", ""), "author": {"@type": "Organization", "name": IDENTITY["name"]}, "publisher": {"@type": "Organization", "name": IDENTITY["name"], "logo": {"@type": "ImageObject", "url": f"https://{DOMAIN}/assets/logo.png"}}, } return { "@context": "https://schema.org", "@type": "WebSite", "name": IDENTITY["name"], "url": f"https://{DOMAIN}", "description": IDENTITY["description"], "potentialAction": { "@type": "SearchAction", "target": f"https://{DOMAIN}/search?q={{search_term_string}}", "query-input": "required name=search_term_string", }, } def _xml(s): return str(s).replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """) # ─── Routes ──────────────────────────────────────────────────────── @app.route("/") def home(): db = get_db() articles = db.execute( "SELECT * FROM articles WHERE status='published' ORDER BY published_at DESC LIMIT 20" ).fetchall() articles = [dict(a) for a in articles] popular = get_popular(6) total = db.execute("SELECT COUNT(*) FROM articles WHERE status='published'").fetchone()[0] total_views = db.execute("SELECT COUNT(*) FROM pageviews").fetchone()[0] record_view("/") return render_template_string(HOME_TEMPLATE, **IDENTITY, articles=articles, popular=popular, total=total, total_views=total_views, domain=DOMAIN, vertical=VERTICAL, json_ld=json.dumps(build_json_ld())) @app.route("/articles/") def article(slug): db = get_db() row = db.execute("SELECT * FROM articles WHERE slug=? AND status='published'", (slug,)).fetchone() if not row: abort(404) article = dict(row) related = get_related(article["id"]) record_view(f"/articles/{slug}", article["id"]) return render_template_string(ARTICLE_TEMPLATE, **IDENTITY, article=article, related=related, domain=DOMAIN, vertical=VERTICAL, json_ld=json.dumps(build_json_ld(article))) @app.route("/search") def search(): q = request.args.get("q", "").strip() results = [] if q: results = search_articles(q) record_view("/search") return render_template_string(SEARCH_TEMPLATE, **IDENTITY, query=q, results=results, domain=DOMAIN, vertical=VERTICAL) @app.route("/rss.xml") def rss(): return Response(build_rss(), mimetype="application/rss+xml") @app.route("/sitemap.xml") def sitemap(): return Response(build_sitemap_xml(), mimetype="application/xml") @app.route("/tag/") def tag_page(tag): """Aggregate all articles with a given tag.""" db = get_db() rows = db.execute(""" SELECT * FROM articles WHERE status='published' AND keywords LIKE ? ORDER BY published_at DESC LIMIT 30 """, (f"%{tag}%",)).fetchall() articles = [dict(r) for r in rows] record_view(f"/tag/{tag}") return render_template_string(TAG_TEMPLATE, **IDENTITY, tag=tag, articles=articles, domain=DOMAIN, vertical=VERTICAL) # ─── Analytics Collector ────────────────────────────────────────── @app.route("/a/ping") def analytics_ping(): """Invisible tracking pixel.""" path = request.args.get("p", "/") record_view(path) # 1x1 transparent GIF return Response( b"\x47\x49\x46\x38\x39\x61\x01\x00\x01\x00\x80\x00\x00\xff\xff\xff\x00\x00\x00\x21\xf9\x04\x01\x00\x00\x00\x00\x2c\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x02\x44\x01\x00\x3b", mimetype="image/gif", headers={"Cache-Control": "no-cache, no-store, must-revalidate"}, ) # ─── Content API (for orchestrator) ──────────────────────────────── @app.route("/api/publish", methods=["POST"]) def api_publish(): """Receive a new article from the orchestrator.""" auth = request.headers.get("Authorization", "") if auth != f"Bearer {SECRET}": return jsonify({"error": "unauthorized"}), 401 data = request.json if not data: return jsonify({"error": "no data"}), 400 db = get_db() slug = data.get("slug", "") title = data.get("title", "") content_html = data.get("content_html", data.get("content_md", "")) content_md = data.get("content_md", "") excerpt = data.get("excerpt", data.get("seo_description", "")) seo_title = data.get("seo_title", title) seo_description = data.get("seo_description", "") og_image = data.get("og_image", "") keywords = json.dumps(data.get("keywords", [])) word_count = data.get("word_count", len(content_md.split())) reading_time = data.get("reading_time", max(1, word_count // 200)) # Upsert existing = db.execute("SELECT id FROM articles WHERE slug=?", (slug,)).fetchone() if existing: db.execute(""" UPDATE articles SET title=?, content_html=?, content_md=?, excerpt=?, seo_title=?, seo_description=?, og_image=?, keywords=?, word_count=?, reading_time=?, updated_at=CURRENT_TIMESTAMP WHERE slug=? """, (title, content_html, content_md, excerpt, seo_title, seo_description, og_image, keywords, word_count, reading_time, slug)) else: db.execute(""" INSERT INTO articles (title, slug, content_html, content_md, excerpt, seo_title, seo_description, og_image, keywords, word_count, reading_time, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'published') """, (title, slug, content_html, content_md, excerpt, seo_title, seo_description, og_image, keywords, word_count, reading_time)) db.commit() return jsonify({"status": "published", "slug": slug}), 201 @app.route("/api/articles") def api_articles(): db = get_db() limit = request.args.get("limit", 50) rows = db.execute( "SELECT * FROM articles WHERE status='published' ORDER BY published_at DESC LIMIT ?", (limit,) ).fetchall() return jsonify([dict(r) for r in rows]) @app.route("/api/stats") def api_stats(): db = get_db() return jsonify({ "total_articles": db.execute("SELECT COUNT(*) FROM articles WHERE status='published'").fetchone()[0], "total_views": db.execute("SELECT COUNT(*) FROM pageviews").fetchone()[0], "today_views": db.execute( "SELECT COUNT(*) FROM pageviews WHERE created_at >= date('now')" ).fetchone()[0], "popular": get_popular(5), }) @app.route("/api/learn", methods=["POST"]) def api_learn(): """Nightly learning: analyze performance and update internal models.""" db = get_db() # Aggregate keyword performance rows = db.execute(""" SELECT a.keywords, COUNT(p.id) as views FROM articles a LEFT JOIN pageviews p ON a.id = p.article_id WHERE a.status = 'published' GROUP BY a.id ORDER BY views DESC LIMIT 20 """).fetchall() keyword_views = {} for row in rows: try: kws = json.loads(row["keywords"]) if isinstance(row["keywords"], str) else (row["keywords"] or []) except (json.JSONDecodeError, TypeError): kws = [] for kw in kws: keyword_views[kw] = keyword_views.get(kw, 0) + (row["views"] or 0) top_keywords = sorted(keyword_views.items(), key=lambda x: x[1], reverse=True)[:15] # Best performing word count range wc_row = db.execute(""" SELECT AVG(a.word_count) as avg_wc, AVG(a.reading_time) as avg_rt FROM articles a LEFT JOIN pageviews p ON a.id = p.article_id WHERE a.status = 'published' GROUP BY a.id HAVING COUNT(p.id) > 0 ORDER BY COUNT(p.id) DESC LIMIT 10 """).fetchone() # Store learning db.execute(""" INSERT OR REPLACE INTO learning (metric, value, recorded_at) VALUES ('top_keywords', ?, datetime('now')) """, (json.dumps(top_keywords),)) db.commit() return jsonify({ "status": "learned", "top_keywords": [{"keyword": kw, "views": v} for kw, v in top_keywords[:10]], "optimal_word_count": round(wc_row["avg_wc"]) if wc_row and wc_row["avg_wc"] else None, "optimal_reading_time": round(wc_row["avg_rt"]) if wc_row and wc_row["avg_rt"] else None, "total_articles_analyzed": db.execute("SELECT COUNT(*) FROM articles WHERE status='published'").fetchone()[0], }) @app.route("/health") def health(): db = get_db() count = db.execute("SELECT COUNT(*) FROM articles").fetchone()[0] return jsonify({ "status": "healthy", "vertical": VERTICAL, "articles": count, "timestamp": datetime.now().isoformat(), }) # ─── Error Handlers ────────────────────────────────────────────── @app.errorhandler(404) def not_found(e): return render_template_string(NOT_FOUND_TEMPLATE, **IDENTITY, domain=DOMAIN, vertical=VERTICAL), 404 # ─── Templates ───────────────────────────────────────────────────── HOME_TEMPLATE = """ {{ name }} — {{ tagline }}

{{ tagline }}

{{ description }}

{{ total }} articles {{ total_views }} readers Updated daily
""" ARTICLE_TEMPLATE = """ {{ article.seo_title or article.title }}

{{ article.title }}

{{ article.content_html|safe }}
{% if article.keywords %}
{% for kw in article.keywords|from_json %} {{ kw }} {% endfor %}
{% endif %}
{% if related %} {% endif %}
""" SEARCH_TEMPLATE = """ Search — {{ name }}

Search {{ name }}

{% if results %}

{{ results|length }} results for "{{ query }}"

{% for r in results %}

{{ r.title }}

{{ r.excerpt[:200] }}

{% endfor %} {% elif query %}

No results for "{{ query }}"

{% endif %}
© {{ domain }}
""" TAG_TEMPLATE = """ {{ tag }} — {{ name }}

#{{ tag }}

{{ articles|length }} article{% if articles|length != 1 %}s{% endif %}

{% for a in articles %}

{{ a.title }}

{{ a.excerpt[:180] }}

{{ a.reading_time }} min read · {{ a.published_at[:10] }}
{% endfor %} {% if not articles %}

No articles tagged "{{ tag }}" yet.

{% endif %}
© {{ domain }}
""" NOT_FOUND_TEMPLATE = """ 404 — {{ name }}

404

This page doesn't exist yet. Maybe it will tomorrow.

← Back to {{ name }}
""" # ─── Jinja filter ────────────────────────────────────────────────── @app.template_filter("from_json") def from_json_filter(s): try: return json.loads(s) except (json.JSONDecodeError, TypeError): return [] # ─── Main ────────────────────────────────────────────────────────── if __name__ == "__main__": import argparse ap = argparse.ArgumentParser() ap.add_argument("--port", type=int, default=5000) ap.add_argument("--host", type=str, default="0.0.0.0") args = ap.parse_args() with app.app_context(): init_db() print(f"🚀 {IDENTITY['name']} → http://{args.host}:{args.port}") app.run(host=args.host, port=args.port, debug=False)