#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Daily Briefing Generator Generuje codzienne podsumowanie (maile, NBA, pogoda, kolarstwo, tech) jako MP3 """ import sys import os from datetime import datetime, timedelta import mysql.connector import requests import random import xml.etree.ElementTree as ET import html # Dodaj backend do ścieżki BACKEND_DIR = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, BACKEND_DIR) import config class DailyBriefingGenerator: def __init__(self): self.output_file = os.path.join(config.OUTPUT_DIR, 'daily_briefing.mp3') self.script_text = "" # Różne warianty powitań self.greetings_morning = [ "Dzień dobry Sławku!", "Cześć Sławku, miłego poranku!", "Witaj Sławku!", "Hej Sławku, dzień dobry!", "Dzień dobry! Jak się masz dziś rano?" ] self.greetings_afternoon = [ "Cześć Sławku!", "Witaj Sławku!", "Hej Sławku!", "Dzień dobry Sławku!", "Siema Sławku!" ] self.greetings_evening = [ "Dobry wieczór Sławku!", "Cześć Sławku, dobry wieczór!", "Witaj Sławku!", "Hej Sławku, dobry wieczór!", "Witaj! Miłego wieczoru Sławku!" ] # Różne warianty wprowadzenia do podsumowania self.intro_phrases = [ "Oto Twoje dzisiejsze podsumowanie.", "Mam dla Ciebie kilka informacji.", "Przygotowałem dla Ciebie podsumowanie dnia.", "Sprawdźmy co się dzieje.", "Zobacz co się wydarzyło.", "Czas na podsumowanie dnia." ] # Różne warianty zakończenia self.closing_phrases = [ "To wszystko na teraz. Miłego dnia!", "To by było na tyle. Trzymaj się!", "To wszystko. Życzę produktywnego dnia!", "I to już wszystkie wiadomości. Powodzenia!", "Na tym kończymy. Do usłyszenia!", "To wszystko z mojej strony. Miłego dnia!" ] def get_jarvis_db(self): return mysql.connector.connect( host=config.JARVIS_DB_HOST, user=config.JARVIS_DB_USER, password=config.JARVIS_DB_PASSWORD, database=config.JARVIS_DB_NAME, charset='utf8mb4' ) def get_nba_db(self): return mysql.connector.connect( host=config.NBA_DB_HOST, user=config.NBA_DB_USER, password=config.NBA_DB_PASSWORD, database=config.NBA_DB_NAME, charset='utf8mb4' ) def fetch_rss_headlines(self, url, limit=3): """Pomocnicza funkcja do pobierania i parsowania nagłówków RSS bez zewnętrznych bibliotek""" try: response = requests.get(url, timeout=10, headers={"User-Agent": "Mozilla/5.0"}) if response.status_code != 200: return [] root = ET.fromstring(response.content) headlines = [] for item in root.findall('.//item')[:limit]: title = item.find('title').text if title: # Oczyszczanie tagów HTML i encji typu " title = html.unescape(title) headlines.append(title.strip()) return headlines except Exception as e: print(f"Błąd RSS ({url}): {e}") return [] # ========================= # SEKCJA: KALENDARZ # ========================= def generate_calendar_summary(self): """Generuje informację o dacie i dniu tygodnia""" dni_tygodnia = ["w poniedziałek", "we wtorek", "w środę", "we czwartek", "w piątek", "w sobotę", "w niedzielę"] miesiace = ["stycznia", "lutego", "marca", "kwietnia", "maja", "czerwca", "lipca", "sierpnia", "września", "października", "listopada", "grudnia"] now = datetime.now() dzien_tyg = dni_tygodnia[now.weekday()] miesiac = miesiace[now.month - 1] return f"Mamy dziś {dzien_tyg}, {now.day} {miesiac}. " # ========================= # SEKCJA 1: MAILE # ========================= def get_recent_emails(self, hours=12): """Pobiera najważniejsze emaile z ostatnich 12 godzin""" conn = self.get_jarvis_db() cursor = conn.cursor(dictionary=True) time_threshold = datetime.now() - timedelta(hours=hours) query = """ SELECT sender_name, subject, category, priority FROM email_history WHERE received_at >= %s AND category NOT IN ('SPAM', 'REKLAMA', 'NEWSLETTER') ORDER BY CASE priority WHEN 'WYSOKI' THEN 1 WHEN 'ŚREDNI' THEN 2 ELSE 3 END, received_at DESC LIMIT 5 """ cursor.execute(query, (time_threshold,)) emails = cursor.fetchall() cursor.close() conn.close() return emails def generate_emails_summary(self): """Generuje podsumowanie emaili z losowymi wariantami""" emails = self.get_recent_emails(hours=12) if not emails: no_emails_variants = [ "Nie masz nowych ważnych wiadomości email.", "Skrzynka jest pusta. Brak nowych maili.", "Żadnych nowych wiadomości nie otrzymałeś.", "Cisza na poczcie. Żadnych nowych maili.", "Spokojnie na froncie mailowym. Nic nowego." ] return random.choice(no_emails_variants) count = len(emails) intro_variants = [ f"Masz {count} ", f"Otrzymałeś {count} ", f"W skrzynce czeka {count} ", f"Dotarło {count} " ] summary = random.choice(intro_variants) if count == 1: summary += random.choice(["nową wiadomość. ", "nowy mail. ", "nową wiadomość email. "]) elif count in [2, 3, 4]: summary += random.choice(["nowe wiadomości. ", "nowe maile. ", "nowe emaile. "]) else: summary += random.choice(["nowych wiadomości. ", "nowych maili. ", "nowych emaili. "]) for i, email in enumerate(emails, 1): priority_text = "" if email['priority'] == 'WYSOKI': priority_variants = ["Pilne! ", "Ważne! ", "Uwaga! ", "Priorytet! "] priority_text = random.choice(priority_variants) sender = email['sender_name'] if email['sender_name'] else "Nieznany nadawca" email_variants = [ f"{priority_text}{sender} pisze: {email['subject']}. ", f"{priority_text}Od {sender}: {email['subject']}. ", f"{priority_text}Mail od {sender} w sprawie: {email['subject']}. ", f"{priority_text}{sender} wysłał: {email['subject']}. " ] summary += random.choice(email_variants) return summary # ========================= # SEKCJA 2: NBA # ========================= def get_favorite_players_stats(self, days=2): """Pobiera statystyki ulubionych graczy""" conn = self.get_nba_db() cursor = conn.cursor(dictionary=True) favorite_players = getattr(config, 'FAVORITE_NBA_PLAYERS', []) if not favorite_players: cursor.close() conn.close() return [] date_from = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') placeholders = ','.join(['%s'] * len(favorite_players)) query = f""" SELECT p.first_name, p.last_name, p.player_id, pgs.points, pgs.rebounds, pgs.assists, pgs.minutes_played, g.game_date, ht.abbreviation as home_abbr, at.abbreviation as away_abbr, g.home_score, g.away_score, pgs.team_id, g.home_team_id, g.game_status FROM player_game_stats pgs JOIN players p ON pgs.player_id = p.player_id JOIN games g ON pgs.game_id = g.game_id JOIN team_info ht ON g.home_team_id = ht.team_id JOIN team_info at ON g.away_team_id = at.team_id WHERE pgs.player_id IN ({placeholders}) AND g.game_date >= %s AND g.game_status = 'Final' ORDER BY g.game_date DESC """ cursor.execute(query, favorite_players + [date_from]) stats = cursor.fetchall() cursor.close() conn.close() return stats def generate_nba_summary(self): """Generuje podsumowanie NBA z losowymi wariantami (Pomiń poza sezonem)""" # Sprawdzenie off-seasonu (Lipiec=7, Sierpień=8, Wrzesień=9) if datetime.now().month in [7, 8, 9]: print("🏀 Off-season NBA. Sekcja pominięta.") return "" stats = self.get_favorite_players_stats(days=2) if not stats: no_games_variants = [ "Twoi ulubieni gracze NBA nie rozegrali ostatnio meczów.", "W NBA cisza. Twoi faworyci nie grali.", "Brak meczów Twoich ulubionych zawodników NBA." ] return random.choice(no_games_variants) players_latest = {} for stat in stats: player_key = f"{stat['first_name']} {stat['last_name']}" if player_key not in players_latest: players_latest[player_key] = stat player_count = len(players_latest) intro_variants = ["W NBA, ostatnio ", "Jeśli chodzi o NBA, to ", "Na parkietach NBA ", "W lidze NBA "] summary = random.choice(intro_variants) if player_count == 1: single_variants = ["zagrał Twój ulubiony zawodnik. ", "wystąpił Twój fawor. ", "grał Twój ulubieniec. "] summary += random.choice(single_variants) else: multiple_variants = [f"zagrało {player_count} Twoich ulubionych zawodników. ", f"wystąpiło {player_count} Twoich faworytów. "] summary += random.choice(multiple_variants) for player_name, stat in players_latest.items(): is_home = stat['team_id'] == stat['home_team_id'] team_score = stat['home_score'] if is_home else stat['away_score'] opp_score = stat['away_score'] if is_home else stat['home_score'] if team_score > opp_score: result = random.choice(["wygrał", "zwyciężył", "pokonał", "ograł"]) else: result = random.choice(["przegrał", "uległ", "nie dał rady", "poniósł porażkę"]) opponent = stat['away_abbr'] if is_home else stat['home_abbr'] game_desc_variants = [ f"{player_name} {result} z {opponent}, ", f"{player_name} w meczu z {opponent} {result}, " ] summary += random.choice(game_desc_variants) stats_variants = [ f"zdobywając {stat['points']} punktów, {stat['rebounds']} zbiórek i {stat['assists']} asyst. ", f"notując {stat['points']} punktów, {stat['rebounds']} zbiórek oraz {stat['assists']} asyst. " ] summary += random.choice(stats_variants) return summary # ========================= # SEKCJA 3: POGODA + KOLARSTWO # ========================= def get_weather(self): """Pobiera pogodę""" try: url = f"https://api.openweathermap.org/data/2.5/weather?lat={config.WEATHER_LAT}&lon={config.WEATHER_LON}&appid={config.OPENWEATHER_API_KEY}&units=metric&lang=pl" response = requests.get(url, timeout=10) data = response.json() temp = round(data['main']['temp']) feels_like = round(data['main']['feels_like']) description = data['weather'][0]['description'] wind_speed = data.get('wind', {}).get('speed', 0) * 3.6 # m/s na km/h # Sprawdzenie opadów has_rain = 'rain' in data or 'snow' in data or "deszcz" in description or "mżawka" in description return { 'temp': temp, 'feels_like': feels_like, 'description': description, 'wind_speed': round(wind_speed), 'has_rain': has_rain } except Exception as e: print(f"Błąd pobierania pogody: {e}") return None def generate_weather_summary(self): """Generuje podsumowanie pogody oraz kolarski komentarz warunków szosowych""" weather = self.get_weather() if not weather: return "Niestety nie mam aktualnych informacji o pogodzie. " intro_variants = [ "Pogoda w Jeleniej Górze: ", "Jeśli chodzi o warunki w Jeleniej Górze: ", "Na zewnątrz w Jeleniej Górze: " ] summary = random.choice(intro_variants) summary += f"{weather['description']}, " temp_variants = [ f"termometry pokazują {weather['temp']} stopni, ", f"temperatura wynosi {weather['temp']} stopni Celsjusza, " ] summary += random.choice(temp_variants) summary += f"a odczuwalna to {weather['feels_like']} stopni. Wiatr wieje z prędkością {weather['wind_speed']} kilometrów na godzinę. " # --- ANALIZA KOLARSKA (SZOSA) --- summary += "Warunki na szosę: " if weather['has_rain']: summary += "Asfalt może być mokry i śliski, zapowiada się deszczowy trening. Może lepiej wybrać trenażer? " elif weather['wind_speed'] > 28: summary += "Mocno dmucha! Przy {weather['wind_speed']} kilometrach na godzinę boczny wiatr na karkonoskich zjazdach może dać popalić. Trzymaj mocno kierownicę. " elif weather['temp'] < 8: summary += "Jest zimno, upewnij się, że dobrze zabezpieczyłeś kolana i ubrałeś ciepłe skarpety. " elif weather['temp'] >= 18 and not weather['has_rain']: summary += "Warunki są idealne! Krótki rękawek, czysty asfalt i w drogę. Udanej jazdy! " else: summary += "Warunki są stabilne, można śmiało planować trasę. " return summary # ========================= # SEKCJA 4: NEWSY (RSS) # ========================= def generate_tech_summary(self): """Pobiera i generuje przegląd technologii z portalu Niebezpiecznik""" headlines = self.fetch_rss_headlines("https://niebezpiecznik.pl/feed/", limit=3) if not headlines: return "W świecie technologii brak nowych doniesień. " summary = "Z technologii i bezpieczeństwa, Niebezpiecznik pisze o: " summary += ", oraz: ".join(headlines) + ". " return summary def generate_pro_cycling_summary(self): """Pobiera i generuje przegląd zawodowego kolarstwa szosowego z Naszosie.pl""" headlines = self.fetch_rss_headlines("https://feeds.feedburner.com/niebezpiecznik/", limit=3) if not headlines: return "W zawodowym peletonie szosowym bez zmian. " summary = "Ze świata zawodowego kolarstwa szosowego: " summary += ", kolejny temat to: ".join(headlines) + ". " return summary # ========================= # GENEROWANIE FINALNEGO MP3 # ========================= def generate_full_briefing(self): """Generuje pełne podsumowanie dnia ze wszystkich aktywnych sekcji""" print("=" * 60) print("🎙️ GENEROWANIE PEŁNEGO BRIEFINGU") print("=" * 60) hour = datetime.now().hour if hour < 12: greeting = random.choice(self.greetings_morning) elif hour < 18: greeting = random.choice(self.greetings_afternoon) else: greeting = random.choice(self.greetings_evening) intro = random.choice(self.intro_phrases) briefing = f"{greeting} {intro} " # Sekcja: Kalendarz na start briefing += self.generate_calendar_summary() # Sekcja 1: Emaile print("\n📧 Generuję podsumowanie emaili...") email_summary = self.generate_emails_summary() briefing += email_summary + " " # Sekcja 2: NBA (Zwróci "" w lipcu, sierpniu i wrześniu) print("\n🏀 Analizuję NBA...") nba_summary = self.generate_nba_summary() if nba_summary: briefing += nba_summary + " " print(f" ✓ {nba_summary}") # Sekcja 3: Pogoda i Szosa print("\n🌤️ Generuję pogodę i analizę rowerową...") weather_summary = self.generate_weather_summary() briefing += weather_summary + " " print(f" ✓ {weather_summary}") # Sekcja 4: Kolarstwo Zawodowe (RSS) print("\n🚴 Pobieram newsy z pro-peletonu...") cycling_summary = self.generate_pro_cycling_summary() briefing += cycling_summary + " " print(f" ✓ {cycling_summary}") # Sekcja 5: Tech (RSS) print("\n💻 Pobieram tech-newsy...") tech_summary = self.generate_tech_summary() briefing += tech_summary + " " print(f" ✓ {tech_summary}") # Zakończenie closing = random.choice(self.closing_phrases) briefing += closing self.script_text = briefing return briefing def create_mp3_elevenlabs(self): """Tworzy plik MP3 z syntezą mowy przy użyciu ElevenLabs API""" briefing_text = self.generate_full_briefing() print("\n" + "=" * 60) print("📝 PEŁNA TREŚĆ BRIEFINGU:") print("=" * 60) print(briefing_text) print("=" * 60) print("\n🎵 Generuję plik MP3 z ElevenLabs...") try: url = f"https://api.elevenlabs.io/v1/text-to-speech/{config.ELEVENLABS_VOICE_ID}" headers = { "Accept": "audio/mpeg", "Content-Type": "application/json", "xi-api-key": config.ELEVENLABS_API_KEY } data = { "text": briefing_text, "model_id": "eleven_multilingual_v2", "voice_settings": { "stability": 0.5, "similarity_boost": 0.75, "style": 0.0, "use_speaker_boost": True } } response = requests.post(url, json=data, headers=headers, timeout=60) if response.status_code == 200: with open(self.output_file, 'wb') as f: f.write(response.content) file_size = os.path.getsize(self.output_file) file_size_kb = file_size / 1024 print(f"\n✅ MP3 WYGENEROWANY POMYŚLNIE!") print(f" 📂 Lokalizacja: {self.output_file}") print(f" 📊 Rozmiar: {file_size_kb:.2f} KB") print(f" 🕒 Czas: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print("=" * 60) return self.output_file else: print(f"❌ Błąd API ElevenLabs: {response.status_code}") print(f" Odpowiedź: {response.text}") raise Exception(f"ElevenLabs API error: {response.status_code}") except Exception as e: print(f"\n❌ BŁĄD ElevenLabs: {e}") raise def create_mp3(self): return self.create_mp3_elevenlabs() def run(self): """Główna funkcja - uruchom generator""" try: self.create_mp3() return True except Exception as e: print(f"\n❌ BŁĄD: {e}") import traceback traceback.print_exc() return False if __name__ == "__main__": generator = DailyBriefingGenerator() success = generator.run() if success: print("\n🎉 Daily briefing gotowy!") sys.exit(0) else: print("\n💥 Nie udało się wygenerować briefingu") sys.exit(1)