#!/usr/bin/env python3 """ Uruguay News Monitoring System - cPanel Version Monitors news websites and uploads results to cPanel via FTP """ import requests from bs4 import BeautifulSoup import re import json import datetime from typing import List, Dict, Any import time import logging import ftplib import os from urllib.parse import urljoin, urlparse import hashlib # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class CpanelNewsMonitor: def __init__(self, config_file='config.json'): self.load_config(config_file) self.websites = { 'elpais': { 'url': 'https://www.elpais.com.uy/', 'name': 'El País Uruguay', 'priority': 'medium' }, 'montevideo': { 'url': 'https://www.montevideo.com.uy/', 'name': 'Montevideo Portal', 'priority': 'high' }, 'elobservador': { 'url': 'https://www.elobservador.com.uy/', 'name': 'El Observador', 'priority': 'medium' }, 'uypress': { 'url': 'https://www.uypress.net/home', 'name': 'UyPress', 'priority': 'high' }, 'telenoche': { 'url': 'https://www.telenoche.com.uy/', 'name': 'Telenoche', 'priority': 'medium' }, 'subrayado': { 'url': 'https://www.subrayado.com.uy/', 'name': 'Subrayado', 'priority': 'high' }, 'ladiaria': { 'url': 'https://ladiaria.com.uy/', 'name': 'La Diaria', 'priority': 'low' }, 'portal180': { 'url': 'https://www.180.com.uy/', 'name': 'Portal 180', 'priority': 'low' } } # Enhanced keywords with variations and related terms self.keywords = { 'economia_sociedad': { 'keywords': [ 'costo de vida', 'tarifas', 'UTE', 'OSE', 'ANTEL', 'precio', 'seguridad ciudadana', 'crisis habitacional', 'vivienda', 'MEVIR', 'precariedad', 'pobreza', 'inflación', 'salario', 'trabajo', 'economía', 'social', 'familia', 'hogar', 'barrio' ], 'weight': 0.7, 'emoji': '📈' }, 'tech_ia_domotica': { 'keywords': [ 'hogar inteligente', 'automatización', 'inteligencia artificial', 'IA', 'cámaras', 'domótica', 'tecnología', 'solar', 'smart', 'tecnológico', 'digital', 'innovación', 'app', 'sistema', 'dispositivo', 'conectividad', 'internet' ], 'weight': 0.8, 'emoji': '📲' }, 'seguridad_urbana': { 'keywords': [ 'asalto', 'robo', 'robos', 'alarma', 'cámaras', 'videovigilancia', 'policía', 'patrullaje', 'seguridad', 'delincuencia', 'crimen', 'violencia', 'inseguridad', 'peligroso', 'zona roja', 'delincuente', 'ladrón', 'criminal', 'hurto', 'rapiña', 'mano armada', 'arma', 'pistola', 'revólver', 'cuchillo' ], 'weight': 0.9, 'emoji': '🚨' }, 'drogas_narcotrafico': { 'keywords': [ 'droga', 'drogas', 'cocaína', 'marihuana', 'pasta base', 'narcotráfico', 'narco', 'incautación', 'incautaron', 'decomiso', 'operativo', 'antidrogas', 'tráfico', 'dealer', 'estupefacientes', 'sustancias', 'psicoactivas', 'ilegal', 'contrabando', 'banda', 'organización criminal' ], 'weight': 1.0, 'emoji': '💊' }, 'agro_campo': { 'keywords': [ 'ganado', 'agropecuaria', 'rural', 'campo', 'productores', 'feria', 'precio', 'emergencia', 'conectividad', 'agricultura', 'ganadería', 'vacas', 'ovejas', 'soja', 'trigo', 'arroz', 'tambero', 'estancia' ], 'weight': 0.6, 'emoji': '🧑‍🌾' }, 'seguridad_rural': { 'keywords': [ 'abigeato', 'robo de ganado', 'campo', 'rural', 'productores', 'tranqueras', 'alambrado', 'estancia', 'cuatrerismo', 'ganado robado', 'vacas robadas', 'delincuencia rural', 'inseguridad rural' ], 'weight': 0.8, 'emoji': '🐑' }, 'armas_violencia': { 'keywords': [ 'arma', 'armas', 'pistola', 'revólver', 'escopeta', 'rifle', 'bala', 'disparo', 'tiroteo', 'balacera', 'guerra', 'munición', 'calibre', 'armamento', 'arsenal', 'violento' ], 'weight': 0.95, 'emoji': '🔫' }, 'seguridad_negocios': { 'keywords': [ 'comercio', 'negocio', 'tienda', 'local', 'almacén', 'farmacia', 'ferretería', 'carnicería', 'emprendedor', 'empresa', 'establecimiento', 'comerciante', 'vendedor' ], 'weight': 0.8, 'emoji': '🏪' }, 'seguridad_hogares': { 'keywords': [ 'casa', 'hogar', 'domicilio', 'vivienda', 'entradera', 'vecino', 'vecinos', 'barrio', 'residencial', 'domiciliario', 'residencia', 'apartamento', 'edificio' ], 'weight': 0.9, 'emoji': '🏠' } } self.session = requests.Session() self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' }) def load_config(self, config_file): """Load configuration from JSON file""" try: with open(config_file, 'r') as f: self.config = json.load(f) except FileNotFoundError: logger.error(f"Configuration file {config_file} not found!") self.config = {} def scrape_website(self, site_key: str) -> List[Dict[str, Any]]: """Scrape a specific website for articles with improved extraction""" site_info = self.websites[site_key] articles = [] try: logger.info(f"Scraping {site_info['name']}...") response = self.session.get(site_info['url'], timeout=15) response.raise_for_status() soup = BeautifulSoup(response.content, 'html.parser') # Site-specific extraction strategies if site_key == 'montevideo': articles = self._extract_montevideo(soup, site_info) elif site_key == 'subrayado': articles = self._extract_subrayado(soup, site_info) elif site_key == 'uypress': articles = self._extract_uypress(soup, site_info) else: articles = self._extract_generic(soup, site_info) except Exception as e: logger.error(f"Error scraping {site_info['name']}: {e}") logger.info(f"Found {len(articles)} articles from {site_info['name']}") return articles def _extract_montevideo(self, soup, site_info): """Extract articles from Montevideo Portal""" articles = [] selectors = ['a[href*="/articulo/"]', '.titulo a', 'h1 a', 'h2 a', 'h3 a', '.noticia a'] found_links = set() for selector in selectors: elements = soup.select(selector) for elem in elements: href = elem.get('href') if href and href not in found_links: found_links.add(href) title = elem.get_text(strip=True) if len(title) > 10: articles.append({ 'title': title, 'link': urljoin(site_info['url'], href), 'summary': '', 'source': site_info['name'], 'source_key': 'montevideo', 'scraped_at': datetime.datetime.now().isoformat(), 'priority': site_info['priority'] }) return articles[:30] def _extract_subrayado(self, soup, site_info): """Extract articles from Subrayado""" articles = [] selectors = ['h1', 'h2', 'h3', '.titulo', 'a[href*="/noticia/"]', 'a[href*="/articulo/"]'] found_titles = set() for selector in selectors: elements = soup.select(selector) for elem in elements: title = elem.get_text(strip=True) if len(title) > 15 and title not in found_titles: found_titles.add(title) link = None if elem.name == 'a': link = elem.get('href') else: link_elem = elem.find('a') or elem.find_parent('a') if link_elem: link = link_elem.get('href') if link: link = urljoin(site_info['url'], link) articles.append({ 'title': title, 'link': link, 'summary': '', 'source': site_info['name'], 'source_key': 'subrayado', 'scraped_at': datetime.datetime.now().isoformat(), 'priority': site_info['priority'] }) return articles[:30] def _extract_uypress(self, soup, site_info): """Extract articles from UyPress""" articles = [] selectors = ['h1', 'h2', 'h3', 'h4', '.titulo', 'a[href*="/articulo/"]'] found_titles = set() for selector in selectors: elements = soup.select(selector) for elem in elements: title = elem.get_text(strip=True) if len(title) > 15 and title not in found_titles: found_titles.add(title) link = None if elem.name == 'a': link = elem.get('href') else: link_elem = elem.find('a') or elem.find_parent('a') if link_elem: link = link_elem.get('href') if link: link = urljoin(site_info['url'], link) articles.append({ 'title': title, 'link': link, 'summary': '', 'source': site_info['name'], 'source_key': 'uypress', 'scraped_at': datetime.datetime.now().isoformat(), 'priority': site_info['priority'] }) return articles[:30] def _extract_generic(self, soup, site_info): """Generic article extraction for other sites""" articles = [] selectors = ['h1', 'h2', 'h3', 'article h1', 'article h2', 'article h3', '.headline', '.title', '.titulo', 'a[href*="/noticia/"]', 'a[href*="/articulo/"]'] found_titles = set() for selector in selectors: elements = soup.select(selector) for elem in elements: title = elem.get_text(strip=True) if len(title) > 15 and title not in found_titles: found_titles.add(title) link = None if elem.name == 'a': link = elem.get('href') else: link_elem = elem.find('a') or elem.find_parent('a') if link_elem: link = link_elem.get('href') if link: link = urljoin(site_info['url'], link) articles.append({ 'title': title, 'link': link, 'summary': '', 'source': site_info['name'], 'source_key': site_info.get('source_key', 'generic'), 'scraped_at': datetime.datetime.now().isoformat(), 'priority': site_info['priority'] }) return articles[:25] def analyze_article(self, article: Dict[str, Any]) -> Dict[str, Any]: """Enhanced article analysis with flexible keyword matching""" text = f"{article['title']} {article['summary']}".lower() matches = [] total_score = 0 for category, info in self.keywords.items(): category_matches = [] for keyword in info['keywords']: if keyword.lower() in text: category_matches.append(keyword) total_score += info['weight'] if category_matches: matches.append({ 'category': category, 'emoji': info['emoji'], 'keywords': category_matches, 'weight': info['weight'] }) # Calculate engagement score (0-100) engagement_score = min(100, total_score * 8) # Boost score for high-priority sources if article['priority'] == 'high': engagement_score *= 1.3 elif article['priority'] == 'medium': engagement_score *= 1.1 engagement_score = min(100, engagement_score) article['keyword_matches'] = matches article['engagement_score'] = round(engagement_score, 1) article['has_matches'] = len(matches) > 0 return article def generate_social_media_suggestions(self, article: Dict[str, Any]) -> Dict[str, Any]: """Generate enhanced social media content suggestions""" if not article['has_matches']: return {} suggestions = {} title = article['title'] matches = article['keyword_matches'] # Determine primary category for better suggestions primary_category = max(matches, key=lambda x: x['weight'])['category'] # Instagram suggestion instagram_hashtags = ['#Uruguay', '#Noticias'] if 'seguridad' in primary_category or 'drogas' in primary_category: instagram_hashtags.extend(['#SeguridadUruguay', '#Seguridad', '#Montevideo']) elif 'tech' in primary_category: instagram_hashtags.extend(['#TecnologiaUruguay', '#Innovacion', '#TechUY']) elif 'economia' in primary_category: instagram_hashtags.extend(['#EconomiaUruguay', '#Sociedad', '#UruguayHoy']) elif 'agro' in primary_category or 'rural' in primary_category: instagram_hashtags.extend(['#CampoUruguayo', '#Agro', '#Rural']) suggestions['instagram'] = { 'caption': f"🚨 {title}\n\n¿Qué opinas sobre esta situación? 👇\n\n{' '.join(instagram_hashtags[:8])}", 'engagement_tips': 'Usar Stories para hacer encuestas, crear carrusel con datos relevantes, hacer preguntas en comentarios' } # Facebook suggestion facebook_post = f"{title}\n\n" if 'seguridad' in primary_category: facebook_post += "¿Cómo podemos mejorar la seguridad en nuestros barrios? Comparte tu experiencia." elif 'economia' in primary_category: facebook_post += "¿Cómo te afecta esta situación económica? Cuéntanos tu opinión." else: facebook_post += "¿Qué opinas sobre esta noticia? Comparte tu punto de vista." suggestions['facebook'] = { 'post': facebook_post, 'engagement_tips': 'Hacer preguntas abiertas, compartir en grupos locales relevantes, responder a todos los comentarios' } # X (Twitter) suggestion tweet_text = title if len(tweet_text) > 200: tweet_text = tweet_text[:197] + "..." if 'seguridad' in primary_category: tweet_text += "\n\n¿Tu barrio es seguro? #UruguaySeguro #Seguridad" elif 'economia' in primary_category: tweet_text += "\n\n¿Cómo te afecta? #EconomiaUY #Uruguay" else: tweet_text += "\n\n¿Qué opinas? #UruguayHoy #Noticias" suggestions['x'] = { 'tweet': tweet_text, 'engagement_tips': 'Crear hilos con más contexto, usar hashtags trending, responder rápidamente a menciones' } return suggestions def monitor_all_sites(self) -> List[Dict[str, Any]]: """Monitor all websites and return analyzed articles""" all_articles = [] for site_key in self.websites.keys(): articles = self.scrape_website(site_key) for article in articles: analyzed_article = self.analyze_article(article) if analyzed_article['has_matches']: social_suggestions = self.generate_social_media_suggestions(analyzed_article) analyzed_article['social_media_suggestions'] = social_suggestions all_articles.append(analyzed_article) # Be respectful with requests time.sleep(1) # Sort by engagement score all_articles.sort(key=lambda x: x['engagement_score'], reverse=True) return all_articles def save_results(self, articles: List[Dict[str, Any]], filename: str = None): """Save monitoring results to JSON file""" if filename is None: timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"news_monitoring_{timestamp}.json" results = { 'timestamp': datetime.datetime.now().isoformat(), 'total_articles': len(articles), 'max_score': max([a['engagement_score'] for a in articles]) if articles else 0, 'articles': articles } with open(filename, 'w', encoding='utf-8') as f: json.dump(results, f, ensure_ascii=False, indent=2) logger.info(f"Results saved to {filename}") return filename, results def upload_to_cpanel(self, local_file, remote_file): """Upload file to cPanel via FTP""" if 'ftp' not in self.config: logger.error("FTP configuration not found in config file") return False try: ftp_config = self.config['ftp'] with ftplib.FTP(ftp_config['host']) as ftp: ftp.login(ftp_config['username'], ftp_config['password']) # Change to data directory try: ftp.cwd('public_html/data') except: # Create directory if it doesn't exist ftp.mkd('public_html/data') ftp.cwd('public_html/data') # Upload file with open(local_file, 'rb') as f: ftp.storbinary(f'STOR {remote_file}', f) logger.info(f"Successfully uploaded {local_file} to {remote_file}") return True except Exception as e: logger.error(f"Error uploading to cPanel: {e}") return False def update_history_index(self, new_file_info): """Update history index file""" history_file = 'history_index.json' try: # Load existing history if os.path.exists(history_file): with open(history_file, 'r') as f: history = json.load(f) else: history = {'files': []} # Add new file info history['files'].insert(0, new_file_info) # Insert at beginning # Keep only last 30 entries history['files'] = history['files'][:30] # Save updated history with open(history_file, 'w') as f: json.dump(history, f, indent=2) # Upload to cPanel self.upload_to_cpanel(history_file, 'history_index.json') except Exception as e: logger.error(f"Error updating history index: {e}") def run_monitoring(self): """Run complete monitoring cycle""" logger.info("Starting Uruguay News Monitoring System for cPanel...") # Monitor all sites articles = self.monitor_all_sites() logger.info(f"Found {len(articles)} relevant articles") # Save results locally timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") filename, results = self.save_results(articles, f"news_monitoring_{timestamp}.json") # Upload latest.json to cPanel if self.upload_to_cpanel(filename, 'latest.json'): logger.info("Successfully uploaded latest results to cPanel") # Also upload with timestamp for history if self.upload_to_cpanel(filename, filename): logger.info(f"Successfully uploaded {filename} to cPanel history") # Update history index file_info = { 'filename': filename, 'timestamp': results['timestamp'], 'total_articles': results['total_articles'], 'max_score': results['max_score'] } self.update_history_index(file_info) # Print summary print(f"\n🔥 MONITOREO COMPLETADO - {len(articles)} ARTÍCULOS RELEVANTES:") print("=" * 80) for i, article in enumerate(articles[:10], 1): print(f"{i}. [{article['engagement_score']:.1f}] {article['title']}") print(f" Source: {article['source']}") categories = ', '.join([m['emoji'] + ' ' + m['category'] for m in article['keyword_matches']]) print(f" Categories: {categories}") print() return filename def main(): """Main function""" monitor = CpanelNewsMonitor() try: filename = monitor.run_monitoring() print(f"\n✅ Monitoreo completado exitosamente!") print(f"📁 Archivo generado: {filename}") print(f"🌐 Datos subidos a cPanel") except Exception as e: logger.error(f"Error during monitoring: {e}") print(f"\n❌ Error durante el monitoreo: {e}") if __name__ == "__main__": main()