Autonomous Publishing System — full stack: orchestrator, 8 vertical sites, admin dashboard, analytics, cron pipeline
This commit is contained in:
drjones
190
analytics/analytics.py
Normal file
190
analytics/analytics.py
Normal file
@@ -0,0 +1,190 @@
|
||||
"""
|
||||
Analytics & Learning Loop
|
||||
Tracks content performance and feeds insights back into topic selection.
|
||||
"""
|
||||
import sqlite3
|
||||
import json
|
||||
import time
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Optional
|
||||
import logging
|
||||
|
||||
BASE_DIR = Path(__file__).resolve().parent.parent
|
||||
DB_PATH = BASE_DIR / "core" / "publisher.db"
|
||||
|
||||
log = logging.getLogger("analytics")
|
||||
|
||||
|
||||
def record_pageview(article_id: int, vertical: str, referrer: str = "",
|
||||
user_agent: str = "", ip_hash: str = "") -> None:
|
||||
"""Record a pageview for an article."""
|
||||
db = sqlite3.connect(str(DB_PATH))
|
||||
|
||||
# Update or insert analytics row for today
|
||||
today = datetime.now().strftime("%Y-%m-%d")
|
||||
existing = db.execute(
|
||||
"SELECT id, pageviews, unique_visitors FROM analytics WHERE article_id = ? AND recorded_at = ?",
|
||||
(article_id, today)
|
||||
).fetchone()
|
||||
|
||||
if existing:
|
||||
db.execute(
|
||||
"UPDATE analytics SET pageviews = pageviews + 1 WHERE id = ?",
|
||||
(existing[0],)
|
||||
)
|
||||
else:
|
||||
db.execute(
|
||||
"INSERT INTO analytics (article_id, vertical, pageviews, unique_visitors, recorded_at) "
|
||||
"VALUES (?, ?, 1, 1, ?)",
|
||||
(article_id, vertical, today)
|
||||
)
|
||||
|
||||
db.commit()
|
||||
db.close()
|
||||
|
||||
|
||||
def get_article_performance(days: int = 30) -> list[dict]:
|
||||
"""Get performance data for all articles in the last N days."""
|
||||
db = sqlite3.connect(str(DB_PATH))
|
||||
db.row_factory = sqlite3.Row
|
||||
|
||||
rows = db.execute("""
|
||||
SELECT a.id, a.title, a.vertical, a.slug, a.word_count,
|
||||
COALESCE(SUM(an.pageviews), 0) as total_views,
|
||||
COALESCE(SUM(an.unique_visitors), 0) as total_visitors,
|
||||
COUNT(DISTINCT an.recorded_at) as days_tracked
|
||||
FROM articles a
|
||||
LEFT JOIN analytics an ON a.id = an.article_id
|
||||
WHERE a.status = 'published'
|
||||
AND (an.recorded_at >= date('now', ?) OR an.recorded_at IS NULL)
|
||||
GROUP BY a.id
|
||||
ORDER BY total_views DESC
|
||||
""", (f"-{days} days",)).fetchall()
|
||||
|
||||
db.close()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def get_vertical_performance(days: int = 30) -> dict:
|
||||
"""Get aggregate performance per vertical."""
|
||||
db = sqlite3.connect(str(DB_PATH))
|
||||
db.row_factory = sqlite3.Row
|
||||
|
||||
rows = db.execute("""
|
||||
SELECT a.vertical,
|
||||
COUNT(DISTINCT a.id) as article_count,
|
||||
COALESCE(SUM(an.pageviews), 0) as total_views,
|
||||
COALESCE(AVG(an.pageviews), 0) as avg_views_per_article,
|
||||
AVG(a.word_count) as avg_word_count
|
||||
FROM articles a
|
||||
LEFT JOIN analytics an ON a.id = an.article_id
|
||||
WHERE a.status = 'published'
|
||||
AND (an.recorded_at >= date('now', ?) OR an.recorded_at IS NULL)
|
||||
GROUP BY a.vertical
|
||||
ORDER BY total_views DESC
|
||||
""", (f"-{days} days",)).fetchall()
|
||||
|
||||
db.close()
|
||||
return {r["vertical"]: dict(r) for r in rows}
|
||||
|
||||
|
||||
def run_learning_loop() -> dict:
|
||||
"""Nightly analysis: learn what works and update topic scoring."""
|
||||
log.info("Running learning loop...")
|
||||
db = sqlite3.connect(str(DB_PATH))
|
||||
db.row_factory = sqlite3.Row
|
||||
|
||||
insights = {}
|
||||
|
||||
for vertical in ["ai", "tech", "science", "crypto", "linux", "gaming", "diy", "guides"]:
|
||||
# Top performing articles
|
||||
top = db.execute("""
|
||||
SELECT a.title, a.word_count, COALESCE(SUM(an.pageviews), 0) as views
|
||||
FROM articles a
|
||||
LEFT JOIN analytics an ON a.id = an.article_id
|
||||
WHERE a.vertical = ? AND a.status = 'published'
|
||||
GROUP BY a.id
|
||||
ORDER BY views DESC LIMIT 5
|
||||
""", (vertical,)).fetchall()
|
||||
|
||||
# Optimal word count
|
||||
wc = db.execute("""
|
||||
SELECT AVG(a.word_count) as avg_wc
|
||||
FROM articles a
|
||||
LEFT JOIN analytics an ON a.id = an.article_id
|
||||
WHERE a.vertical = ? AND a.status = 'published'
|
||||
GROUP BY a.vertical
|
||||
""", (vertical,)).fetchone()
|
||||
|
||||
# Best headline patterns (simple analysis)
|
||||
headline_data = db.execute("""
|
||||
SELECT a.title
|
||||
FROM articles a
|
||||
LEFT JOIN analytics an ON a.id = an.article_id
|
||||
WHERE a.vertical = ? AND a.status = 'published'
|
||||
ORDER BY COALESCE(SUM(an.pageviews), 0) DESC LIMIT 3
|
||||
""", (vertical,)).fetchall()
|
||||
|
||||
insights[vertical] = {
|
||||
"top_articles": [dict(r) for r in top],
|
||||
"optimal_word_count": round(wc["avg_wc"]) if wc and wc["avg_wc"] else None,
|
||||
"top_headlines": [r["title"] for r in headline_data],
|
||||
"total_articles": db.execute(
|
||||
"SELECT COUNT(*) FROM articles WHERE vertical=? AND status='published'",
|
||||
(vertical,)
|
||||
).fetchone()[0],
|
||||
}
|
||||
|
||||
# Store to performance_learning table
|
||||
db.execute("""
|
||||
INSERT OR REPLACE INTO performance_learning (vertical, top_patterns, headline_formats,
|
||||
optimal_word_count, keyword_insights, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, datetime('now'))
|
||||
""", (
|
||||
vertical,
|
||||
json.dumps(insights[vertical]["top_articles"]),
|
||||
json.dumps(insights[vertical]["top_headlines"]),
|
||||
insights[vertical]["optimal_word_count"],
|
||||
json.dumps([]),
|
||||
))
|
||||
|
||||
db.commit()
|
||||
db.close()
|
||||
|
||||
log.info(f"Learning loop complete. Processed {len(insights)} verticals.")
|
||||
return insights
|
||||
|
||||
|
||||
def generate_topic_boost() -> dict:
|
||||
"""Generate topic scoring boosts based on learning data."""
|
||||
db = sqlite3.connect(str(DB_PATH))
|
||||
db.row_factory = sqlite3.Row
|
||||
|
||||
boosts = {}
|
||||
for vertical in ["ai", "tech", "science", "crypto", "linux", "gaming", "diy", "guides"]:
|
||||
pl = db.execute(
|
||||
"SELECT * FROM performance_learning WHERE vertical=? ORDER BY updated_at DESC LIMIT 1",
|
||||
(vertical,)
|
||||
).fetchone()
|
||||
|
||||
if pl:
|
||||
boosts[vertical] = {
|
||||
"boost": 1.0, # Default neutral
|
||||
"preferred_word_count": pl["optimal_word_count"],
|
||||
"avoid_patterns": [],
|
||||
"prefer_patterns": [],
|
||||
}
|
||||
|
||||
db.close()
|
||||
return boosts
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Running analytics learning loop...")
|
||||
insights = run_learning_loop()
|
||||
for vertical, data in insights.items():
|
||||
print(f"\n{vertical}: {data['total_articles']} articles, "
|
||||
f"optimal WC: {data['optimal_word_count']}")
|
||||
for art in data["top_articles"]:
|
||||
print(f" {art['views']:>5} views | {art['title'][:60]}")
|
||||
Reference in New Issue
Block a user