88 lines
3.5 KiB
Python
88 lines
3.5 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
|
|
|
|
|
|
def cmd_services_list(cli: "NimbusFlowCLI", args) -> None:
|
|
"""List services with optional filters."""
|
|
print("Listing services...")
|
|
|
|
# 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"❌ Invalid date format '{args.date}'. Use YYYY-MM-DD")
|
|
return
|
|
else:
|
|
services = cli.service_repo.list_all()
|
|
if args.limit:
|
|
services = services[:args.limit]
|
|
|
|
if not services:
|
|
print("No services found.")
|
|
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 header
|
|
print(f"\n{'ID':<3} | {'Date':<12} | {'Service Type':<12} | {'Total':<5} | {'Pending':<7} | {'Accepted':<8} | {'Declined'}")
|
|
print("-" * 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})
|
|
|
|
# Format date properly
|
|
date_str = str(service.ServiceDate) if service.ServiceDate else "N/A"
|
|
|
|
print(f"{service.ServiceId:<3} | {date_str:<12} | {type_name:<12} | "
|
|
f"{counts['total']:<5} | {counts['pending']:<7} | {counts['accepted']:<8} | {counts['declined']}")
|
|
|
|
print(f"\nTotal: {len(services)} services")
|
|
|
|
|
|
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") |