""" Output formatting for flight comparison results. Supports table, JSON, and CSV formats. """ import json import csv import sys from typing import Optional from datetime import datetime try: from rich.console import Console from rich.table import Table from rich import box HAS_RICH = True except ImportError: HAS_RICH = False def format_duration(minutes: int) -> str: """ Format duration in minutes to human-readable format. Args: minutes: Duration in minutes Returns: Formatted string like "9h 30m" """ if minutes == 0: return "—" hours = minutes // 60 mins = minutes % 60 if mins == 0: return f"{hours}h" return f"{hours}h {mins}m" def format_table_single_date( results: dict[str, list[dict]], destination: str, country: str, date: str, seat_class: str, sort_by: str, total_airports: int, elapsed_time: float, ) -> None: """ Format and print results for single-date mode as a table. Args: results: Dict mapping airport IATA codes to lists of flights destination: Destination airport code country: Origin country code date: Query date seat_class: Cabin class sort_by: Sort criterion (price or duration) total_airports: Total number of airports scanned elapsed_time: Total execution time in seconds """ if not HAS_RICH: print("Rich library not installed, using plain text output") _format_plain_single_date(results, destination, country, date, seat_class, sort_by, total_airports, elapsed_time) return console = Console() # Print header console.print() console.print( f"Flight Comparator: {country} → {destination} | {date} | " f"{seat_class.title()} | Sorted by: {sort_by.title()}", style="bold cyan" ) console.print() # Create table table = Table(box=box.DOUBLE_EDGE, show_header=True, header_style="bold magenta") table.add_column("#", justify="right", style="dim") table.add_column("From", style="cyan") table.add_column("Airline", style="green") table.add_column("Depart", justify="center") table.add_column("Arrive", justify="center") table.add_column("Duration", justify="center") table.add_column("Price", justify="right", style="yellow") table.add_column("Market", justify="center") # Flatten and sort results all_flights = [] for airport_code, flights in results.items(): for flight in flights: flight['airport_code'] = airport_code all_flights.append(flight) # Sort by price or duration if sort_by == "duration": all_flights.sort(key=lambda f: f.get('duration_minutes', 999999)) else: # price all_flights.sort(key=lambda f: f.get('price', 999999)) # Add rows airports_with_flights = set() for idx, flight in enumerate(all_flights, 1): airport_code = flight['airport_code'] airports_with_flights.add(airport_code) from_text = f"{airport_code} {flight.get('city', '')}" airline = flight.get('airline', 'Unknown') depart = flight.get('departure_time', '—') arrive = flight.get('arrival_time', '—') duration = format_duration(flight.get('duration_minutes', 0)) price = f"{flight.get('currency', '€')}{flight.get('price', 0):.0f}" # Simple market indicator (Low/Typical/High) # In a real implementation, this would compare against price distribution market = "Typical" table.add_row( str(idx), from_text, airline, depart, arrive, duration, price, market ) # Add rows for airports with no direct flights no_direct_count = 0 for airport_code in results.keys(): if airport_code not in airports_with_flights: from_text = f"{airport_code}" table.add_row( "—", from_text, "—", "—", "—", "no direct flights found", "—", "—", style="dim" ) no_direct_count += 1 console.print(table) console.print() # Summary console.print( f"Scanned {total_airports} airports • " f"{len(airports_with_flights)} with direct flights • " f"Done in {elapsed_time:.1f}s", style="dim" ) console.print() def format_table_seasonal( results_by_month: dict[str, dict[str, list[dict]]], new_connections: dict[str, str], destination: str, country: str, seat_class: str, total_airports: int, elapsed_time: float, ) -> None: """ Format and print results for seasonal scan mode. Args: results_by_month: Dict mapping month strings to airport->flights dicts new_connections: Dict mapping route keys to first appearance month destination: Destination airport code country: Origin country code seat_class: Cabin class total_airports: Total airports scanned elapsed_time: Execution time in seconds """ if not HAS_RICH: print("Rich library not installed, using plain text output") _format_plain_seasonal(results_by_month, new_connections, destination, country, seat_class, total_airports, elapsed_time) return console = Console() # Print header console.print() console.print( f"Flight Comparator: {country} → {destination} | " f"Seasonal scan: {len(results_by_month)} months | {seat_class.title()}", style="bold cyan" ) console.print() # Process each month for month in sorted(results_by_month.keys()): month_results = results_by_month[month] # Create month header console.print(f"[bold yellow]{month.upper()}[/bold yellow]") console.print() # Flatten flights for this month all_flights = [] for airport_code, flights in month_results.items(): for flight in flights: flight['airport_code'] = airport_code all_flights.append(flight) # Sort by price all_flights.sort(key=lambda f: f.get('price', 999999)) # Show top results for idx, flight in enumerate(all_flights[:5], 1): # Top 5 per month airport_code = flight['airport_code'] from_text = f"{airport_code} {flight.get('city', '')}" airline = flight.get('airline', 'Unknown') price = f"{flight.get('currency', '€')}{flight.get('price', 0):.0f}" # Check if this is a new connection route_key = f"{airport_code}->{destination}" is_new = route_key in new_connections and new_connections[route_key] == month market = "✨ NEW" if is_new else "Typical" console.print( f"{idx} │ {from_text:20} │ {airline:15} │ {price:8} │ {market}" ) console.print() console.print() # Summary if new_connections: console.print("[bold]New connections detected:[/bold]") for route, first_month in sorted(new_connections.items()): console.print(f" • {route} (from {first_month})", style="green") console.print() console.print( f"Scanned {len(results_by_month)} months × {total_airports} airports • " f"Done in {elapsed_time:.1f}s", style="dim" ) console.print() def _format_plain_single_date(results, destination, country, date, seat_class, sort_by, total_airports, elapsed_time): """Plain text fallback for single-date mode.""" print() print(f"Flight Comparator: {country} → {destination} | {date} | {seat_class.title()} | Sorted by: {sort_by.title()}") print("=" * 80) print() all_flights = [] for airport_code, flights in results.items(): for flight in flights: flight['airport_code'] = airport_code all_flights.append(flight) if sort_by == "duration": all_flights.sort(key=lambda f: f.get('duration_minutes', 999999)) else: all_flights.sort(key=lambda f: f.get('price', 999999)) for idx, flight in enumerate(all_flights, 1): print(f"{idx}. {flight['airport_code']} - {flight.get('airline', 'Unknown')} - " f"{flight.get('currency', '€')}{flight.get('price', 0):.0f} - " f"{format_duration(flight.get('duration_minutes', 0))}") print() print(f"Scanned {total_airports} airports • Done in {elapsed_time:.1f}s") print() def _format_plain_seasonal(results_by_month, new_connections, destination, country, seat_class, total_airports, elapsed_time): """Plain text fallback for seasonal mode.""" print() print(f"Flight Comparator: {country} → {destination} | Seasonal scan: {len(results_by_month)} months | {seat_class.title()}") print("=" * 80) for month in sorted(results_by_month.keys()): print(f"\n{month.upper()}") print("-" * 40) month_results = results_by_month[month] all_flights = [] for airport_code, flights in month_results.items(): for flight in flights: flight['airport_code'] = airport_code all_flights.append(flight) all_flights.sort(key=lambda f: f.get('price', 999999)) for idx, flight in enumerate(all_flights[:5], 1): route_key = f"{flight['airport_code']}->{destination}" is_new = route_key in new_connections and new_connections[route_key] == month new_tag = " ✨ NEW" if is_new else "" print(f"{idx}. {flight['airport_code']} - {flight.get('airline', 'Unknown')} - " f"{flight.get('currency', '€')}{flight.get('price', 0):.0f}{new_tag}") print() if new_connections: print("New connections detected:") for route, first_month in sorted(new_connections.items()): print(f" • {route} (from {first_month})") print() print(f"Scanned {len(results_by_month)} months × {total_airports} airports • Done in {elapsed_time:.1f}s") print() def format_json(results, **kwargs) -> None: """Format results as JSON.""" print(json.dumps(results, indent=2)) def format_csv(results: dict[str, list[dict]], **kwargs) -> None: """Format results as CSV.""" writer = csv.writer(sys.stdout) writer.writerow(['Origin', 'Destination', 'Airline', 'Departure', 'Arrival', 'Duration_Min', 'Price', 'Currency']) for airport_code, flights in results.items(): for flight in flights: writer.writerow([ airport_code, flight.get('destination', ''), flight.get('airline', ''), flight.get('departure_time', ''), flight.get('arrival_time', ''), flight.get('duration_minutes', 0), flight.get('price', 0), flight.get('currency', ''), ])