612 lines
22 KiB
Python
Executable File
612 lines
22 KiB
Python
Executable File
import logging
|
|
from flask import request
|
|
from flask_login import login_required
|
|
from geopy.geocoders import Nominatim
|
|
from app.customer import customer
|
|
from app import db
|
|
from app.common.decorators import login_required as common_login_required
|
|
from app.common.responses import error_response, success_response
|
|
|
|
logger = logging.getLogger(__name__)
|
|
from datetime import datetime
|
|
from app.classes.cards import Card_Card
|
|
from app.classes.customer import \
|
|
Customer_Customer, \
|
|
Customer_Customer_schema,\
|
|
Customer_Description, \
|
|
Customer_Description_schema,\
|
|
Customer_Tank_Inspection_schema,\
|
|
Customer_Tank_Inspection
|
|
from app.classes.service import Service_Parts
|
|
from app.classes.admin import Admin_Company
|
|
from app.classes.auto import Auto_Delivery,Auto_Delivery_schema
|
|
from app.classes.stats_customer import Stats_Customer
|
|
from app.schemas import CreateCustomerSchema, UpdateCustomerSchema, validate_request
|
|
from app.constants import DEFAULT_TANK_SIZE_GALLONS, DEFAULT_PAGE_SIZE
|
|
import string
|
|
import secrets
|
|
|
|
|
|
def generate_random_number_string(length):
|
|
# Ensure the length is at least 1
|
|
if length < 1:
|
|
raise ValueError("Length must be at least 1")
|
|
|
|
# Generate a cryptographically secure random number string
|
|
random_number = ''.join(secrets.choice(string.digits) for _ in range(length))
|
|
|
|
return random_number
|
|
|
|
|
|
|
|
@customer.route("/all", methods=["GET"])
|
|
@common_login_required
|
|
def all_customers_around():
|
|
logger.info("GET /customer/all - Fetching all customers")
|
|
customer_list = db.session \
|
|
.query(Customer_Customer) \
|
|
.all()
|
|
customer_schema = Customer_Customer_schema(many=True)
|
|
return success_response({"customers": customer_schema.dump(customer_list)})
|
|
|
|
|
|
@customer.route("/all/<int:page>", methods=["GET"])
|
|
@common_login_required
|
|
def all_customers(page):
|
|
"""
|
|
pagination all customers
|
|
"""
|
|
logger.info(f"GET /customer/all/{page} - Fetching customers page {page}")
|
|
per_page_amount = DEFAULT_PAGE_SIZE
|
|
if page is None:
|
|
offset_limit = 0
|
|
elif page == 1:
|
|
offset_limit = 0
|
|
else:
|
|
offset_limit = (per_page_amount * page) - per_page_amount
|
|
|
|
customer_list = db.session \
|
|
.query(Customer_Customer) \
|
|
.order_by(Customer_Customer.id.desc()) \
|
|
.limit(per_page_amount).offset(offset_limit)
|
|
|
|
customer_schema = Customer_Customer_schema(many=True)
|
|
return success_response({"customers": customer_schema.dump(customer_list)})
|
|
|
|
|
|
@customer.route("/<int:customer_id>", methods=["GET"])
|
|
@common_login_required
|
|
def get_a_customer(customer_id):
|
|
"""
|
|
"""
|
|
logger.info(f"GET /customer/{customer_id} - Fetching customer")
|
|
get_customer = (db.session
|
|
.query(Customer_Customer)
|
|
.filter(Customer_Customer.id == customer_id)
|
|
.first())
|
|
customer_schema = Customer_Customer_schema(many=False)
|
|
return success_response({"customer": customer_schema.dump(get_customer)})
|
|
|
|
|
|
@customer.route("/description/<int:customer_id>", methods=["GET"])
|
|
@common_login_required
|
|
def get_a_customer_description(customer_id):
|
|
"""
|
|
|
|
"""
|
|
logger.info(f"GET /customer/description/{customer_id} - Fetching customer description")
|
|
get_customer_description = (db.session
|
|
.query(Customer_Description)
|
|
.filter(Customer_Description.customer_id == customer_id)
|
|
.first())
|
|
if get_customer_description is None:
|
|
get_customer = (db.session
|
|
.query(Customer_Customer)
|
|
.filter(Customer_Customer.id == customer_id)
|
|
.first())
|
|
|
|
new_description = Customer_Description(
|
|
customer_id = customer_id,
|
|
account_number = get_customer.account_number,
|
|
company_id = get_customer.company_id,
|
|
fill_location = None,
|
|
description = None,
|
|
)
|
|
db.session.add(new_description)
|
|
db.session.commit()
|
|
get_customer_description = (db.session
|
|
.query(Customer_Description)
|
|
.filter(Customer_Description.customer_id == customer_id)
|
|
.first())
|
|
customer_schema = Customer_Description_schema(many=False)
|
|
return success_response({"description": customer_schema.dump(get_customer_description)})
|
|
|
|
|
|
@customer.route("/tank/<int:customer_id>", methods=["GET"])
|
|
@common_login_required
|
|
def get_a_customer_tank(customer_id):
|
|
"""
|
|
|
|
"""
|
|
logger.info(f"GET /customer/tank/{customer_id} - Fetching customer tank info")
|
|
get_customer_tank = (db.session
|
|
.query(Customer_Tank_Inspection)
|
|
.filter(Customer_Tank_Inspection.customer_id == customer_id)
|
|
.first())
|
|
if get_customer_tank is None:
|
|
|
|
new_tank = Customer_Tank_Inspection(
|
|
customer_id = customer_id,
|
|
last_tank_inspection = None,
|
|
tank_status = False,
|
|
outside_or_inside = True,
|
|
tank_size = DEFAULT_TANK_SIZE_GALLONS,
|
|
)
|
|
db.session.add(new_tank)
|
|
db.session.commit()
|
|
get_customer_tank = (db.session
|
|
.query(Customer_Tank_Inspection)
|
|
.filter(Customer_Tank_Inspection.customer_id == customer_id)
|
|
.first())
|
|
customer_schema = Customer_Tank_Inspection_schema(many=False)
|
|
return success_response({"tank": customer_schema.dump(get_customer_tank)})
|
|
|
|
|
|
@customer.route("/create", methods=["POST"])
|
|
@validate_request(CreateCustomerSchema)
|
|
@common_login_required
|
|
def create_customer():
|
|
"""
|
|
Create a new customer with validated input data.
|
|
"""
|
|
logger.info("POST /customer/create - Creating new customer")
|
|
# Get validated data from request
|
|
data = request.validated_data
|
|
|
|
now = datetime.utcnow()
|
|
get_company = (db.session
|
|
.query(Admin_Company)
|
|
.filter(Admin_Company.id == 1)
|
|
.first())
|
|
|
|
random_string = generate_random_number_string(6)
|
|
|
|
made_account_number = str(get_company.account_prefix) + '-' + str(random_string)
|
|
see_if_exists = (db.session.query(Customer_Customer).filter(Customer_Customer.account_number == made_account_number).first())
|
|
|
|
if see_if_exists is not None:
|
|
random_string = generate_random_number_string(10)
|
|
made_account_number = str(get_company.account_prefix) + '-' + str(random_string)
|
|
see_if_exists = (db.session.query(Customer_Customer).filter(Customer_Customer.account_number == made_account_number).first())
|
|
|
|
if see_if_exists is not None:
|
|
random_string = generate_random_number_string(10)
|
|
made_account_number = str(get_company.account_prefix) + '-' + str(random_string)
|
|
|
|
# Use validated data instead of direct request.json access
|
|
response_customer_last_name = data["customer_last_name"]
|
|
response_customer_first_name = data["customer_first_name"]
|
|
response_customer_town = data["customer_town"]
|
|
response_customer_state = data["customer_state"]
|
|
response_customer_zip = str(data["customer_zip"])
|
|
response_customer_email = data.get("customer_email")
|
|
response_customer_home_type = data["customer_home_type"]
|
|
customer_phone_number = data.get("customer_phone_number")
|
|
customer_address = data["customer_address"]
|
|
customer_apt = data.get("customer_apt")
|
|
customer_description_msg = data.get("customer_description")
|
|
|
|
int_customer_home_type = int(response_customer_home_type)
|
|
response_customer_state = int(response_customer_state)
|
|
|
|
|
|
if response_customer_state == 0:
|
|
the_state = 'MA'
|
|
elif response_customer_state == 1:
|
|
the_state = 'RI'
|
|
elif response_customer_state == 2:
|
|
the_state = 'NH'
|
|
else:
|
|
the_state = 'MA'
|
|
|
|
|
|
|
|
|
|
geolocator = Nominatim(user_agent="auburnoil")
|
|
address_string = customer_address + ' ' + response_customer_town+ ' ' + the_state
|
|
try:
|
|
location = geolocator.geocode(address_string)
|
|
user_lat = location.latitude
|
|
user_long = location.longitude
|
|
cor_ad = True
|
|
except Exception:
|
|
user_lat = None
|
|
user_long = None
|
|
cor_ad = False
|
|
|
|
|
|
|
|
new_customer = Customer_Customer(
|
|
account_number=made_account_number,
|
|
customer_last_name=response_customer_last_name,
|
|
customer_first_name=response_customer_first_name,
|
|
customer_town=response_customer_town,
|
|
customer_state=response_customer_state,
|
|
customer_zip=response_customer_zip,
|
|
customer_first_call=now,
|
|
customer_email=response_customer_email,
|
|
customer_automatic=0,
|
|
customer_home_type=int_customer_home_type,
|
|
customer_phone_number=customer_phone_number,
|
|
customer_address=customer_address,
|
|
customer_apt=customer_apt,
|
|
company_id=1,
|
|
customer_latitude=user_lat,
|
|
customer_longitude=user_long,
|
|
correct_address=cor_ad
|
|
)
|
|
db.session.add(new_customer)
|
|
db.session.flush()
|
|
|
|
create_stats_customer = Stats_Customer(
|
|
customer_id = new_customer.id,
|
|
total_calls = 0,
|
|
service_calls_total = 0,
|
|
service_calls_total_spent = 0,
|
|
service_calls_total_profit = 0,
|
|
oil_deliveries = 0,
|
|
oil_total_gallons = 0,
|
|
oil_total_spent = 0,
|
|
oil_total_profit = 0,
|
|
)
|
|
db.session.add(create_stats_customer)
|
|
|
|
new_description = Customer_Description(
|
|
customer_id = new_customer.id,
|
|
account_number = made_account_number,
|
|
description = customer_description_msg,
|
|
fill_location=None,
|
|
company_id=1,
|
|
)
|
|
db.session.add(new_description)
|
|
|
|
new_tank = Customer_Tank_Inspection(
|
|
customer_id = new_customer.id,
|
|
last_tank_inspection=None,
|
|
tank_status = False,
|
|
outside_or_inside = True,
|
|
tank_size=DEFAULT_TANK_SIZE_GALLONS,
|
|
)
|
|
db.session.add(new_tank)
|
|
|
|
|
|
db.session.commit()
|
|
|
|
return success_response({
|
|
'user': {
|
|
'user_id': new_customer.id,
|
|
'user_name': new_customer.customer_last_name,
|
|
'user_email': new_customer.customer_email,
|
|
},
|
|
})
|
|
|
|
|
|
@customer.route("/edit/<int:customer_id>", methods=["PUT"])
|
|
@login_required
|
|
@validate_request(UpdateCustomerSchema)
|
|
def edit_customer(customer_id):
|
|
"""
|
|
"""
|
|
logger.info(f"PUT /customer/edit/{customer_id} - Editing customer")
|
|
get_customer = (db.session
|
|
.query(Customer_Customer)
|
|
.filter(Customer_Customer.id == customer_id)
|
|
.first())
|
|
|
|
if not get_customer:
|
|
return error_response("Customer not found", 404)
|
|
|
|
get_customer_description = (db.session
|
|
.query(Customer_Description)
|
|
.filter(Customer_Description.customer_id == customer_id)
|
|
.first())
|
|
|
|
data = request.validated_data
|
|
response_customer_last_name = data.get("customer_last_name")
|
|
response_customer_first_name = data.get("customer_first_name")
|
|
response_customer_town = data.get("customer_town")
|
|
response_customer_state = data.get("customer_state")
|
|
response_customer_zip = data.get("customer_zip")
|
|
response_customer_phone_number = data.get("customer_phone_number")
|
|
response_customer_email = data.get("customer_email")
|
|
response_customer_home_type = data.get("customer_home_type")
|
|
response_customer_address = data.get("customer_address")
|
|
response_customer_apt = data.get("customer_apt")
|
|
response_customer_description = data.get("customer_description")
|
|
response_customer_fill_location = data.get("customer_fill_location")
|
|
|
|
|
|
# Update description if provided
|
|
if get_customer_description is not None:
|
|
if response_customer_description is not None:
|
|
get_customer_description.description = response_customer_description
|
|
if response_customer_fill_location is not None:
|
|
get_customer_description.fill_location = response_customer_fill_location
|
|
db.session.add(get_customer_description)
|
|
|
|
# Only update fields that were provided in the request
|
|
if response_customer_last_name is not None:
|
|
get_customer.customer_last_name = response_customer_last_name
|
|
if response_customer_first_name is not None:
|
|
get_customer.customer_first_name = response_customer_first_name
|
|
if response_customer_town is not None:
|
|
get_customer.customer_town = response_customer_town
|
|
if response_customer_state is not None:
|
|
get_customer.customer_state = response_customer_state
|
|
if response_customer_zip is not None:
|
|
get_customer.customer_zip = response_customer_zip
|
|
if response_customer_phone_number is not None:
|
|
get_customer.customer_phone_number = response_customer_phone_number
|
|
if response_customer_email is not None:
|
|
get_customer.customer_email = response_customer_email
|
|
if response_customer_home_type is not None:
|
|
get_customer.customer_home_type = response_customer_home_type
|
|
if response_customer_apt is not None:
|
|
get_customer.customer_apt = response_customer_apt
|
|
|
|
# Re-geocode if address fields changed
|
|
if response_customer_address is not None or response_customer_town is not None or response_customer_state is not None:
|
|
get_customer.customer_address = response_customer_address if response_customer_address is not None else get_customer.customer_address
|
|
|
|
state_code = response_customer_state if response_customer_state is not None else get_customer.customer_state
|
|
if state_code == 0:
|
|
the_state = 'MA'
|
|
elif state_code == 1:
|
|
the_state = 'RI'
|
|
elif state_code == 2:
|
|
the_state = 'NH'
|
|
else:
|
|
the_state = 'MA'
|
|
|
|
town = response_customer_town if response_customer_town is not None else get_customer.customer_town
|
|
address = get_customer.customer_address
|
|
|
|
geolocator = Nominatim(user_agent="auburnoil")
|
|
address_string = address + ' ' + town + ' ' + the_state
|
|
try:
|
|
location = geolocator.geocode(address_string, timeout=10)
|
|
get_customer.customer_latitude = location.latitude
|
|
get_customer.customer_longitude = location.longitude
|
|
get_customer.correct_address = True
|
|
except Exception:
|
|
get_customer.customer_latitude = None
|
|
get_customer.customer_longitude = None
|
|
get_customer.correct_address = False
|
|
|
|
|
|
db.session.add(get_customer)
|
|
db.session.commit()
|
|
|
|
return success_response({
|
|
'user': {
|
|
'user_name': get_customer.customer_last_name,
|
|
'user_email': get_customer.customer_email,
|
|
},
|
|
})
|
|
|
|
|
|
@customer.route("/delete/<int:customer_id>", methods=["DELETE"])
|
|
@login_required
|
|
def delete_customer(customer_id):
|
|
"""
|
|
"""
|
|
logger.info(f"DELETE /customer/delete/{customer_id} - Deleting customer")
|
|
get_customer = (db.session
|
|
.query(Customer_Customer)
|
|
.filter(Customer_Customer.id == customer_id)
|
|
.first())
|
|
|
|
get_cards = (db.session
|
|
.query(Card_Card)
|
|
.filter(Card_Card.user_id == get_customer.id)
|
|
.first())
|
|
|
|
if get_cards is not None:
|
|
db.session.delete(get_cards)
|
|
|
|
db.session.delete(get_customer)
|
|
db.session.commit()
|
|
return success_response({
|
|
'user': {
|
|
'user_name': get_customer.customer_last_name,
|
|
'user_email': get_customer.customer_email,
|
|
},
|
|
})
|
|
|
|
|
|
@customer.route("/count", methods=["GET"])
|
|
@login_required
|
|
def customer_count():
|
|
"""
|
|
"""
|
|
logger.info("GET /customer/count - Getting customer count")
|
|
get_customer = (db.session
|
|
.query(Customer_Customer)
|
|
.count())
|
|
|
|
return success_response({'count': get_customer})
|
|
|
|
|
|
@customer.route("/automatic/status/<int:customer_id>", methods=["GET"])
|
|
@login_required
|
|
def customer_automatic_status(customer_id):
|
|
"""
|
|
"""
|
|
logger.info(f"GET /customer/automatic/status/{customer_id} - Checking auto delivery status")
|
|
get_customer = (db.session
|
|
.query(Customer_Customer)
|
|
.filter(Customer_Customer.id == customer_id)
|
|
.first())
|
|
|
|
if get_customer.customer_automatic == 1:
|
|
status = 1
|
|
|
|
if get_customer.customer_automatic == 0:
|
|
status = 0
|
|
|
|
|
|
return success_response({'status': status})
|
|
|
|
|
|
@customer.route("/automatic/deliveries", methods=["GET"])
|
|
@login_required
|
|
def get_all_automatic_deliveries():
|
|
"""
|
|
Get all automatic deliveries for the table.
|
|
"""
|
|
logger.info("GET /customer/automatic/deliveries - Fetching all auto deliveries")
|
|
try:
|
|
deliveries = Auto_Delivery.query.all()
|
|
schema = Auto_Delivery_schema(many=True)
|
|
return success_response({"deliveries": schema.dump(deliveries)})
|
|
except Exception as e:
|
|
return error_response(str(e), 500)
|
|
|
|
|
|
|
|
|
|
@customer.route("/automatic/assign/<int:customer_id>", methods=["GET"])
|
|
@login_required
|
|
def customer_automatic_assignment(customer_id):
|
|
"""
|
|
"""
|
|
logger.info(f"GET /customer/automatic/assign/{customer_id} - Toggling auto delivery assignment")
|
|
get_customer = (db.session
|
|
.query(Customer_Customer)
|
|
.filter(Customer_Customer.id == customer_id)
|
|
.first())
|
|
get_auto = (db.session
|
|
.query(Auto_Delivery)
|
|
.filter(Auto_Delivery.customer_id == customer_id)
|
|
.first())
|
|
|
|
get_main_credit_card = (db.session
|
|
.query(Card_Card)
|
|
.filter(Card_Card.user_id == customer_id)
|
|
.filter(Card_Card.main_card == True)
|
|
.first())
|
|
|
|
get_customer_tank = (db.session
|
|
.query(Customer_Tank_Inspection)
|
|
.filter(Customer_Tank_Inspection.customer_id == customer_id)
|
|
.first())
|
|
|
|
get_service_parts = (db.session
|
|
.query(Service_Parts)
|
|
.filter(Service_Parts.customer_id == customer_id)
|
|
.first())
|
|
|
|
if get_customer.customer_automatic == 1:
|
|
# customer becomes will call
|
|
get_customer.customer_automatic = 0
|
|
|
|
db.session.add(get_customer)
|
|
if get_auto is not None:
|
|
db.session.delete(get_auto)
|
|
|
|
status = 3
|
|
else:
|
|
|
|
if get_main_credit_card is None:
|
|
status = 2
|
|
return success_response({'status': status})
|
|
|
|
# customer becomes an automatic
|
|
if get_auto is None:
|
|
hot_water_value = get_service_parts.hot_water_tank if get_service_parts and get_service_parts.hot_water_tank is not None else 0
|
|
create_auto = Auto_Delivery(customer_id=customer_id,
|
|
customer_full_name=get_customer.customer_first_name + ' ' + get_customer.customer_last_name,
|
|
account_number=get_customer.account_number,
|
|
customer_town=get_customer.customer_town,
|
|
customer_state=get_customer.customer_state,
|
|
customer_zip=get_customer.customer_zip,
|
|
customer_address=get_customer.customer_address,
|
|
last_fill=None,
|
|
last_updated=None,
|
|
estimated_gallons_left=0,
|
|
estimated_gallons_left_prev_day=0,
|
|
tank_height=0,
|
|
tank_size=get_customer_tank.tank_size,
|
|
house_factor=1,
|
|
auto_status=1,
|
|
days_since_last_fill=0,
|
|
hot_water_summer=hot_water_value
|
|
)
|
|
|
|
db.session.add(create_auto)
|
|
|
|
|
|
get_customer.customer_automatic = 1
|
|
db.session.add(get_customer)
|
|
|
|
status = 1
|
|
|
|
db.session.commit()
|
|
|
|
return success_response({'status': status})
|
|
|
|
|
|
@customer.route("/edit/tank/<int:customer_id>", methods=["PUT"])
|
|
@login_required
|
|
def edit_customer_tank(customer_id):
|
|
"""
|
|
Safely edits or creates tank and description details for a customer.
|
|
"""
|
|
logger.info(f"PUT /customer/edit/tank/{customer_id} - Editing customer tank info")
|
|
get_customer = db.session.query(Customer_Customer).filter(Customer_Customer.id == customer_id).one_or_none()
|
|
if not get_customer:
|
|
return error_response("Customer not found", 404)
|
|
|
|
get_customer_description = db.session.query(Customer_Description).filter(Customer_Description.customer_id == customer_id).first()
|
|
if not get_customer_description:
|
|
get_customer_description = Customer_Description(customer_id=customer_id)
|
|
db.session.add(get_customer_description)
|
|
|
|
get_customer_tank = db.session.query(Customer_Tank_Inspection).filter(Customer_Tank_Inspection.customer_id == customer_id).first()
|
|
if not get_customer_tank:
|
|
get_customer_tank = Customer_Tank_Inspection(customer_id=customer_id)
|
|
db.session.add(get_customer_tank)
|
|
|
|
data = request.get_json()
|
|
|
|
if 'tank_status' in data:
|
|
get_customer_tank.tank_status = data["tank_status"]
|
|
|
|
if 'outside_or_inside' in data:
|
|
get_customer_tank.outside_or_inside = data["outside_or_inside"]
|
|
|
|
response_last_tank_inspection = data.get("last_tank_inspection", None)
|
|
response_tank_size = data.get("tank_size", 0)
|
|
|
|
# --- FIX APPLIED HERE ---
|
|
# 1. Get the value from the request. Default to 0 if it's missing.
|
|
response_customer_fill_location = data.get("fill_location", 0)
|
|
|
|
# 2. Add a safety check: if the frontend sent an empty string, convert it to 0.
|
|
if response_customer_fill_location == "":
|
|
response_customer_fill_location = 0
|
|
|
|
get_customer_tank.last_tank_inspection = response_last_tank_inspection
|
|
get_customer_tank.tank_size = response_tank_size
|
|
get_customer_description.fill_location = response_customer_fill_location
|
|
|
|
if get_customer.customer_automatic == 1:
|
|
get_auto_info = db.session.query(Auto_Delivery).filter(Auto_Delivery.customer_id == customer_id).first()
|
|
if get_auto_info:
|
|
get_auto_info.tank_size = response_tank_size
|
|
db.session.add(get_auto_info)
|
|
|
|
db.session.commit()
|
|
|
|
return success_response()
|