84 lines
3.4 KiB
Python
84 lines
3.4 KiB
Python
"""
|
|
Service-related CLI commands.
|
|
"""
|
|
|
|
import argparse
|
|
from datetime import date
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from backend.cli.base import NimbusFlowCLI
|
|
|
|
from backend.models.enums import ScheduleStatus
|
|
from backend.cli.utils import format_service_row, create_table_header, create_table_separator, TableColors
|
|
|
|
|
|
def cmd_services_list(cli: "NimbusFlowCLI", args) -> None:
|
|
"""List services with optional filters."""
|
|
# Apply filters
|
|
if args.upcoming:
|
|
services = cli.service_repo.upcoming(limit=args.limit or 50)
|
|
elif args.date:
|
|
try:
|
|
target_date = date.fromisoformat(args.date)
|
|
# Get services for specific date
|
|
all_services = cli.service_repo.list_all()
|
|
services = [s for s in all_services if s.ServiceDate == target_date]
|
|
except ValueError:
|
|
print(f"{TableColors.ERROR}Invalid date format '{args.date}'. Use YYYY-MM-DD{TableColors.RESET}")
|
|
return
|
|
else:
|
|
services = cli.service_repo.list_all()
|
|
if args.limit:
|
|
services = services[:args.limit]
|
|
|
|
if not services:
|
|
print(f"{TableColors.DIM}No services found.{TableColors.RESET}")
|
|
return
|
|
|
|
# Get service type names for display
|
|
service_type_map = {st.ServiceTypeId: st.TypeName for st in cli.service_type_repo.list_all()}
|
|
|
|
# Get schedule counts for each service
|
|
schedule_counts = {}
|
|
for service in services:
|
|
schedules = cli.schedule_repo.list_all()
|
|
service_schedules = [s for s in schedules if s.ServiceId == service.ServiceId]
|
|
|
|
pending_count = len([s for s in service_schedules if s.Status == ScheduleStatus.PENDING.value])
|
|
accepted_count = len([s for s in service_schedules if s.Status == ScheduleStatus.ACCEPTED.value])
|
|
declined_count = len([s for s in service_schedules if s.Status == ScheduleStatus.DECLINED.value])
|
|
|
|
schedule_counts[service.ServiceId] = {
|
|
'pending': pending_count,
|
|
'accepted': accepted_count,
|
|
'declined': declined_count,
|
|
'total': len(service_schedules)
|
|
}
|
|
|
|
# Print styled header
|
|
print()
|
|
print(create_table_header("ID ", "Date ", "Service Type ", "Total", "Pending", "Accepted", "Declined"))
|
|
print(create_table_separator(85))
|
|
|
|
# Print services
|
|
for service in sorted(services, key=lambda s: s.ServiceDate):
|
|
type_name = service_type_map.get(service.ServiceTypeId, "Unknown")
|
|
counts = schedule_counts.get(service.ServiceId, {'total': 0, 'pending': 0, 'accepted': 0, 'declined': 0})
|
|
print(format_service_row(service, type_name, counts))
|
|
|
|
print()
|
|
print(f"{TableColors.SUCCESS}Total: {len(services)} services{TableColors.RESET}")
|
|
|
|
|
|
def setup_services_parser(subparsers) -> None:
|
|
"""Set up service-related command parsers."""
|
|
# Services commands
|
|
services_parser = subparsers.add_parser("services", help="Manage services")
|
|
services_subparsers = services_parser.add_subparsers(dest="services_action", help="Services actions")
|
|
|
|
# services list
|
|
services_list_parser = services_subparsers.add_parser("list", help="List services")
|
|
services_list_parser.add_argument("--date", type=str, help="Filter by specific date (YYYY-MM-DD)")
|
|
services_list_parser.add_argument("--upcoming", action="store_true", help="Show only upcoming services")
|
|
services_list_parser.add_argument("--limit", type=int, help="Limit number of results") |