News Synthesis Tool Setup

from __future__ import annotations from typing import List, Dict, Tuple, Optional, Any import feedparser import asyncio import re import aiohttp from playwright.async_api import async_playwright import datetime import json import streamlit as st from streamlit_extras.colored_header import colored_header from streamlit_extras.add_vertical_space import add_vertical_space from streamlit_lottie import st_lottie import logging import hashlib import sqlite3 import requests from datetime import datetime class NewsSynthesizer: def __init__(self): self.rss_feeds = [ ('BBC News', 'http://feeds.bbci.co.uk/news/rss.xml'), ('Reuters', 'https://www.reutersagency.com/feed/?taxonomy=best-topics&post_type=best'), # Updated Reuters feed ('AP News', 'https://apnews.com/hub/ap-top-news.rss'), ('NY Times', 'https://rss.nytimes.com/services/xml/rss/nyt/HomePage.xml'), ('Al Jazeera', 'https://www.aljazeera.com/xml/rss/all.xml') ] self.user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" self.scrape_timeout = 30 self.logger = logging.getLogger(__name__) self.setup_database() def setup_database(self): """Initialize SQLite database with necessary tables""" try: with sqlite3.connect('news_synthesis.db') as conn: cursor = conn.cursor() # Create articles table cursor.execute(''' CREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, source TEXT NOT NULL, url TEXT UNIQUE NOT NULL, content TEXT, published_date TEXT, hash TEXT UNIQUE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Create reports table cursor.execute(''' CREATE TABLE IF NOT EXISTS reports ( id INTEGER PRIMARY KEY AUTOINCREMENT, headline TEXT NOT NULL, sources TEXT NOT NULL, urls TEXT NOT NULL, published_dates TEXT NOT NULL, report_content TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') conn.commit() self.logger.info("Database setup completed successfully") except sqlite3.Error as e: self.logger.error(f"Database setup error: {str(e)}") raise Exception(f"Failed to setup database: {str(e)}") async def _analyze_cluster(self, cluster: List[Dict]) -> Optional[str]: """ Analyze a cluster of related articles and generate a synthesized report. Args: cluster: List of article dictionaries containing content from different sources Returns: Optional[str]: A synthesized report or None if analysis fails """ try: # Extract titles and content titles = [article['title'] for article in cluster] contents = [article['content'] for article in cluster] # Combine all content for analysis combined_content = " ".join(contents) # Basic report structure report_sections = [] # Add overview from titles report_sections.append("## Overview\n") report_sections.append(f"This story is being reported by {len(cluster)} news sources. ") report_sections.append(f"The primary headline from {cluster[0]['source']} reads: '{titles[0]}'\n") # Extract key information # Find common locations mentioned location_pattern = r'in ([A-Z][a-zA-Z]+(?: [A-Z][a-zA-Z]+)*)' locations = set(re.findall(location_pattern, combined_content)) if locations: report_sections.append("\n## Location\n") report_sections.append(f"This story takes place in {', '.join(list(locations)[:3])}.\n") # Find dates and times date_pattern = r'(?:January|February|March|April|May|June|July|August|September|October|November|December) \d{1,2}(?:st|nd|rd|th)?(?:,? \d{4})?' dates = set(re.findall(date_pattern, combined_content)) if dates: report_sections.append("\n## Timeline\n") report_sections.append(f"Key dates mentioned: {', '.join(list(dates)[:3])}.\n") # Extract quotes quote_pattern = r'"([^"]*)"' quotes = re.findall(quote_pattern, combined_content) if quotes: report_sections.append("\n## Key Quotes\n") for quote in quotes[:2]: # Limit to 2 most relevant quotes if len(quote) > 20: # Filter out short quotes report_sections.append(f'- "{quote}"\n') # Compare perspectives report_sections.append("\n## Source Comparison\n") for article in cluster: report_sections.append(f"- {article['source']}'s coverage focuses on: {article['title']}\n") # Combine sections into final report final_report = "\n".join(report_sections) return final_report except Exception as e: self.logger.error(f"Cluster analysis error: {str(e)}") return None def _store_article(self, article: Dict) -> bool: """Store article in database if it doesn't exist""" try: content_hash = hashlib.sha256( f"{article['title']}{article['url']}".encode() ).hexdigest() with sqlite3.connect('news_synthesis.db') as conn: cursor = conn.cursor() cursor.execute(''' INSERT OR IGNORE INTO articles (title, source, url, content, published_date, hash) VALUES (?, ?, ?, ?, ?, ?) ''', ( article['title'], article['source'], article['url'], article.get('content', ''), article.get('published', 'N/A'), content_hash )) return cursor.rowcount > 0 except sqlite3.Error as e: self.logger.error(f"Article storage error: {str(e)}") return False def _store_report(self, report: Dict) -> bool: """Store generated report in database""" try: with sqlite3.connect('news_synthesis.db') as conn: cursor = conn.cursor() cursor.execute(''' INSERT INTO reports (headline, sources, urls, published_dates, report_content) VALUES (?, ?, ?, ?, ?) ''', ( report['headline'], json.dumps(report['sources']), json.dumps(report['urls']), json.dumps(report['published_dates']), report.get('report', '') )) return cursor.rowcount > 0 except sqlite3.Error as e: self.logger.error(f"Report storage error: {str(e)}") return False async def _parse_feed_with_retry(self, session: aiohttp.ClientSession, feed_url: str, retries=3) -> Optional[Dict]: """Parse RSS feed with improved error handling and retry logic""" for attempt in range(retries): try: async with session.get(feed_url, timeout=self.scrape_timeout) as response: if response.status == 200: content = await response.text() feed_data = feedparser.parse(content) if feed_data.entries: return feed_data await asyncio.sleep(2 ** attempt) except Exception as e: self.logger.error(f"Feed parsing error (attempt {attempt + 1}): {str(e)}") if attempt < retries - 1: await asyncio.sleep(2 ** attempt) return None async def _scrape_article(self, url: str) -> Optional[str]: try: async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context( user_agent=self.user_agent, viewport={'width': 1920, 'height': 1080} ) page = await context.new_page() await page.route("**/*", lambda route: route.continue_() if route.request.resource_type == "document" else route.abort()) try: await page.goto(url, timeout=self.scrape_timeout * 1000) content = await page.evaluate('''() => { const selectors = [ 'article', '[itemprop="articleBody"]', '.article-body', '.story-body' ]; for (const selector of selectors) { const element = document.querySelector(selector); if (element?.textContent?.length > 500) { return element.innerText; } } return document.body.innerText; }''') return re.sub(r'\s+', ' ', content).strip()[:4000] if content else None finally: await browser.close() except Exception as e: self.logger.error(f"Scraping error for {url}: {str(e)}") return None async def process_feeds(self, progress_bar) -> List[Dict]: """Process all feeds with improved concurrency and error handling""" async with aiohttp.ClientSession(headers={'User-Agent': self.user_agent}) as session: all_articles = [] for idx, (feed_name, feed_url) in enumerate(self.rss_feeds): try: progress_bar.progress((idx + 0.5)/len(self.rss_feeds), f"Processing {feed_name}") feed_data = await self._parse_feed_with_retry(session, feed_url) if not feed_data: continue scrape_tasks = [ self._scrape_article(entry.link) for entry in feed_data.entries[:3] ] contents = await asyncio.gather(*scrape_tasks) valid_articles = [ { 'title': entry.title, 'source': feed_name, 'content': content, 'url': entry.link, 'published': getattr(entry, 'published', 'N/A') } for entry, content in zip(feed_data.entries[:3], contents) if content and len(content) > 500 ] # Store valid articles in database for article in valid_articles: if self._store_article(article): all_articles.append(article) st.toast(f"Added article from {feed_name}") progress_bar.progress((idx + 1)/len(self.rss_feeds)) except Exception as e: self.logger.error(f"Error processing {feed_name}: {str(e)}") st.error(f"Failed to process {feed_name}") return await self._generate_reports(all_articles) async def _generate_reports(self, articles: List[Dict]) -> List[Dict]: """Generate reports from collected articles""" # Group articles by source source_map = {source: [] for source, _ in self.rss_feeds} for article in articles: source_map[article['source']].append(article) # Find minimum number of articles across sources min_articles = min(len(articles) for articles in source_map.values() if articles) reports = [] for i in range(min_articles): try: cluster = [] for source in source_map: if i < len(source_map[source]): cluster.append(source_map[source][i]) if len(cluster) >= 2: # Only process if we have at least 2 sources analysis = await self._analyze_cluster(cluster) if analysis: report = { 'headline': cluster[0]['title'], 'sources': [a['source'] for a in cluster], 'urls': [a['url'] for a in cluster], 'published_dates': [a['published'] for a in cluster], 'report': analysis } if self._store_report(report): reports.append(report) except Exception as e: self.logger.error(f"Report generation error: {str(e)}") return reports def display_report(reports: List[Dict]): # Sidebar with st.sidebar: st.image("https://via.placeholder.com/150x150.png?text=NS", width=150) colored_header(label="Report Information", description="", color_name="blue-70") st.write(f"📅 **Generated:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") st.write(f"🤖 **AI Model:** Phi-4") st.write("📰 **News Sources:**") for source in ["BBC News", "Reuters", "AP News", "The New York Times", "Al Jazeera"]: st.write(f" • {source}") add_vertical_space(2) with st.expander("ℹ️ About", expanded=False): st.write(""" This news synthesizer uses advanced AI to analyze and summarize news from multiple sources. It identifies related stories, cross-references facts, and generates comprehensive reports. """) # Main content col1, col2, col3 = st.columns([1,3,1]) with col2: st.title("🌐 News Synthesis Hub") if not reports: # Display loading animation lottie_news = load_lottie_url("https://assets4.lottiefiles.com/packages/lf20_9qshezyl.json") if lottie_news: st_lottie(lottie_news, height=400) st.warning("No reports available. Generate a new report to begin.") return # Display reports # ... [rest of display_report function remains the same] def load_lottie_url(url: str): try: r = requests.get(url) if r.status_code != 200: return None return r.json() except: return None async def main(): st.set_page_config( page_title="News Synthesis Hub", page_icon="🌐", layout="wide", initial_sidebar_state="expanded" ) if st.button("🔄 Generate New Report"): progress_bar = st.progress(0) try: with st.spinner("Analyzing news feeds..."): synthesizer = NewsSynthesizer() reports = await synthesizer.process_feeds(progress_bar) display_report(reports) except Exception as e: st.error(f"Application error: {str(e)[:200]}") finally: progress_bar.empty() else: display_report([]) if __name__ == "__main__": asyncio.run(main())

Looks like you've built a solid news synthesis tool! You're pulling RSS feeds, scraping content, clustering articles, analyzing key details, and storing reports in a database. Is there something specific you'd like to refine or improve? Maybe optimizing the scraping, enhancing clustering logic, or adding AI-powered summarization?

it does not output the final report

It looks like the function `_generate_reports` is responsible for creating reports but doesn't seem to return or store them properly at the end. A few possible reasons why the final report might not be outputted: 1. **Missing `return reports`** - Ensure `_generate_reports` explicitly returns the list of reports. 2. **Failed Cluster Analysis** - The `_analyze_cluster` method might return `None` if it encounters an issue. Check logs for errors. 3. **Database Storage Issue** - The `_store_report` function might fail due to SQLite constraints (like duplicates). ### Fix: Modify `_generate_reports` to explicitly store reports and return them: ```python async def _generate_reports(self, articles: List[Dict]) -> List[Dict]: """Generate reports from collected articles""" source_map = {source: [] for source, _ in self.rss_feeds} for article in articles: source_map[article['source']].append(article) min_articles = min(len(articles) for articles in source_map.values() if articles) reports = [] for i in range(min_articles): try: cluster = [source_map[source][i] for source in source_map if i < len(source_map[source])] if len(cluster) >= 2: analysis = await self._analyze_cluster(cluster) if analysis: report = { 'headline': cluster[0]['title'], 'sources': [a['source'] for a in cluster], 'urls': [a['url'] for a in cluster], 'published_dates': [a['published'] for a in cluster], 'report': analysis } if self._store_report(report): reports.append(report) except Exception as e: self.logger.error(f"Error generating report: {str(e)}") return reports # Ensure reports are returned ``` After calling `process_feeds()`, explicitly print or display the results: ```python reports = await news_synthesizer.process_feeds(progress_bar) if reports: for report in reports: st.write("### " + report['headline']) st.write(report['report']) else: st.write("No reports generated.") ``` This should ensure the final reports are stored and displayed. Let me know if you're still having issues! 🚀