from datetime import datetime, timedelta, timezone
import os
import pytz
from sqlalchemy.orm import Session
from app.common.constants import SALES_EMAIL_CREDENTIALS
from app.common.models import Advisor, Settings, SettingsHistory, User
from app.common.services.email_service import send_email
from app.common.utils import get_current_time_in_timezone
from app.repositories.base_repo import BaseRepo
from datetime import datetime
import typing
import json
import phonenumbers
from phonenumbers import (
parse,
format_number,
PhoneNumberFormat,
NumberParseException,
is_valid_number,
)
class SettingsRepo(BaseRepo):
def __init__(self, db):
super().__init__(db)
def format_phone_number(self, phone: str) -> str:
try:
phone_number = parse(phone, None)
if not is_valid_number(phone_number):
return phone
# Get raw international format: +14248423103
raw = format_number(phone_number, PhoneNumberFormat.E164)
# Format to +1 (424) 842-3103 manually
if raw.startswith("+1") and len(raw) == 12:
country_code = raw[:2] # +1
area_code = raw[2:5] # 424
central_office = raw[5:8] # 842
line_number = raw[8:] # 3103
return f"{country_code} ({area_code}) {central_office}-{line_number}"
# For non-US numbers, fallback to default INTERNATIONAL format
return format_number(phone_number, PhoneNumberFormat.INTERNATIONAL)
except NumberParseException:
return phone
# def send_appointment_email(self, appointment):
# appSchedule = datetime.fromisoformat(appointment["schedule"])
# # managers = self.db.query(Advisor).filter(Advisor.type == "manager").all()
# # manager_emails = [manager.email for manager in managers]
# # manager_emails.append("shahzad@fluten.ai")
# # manager_emails = ''
# # manager_emails.append("hananali423@gmail.com")
# # manager_emails.append("smalik@fluten.ai")
# # manager_emails.append("mimj1800@gmail.com")
# # manager_emails_str = "shahzad@fluten.ai, smalik@fluten.ai, zaidnexrupt@gmail.com, hananali423@gmail.com"
# if os.environ.get("ENVIRONMENT") == "LIVE":
# manager_emails_str = "ninda@surfcitynissan.com, Kgomez@puentehillsnissan.com, Kbarrita@coronanissan.com, dlopez@surfcitynissan.com, shahzad@fluten.ai"
# print("Live emails")
# else:
# manager_emails_str = "shahzad@fluten.ai, smalik@fluten.ai, zaidnexrupt@gmail.com, hananali423@gmail.com"
# print("Dev emails")
# print(manager_emails_str,"------------------------emails")
# pst = pytz.timezone("America/Los_Angeles")
# current_time = datetime.now(pst).strftime("%b %d, %Y %I:%M %p")
# formatted_schedule_date = appSchedule.strftime("%b %d, %Y %I:%M %p")
# email_body = (
# f"""Dear Team, A new appointment has been scheduled on DealerPulse"""
# )
# conv = self.format_phone_number(appointment["phone"])
# html_body = f"""
#
# Dear Team,
# A new appointment has been scheduled on DealerPulse:
# Timestamp: {current_time}
# Name: {appointment['name']}
# Phone: {conv}
# Schedule: {formatted_schedule_date}
#
# Regards,
# DealerPulse Team
#
# """
# send_email(
# recipient=manager_emails_str,
# subject="DealerPulse Appointment Schedule Alert [Surfcity Nissan]",
# body=email_body,
# html_body=html_body,
# credentials=SALES_EMAIL_CREDENTIALS
# )
# return True
def create_settings_history(self, name, updated_by):
self.create(SettingsHistory, {"name": name, "updated_by": updated_by})
def get_settings_history_paginated(self, names: list, page, page_size):
filter_condition = SettingsHistory.name.in_(names)
history, total_pages = self.get_all_paginated(
SettingsHistory,
page=page,
page_size=page_size,
order_by="created_at",
order_direction="desc",
filter_condition=filter_condition,
)
history_with_user = []
for record in history:
user = self.get_by_id(User, record.updated_by)
history_with_user.append(
{
"id": record.id,
"name": record.name,
"updated_by": user.name,
"updated_at": record.updated_at,
}
)
return history_with_user, total_pages
def format_setting_name(self, name: str) -> str:
"""Convert setting name from underscore format to title case with spaces"""
return name.replace("_", " ").title()
def send_advance_setting_update_alert(self, name, updated_by):
managers = self.db.query(Advisor).filter(Advisor.type == "manager").all()
manager_emails = [manager.email for manager in managers]
manager_emails.append("shahzad@fluten.ai")
manager_emails.append("hananali423@gmail.com")
manager_emails.append("smalik@fluten.ai")
manager_emails.append("mimj1800@gmail.com")
manager_emails_str = ",".join(manager_emails)
pst = pytz.timezone("America/Los_Angeles")
current_time = datetime.now(pst).strftime("%b %d, %Y %I:%M %p")
formatted_name = self.format_setting_name(name)
email_body = f"""Dear Team,
Please be informed that an update has been made on DealerPulse.
{current_time} – {updated_by} updated the {formatted_name}.
Regards,
DealerPulse Team"""
html_body = f"""
Dear Team,
Please be informed that an update has been made on DealerPulse.
Timestamp: {current_time}
User: {updated_by}
Event:
- Updated Advance Settings.
Regards,
DealerPulse Team
"""
send_email(
recipient=manager_emails_str,
subject="DealerPulse Setting Update Alert",
body=email_body,
html_body=html_body,
)
return True
def send_setting_update_alert(self, name, updated_by, change_timings):
managers = self.db.query(Advisor).filter(Advisor.type == "manager").all()
manager_emails = [manager.email for manager in managers]
manager_emails.append("shahzad@fluten.ai")
manager_emails.append("hananali423@gmail.com")
manager_emails.append("smalik@fluten.ai")
manager_emails.append("mimj1800@gmail.com")
manager_emails_str = ",".join(manager_emails)
pst = pytz.timezone("America/Los_Angeles")
current_time = datetime.now(pst).strftime("%b %d, %Y %I:%M %p")
formatted_name = self.format_setting_name(name)
if name == "OFFICE_TIMING":
formatted_name = "Office Schedule"
email_body = f"""Dear Team,
Please be informed that an update has been made on DealerPulse.
{current_time} – {updated_by} updated the {formatted_name}.
Regards,
DealerPulse Team"""
html_body = f"""
Dear Team,
Please be informed that an update has been made on DealerPulse.
Timestamp: {current_time}
User: {updated_by}
Event:
{''.join(f'- {msg}
' for msg in change_timings)}
Regards,
DealerPulse Team
"""
send_email(
recipient=manager_emails_str,
subject="DealerPulse Setting Update Alert",
body=email_body,
html_body=html_body,
)
return True
def is_company_online(self):
timezone_setting = (
self.db.query(Settings).filter(Settings.name == "TIMEZONE_OFFSET").first()
)
timezone_offset = int(timezone_setting.value)
current_time = get_current_time_in_timezone(timezone_offset)
return self.is_office_hours(current_time, self.db)
def is_sales_company_online(self):
timezone_setting = (
self.db.query(Settings).filter(Settings.name == "TIMEZONE_OFFSET").first()
)
print(timezone_setting.value, "------")
timezone_offset = int(timezone_setting.value)
print(timezone_offset, "------")
current_time = get_current_time_in_timezone(timezone_offset)
return self.is_office_hours_sales(current_time, self.db)
def is_office_hours_sales(self, current_time: datetime, db: Session) -> bool:
"""
Check if current time is within Hazel's business hours based on SALES_DEPARTMENT setting
Format expected: {'hazel': {'end_time': '14:00', 'start_time': '09:00'}}
"""
try:
# Get the sales department timing setting
office_timing = (
db.query(Settings).filter(Settings.name == "SALES_DEPARTMENT").first()
)
if not office_timing or not office_timing.value:
return False
timing = json.loads(office_timing.value)
# Extract Hazel's working hours
hazel_hours = timing.get("hazel")
if not hazel_hours:
return False
# Convert string times to time objects
start_time = datetime.strptime(hazel_hours["start_time"], "%H:%M").time()
end_time = datetime.strptime(hazel_hours["end_time"], "%H:%M").time()
# Get current time
current_time_only = current_time.time()
# Check if current time is within working hours
return start_time <= current_time_only <= end_time
except Exception as e:
print(f"Error checking office hours: {e}")
return False
def is_office_hours(self, current_time: datetime, db: Session) -> bool:
"""
Check if current time is outside business hours based on OFFICE_TIMING setting
"""
try:
office_timing = (
self.db.query(Settings).filter(Settings.name == "OFFICE_TIMING").first()
)
if not office_timing:
return False
timing = json.loads(office_timing.value)
current_day = current_time.strftime("%A").lower()
current_time_only = current_time.time()
if current_day in timing.get("weekend", []):
return False
day_timing = timing.get(current_day)
if not day_timing:
return False
start_time = datetime.strptime(day_timing["start_time"], "%H:%M").time()
end_time = datetime.strptime(day_timing["end_time"], "%H:%M").time()
return start_time <= current_time_only <= end_time
except Exception as e:
return False