Implementation Details

Feed URL → AdvancedFeedFetcher → FeedMetadata + Items
                                        ↓
                                  DatabaseManager
                                        ↓
                                  FeedAnalytics
                                        ↓
                                  CLI Commands

from ai_web_feeds.fetcher import AdvancedFeedFetcher
from ai_web_feeds.storage import DatabaseManager

# Initialize
db = DatabaseManager("sqlite:///data/aiwebfeeds.db")
fetcher = AdvancedFeedFetcher()

# Fetch feed
fetch_log, metadata, items = await fetcher.fetch_feed(
    "https://example.com/feed.xml"
)

# Access quality scores
print(f"Completeness: {metadata.completeness_score:.2f}")
print(f"Richness: {metadata.richness_score:.2f}")
print(f"Structure: {metadata.structure_score:.2f}")

# Access metadata
print(f"Update frequency: {metadata.estimated_update_frequency}")
print(f"Total items: {metadata.total_items}")
print(f"Found {len(items)} items")

# Save to database
session = db.get_session()
session.add(fetch_log)
session.commit()

# Use ETag and Last-Modified from previous fetch
fetch_log, metadata, items = await fetcher.fetch_feed(
    url="https://example.com/feed.xml",
    etag="33a64df551425fcc55e4d42a148795d9f25f89d4",
    last_modified="Wed, 15 Nov 2023 12:00:00 GMT"
)

# Returns 304 Not Modified if feed hasn't changed
if fetch_log.status_code == 304:
    print("Feed unchanged")

# Automatic retries (configured via tenacity)
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def fetch_with_retry(url):
    # Will retry up to 3 times
    # Waits 2s, 4s, 8s between attempts
    pass

# Export everything to JSON
report = analytics.generate_full_report()

# Save to file
import json
with open("analytics.json", "w") as f:
    json.dump(report, f, indent=2)

# Report includes all 8 analytics views

class FeedFetchLog(SQLModel, table=True):
    # ... existing fields ...

    # Enhanced usage of extra_data
    extra_data: Optional[Dict[str, Any]] = Field(
        default=None,
        sa_column=Column(JSON)
    )
    # Now stores:
    # - Complete HTTP headers
    # - Detailed error information
    # - Item statistics
    # - Quality scores
    # - Extension metadata

class FeedItem(SQLModel, table=True):
    # ... existing fields ...

    # Enhanced usage of extra_data
    extra_data: Optional[Dict[str, Any]] = Field(
        default=None,
        sa_column=Column(JSON)
    )
    # Now stores:
    # - Extension metadata (iTunes, Media RSS, etc.)
    # - Multiple categories
    # - Enclosure metadata
    # - Author details

# Store from previous fetch
etag = fetch_log.etag
last_modified = fetch_log.last_modified

# Use in next fetch
new_log, metadata, items = await fetcher.fetch_feed(
    url=feed_url,
    etag=etag,
    last_modified=last_modified
)

# Server returns 304 Not Modified if unchanged
if new_log.status_code == 304:
    # No processing needed
    return

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential
)

@retry(
    stop=stop_after_attempt(3),  # Max 3 attempts
    wait=wait_exponential(
        multiplier=1,
        min=2,    # Wait 2s after first failure
        max=10    # Wait max 10s
    )
)
async def fetch_with_retry(url):
    # Automatic retry on failure
    pass

# Configurable timeout (default 30s)
fetcher = AdvancedFeedFetcher(timeout=30.0)

# Per-request timeout
fetch_log, metadata, items = await fetcher.fetch_feed(
    url=feed_url,
    timeout=60.0  # Override for slow feed
)

packages/ai_web_feeds/src/ai_web_feeds/
├── fetcher.py          # AdvancedFeedFetcher class
│   ├── FeedMetadata    # Metadata container (100+ fields)
│   ├── fetch_feed()    # Main fetch method
│   ├── _extract_*()    # Extraction helpers
│   └── _calculate_*()  # Quality scoring
│
├── analytics.py        # FeedAnalytics class
│   ├── get_overview_stats()
│   ├── get_*_distribution()
│   ├── get_quality_metrics()
│   ├── get_fetch_performance_stats()
│   ├── get_content_statistics()
│   ├── get_publishing_trends()
│   ├── get_feed_health_report()
│   ├── get_top_contributors()
│   └── generate_full_report()
│
apps/cli/ai_web_feeds/cli/commands/
├── fetch.py            # Fetch CLI commands
│   ├── fetch_one()     # Single feed fetch
│   └── fetch_all()     # Batch fetch
│
└── analytics.py        # Analytics CLI commands
    ├── show_overview()
    ├── show_distributions()
    ├── show_quality()
    ├── show_performance()
    ├── show_content()
    ├── show_trends()
    ├── show_health()
    ├── show_contributors()
    └── generate_report()