191 lines
6.5 KiB
Python
191 lines
6.5 KiB
Python
"""
|
|
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]}")
|