Building NewsInsight: The Backend - Part B

# Structured logging with visual indicators
print(f"📡 Fetching from NewsAPI: {topic}")
print(f"📰 NewsAPI returned {len(articles)} articles")
print(f"⚡ Cache hit for '{topic}' - returning {len(cached_result['articles'])} cached articles")
print(f"🔄 Cache miss for '{topic}' - searching database")
print(f"❌ NewsAPI fetch failed: {e}")
print(f"✅ Stored article: {doc_id}")

# AWS client initialization with fallback
try:
    aws_access_key = os.getenv("AWS_ACCESS_KEY_ID")
    aws_secret_key = os.getenv("AWS_SECRET_ACCESS_KEY")
    
    if aws_access_key and aws_secret_key:
        session = boto3.Session(
            aws_access_key_id=aws_access_key,
            aws_secret_access_key=aws_secret_key,
            region_name=AWS_REGION
        )
        print("🔑 Using AWS credentials from environment variables")
    else:
        session = boto3.Session(region_name=AWS_REGION)
        print("🔑 Using default AWS profile")
        
except Exception as e:
    print(f"⚠️ AWS initialization failed: {e}")
    if not DEBUG_MODE:
        table = None
        s3 = None
        bedrock = None

def search_articles_ddb(topic: Optional[str] = None, limit: int = 6) -> List[Dict]:
    if not table:
        print("⚠️ DynamoDB table not available")
        return []  # Return empty list, not error
    
    try:
        items = []
        resp = table.scan(Limit=200)
        items.extend(resp.get("Items", []) or [])
        # ... processing
        return result
    except Exception as e:
        print(f"❌ DDB scan error: {e}")
        return []  # Graceful degradation

@app.exception_handler(404)
async def not_found_handler(request, exc):
    return {"error": "Not found", "detail": str(exc)}

@app.exception_handler(500)
async def internal_error_handler(request, exc):
    return {"error": "Internal server error", "detail": "An unexpected error occurred"}

# Global cache state
_search_cache = {}
_cache_timestamp = datetime.utcnow()
_popular_topics = ["technology", "politics", "business", "science", 
                   "health", "economy", "AI", "climate", "market", "innovation"]

def get_cache_key(topic: str, limit: int) -> str:
    """Generate cache key for search results"""
    # Key becomes "ai:6" or "ai:12"
    return f"{topic.lower().strip()}:{limit}"

def is_cache_valid(cache_entry: dict) -> bool:
    """Check if cache entry is still valid (24 hours)"""
    if not cache_entry:
        return False
    
    cache_time = datetime.fromisoformat(cache_entry.get('timestamp', ''))
    age_hours = (datetime.utcnow() - cache_time).total_seconds() / 3600
    return age_hours < 24  # Cache valid for 24 hours

def clear_search_cache():
    """Clear the search cache when new articles are added"""
    global _search_cache, _cache_timestamp
    _search_cache = {}
    _cache_timestamp = datetime.utcnow()
    print("🗑️ Search cache cleared")

def should_prefetch_topic(topic: str) -> bool:
    """Determine if topic should be prefetched"""
    if not topic:
        return False
    
    topic_lower = topic.lower().strip()
    
    # Always prefetch popular topics
    if topic_lower in _popular_topics:
        return True
    
    # Prefetch if topic contains popular keywords
    return any(popular in topic_lower for popular in _popular_topics)

@app.post("/api/articles/prefetch")
async def prefetch_popular_topics():
    """Prefetch and cache popular topics for faster searches"""
    prefetched = []
    
    for topic in _popular_topics:
        articles = search_articles_ddb(topic, 6, use_cache=True)
        prefetched.append({"topic": topic, "cached_articles": len(articles)})
    
    return {
        "message": f"Prefetched {len(prefetched)} popular topics",
        "topics": prefetched,
        "cache_size": len(_search_cache)
    }

class ContentFilter:
    def __init__(self, session: boto3.Session):
        self.ddb = session.resource("dynamodb")
        self.blacklist_table = self.ddb.Table("content_blacklist")
        
        # Content quality thresholds
        self.MIN_WORDS = 200
        self.MAX_WORDS = 10000
        self.MIN_QUALITY_SCORE = 70
        
        # Known low-quality indicators
        self.SPAM_KEYWORDS = [
            "click here", "limited time", "act now", "free trial",
            "make money", "work from home", "get rich", "miracle cure"
        ]
        
        self.AD_INDICATORS = [
            "sponsored", "advertisement", "promoted", "paid content",
            "affiliate", "partner content", "brand story"
        ]

def _check_article_age(self, article: Dict) -> Tuple[bool, str]:
    """Check if article is within acceptable age limit (2 days)"""
    MAX_AGE_DAYS = 2
    
    article_date_str = article.get("date") or article.get("publishedAt")
    
    if not article_date_str:
        return False, "No publication date found"
    
    article_date = self._parse_article_date(article_date_str)
    if not article_date:
        return False, f"Invalid date format: {article_date_str}"
    
    age_delta = datetime.utcnow() - article_date
    age_days = age_delta.total_seconds() / (24 * 3600)
    
    if age_days > MAX_AGE_DAYS:
        return False, f"Too old: {age_days:.1f} days (max: {MAX_AGE_DAYS} days)"
    
    return True, f"Recent: {age_days:.1f} days old"

# DynamoDB table structure
{
    "type": "source",      # HASH key
    "value": "buzzfeed",   # RANGE key
    "reason": "Entertainment/satire",
    "added_date": "2024-01-15T10:30:00Z",
    "added_by": "system"
}

INITIAL_BLACKLIST = {
    "sources": [
        "buzzfeed", "clickhole", "theonion", "babylonbee",  # Satire
        "infowars", "breitbart", "naturalnews",  # Low credibility
        "taboola", "outbrain", "revcontent"  # Ad networks
    ],
    "domains": [
        "ads.yahoo.com", "googleads.com", "doubleclick.net",
        "facebook.com/tr", "analytics.google.com"
    ],
    "keywords": [
        "sponsored content", "paid promotion", "advertisement"
    ]
}

def ai_classify_content(self, article: Dict, bedrock_client) -> Dict:
    """AI-powered content classification using AWS Bedrock"""
    
    classification_prompt = f"""
    Analyze this article and provide a JSON response:
    
    {{
        "category": "news_article|advertisement|opinion_piece|press_release|clickbait|spam",
        "quality_score": 0-100,
        "credibility_indicators": {{
            "has_sources": true/false,
            "has_quotes": true/false,
            "factual_tone": true/false,
            "proper_attribution": true/false
        }},
        "content_flags": ["promotional_language", "sensationalized_headline", ...],
        "recommendation": "accept|review|reject",
        "reasoning": "Brief explanation"
    }}
    
    Article Title: {article.get('headline', '')}
    Article Content: {content[:2000]}
    """

def should_store_article(self, classification: Dict) -> Tuple[str, str]:
    """Determine storage destination based on classification"""
    
    quality_score = classification.get("quality_score", 0)
    category = classification.get("category", "unknown")
    recommendation = classification.get("recommendation", "review")
    
    # High quality - store in main database
    if quality_score >= 70 and category == "news_article" and recommendation == "accept":
        return "news_metadata", "High quality news article"
    
    # Medium quality - queue for review
    elif quality_score >= 50 and recommendation in ["accept", "review"]:
        return "content_review_queue", "Needs human review"
    
    # Low quality - reject
    else:
        return "content_rejected", f"Low quality: {classification.get('reasoning')}"

# In backend.py - Content filter initialization
content_filter = None
if ContentFilter:
    try:
        content_filter = ContentFilter(session)
        print("✅ Content filtering system initialized")
    except Exception as e:
        print(f"⚠️ Content filter initialization failed: {e}")
        print("   Continuing without content filtering...")

def ingest_topic_with_filtering(topic: str, content_filter: ContentFilter) -> Dict[str, int]:
    """Enhanced ingestion with comprehensive content filtering"""
    
    stats = {
        "fetched": 0,
        "age_rejected": 0,
        "word_count_rejected": 0,
        "blacklist_rejected": 0,
        "spam_rejected": 0,
        "ai_rejected": 0,
        "processed": 0,
        "stored": 0
    }
    
    # ... processing logic ...
    
    # Print summary
    print(f"📊 Ingestion Summary for '{topic}':")
    print(f"   📥 Fetched: {stats['fetched']}")
    print(f"   🕒 Age rejected: {stats['age_rejected']} ({stats['age_rejected']/stats['fetched']*100:.1f}%)")
    print(f"   📝 Word count rejected: {stats['word_count_rejected']}")
    print(f"   🚫 Blacklist rejected: {stats['blacklist_rejected']}")
    print(f"   🗑️ Spam rejected: {stats['spam_rejected']}")
    print(f"   🤖 AI rejected: {stats['ai_rejected']}")
    print(f"   ✅ Processed: {stats['processed']}")
    print(f"   💾 Stored: {stats['stored']}")
    
    return stats

def calculate_relevance_score(item):
    """Calculate relevance score for ranking"""
    score = 0
    
    # Check entities (highest priority - 10 points)
    entities = item.get("entities", [])
    if isinstance(entities, list):
        for entity in entities:
            entity_text = ""
            if isinstance(entity, dict):
                entity_text = (entity.get("text") or "").lower()
            elif isinstance(entity, str):
                entity_text = entity.lower()
            
            if t_lower in entity_text or entity_text in t_lower:
                score += 10  # Exact entity match
            elif any(word in entity_text for word in t_lower.split()):
                score += 5   # Partial entity match
    
    # Check headline (medium priority - 8 points)
    headline = (item.get("headline") or "").lower()
    if t_lower in headline:
        score += 8
    elif any(word in headline for word in t_lower.split()):
        score += 3
    
    # Check summary (lower priority - 4 points)
    summary = (item.get("summary") or "").lower()
    if t_lower in summary:
        score += 4
    elif any(word in summary for word in t_lower.split()):
        score += 1
    
    return score

# Calculate scores and filter
scored_items = []
for item in items:
    score = calculate_relevance_score(item)
    if score > 0:  # Only include items with some relevance
        scored_items.append((item, score))

# Sort by relevance score (descending) then by date
scored_items.sort(
    key=lambda x: (x[1], _to_dt(x[0].get("date", "")) or datetime.min), 
    reverse=True
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        # Local development
        "http://localhost:3000", 
        "http://localhost:3001",
        "http://localhost:3002",
        # Production
        "https://news-insight-ai-tawny.vercel.app",
        "https://*.vercel.app",  # Preview deployments
    ],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

def format_article(article: Dict[str, Any]) -> Dict[str, Any]:
    def convert_decimal(obj):
        if isinstance(obj, Decimal):
            return float(obj)
        return obj
    
    formatted = {}
    for key, value in article.items():
        if isinstance(value, dict):
            formatted[key] = {k: convert_decimal(v) for k, v in value.items()}
        else:
            formatted[key] = convert_decimal(value)
    return formatted

def _store_processed_article(article, analysis):
    # ... store logic ...
    table.put_item(Item=item)
    clear_search_cache()  # Invalidate cache

return StreamingResponse(
    generate_stream(),
    media_type="text/event-stream",
    headers={
        "Cache-Control": "no-cache",
        "Connection": "keep-alive",
        "Access-Control-Allow-Origin": "*",
    }
)