""" sheets.py — Google Sheets integration for the Psychotherapist Booking Bot. Manages four worksheets: - "Appointments" : Booked sessions log [Name, Phone, Date, Time, Status, Notes] - "Availability" : Doctor session availability schedule [Day, Available Slots] - "Patients" : Patient records [Name, Phone, First Visit, Total Sessions] - "Messages" : Message log [Timestamp, Phone, Direction, Message, Response] """ import asyncio import json import logging import os from datetime import datetime, timedelta from typing import Optional import gspread import pytz from google.oauth2.service_account import Credentials from core.config import ( GOOGLE_SERVICE_ACCOUNT_JSON, GOOGLE_SHEETS_ID, ) from core.constants import ( SHEET_APPOINTMENTS, SHEET_AVAILABILITY, SHEET_MESSAGES, SHEET_PATIENTS, TIMEZONE, ) from core.schemas import AppointmentRecord, MessageLog, PatientRecord logger = logging.getLogger(__name__) _SCOPES = [ "https://www.googleapis.com/auth/spreadsheets", "https://www.googleapis.com/auth/drive", ] _gc: gspread.Client | None = None _appointments_ws: gspread.Worksheet | None = None _availability_ws: gspread.Worksheet | None = None _patients_ws: gspread.Worksheet | None = None _messages_ws: gspread.Worksheet | None = None _known_patients: set[str] = set() _availability_text: str = "" def _get_credentials() -> Credentials: value = GOOGLE_SERVICE_ACCOUNT_JSON.strip() if not value: raise ValueError("GOOGLE_SERVICE_ACCOUNT_JSON is not set.") if value.startswith("{"): info = json.loads(value) if "private_key" in info and "\\n" in info["private_key"]: info["private_key"] = info["private_key"].replace("\\n", "\n") return Credentials.from_service_account_info(info, scopes=_SCOPES) if os.path.isfile(value): return Credentials.from_service_account_file(value, scopes=_SCOPES) raise FileNotFoundError(f"Service account file not found: {value}") def _get_or_create_worksheet( spreadsheet: gspread.Spreadsheet, title: str, headers: list[str], rows: int = 1000, cols: int = 10, ) -> gspread.Worksheet | None: try: return spreadsheet.worksheet(title) except gspread.WorksheetNotFound: try: ws = spreadsheet.add_worksheet(title=title, rows=rows, cols=cols) ws.append_row(headers, value_input_option="USER_ENTERED") logger.info("Created '%s' worksheet", title) return ws except Exception as exc: logger.warning("Could not create worksheet '%s': %s", title, exc) try: return spreadsheet.sheet1 except Exception: return None def _connect() -> None: global _gc, _appointments_ws, _availability_ws, _patients_ws, _messages_ws global _known_patients, _availability_text try: creds = _get_credentials() _gc = gspread.authorize(creds) spreadsheet = _gc.open_by_key(GOOGLE_SHEETS_ID) except Exception as exc: logger.warning("❌ Google Sheets connection skipped or failed: %s", exc) return _appointments_ws = _get_or_create_worksheet( spreadsheet, SHEET_APPOINTMENTS, ["Day", "Branch", "Date", "Time", "Status", "Name", "Phone", "Notes"], rows=2000, cols=8, ) _availability_ws = _get_or_create_worksheet( spreadsheet, SHEET_AVAILABILITY, ["Day", "Available Slots"], rows=100, cols=2, ) _patients_ws = _get_or_create_worksheet( spreadsheet, SHEET_PATIENTS, ["Name", "Phone", "First Visit", "Total Sessions"], rows=1000, cols=4, ) _messages_ws = _get_or_create_worksheet( spreadsheet, SHEET_MESSAGES, ["Timestamp", "Phone", "Direction", "Message", "Response"], rows=5000, cols=5, ) # Populate known patients cache if _patients_ws is not None: try: records = _patients_ws.get_all_records() for row in records: phone = str(row.get("Phone", "")).strip() if phone: _known_patients.add(phone) except Exception as exc: logger.warning("Could not pre-fill patients cache: %s", exc) ensure_upcoming_weeks_seeded(weeks_ahead=2) auto_merge_appointments_sheet() sync_availability_from_appointments() logger.info("Connected to Google Sheets. Known patients: %d", len(_known_patients)) def auto_merge_appointments_sheet() -> None: """Dynamically merge vertically Day (Col A), Branch (Col B), and Date (Col C) for whatever number of sessions exist for each date in ONE batch API request.""" if not _appointments_ws: return try: records = _appointments_ws.get_all_records() if not records: return # Pre-resolve exact (Date, Day) for every row using carry-forward resolved_dates = [] resolved_days = [] last_d = "" last_day = "" for r in records: d = str(r.get("Date", "")).strip() day = str(r.get("Day", "")).strip() if d: last_d = d if day: last_day = day resolved_dates.append(last_d) resolved_days.append(last_day) sheet_id = _appointments_ws.id requests = [] i = 0 n = len(records) while i < n: target_date = resolved_dates[i] target_day = resolved_days[i] if not (target_date and target_day): i += 1 continue j = i + 1 while j < n and resolved_dates[j] == target_date and resolved_days[j] == target_day: j += 1 row_count = j - i if row_count > 1: start_row_idx = i + 1 # 0-indexed: record 0 is row index 1 (row index 0 is header) end_row_idx = j + 1 # 0-indexed exclusive: record j-1 is row index j, so end index is j+1 # Merge Column A (0), Column B (1), Column C (2) vertically (each column separately) for col_idx in range(3): # A=0, B=1, C=2 requests.append({ "mergeCells": { "range": { "sheetId": sheet_id, "startRowIndex": start_row_idx, "endRowIndex": end_row_idx, "startColumnIndex": col_idx, "endColumnIndex": col_idx + 1, }, "mergeType": "MERGE_ALL", } }) # Format Columns A-C: Middle vertical alignment, Center horizontal alignment requests.append({ "repeatCell": { "range": { "sheetId": sheet_id, "startRowIndex": start_row_idx, "endRowIndex": end_row_idx, "startColumnIndex": 0, "endColumnIndex": 3, }, "cell": { "userEnteredFormat": { "verticalAlignment": "MIDDLE", "horizontalAlignment": "CENTER", } }, "fields": "userEnteredFormat(verticalAlignment,horizontalAlignment)", } }) i = j if requests: _appointments_ws.spreadsheet.batch_update({"requests": requests}) logger.info("Auto-merged and formatted %d session day blocks in 1 batch API call.", len(requests) // 2) except Exception as exc: logger.warning("Could not auto-merge Appointments sheet cells: %s", exc) def ensure_upcoming_weeks_seeded(weeks_ahead: int = 2) -> None: """ Ensure the Appointments sheet has pre-filled session rows for the current week and up to `weeks_ahead` future weeks. Accurately maps Day names (السبت, الأحد, ...) to their actual calendar dates. """ if not _appointments_ws: return try: default_times = ["5:30 - 6:30 PM", "6:30 - 7:30 PM", "7:30 - 8:30 PM", "8:30 - 9:30 PM"] friday_times = ["5:30 - 6:30 PM", "6:30 - 7:30 PM", "7:30 - 8:30 PM"] weekly_schedule = [ ("السبت", "مستشفى الأمل", 5, default_times), ("الأحد", "مستشفى الأمل", 6, default_times), ("الإثنين", "مستشفى الخير والبركة", 0, default_times), ("الثلاثاء", "مستشفى الخير والبركة", 1, default_times), ("الأربعاء", "مستشفى الخير والبركة", 2, default_times), ("الخميس", "مستشفى الخير والبركة", 3, default_times), ("الجمعة", "أونلاين", 4, friday_times), ] tz = pytz.timezone(TIMEZONE) today = datetime.now(tz).date() # Find current Saturday (start of current clinic week in Egypt) days_since_saturday = (today.weekday() - 5) % 7 current_saturday = today - timedelta(days=days_since_saturday) # Get existing (Date, Time) pairs in Appointments sheet to avoid duplicates records = _appointments_ws.get_all_records() existing_slots = set() last_d = "" for r in records: raw_d = str(r.get("Date", "")).strip() if raw_d: last_d = raw_d d = last_d t = str(r.get("Time", "")).strip() if d and t: existing_slots.add((d, t)) new_rows = [] for week_offset in range(weeks_ahead): week_start_sat = current_saturday + timedelta(weeks=week_offset) for day_name, branch, weekday_idx, day_times in weekly_schedule: offset_from_sat = (weekday_idx - 5) % 7 slot_date = (week_start_sat + timedelta(days=offset_from_sat)).strftime("%Y-%m-%d") for t in day_times: if (slot_date, t) not in existing_slots: new_rows.append([day_name, branch, slot_date, t, "Available", "", "", ""]) if new_rows: for row in new_rows: _appointments_ws.append_row(row, value_input_option="USER_ENTERED") logger.info("Auto-seeded %d session rows into Appointments sheet.", len(new_rows)) except Exception as exc: logger.warning("Could not auto-seed upcoming weeks: %s", exc) def auto_complete_past_appointments() -> None: """ Check all 'Reserved' appointments in the Appointments sheet. If the appointment's end time has passed, automatically update Status to 'Completed'. """ if not _appointments_ws: return try: records = _appointments_ws.get_all_records() if not records: return tz = pytz.timezone(TIMEZONE) now = datetime.now(tz) last_d = "" completed_indices = [] for idx, r in enumerate(records, start=2): # Row 1 is header status = str(r.get("Status", "")).strip().lower() if status in ("reserved", "محجوز"): raw_d = str(r.get("Date", "")).strip() if raw_d: last_d = raw_d date_str = last_d time_str = str(r.get("Time", "")).strip() if not (date_str and time_str): continue try: if "-" in time_str: end_time_part = time_str.split("-")[-1].strip() else: end_time_part = time_str.strip() dt_str = f"{date_str} {end_time_part}" slot_end_dt = None for fmt in ("%Y-%m-%d %I:%M %p", "%Y-%m-%d %H:%M", "%Y-%m-%d %I:%M%p"): try: naive_dt = datetime.strptime(dt_str, fmt) slot_end_dt = tz.localize(naive_dt) break except ValueError: continue if slot_end_dt and now > slot_end_dt: completed_indices.append(idx) except Exception as parse_err: logger.debug("Error parsing appointment time '%s %s': %s", date_str, time_str, parse_err) for row_idx in completed_indices: try: _appointments_ws.update_cell(row_idx, 5, "Completed") logger.info("Auto-completed past appointment at row %d.", row_idx) except Exception as update_err: logger.warning("Could not update row %d to Completed: %s", row_idx, update_err) except Exception as exc: logger.warning("Error in auto_complete_past_appointments: %s", exc) def sync_availability_from_appointments() -> str: """ Automatically detect available dates/slots from the 'Appointments' sheet (Status == 'Available' or 'متاح' or empty), group them by (Day, Branch, Date), update the 'Availability' sheet, and refresh _availability_text for the bot. Handles merged cells gracefully and ensures upcoming weeks are seeded. """ global _availability_text if _appointments_ws is None: return _availability_text try: ensure_upcoming_weeks_seeded(weeks_ahead=2) auto_complete_past_appointments() records = _appointments_ws.get_all_records() if not records: return _availability_text # Map: (Day, Branch, Date) -> list of available time slots avail_map: dict[tuple[str, str, str], list[str]] = {} last_day = "" last_branch = "" last_date = "" tz = pytz.timezone(TIMEZONE) today_date = datetime.now(tz).date() max_date = today_date + timedelta(days=6) # Upcoming 7 days (today through today+6) for r in records: raw_day = str(r.get("Day", "")).strip() raw_branch = str(r.get("Branch", "")).strip() raw_date = str(r.get("Date", "")).strip() if raw_day: last_day = raw_day if raw_branch: last_branch = raw_branch if raw_date: last_date = raw_date day = last_day branch = last_branch date = last_date time_slot = str(r.get("Time", "")).strip() status = str(r.get("Status", "Available")).strip().lower() # Filter to 7-day window regardless of status if date: try: slot_d = datetime.strptime(date, "%Y-%m-%d").date() if not (today_date <= slot_d <= max_date): continue except ValueError: pass if status in ("available", "متاح", ""): key = (day, branch, date) if key not in avail_map: avail_map[key] = [] if time_slot: avail_map[key].append(time_slot) if not avail_map: _availability_text = "لا توجد مواعيد متاحة حالياً." rows_to_write = [] else: text_lines = [] rows_to_write = [] for (day, branch, date), slots in avail_map.items(): slots_str = ", ".join(slots) header_parts = [] if day: header_parts.append(day) if date: header_parts.append(date) if branch: header_parts.append(f"({branch})") label = " ".join(header_parts) text_lines.append(f"{label}: {slots_str}") rows_to_write.append([day, branch, date, slots_str]) _availability_text = " | ".join(text_lines) # Update the 'Availability' worksheet in 1 fast batch call if _availability_ws is not None: try: _availability_ws.clear() all_avail_rows = [["Day", "Branch", "Date", "Available Slots"]] if rows_to_write: all_avail_rows.extend(rows_to_write) else: all_avail_rows.append(["-", "-", "-", "لا توجد مواعيد متاحة حالياً"]) _availability_ws.append_rows(all_avail_rows, value_input_option="USER_ENTERED") logger.info("Synced %d available date groups to 'Availability' sheet.", len(rows_to_write)) except Exception as sync_err: logger.warning("Could not sync to 'Availability' worksheet: %s", sync_err) except Exception as exc: logger.error("Failed to sync availability from Appointments: %s", exc) return _availability_text def _ensure_connected() -> None: if _appointments_ws is None or _patients_ws is None: _connect() async def initialize() -> None: await asyncio.to_thread(_ensure_connected) logger.info("Google Sheets initialized.") def get_availability_text() -> str: sync_availability_from_appointments() return _availability_text def _clean_digits(s: str) -> str: return "".join(c for c in str(s) if c.isdigit()) def is_existing_patient(phone: str) -> bool: target = _clean_digits(phone) for p in _known_patients: if _clean_digits(p) == target: return True return False def get_patient_details(phone: str) -> Optional[dict]: if _patients_ws is None: return None try: target_digits = _clean_digits(phone) if not target_digits: return None records = _patients_ws.get_all_records() for row in records: if _clean_digits(row.get("Phone", "")) == target_digits: return { "name": str(row.get("Name", "")), "phone": str(row.get("Phone", "")), "first_visit": str(row.get("First Visit", "")), "total_sessions": row.get("Total Sessions", 1), } except Exception as exc: logger.warning("Error fetching patient details for %s: %s", phone, exc) return None async def save_appointment( name: str, phone: str, date: str, time: str, is_first: bool = True, notes: str = "", ) -> None: """Save appointment to Appointments tab and update/create Patients tab record.""" def _write() -> None: _ensure_connected() note_str = "1st session" if is_first else "Returning patient" if notes: note_str += f" | {notes}" matched_row_idx = None if _appointments_ws: try: records = _appointments_ws.get_all_records() last_d = "" for idx, r in enumerate(records, start=2): raw_d = str(r.get("Date", "")).strip() if raw_d: last_d = raw_d r_date = last_d r_time = str(r.get("Time", "")).strip() if r_date and r_time and (r_date == date and (r_time in time or time in r_time)): matched_row_idx = idx break except Exception as exc: logger.warning("Error searching Appointments tab: %s", exc) if matched_row_idx: _appointments_ws.update_cell(matched_row_idx, 5, "Reserved") _appointments_ws.update_cell(matched_row_idx, 6, name) _appointments_ws.update_cell(matched_row_idx, 7, phone) _appointments_ws.update_cell(matched_row_idx, 8, note_str) else: _appointments_ws.append_row(["", "", date, time, "Reserved", name, phone, note_str], value_input_option="USER_ENTERED") # Update or add in Patients tab target_digits = _clean_digits(phone) records = _patients_ws.get_all_records() patient_row_idx = None current_sessions = 0 for idx, row in enumerate(records, start=2): # Header is row 1 if _clean_digits(row.get("Phone", "")) == target_digits: patient_row_idx = idx try: current_sessions = int(row.get("Total Sessions", 0)) except ValueError: current_sessions = 0 break if patient_row_idx: # Update existing patient's total sessions _patients_ws.update_cell(patient_row_idx, 4, current_sessions + 1) else: # Create new patient record pat = PatientRecord( name=name, phone=phone, first_visit=date if date else datetime.now().strftime("%Y-%m-%d"), total_sessions=1, ) _patients_ws.append_row(pat.to_row(), value_input_option="USER_ENTERED") _known_patients.add(phone) logger.info("Appointment booked for %s (%s) on %s %s", name, phone, date, time) sync_availability_from_appointments() await asyncio.to_thread(_write) async def log_message( phone: str, direction: str, message: str, response: str, timestamp: str, ) -> None: def _write() -> None: _ensure_connected() log = MessageLog( timestamp=timestamp, phone=phone, direction=direction, message=message, response=response, ) _messages_ws.append_row(log.to_row(), value_input_option="USER_ENTERED") await asyncio.to_thread(_write) async def doctor_cancel_appointment(date: str = "", time: str = "", identifier: str = "") -> bool: """Cancel an appointment matching date, time, or patient name/phone.""" def _do_cancel() -> bool: _ensure_connected() if not _appointments_ws: return False try: records = _appointments_ws.get_all_records() clean_id = _clean_digits(identifier) last_d = "" for idx, r in enumerate(records, start=2): raw_d = str(r.get("Date", "")).strip() if raw_d: last_d = raw_d r_date = last_d r_time = str(r.get("Time", "")).strip() r_name = str(r.get("Name", "")).strip().lower() r_phone = _clean_digits(r.get("Phone", "")) match_id = (clean_id and clean_id in r_phone) or (identifier and identifier.lower() in r_name) match_dt = (date and date == r_date) and (not time or time in r_time or r_time in time) if match_id or match_dt: _appointments_ws.update_cell(idx, 5, "Cancelled") logger.info("Doctor cancelled appointment at row %d", idx) sync_availability_from_appointments() return True except Exception as exc: logger.error("Error in doctor_cancel_appointment: %s", exc) return False return await asyncio.to_thread(_do_cancel) async def doctor_block_slot(date: str = "", time: str = "", day: str = "") -> int: """Block single or all slots for a specific date or day.""" def _do_block() -> int: _ensure_connected() if not _appointments_ws: return 0 blocked_count = 0 try: records = _appointments_ws.get_all_records() last_d = "" last_day_name = "" for idx, r in enumerate(records, start=2): raw_d = str(r.get("Date", "")).strip() raw_day = str(r.get("Day", "")).strip().lower() if raw_d: last_d = raw_d if raw_day: last_day_name = raw_day r_date = last_d r_day = last_day_name r_time = str(r.get("Time", "")).strip() match_day = day and (day.lower() in r_day or r_day in day.lower()) match_date = date and (date == r_date) match_time = not time or (time in r_time or r_time in time) if (match_date or match_day) and match_time: _appointments_ws.update_cell(idx, 5, "Blocked") blocked_count += 1 if blocked_count > 0: sync_availability_from_appointments() except Exception as exc: logger.error("Error in doctor_block_slot: %s", exc) return blocked_count return await asyncio.to_thread(_do_block) async def doctor_add_note(identifier: str = "", date: str = "", note_text: str = "") -> bool: """Add or append notes to an appointment row.""" def _do_note() -> bool: _ensure_connected() if not _appointments_ws: return False try: records = _appointments_ws.get_all_records() clean_id = _clean_digits(identifier) last_d = "" for idx, r in enumerate(records, start=2): raw_d = str(r.get("Date", "")).strip() if raw_d: last_d = raw_d r_date = last_d r_name = str(r.get("Name", "")).strip().lower() r_phone = _clean_digits(r.get("Phone", "")) match_id = (clean_id and clean_id in r_phone) or (identifier and identifier.lower() in r_name) match_date = date and (date == r_date) if match_id or match_date: existing_notes = str(r.get("Notes", "")).strip() new_notes = f"{existing_notes} | {note_text}" if existing_notes else note_text _appointments_ws.update_cell(idx, 8, new_notes) logger.info("Added note to row %d", idx) return True except Exception as exc: logger.error("Error in doctor_add_note: %s", exc) return False return await asyncio.to_thread(_do_note)