feat: add admin settings system and improve customer/pricing endpoints
Add centralized admin settings (company info, social links, quick calls, sidebar visibility toggles, theme, logo upload) with singleton pattern and full CRUD API. Add active/dedicated customer count endpoints for dashboard stats. Fix automatic assignment route to use PUT instead of GET. Refactor oil price endpoint to use schema serialization with null safety. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,4 +1,6 @@
|
||||
import logging
|
||||
import base64
|
||||
import json
|
||||
from flask import request
|
||||
from flask_login import current_user, logout_user, login_user, login_required
|
||||
from app.admin import admin
|
||||
@@ -8,8 +10,8 @@ from datetime import datetime
|
||||
from app.classes.pricing import (
|
||||
Pricing_Oil_Oil,
|
||||
Pricing_Oil_Oil_schema)
|
||||
from app.classes.admin import Admin_Company, Admin_Company_schema, Call
|
||||
from app.common.decorators import admin_required
|
||||
from app.classes.admin import Admin_Company, Admin_Company_schema, Call, Admin_Settings, Admin_Settings_schema
|
||||
from app.common.decorators import admin_required, login_required as custom_login_required
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -129,3 +131,119 @@ def get_voip_routing():
|
||||
return success_response({"current_phone": latest_call.current_phone})
|
||||
else:
|
||||
return error_response("No VoIP routing found", 404)
|
||||
|
||||
|
||||
# ============================================
|
||||
# Admin Settings Endpoints
|
||||
# ============================================
|
||||
|
||||
DEFAULT_QUICK_CALLS = json.dumps([
|
||||
{"name": "WB Hill Tank Springfield", "phone": "(413) 525-3678"},
|
||||
{"name": "LW Tank Uxbridge", "phone": "(508) 234-6000"},
|
||||
{"name": "Trask Tank Worcester", "phone": "(508) 791-5064"},
|
||||
{"name": "David Mechanic", "phone": "(774) 239-3776"},
|
||||
{"name": "Spring Rebuilders", "phone": "(508) 799-9342"},
|
||||
])
|
||||
|
||||
def _get_or_create_settings():
|
||||
"""Get the singleton settings row, creating it with defaults on first access."""
|
||||
settings = db.session.query(Admin_Settings).first()
|
||||
if not settings:
|
||||
settings = Admin_Settings(
|
||||
company_name='Auburn Oil',
|
||||
link_facebook='https://www.facebook.com/auburnoil',
|
||||
link_google='https://www.google.com/search?client=firefox-b-1-d&sca_esv=02c44965d6d4b280&sca_upv=1&cs=1&output=search&kgmid=/g/11wcbqrx5l&q=Auburn+Oil&shndl=30&shem=lsde&source=sh/x/loc/act/m1/1&kgs=52995d809762cd61',
|
||||
link_website='https://auburnoil.com',
|
||||
link_google_review='https://g.page/r/CZHnPQ85LsMUEBM/review',
|
||||
quick_calls=DEFAULT_QUICK_CALLS,
|
||||
show_automatics=True,
|
||||
show_stats=True,
|
||||
show_service=True,
|
||||
show_ticker=True,
|
||||
default_theme='ocean',
|
||||
)
|
||||
db.session.add(settings)
|
||||
db.session.commit()
|
||||
return settings
|
||||
|
||||
|
||||
@admin.route("/settings", methods=["GET"])
|
||||
@custom_login_required
|
||||
def get_settings():
|
||||
"""Get admin settings (any logged-in user can read)."""
|
||||
logger.info("GET /admin/settings")
|
||||
settings = _get_or_create_settings()
|
||||
schema = Admin_Settings_schema(many=False)
|
||||
return success_response({"settings": schema.dump(settings)})
|
||||
|
||||
|
||||
@admin.route("/settings", methods=["PUT"])
|
||||
@admin_required
|
||||
def update_settings():
|
||||
"""Update admin settings (admin only). Does not update logo."""
|
||||
logger.info("PUT /admin/settings")
|
||||
settings = _get_or_create_settings()
|
||||
data = request.json
|
||||
|
||||
allowed_fields = [
|
||||
'company_name', 'link_facebook', 'link_google', 'link_website',
|
||||
'link_google_review', 'quick_calls', 'show_automatics', 'show_stats',
|
||||
'show_service', 'show_ticker', 'default_theme',
|
||||
]
|
||||
for field in allowed_fields:
|
||||
if field in data:
|
||||
value = data[field]
|
||||
if field == 'quick_calls' and isinstance(value, list):
|
||||
value = json.dumps(value)
|
||||
setattr(settings, field, value)
|
||||
|
||||
settings.updated_at = datetime.utcnow()
|
||||
db.session.commit()
|
||||
|
||||
schema = Admin_Settings_schema(many=False)
|
||||
return success_response({"settings": schema.dump(settings)})
|
||||
|
||||
|
||||
@admin.route("/settings/logo", methods=["POST"])
|
||||
@admin_required
|
||||
def upload_logo():
|
||||
"""Upload a logo image (admin only). Max 2MB."""
|
||||
logger.info("POST /admin/settings/logo")
|
||||
if 'logo' not in request.files:
|
||||
return error_response("No file provided", 400)
|
||||
|
||||
file = request.files['logo']
|
||||
if file.filename == '':
|
||||
return error_response("No file selected", 400)
|
||||
|
||||
# 2MB limit
|
||||
file_data = file.read()
|
||||
if len(file_data) > 2 * 1024 * 1024:
|
||||
return error_response("File too large. Maximum size is 2MB.", 400)
|
||||
|
||||
mime_type = file.content_type or 'image/png'
|
||||
encoded = base64.b64encode(file_data).decode('utf-8')
|
||||
|
||||
settings = _get_or_create_settings()
|
||||
settings.logo_base64 = encoded
|
||||
settings.logo_mime_type = mime_type
|
||||
settings.updated_at = datetime.utcnow()
|
||||
db.session.commit()
|
||||
|
||||
schema = Admin_Settings_schema(many=False)
|
||||
return success_response({"settings": schema.dump(settings)})
|
||||
|
||||
|
||||
@admin.route("/settings/logo", methods=["DELETE"])
|
||||
@admin_required
|
||||
def delete_logo():
|
||||
"""Delete the custom logo (admin only)."""
|
||||
logger.info("DELETE /admin/settings/logo")
|
||||
settings = _get_or_create_settings()
|
||||
settings.logo_base64 = None
|
||||
settings.logo_mime_type = None
|
||||
settings.updated_at = datetime.utcnow()
|
||||
db.session.commit()
|
||||
|
||||
schema = Admin_Settings_schema(many=False)
|
||||
return success_response({"settings": schema.dump(settings)})
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
from app import db, ma
|
||||
from datetime import datetime
|
||||
import json
|
||||
|
||||
|
||||
class Admin_Company(db.Model):
|
||||
@@ -30,4 +31,30 @@ class Call(db.Model):
|
||||
|
||||
id = db.Column(db.Integer, primary_key=True, index=True)
|
||||
current_phone = db.Column(db.String(500))
|
||||
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
||||
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
||||
|
||||
|
||||
class Admin_Settings(db.Model):
|
||||
__tablename__ = 'admin_settings'
|
||||
__table_args__ = {"schema": "public"}
|
||||
|
||||
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
|
||||
updated_at = db.Column(db.TIMESTAMP(), default=datetime.utcnow, onupdate=datetime.utcnow)
|
||||
logo_base64 = db.Column(db.Text, nullable=True)
|
||||
logo_mime_type = db.Column(db.VARCHAR(50), nullable=True)
|
||||
company_name = db.Column(db.VARCHAR(250), default='Auburn Oil')
|
||||
link_facebook = db.Column(db.VARCHAR(500), nullable=True)
|
||||
link_google = db.Column(db.VARCHAR(500), nullable=True)
|
||||
link_website = db.Column(db.VARCHAR(500), nullable=True)
|
||||
link_google_review = db.Column(db.VARCHAR(500), nullable=True)
|
||||
quick_calls = db.Column(db.Text, nullable=True)
|
||||
show_automatics = db.Column(db.Boolean, default=True)
|
||||
show_stats = db.Column(db.Boolean, default=True)
|
||||
show_service = db.Column(db.Boolean, default=True)
|
||||
show_ticker = db.Column(db.Boolean, default=True)
|
||||
default_theme = db.Column(db.VARCHAR(50), default='ocean')
|
||||
|
||||
|
||||
class Admin_Settings_schema(ma.SQLAlchemyAutoSchema):
|
||||
class Meta:
|
||||
model = Admin_Settings
|
||||
@@ -8,7 +8,8 @@ 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 datetime import datetime, timedelta
|
||||
from sqlalchemy import func
|
||||
from app.classes.cards import Card_Card
|
||||
from app.classes.customer import \
|
||||
Customer_Customer, \
|
||||
@@ -18,6 +19,7 @@ from app.classes.customer import \
|
||||
Customer_Tank_Inspection_schema,\
|
||||
Customer_Tank_Inspection
|
||||
from app.classes.service import Service_Parts
|
||||
from app.classes.delivery import Delivery_Delivery
|
||||
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
|
||||
@@ -437,6 +439,38 @@ def customer_count():
|
||||
return success_response({'count': get_customer})
|
||||
|
||||
|
||||
@customer.route("/count/active", methods=["GET"])
|
||||
@login_required
|
||||
def customer_count_active():
|
||||
"""
|
||||
Count distinct customers who had a delivery within the past year.
|
||||
"""
|
||||
logger.info("GET /customer/count/active - Getting active past year count")
|
||||
one_year_ago = datetime.utcnow().date() - timedelta(days=365)
|
||||
count = (db.session
|
||||
.query(func.count(func.distinct(Delivery_Delivery.customer_id)))
|
||||
.filter(Delivery_Delivery.when_delivered >= one_year_ago)
|
||||
.scalar())
|
||||
return success_response({'count': count or 0})
|
||||
|
||||
|
||||
@customer.route("/count/dedicated", methods=["GET"])
|
||||
@login_required
|
||||
def customer_count_dedicated():
|
||||
"""
|
||||
Count distinct customers with more than one finalized delivery.
|
||||
"""
|
||||
logger.info("GET /customer/count/dedicated - Getting dedicated customer count")
|
||||
subquery = (db.session
|
||||
.query(Delivery_Delivery.customer_id)
|
||||
.filter(Delivery_Delivery.delivery_status == 10)
|
||||
.group_by(Delivery_Delivery.customer_id)
|
||||
.having(func.count(Delivery_Delivery.id) > 1)
|
||||
.subquery())
|
||||
count = db.session.query(func.count()).select_from(subquery).scalar()
|
||||
return success_response({'count': count or 0})
|
||||
|
||||
|
||||
@customer.route("/automatic/status/<int:customer_id>", methods=["GET"])
|
||||
@login_required
|
||||
def customer_automatic_status(customer_id):
|
||||
@@ -475,7 +509,7 @@ def get_all_automatic_deliveries():
|
||||
|
||||
|
||||
|
||||
@customer.route("/automatic/assign/<int:customer_id>", methods=["GET"])
|
||||
@customer.route("/automatic/assign/<int:customer_id>", methods=["PUT"])
|
||||
@login_required
|
||||
def customer_automatic_assignment(customer_id):
|
||||
"""
|
||||
|
||||
@@ -52,14 +52,10 @@ def get_oil_price_today():
|
||||
.query(Pricing_Oil_Oil)
|
||||
.order_by(Pricing_Oil_Oil.date.desc())
|
||||
.first())
|
||||
return success_response({
|
||||
'price_from_supplier': get_price_query.price_from_supplier,
|
||||
'price_for_customer': get_price_query.price_for_customer,
|
||||
'price_for_employee': get_price_query.price_for_employee,
|
||||
'price_same_day': get_price_query.price_same_day,
|
||||
'price_prime': get_price_query.price_prime,
|
||||
'price_emergency': get_price_query.price_emergency,
|
||||
})
|
||||
if get_price_query:
|
||||
delivery_schema = Pricing_Oil_Oil_schema(many=False)
|
||||
return success_response(delivery_schema.dump(get_price_query))
|
||||
return error_response("No pricing data found", 404)
|
||||
|
||||
|
||||
@info.route("/price/oil/table", methods=["GET"])
|
||||
|
||||
@@ -2,6 +2,7 @@ import logging
|
||||
from datetime import date
|
||||
from app.stats import stats
|
||||
import datetime
|
||||
from flask import request
|
||||
from app import db
|
||||
from app.common.responses import success_response
|
||||
from app.classes.delivery import Delivery_Delivery
|
||||
|
||||
Reference in New Issue
Block a user