crayonix/nimbusflow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(backend): add ability to schedule member in cli

Browse Source
This commit is contained in:
Giovani
2025-08-27 23:55:29 -04:00
parent 1379998e5b
commit 6763a31a41
7 changed files with 285 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
backend/cli/base.py
View File

@@ -12,6 +12,7 @@ from backend.repositories import (
ScheduleRepository,
ServiceTypeRepository
)
from backend.services.scheduling_service import SchedulingService
class CLIError(Exception):
@@ -39,6 +40,15 @@ class NimbusFlowCLI:
self.availability_repo = ServiceAvailabilityRepository(self.db)
self.schedule_repo = ScheduleRepository(self.db)
self.service_type_repo = ServiceTypeRepository(self.db)
# Initialize scheduling service
self.scheduling_service = SchedulingService(
classification_repo=self.classification_repo,
member_repo=self.member_repo,
service_repo=self.service_repo,
availability_repo=self.availability_repo,
schedule_repo=self.schedule_repo,
)
def close(self):
"""Clean up database connection."""

4
backend/cli/commands/__init__.py
View File

@@ -5,7 +5,7 @@ CLI command modules.
from .members import cmd_members_list, cmd_members_show, setup_members_parser
from .schedules import (
cmd_schedules_list, cmd_schedules_show, cmd_schedules_accept,
cmd_schedules_decline, setup_schedules_parser
cmd_schedules_decline, cmd_schedules_schedule, setup_schedules_parser
)
from .services import cmd_services_list, setup_services_parser
@@ -14,7 +14,7 @@ __all__ = [
"cmd_members_list", "cmd_members_show", "setup_members_parser",
# Schedule commands
"cmd_schedules_list", "cmd_schedules_show", "cmd_schedules_accept",
"cmd_schedules_decline", "setup_schedules_parser",
"cmd_schedules_decline", "cmd_schedules_schedule", "setup_schedules_parser",
# Service commands
"cmd_services_list", "setup_services_parser",
]

180
backend/cli/commands/schedules.py
View File

@@ -357,6 +357,178 @@ def cmd_schedules_decline(cli: "NimbusFlowCLI", args) -> None:
print(f" Reason: {decline_reason}")
def cmd_schedules_schedule(cli: "NimbusFlowCLI", args) -> None:
"""Schedule next member for a service with preview and confirmation."""
# Determine if we're using service ID or date-based selection
service = None
if hasattr(args, 'date') and args.date:
# Date-based selection
try:
target_date = date.fromisoformat(args.date)
except ValueError:
print(f"❌ Invalid date format '{args.date}'. Use YYYY-MM-DD")
return
# Find services for the specified date
all_services = cli.service_repo.list_all()
services_on_date = [s for s in all_services if s.ServiceDate == target_date]
if not services_on_date:
print(f"❌ No services found for {args.date}")
return
# Get service types for display
service_type_map = {st.ServiceTypeId: st.TypeName for st in cli.service_type_repo.list_all()}
# Show available services for the date
print(f"\n📅 Services available on {args.date}:")
print("-" * 50)
for i, svc in enumerate(services_on_date, 1):
type_name = service_type_map.get(svc.ServiceTypeId, "Unknown")
print(f"{i}. {type_name} (Service ID: {svc.ServiceId})")
# Let user select service
try:
choice = input(f"\nSelect service (1-{len(services_on_date)}): ").strip()
if not choice or not choice.isdigit():
print("❌ Invalid selection")
return
service_index = int(choice) - 1
if service_index < 0 or service_index >= len(services_on_date):
print("❌ Invalid selection")
return
service = services_on_date[service_index]
except (KeyboardInterrupt, EOFError):
print("\n🛑 Operation cancelled")
return
elif hasattr(args, 'service_id') and args.service_id:
# Service ID based selection
print(f"Scheduling for service ID {args.service_id}...")
service = cli.service_repo.get_by_id(args.service_id)
if not service:
print(f"❌ Service ID {args.service_id} not found.")
return
else:
print("❌ Either --date or service_id must be provided")
return
# Get service type name
service_type = cli.service_type_repo.get_by_id(service.ServiceTypeId)
service_type_name = service_type.TypeName if service_type else "Unknown"
print(f"\n📅 Selected Service: {service_type_name} on {service.ServiceDate}")
# Get classification constraints if not provided
classification_ids = []
if args.classifications:
# Convert classification names to IDs
all_classifications = cli.classification_repo.list_all()
classification_map = {c.ClassificationName.lower(): c.ClassificationId for c in all_classifications}
for class_name in args.classifications:
class_id = classification_map.get(class_name.lower())
if class_id:
classification_ids.append(class_id)
else:
print(f"❌ Unknown classification: {class_name}")
return
else:
# If no classifications specified, ask user to select
all_classifications = cli.classification_repo.list_all()
print("\n🎵 Available classifications:")
for i, classification in enumerate(all_classifications, 1):
print(f" {i}. {classification.ClassificationName}")
while True:
try:
choice = input(f"\n🎯 Select classification(s) (1-{len(all_classifications)}, comma-separated): ").strip()
if not choice:
continue
selections = [int(x.strip()) for x in choice.split(',')]
if all(1 <= sel <= len(all_classifications) for sel in selections):
classification_ids = [all_classifications[sel-1].ClassificationId for sel in selections]
break
else:
print(f"❌ Please enter numbers between 1 and {len(all_classifications)}")
except ValueError:
print("❌ Please enter valid numbers")
except (KeyboardInterrupt, EOFError):
print("\n🛑 Operation cancelled")
return
# Get selected classification names for display
selected_classifications = [c.ClassificationName for c in cli.classification_repo.list_all()
if c.ClassificationId in classification_ids]
print(f"\n🔍 Looking for eligible members in: {', '.join(selected_classifications)}")
excluded_members = set()
while True:
# Preview who would be scheduled (excluding any we've already shown)
preview_result = cli.scheduling_service.preview_next_member(
classification_ids=classification_ids,
service_id=service.ServiceId,
only_active=True,
exclude_member_ids=excluded_members,
)
if not preview_result:
if excluded_members:
print("❌ No more eligible members found for this service.")
else:
print("❌ No eligible members found for this service.")
return
member_id, first_name, last_name = preview_result
# Show preview
print(f"\n✨ Next available member:")
print(f" 👤 {first_name} {last_name} (ID: {member_id})")
# Confirm scheduling
try:
confirm = input(f"\n🤔 Schedule {first_name} {last_name} for this service? (y/N/q to quit): ").strip().lower()
if confirm in ['y', 'yes']:
# Actually create the schedule
result = cli.scheduling_service.schedule_next_member(
classification_ids=classification_ids,
service_id=service.ServiceId,
only_active=True,
exclude_member_ids=excluded_members,
)
if result:
scheduled_member_id, scheduled_first, scheduled_last, schedule_id = result
print(f"\n✅ Successfully scheduled {scheduled_first} {scheduled_last}!")
print(f" 📋 Schedule ID: {schedule_id}")
print(f" 📧 Status: Pending (awaiting member response)")
else:
print("❌ Failed to create schedule. The member may no longer be eligible.")
return
elif confirm in ['q', 'quit']:
print("🛑 Scheduling cancelled")
return
else:
# User declined this member - add to exclusion list and continue
excluded_members.add(member_id)
print(f"⏭️ Skipping {first_name} {last_name}, looking for next member...")
except (KeyboardInterrupt, EOFError):
print("\n🛑 Operation cancelled")
return
def setup_schedules_parser(subparsers) -> None:
"""Set up schedule-related command parsers."""
# Schedules commands
@@ -381,4 +553,10 @@ def setup_schedules_parser(subparsers) -> None:
schedules_decline_parser = schedules_subparsers.add_parser("decline", help="Decline a scheduled position")
schedules_decline_parser.add_argument("schedule_id", type=int, nargs="?", help="Schedule ID to decline (optional if using --date)")
schedules_decline_parser.add_argument("--date", type=str, help="Interactive mode: select service and member by date (YYYY-MM-DD)")
schedules_decline_parser.add_argument("--reason", type=str, help="Reason for declining")
schedules_decline_parser.add_argument("--reason", type=str, help="Reason for declining")
# schedules schedule
schedules_schedule_parser = schedules_subparsers.add_parser("schedule", help="Schedule next member for a service (cycles through eligible members)")
schedules_schedule_parser.add_argument("service_id", type=int, nargs="?", help="Service ID to schedule for (optional if using --date)")
schedules_schedule_parser.add_argument("--date", type=str, help="Interactive mode: select service by date (YYYY-MM-DD)")
schedules_schedule_parser.add_argument("--classifications", nargs="*", help="Classification names to filter by (e.g., Soprano Alto)")

16
backend/cli/interactive.py
View File

@@ -11,7 +11,7 @@ if TYPE_CHECKING:
from .commands import (
cmd_members_list, cmd_members_show,
cmd_schedules_list, cmd_schedules_show, cmd_schedules_accept, cmd_schedules_decline,
cmd_schedules_list, cmd_schedules_show, cmd_schedules_accept, cmd_schedules_decline, cmd_schedules_schedule,
cmd_services_list,
)
@@ -100,7 +100,8 @@ def display_schedules_menu():
print(" 5️⃣ 👤 Show schedule details")
print(" 6️⃣ ✨ Accept a schedule (interactive)")
print(" 7️⃣ 🚫 Decline a schedule (interactive)")
print(" 8️⃣ 🔙 Back to main menu")
print(" 8️⃣ 📅 Schedule next member for service")
print(" 9️⃣ 🔙 Back to main menu")
print()
@@ -215,7 +216,7 @@ def handle_schedules_menu(cli: "NimbusFlowCLI"):
"""Handle schedules menu interactions."""
while True:
display_schedules_menu()
choice = get_user_choice(8)
choice = get_user_choice(9)
if choice == 1: # List all schedules
print("\n🔍 Listing all schedules...")
@@ -261,7 +262,14 @@ def handle_schedules_menu(cli: "NimbusFlowCLI"):
cmd_schedules_decline(cli, MockArgs(date=date, schedule_id=None, reason=reason))
input("\n⏸️ Press Enter to continue...")
elif choice == 8: # Back to main menu
elif choice == 8: # Schedule next member
date = get_date_input("Enter date to schedule for")
if date:
print(f"\n📅 Starting scheduling for {date}...")
cmd_schedules_schedule(cli, MockArgs(service_id=None, date=date, classifications=None))
input("\n⏸️ Press Enter to continue...")
elif choice == 9: # Back to main menu
break

6
backend/cli/main.py
View File

@@ -11,7 +11,7 @@ from .commands import (
cmd_members_list, cmd_members_show, setup_members_parser,
# Schedule commands
cmd_schedules_list, cmd_schedules_show, cmd_schedules_accept,
cmd_schedules_decline, setup_schedules_parser,
cmd_schedules_decline, cmd_schedules_schedule, setup_schedules_parser,
# Service commands
cmd_services_list, setup_services_parser,
)
@@ -81,8 +81,10 @@ def main():
cmd_schedules_accept(cli, args)
elif args.schedules_action == "decline":
cmd_schedules_decline(cli, args)
elif args.schedules_action == "schedule":
cmd_schedules_schedule(cli, args)
else:
print("❌ Unknown schedules action. Use 'list', 'show', 'accept', or 'decline'")
print("❌ Unknown schedules action. Use 'list', 'show', 'accept', 'decline', or 'schedule'")
elif args.command == "services":
if args.services_action == "list":

2
backend/main.py
View File

@@ -169,7 +169,7 @@ if __name__ == "__main__":
from backend.repositories import MemberRepository, ScheduleRepository, ServiceRepository, ServiceAvailabilityRepository
from backend.services.scheduling_service import SchedulingService
DB_PATH = Path(__file__).parent / "database6_accepts_and_declines2.db"
DB_PATH = Path(__file__).parent / "database6_accepts_and_declines3.db"
# Initialise DB connection (adjust DSN as needed)
db = DatabaseConnection(DB_PATH)

80
backend/services/scheduling_service.py
View File

@@ -7,7 +7,7 @@
from __future__ import annotations
from datetime import datetime, timezone
from typing import Optional, Tuple, List
from typing import Optional, Tuple, List, Iterable
from backend.repositories import (
ClassificationRepository,
@@ -48,7 +48,7 @@ class SchedulingService:
service_id: int,
*,
only_active: bool = True,
boost_seconds: int = 5 * 24 * 60 * 60,
boost_seconds: int = 2 * 24 * 60 * 60,
exclude_member_ids: Iterable[int] | None = None,
) -> Optional[Tuple[int, str, str, int]]:
"""
@@ -224,4 +224,78 @@ class SchedulingService:
status=ScheduleStatus.DECLINED,
reason=reason,
)
return ("created", new_sched.ScheduleId)
return ("created", new_sched.ScheduleId)
def preview_next_member(
self,
classification_ids: Iterable[int],
service_id: int,
*,
only_active: bool = True,
boost_seconds: int = 2 * 24 * 60 * 60,
exclude_member_ids: Iterable[int] | None = None,
) -> Optional[Tuple[int, str, str]]:
"""
Preview who would be scheduled next for a service without actually creating the schedule.
Same logic as schedule_next_member, but doesn't create any records.
Returns
-------
Tuple[member_id, first_name, last_name] | None
The member who would be scheduled next, or None if no eligible member found.
"""
# Same logic as schedule_next_member but without creating records
svc = self.service_repo.get_by_id(service_id)
if svc is None:
return None
service_type_id = svc.ServiceTypeId
target_date = svc.ServiceDate
excluded = set(exclude_member_ids or [])
candidates: List = self.member_repo.candidate_queue(
classification_ids=list(classification_ids),
only_active=only_active,
boost_seconds=boost_seconds,
)
for member in candidates:
member_id = member.MemberId
if member_id in excluded:
continue
if not self.availability_repo.get(member_id, service_type_id):
continue
if self.schedule_repo.has_schedule_on_date(member_id, target_date):
continue
if self.schedule_repo.has_any(
member_id,
service_id,
statuses=[ScheduleStatus.ACCEPTED],
):
continue
if self.schedule_repo.has_any(
member_id,
service_id,
statuses=[ScheduleStatus.PENDING],
):
continue
if self.schedule_repo.has_any(
member_id,
service_id,
statuses=[ScheduleStatus.DECLINED],
):
continue
# Found eligible member - return without creating schedule
return (
member_id,
member.FirstName,
member.LastName,
)
return None