Analytics & Monitoring

ai-web-feeds analytics overview

ai-web-feeds analytics distributions [--limit N]

ai-web-feeds analytics quality

ai-web-feeds analytics performance [--days N]

ai-web-feeds analytics content

ai-web-feeds analytics trends [--days N]

ai-web-feeds analytics health <feed-id>

ai-web-feeds analytics contributors [--limit N]

ai-web-feeds analytics report [--output FILE]

#!/bin/bash
# daily-health-check.sh

# Fetch all verified feeds
ai-web-feeds fetch all --verified-only

# Generate health report
ai-web-feeds analytics overview > daily-overview.txt
ai-web-feeds analytics performance --days 1 > daily-performance.txt

# Check for critical feeds
ai-web-feeds analytics quality | grep -i "poor\|critical"

#!/bin/bash
# weekly-analytics.sh

DATE=$(date +%Y-%m-%d)

# Generate comprehensive report
ai-web-feeds analytics report --output "reports/analytics-${DATE}.json"

# View trends
ai-web-feeds analytics trends --days 7
ai-web-feeds analytics distributions

# Top contributors
ai-web-feeds analytics contributors --limit 10

#!/bin/bash
# check-failures.sh

# Get performance stats
STATS=$(ai-web-feeds analytics performance --days 1)

# Extract success rate
SUCCESS_RATE=$(echo "$STATS" | grep "Success Rate" | awk '{print $3}' | tr -d '%')

if (( $(echo "$SUCCESS_RATE < 90" | bc -l) )); then
    echo "WARNING: Success rate below 90%: ${SUCCESS_RATE}%"
    # Send alert (email, Slack, etc.)
fi

from ai_web_feeds.analytics import FeedAnalytics
from ai_web_feeds.storage import DatabaseManager

# Initialize
db = DatabaseManager("sqlite:///data/aiwebfeeds.db")
analytics = FeedAnalytics(db.get_session())

# Custom query: Find all feeds with quality < 0.5
feeds = db.get_all_feed_sources()
low_quality = [
    f for f in feeds
    if f.quality_score and f.quality_score < 0.5
]

print(f"Found {len(low_quality)} low quality feeds:")
for feed in low_quality:
    print(f"  - {feed.title}: {feed.quality_score:.2f}")

# Generate custom report
report = analytics.generate_full_report()

# Analyze specific dimension
quality_by_type = {}
for feed in feeds:
    if feed.source_type and feed.quality_score:
        type_name = feed.source_type.value
        if type_name not in quality_by_type:
            quality_by_type[type_name] = []
        quality_by_type[type_name].append(feed.quality_score)

# Calculate averages
for source_type, scores in quality_by_type.items():
    avg = sum(scores) / len(scores)
    print(f"{source_type}: {avg:.3f}")

from sqlalchemy import select, func
from ai_web_feeds.models import FeedSource, FeedItem

# Get feeds with most items
stmt = (
    select(FeedSource.id, FeedSource.title, func.count(FeedItem.id))
    .join(FeedItem)
    .group_by(FeedSource.id)
    .order_by(func.count(FeedItem.id).desc())
    .limit(10)
)

results = session.exec(stmt).all()
for feed_id, title, count in results:
    print(f"{title}: {count} items")

{
  "generated_at": "2025-10-15T12:00:00Z",
  "overview": {
    "totals": {
      "feeds": 150,
      "items": 5000,
      "topics": 25
    },
    "feed_status": {
      "verified": 120,
      "active": 100,
      "inactive": 5
    }
  },
  "quality": {
    "average_quality": 0.85,
    "median_quality": 0.87
  }
}

import csv
from ai_web_feeds.storage import DatabaseManager

db = DatabaseManager()
feeds = db.get_all_feed_sources()

with open('feeds-export.csv', 'w', newline='') as f:
    writer = csv.writer(f)
    writer.writerow(['ID', 'Title', 'Type', 'Quality', 'Verified'])

    for feed in feeds:
        writer.writerow([
            feed.id,
            feed.title,
            feed.source_type.value if feed.source_type else '',
            feed.quality_score or '',
            feed.verified
        ])

import json
from datetime import datetime

def export_metrics():
    analytics = FeedAnalytics(session)
    stats = analytics.get_overview_stats()

    metrics = {
        "timestamp": datetime.utcnow().isoformat(),
        "feeds_total": stats["totals"]["feeds"],
        "feeds_active": stats["feed_status"]["active"],
        "items_24h": stats["recent_activity_24h"]["new_items"]
    }

    with open('/var/lib/grafana/metrics/ai-web-feeds.json', 'w') as f:
        json.dump(metrics, f)

from prometheus_client import Gauge, generate_latest

feeds_total = Gauge('ai_web_feeds_total', 'Total number of feeds')
feeds_active = Gauge('ai_web_feeds_active', 'Number of active feeds')

def update_metrics():
    stats = analytics.get_overview_stats()
    feeds_total.set(stats["totals"]["feeds"])
    feeds_active.set(stats["feed_status"]["active"])