""" Autonomous Publishing System — Admin Dashboard Flask app for monitoring and controlling the publishing pipeline. """ import os import sys import json import sqlite3 from pathlib import Path from datetime import datetime, timedelta from flask import Flask, render_template_string, jsonify, request, redirect, url_for BASE_DIR = Path(__file__).resolve().parent.parent DB_PATH = BASE_DIR / "core" / "publisher.db" sys.path.insert(0, str(BASE_DIR / "core")) app = Flask(__name__) app.config["SECRET_KEY"] = os.environ.get("DASHBOARD_SECRET", "auto-publisher-dev") VERTICALS = ["ai", "tech", "science", "crypto", "linux", "gaming", "diy", "guides"] def get_db(): db = sqlite3.connect(str(DB_PATH)) db.row_factory = sqlite3.Row return db # ─── Templates ────────────────────────────────────────────────────── DASHBOARD_HTML = """ Auto Publisher — Admin Dashboard

🤖 Auto Publisher

Autonomous Publishing System — Admin Dashboard

Topics Discovered
{{ stats.topics }}
Articles Written
{{ stats.articles }}
Published
{{ stats.published }}
Last Run
{{ stats.last_run or 'Never' }}
Total Pageviews
{{ stats.total_views }}
Avg Score
{{ stats.avg_score }}

📡 Sites

{% for v in verticals %}
{{ v }}.thetempleofdoom.com
{{ '● Live' if v in live_sites else '○ Pending' }}
{% endfor %}

🎮 Controls

📋 Recent Topics

{% for t in topics %} {% endfor %}
TitleVerticalScoreStatusDiscovered
{{ t.title[:80] }} {{ t.vertical }} {{ t.composite_score }} {{ t.status }} {{ t.created_at[:10] if t.created_at else '-' }}

📝 Recent Articles

{% for a in articles %} {% endfor %}
TitleVerticalWordsStatusCreated
{{ a.title[:80] if a.title else 'Untitled' }} {{ a.vertical }} {{ a.word_count }} {{ a.status }} {{ a.created_at[:10] if a.created_at else '-' }}

📜 Pipeline Log

{% for entry in log_lines %}
[{{ entry.time }}] {{ entry.msg }}
{% endfor %}
""" @app.route("/") def index(): db = get_db() stats = { "topics": db.execute("SELECT COUNT(*) FROM topics").fetchone()[0], "articles": db.execute("SELECT COUNT(*) FROM articles").fetchone()[0], "published": db.execute("SELECT COUNT(*) FROM articles WHERE status='published'").fetchone()[0], "last_run": None, "total_views": db.execute("SELECT COALESCE(SUM(pageviews), 0) FROM analytics").fetchone()[0], "avg_score": round(db.execute("SELECT COALESCE(AVG(composite_score), 0) FROM topics WHERE composite_score > 0").fetchone()[0], 1), } last_run = db.execute("SELECT started_at FROM pipeline_runs ORDER BY id DESC LIMIT 1").fetchone() if last_run: stats["last_run"] = last_run[0] topics = db.execute("SELECT * FROM topics ORDER BY created_at DESC LIMIT 20").fetchall() articles = db.execute("SELECT * FROM articles ORDER BY created_at DESC LIMIT 20").fetchall() # Read log log_path = BASE_DIR / "core" / "orchestrator.log" log_lines = [] if log_path.exists(): for line in log_path.read_text().split("\n")[-30:]: if not line.strip(): continue level = "" if "ERROR" in line: level = "error" elif "✓" in line or "successful" in line.lower(): level = "success" log_lines.append({ "time": line[:19] if len(line) > 19 else "", "msg": line, "level": level, }) live_sites = ["ai", "tech"] # Will be dynamically checked return render_template_string( DASHBOARD_HTML, stats=stats, topics=topics, articles=articles, log_lines=log_lines, verticals=VERTICALS, live_sites=live_sites, ) @app.route("/api/run", methods=["POST"]) def api_run(): """Trigger a pipeline run.""" from orchestrator import run_daily_pipeline import threading def _run(): run_daily_pipeline(max_articles=3) t = threading.Thread(target=_run, daemon=True) t.start() return jsonify({"status": "started", "message": "Pipeline running in background"}) @app.route("/api/discover", methods=["POST"]) def api_discover(): """Trigger trend discovery.""" from orchestrator import discover_trends import threading def _run(): discover_trends() t = threading.Thread(target=_run, daemon=True) t.start() return jsonify({"status": "started", "message": "Trend discovery running"}) @app.route("/api/stats") def api_stats(): db = get_db() return jsonify({ "topics_total": db.execute("SELECT COUNT(*) FROM topics").fetchone()[0], "articles_total": db.execute("SELECT COUNT(*) FROM articles").fetchone()[0], "published": db.execute("SELECT COUNT(*) FROM articles WHERE status='published'").fetchone()[0], "by_vertical": { v: db.execute("SELECT COUNT(*) FROM topics WHERE vertical=?", (v,)).fetchone()[0] for v in VERTICALS }, "pipeline_runs": db.execute("SELECT COUNT(*) FROM pipeline_runs").fetchone()[0], }) @app.route("/health") def health(): return jsonify({"status": "healthy", "timestamp": datetime.now().isoformat()}) if __name__ == "__main__": import argparse ap = argparse.ArgumentParser() ap.add_argument("--port", type=int, default=5106) ap.add_argument("--host", type=str, default="127.0.0.1") args = ap.parse_args() print(f"Admin Dashboard → http://{args.host}:{args.port}") app.run(host=args.host, port=args.port, debug=False)