Gan, Jimmy
6c0d5a4e63
- Flask-based mold cost calculation application - PostgreSQL database with SQLAlchemy ORM - Docker/Podman containerization - Comprehensive documentation in docs/ - Clean, organized codebase without obsolete files - Production-ready deployment configuration - Enhanced pytest configuration with coverage reporting - Master/Development branch structure
383 lines
18 KiB
Python
383 lines
18 KiB
Python
from flask import Blueprint, render_template, redirect, url_for, request, jsonify, current_app, flash
|
|
from flask_login import current_user, login_required, login_user, logout_user
|
|
from app.models import Quotation, User
|
|
from app import db
|
|
from app.forms import LoginForm, QuotationForm
|
|
from app.utils.cost_factor_loader import get_cost_factors
|
|
from app.config import ALLOWED_EMAILS
|
|
import logging
|
|
from datetime import datetime, timezone, timedelta
|
|
from app import limiter, cache
|
|
import json
|
|
from ..utils.calculator import calculate_base_price
|
|
from ..health import get_health_status
|
|
|
|
bp = Blueprint('main', __name__)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
@bp.route('/', methods=['GET', 'POST'])
|
|
def home():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('main.calculator'))
|
|
return redirect(url_for('auth.login'))
|
|
|
|
@bp.route('/calculator')
|
|
@login_required
|
|
def calculator():
|
|
try:
|
|
form = QuotationForm()
|
|
|
|
# Set today's date as default for issue_date
|
|
if not form.issue_date.data:
|
|
form.issue_date.data = datetime.now(timezone.utc).date()
|
|
|
|
# Get cost factors for the form
|
|
cost_factors, tax_rates, default_net_base = get_cost_factors()
|
|
|
|
# Initialize form with default values
|
|
if form.mold_main_type.data in form.MOLD_TYPE_MAPPING:
|
|
form.mold_sub_type.choices = [('', 'Select Sub Type')] + [(x, x) for x in form.MOLD_TYPE_MAPPING[form.mold_main_type.data]]
|
|
|
|
# Get quote history with error handling
|
|
try:
|
|
quote_history = Quotation.query.filter_by(user_id=current_user.id)\
|
|
.order_by(Quotation.created_at.desc())\
|
|
.limit(5)\
|
|
.all()
|
|
except Exception as e:
|
|
current_app.logger.error(f"Error fetching quote history: {str(e)}")
|
|
quote_history = []
|
|
|
|
return render_template('calculator.html',
|
|
form=form,
|
|
quote_history=quote_history,
|
|
tax_rates=tax_rates,
|
|
quotation=None)
|
|
except Exception as e:
|
|
current_app.logger.error(f"Calculator error: {str(e)}", exc_info=True)
|
|
flash('Error loading calculator data', 'danger')
|
|
# Create a new form with default values
|
|
form = QuotationForm()
|
|
if form.mold_main_type.data in form.MOLD_TYPE_MAPPING:
|
|
form.mold_sub_type.choices = [('', 'Select Sub Type')] + [(x, x) for x in form.MOLD_TYPE_MAPPING[form.mold_main_type.data]]
|
|
# Set today's date as default even in error case
|
|
form.issue_date.data = datetime.now(timezone.utc).date()
|
|
return render_template('calculator.html',
|
|
form=form,
|
|
quote_history=[],
|
|
tax_rates={},
|
|
quotation=None)
|
|
|
|
@bp.route('/calculate', methods=['GET', 'POST'])
|
|
@login_required
|
|
@limiter.limit("30 per minute")
|
|
@cache.cached(timeout=300, key_prefix='calculate_form')
|
|
def calculate():
|
|
"""Calculate mold cost based on form data"""
|
|
try:
|
|
form = QuotationForm()
|
|
|
|
# Log incoming request data
|
|
current_app.logger.info('Incoming request data:')
|
|
current_app.logger.info(f'Request method: {request.method}')
|
|
current_app.logger.info(f'Request form data: {dict(request.form)}')
|
|
current_app.logger.info(f'Request JSON data: {request.get_json(silent=True)}')
|
|
|
|
if request.method == 'POST':
|
|
try:
|
|
# Log form data before validation
|
|
current_app.logger.info('Form data before validation:')
|
|
current_app.logger.info(f'Form data: {form.data}')
|
|
current_app.logger.info(f'Form errors: {form.errors}')
|
|
|
|
# Log the selected mold type
|
|
current_app.logger.info(f'Selected mold_main_type: {form.mold_main_type.data}')
|
|
current_app.logger.info(f'Selected mold_sub_type: {form.mold_sub_type.data}')
|
|
current_app.logger.info(f'Selected mold_specific_type: {form.mold_specific_type.data}')
|
|
|
|
# Update mold_sub_type choices based on selected mold_main_type
|
|
main_type = request.form.get('mold_main_type')
|
|
current_app.logger.info(f'Updating sub_type choices for main_type: {main_type}')
|
|
if main_type in form.MOLD_TYPE_MAPPING:
|
|
form.mold_sub_type.choices = [('', 'Select Sub Type')] + [(x, x) for x in form.MOLD_TYPE_MAPPING[main_type]]
|
|
current_app.logger.info(f'Updated sub_type choices: {form.mold_sub_type.choices}')
|
|
|
|
# Update mold_specific_type choices based on selected mold_sub_type
|
|
sub_type = request.form.get('mold_sub_type')
|
|
if sub_type in form.MOLD_SPECIFIC_TYPE_OPTIONS:
|
|
form.mold_specific_type.choices = [('', 'Select Specific Type')] + [(x, x) for x in form.MOLD_SPECIFIC_TYPE_OPTIONS[sub_type]]
|
|
else:
|
|
form.mold_specific_type.choices = [('', 'Select Specific Type')]
|
|
|
|
# Check if specific type is needed
|
|
if sub_type and sub_type in form.MOLD_SPECIFIC_TYPE_OPTIONS:
|
|
# If specific type is needed but not provided, set it to empty string
|
|
if not request.form.get('mold_specific_type'):
|
|
form.mold_specific_type.data = ''
|
|
else:
|
|
# If specific type is not needed, set it to empty string
|
|
form.mold_specific_type.data = ''
|
|
|
|
if not form.validate_on_submit():
|
|
current_app.logger.warning('Form validation failed:')
|
|
current_app.logger.warning(f'Form errors: {form.errors}')
|
|
current_app.logger.warning(f'Form data after validation: {form.data}')
|
|
current_app.logger.warning(f'Request form data: {dict(request.form)}')
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Invalid form data',
|
|
'message': 'Please check the form for errors',
|
|
'errors': form.errors,
|
|
'form_data': dict(request.form)
|
|
}), 400
|
|
|
|
# Get form data
|
|
form_data = form.data
|
|
current_app.logger.info(f'Processing form data: {form_data}')
|
|
|
|
# Calculate costs
|
|
try:
|
|
result = calculate_base_price(
|
|
mold_main_type=form.mold_main_type.data,
|
|
mold_sub_type=form.mold_sub_type.data,
|
|
mold_specific_type=form.mold_specific_type.data,
|
|
sliders_count=int(form.sliders_count.data),
|
|
cavity_count=float(form.cavity_count.data),
|
|
digital_texture_type=form.digital_texture_type.data,
|
|
complexity_high_sidewall=(form.complexity_high_sidewall.data == 'yes')
|
|
)
|
|
current_app.logger.info(f'Calculation result: {result}')
|
|
|
|
# Get cost factors for tax rate
|
|
cost_factors, tax_rates, default_net_base = get_cost_factors()
|
|
|
|
# Get tax rate for the country
|
|
tax_rate = tax_rates.get(form.mold_shop_country.data, 0.0)
|
|
|
|
# Calculate total cost (base price + tax)
|
|
total_cost = result['total_cost'] * (1 + tax_rate)
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'base_price': float(result['base_price']),
|
|
'complexity_cost': float(result['complexity_cost']),
|
|
'texture_cost': float(result['texture_cost']),
|
|
'tax_rate': float(tax_rate),
|
|
'total_cost': float(total_cost),
|
|
'mold_shop_name': form.mold_shop_name.data,
|
|
'mold_shop_country': form.mold_shop_country.data,
|
|
't1_shoe_factory_name': form.t1_shoe_factory_name.data,
|
|
't1_shoe_factory_country': form.t1_shoe_factory_country.data,
|
|
't2_component_factory_name': form.t2_component_factory_name.data,
|
|
't2_component_factory_country': form.t2_component_factory_country.data,
|
|
'model_name': form.model_name.data,
|
|
'tooling_id': form.tooling_id.data,
|
|
'season': form.season.data,
|
|
'stage': form.stage.data,
|
|
'type': form.type.data,
|
|
'issue_date': form.issue_date.data.strftime('%Y-%m-%d'),
|
|
'approved_by_mold_shop': form.approved_by_mold_shop.data,
|
|
'approved_by_t1_tooling': form.approved_by_t1_tooling.data,
|
|
'approved_by_lo_tooling': form.approved_by_lo_tooling.data,
|
|
'mold_main_type': form.mold_main_type.data,
|
|
'mold_sub_type': form.mold_sub_type.data,
|
|
'mold_specific_type': form.mold_specific_type.data,
|
|
'complexity_high_sidewall': 'yes' if form.complexity_high_sidewall.data == 'yes' else 'no',
|
|
'sliders_count': form.sliders_count.data,
|
|
'cavity_count': form.cavity_count.data,
|
|
'digital_texture_type': form.digital_texture_type.data
|
|
})
|
|
except Exception as calc_error:
|
|
current_app.logger.error(f'Error calculating costs: {str(calc_error)}', exc_info=True)
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Calculation error',
|
|
'message': str(calc_error)
|
|
}), 500
|
|
|
|
except Exception as form_error:
|
|
current_app.logger.error(f'Error processing form: {str(form_error)}', exc_info=True)
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Form processing error',
|
|
'message': str(form_error)
|
|
}), 500
|
|
|
|
# GET request - render the form
|
|
return render_template('calculator.html', form=form)
|
|
|
|
except Exception as e:
|
|
current_app.logger.error(f'Unexpected error in calculate route: {str(e)}', exc_info=True)
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Unexpected error',
|
|
'message': str(e)
|
|
}), 500
|
|
|
|
@bp.route('/quotations')
|
|
@login_required
|
|
@limiter.limit("30 per minute")
|
|
@cache.cached(timeout=60, key_prefix='user_quotations')
|
|
def quotations():
|
|
quotations = Quotation.query.filter_by(user_id=current_user.id)\
|
|
.order_by(Quotation.created_at.desc()).all()
|
|
return render_template('quotations.html', quotations=quotations)
|
|
|
|
@bp.route('/quotation/<int:id>')
|
|
@login_required
|
|
@limiter.limit("30 per minute")
|
|
@cache.cached(timeout=60, key_prefix='quotation_detail')
|
|
def quotation_detail(id):
|
|
quotation = Quotation.query.get_or_404(id)
|
|
if quotation.user_id != current_user.id and not current_user.is_admin:
|
|
flash('Access denied.', 'danger')
|
|
return redirect(url_for('main.quotations'))
|
|
return render_template('quotation_detail.html', quotation=quotation)
|
|
|
|
@bp.route('/health')
|
|
def health_check():
|
|
"""Health check endpoint that returns the status of various system components."""
|
|
# Get health status
|
|
health_status = get_health_status()
|
|
|
|
# Set appropriate status code based on overall health
|
|
status_code = 200 if health_status['status'] == 'healthy' else 503
|
|
|
|
# Add rate limiting headers
|
|
response = jsonify(health_status)
|
|
response.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
|
|
response.headers['Pragma'] = 'no-cache'
|
|
response.headers['Expires'] = '0'
|
|
|
|
return response, status_code
|
|
|
|
@bp.route('/api/recent-quotes')
|
|
@login_required
|
|
@limiter.limit("30 per minute")
|
|
def recent_quotes():
|
|
try:
|
|
recent_quotes = Quotation.query.filter_by(user_id=current_user.id)\
|
|
.order_by(Quotation.created_at.desc())\
|
|
.limit(5)\
|
|
.all()
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'quotes': [{
|
|
'id': quote.id,
|
|
'model_name': quote.model_name,
|
|
'tooling_id': quote.tooling_id,
|
|
'mold_sub_type': quote.mold_sub_type,
|
|
'cavity_count': float(quote.cavity_count),
|
|
'sliders_count': quote.sliders_count,
|
|
'created_at': quote.created_at.strftime('%Y-%m-%d %H:%M'),
|
|
'mold_shop_country': quote.mold_shop_country,
|
|
'digital_texture_type': quote.digital_texture_type,
|
|
'season': quote.season,
|
|
'stage': quote.stage,
|
|
'base_price': float(quote.base_price),
|
|
'tax_rate': float(quote.tax_rate) if quote.tax_rate is not None else 0.0,
|
|
'total_cost': float(quote.total_cost)
|
|
} for quote in recent_quotes]
|
|
})
|
|
except Exception as e:
|
|
current_app.logger.error(f'Error fetching recent quotes: {str(e)}', exc_info=True)
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'An error occurred while fetching recent quotes',
|
|
'quotes': []
|
|
}), 500
|
|
|
|
@bp.route('/save-quotation', methods=['POST'])
|
|
@login_required
|
|
@limiter.limit("30 per minute")
|
|
def save_quotation():
|
|
"""Save a quotation to the database"""
|
|
try:
|
|
# Get form data
|
|
form_data = request.get_json()
|
|
|
|
if not form_data:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'No data provided'
|
|
}), 400
|
|
|
|
# Check for duplicate submission (same user, same data within last 30 seconds)
|
|
try:
|
|
recent_duplicate = Quotation.query.filter(
|
|
Quotation.user_id == current_user.id,
|
|
Quotation.model_name == form_data.get('model_name'),
|
|
Quotation.tooling_id == form_data.get('tooling_id'),
|
|
Quotation.mold_main_type == form_data.get('mold_main_type'),
|
|
Quotation.mold_sub_type == form_data.get('mold_sub_type'),
|
|
Quotation.created_at >= datetime.now(timezone.utc) - timedelta(seconds=30)
|
|
).first()
|
|
|
|
if recent_duplicate:
|
|
current_app.logger.warning(f'Duplicate submission detected for user {current_user.id}, tooling_id {form_data.get("tooling_id")}')
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Duplicate submission',
|
|
'message': 'This quotation was already submitted recently. Please wait a moment before trying again.',
|
|
'quotation_id': recent_duplicate.id
|
|
}), 409
|
|
except Exception as duplicate_check_error:
|
|
current_app.logger.error(f'Error checking for duplicates: {str(duplicate_check_error)}')
|
|
# Continue with submission even if duplicate check fails
|
|
|
|
# Ensure mold_specific_type is not empty or missing
|
|
mold_specific_type = form_data.get('mold_specific_type')
|
|
if not mold_specific_type:
|
|
mold_specific_type = 'N/A'
|
|
|
|
# Create new quotation
|
|
quotation = Quotation(
|
|
user_id=current_user.id,
|
|
mold_shop_name=form_data.get('mold_shop_name'),
|
|
mold_shop_country=form_data.get('mold_shop_country'),
|
|
t1_shoe_factory_name=form_data.get('t1_shoe_factory_name'),
|
|
t1_shoe_factory_country=form_data.get('t1_shoe_factory_country'),
|
|
t2_component_factory_name=form_data.get('t2_component_factory_name'),
|
|
t2_component_factory_country=form_data.get('t2_component_factory_country'),
|
|
model_name=form_data.get('model_name'),
|
|
tooling_id=form_data.get('tooling_id'),
|
|
season=form_data.get('season'),
|
|
stage=form_data.get('stage'),
|
|
type=form_data.get('type'),
|
|
issue_date=datetime.strptime(form_data.get('issue_date'), '%Y-%m-%d').date(),
|
|
approved_by_mold_shop=form_data.get('approved_by_mold_shop'),
|
|
approved_by_t1_tooling=form_data.get('approved_by_t1_tooling'),
|
|
approved_by_lo_tooling=form_data.get('approved_by_lo_tooling'),
|
|
mold_main_type=form_data.get('mold_main_type'),
|
|
mold_sub_type=form_data.get('mold_sub_type'),
|
|
mold_specific_type=mold_specific_type,
|
|
complexity_high_sidewall=(form_data.get('complexity_high_sidewall') == 'yes'),
|
|
sliders_count=int(form_data.get('sliders_count')),
|
|
cavity_count=float(form_data.get('cavity_count')),
|
|
digital_texture_type=form_data.get('digital_texture_type'),
|
|
base_price=float(form_data.get('base_price')),
|
|
tax_rate=float(form_data.get('tax_rate')),
|
|
total_cost=float(form_data.get('total_cost'))
|
|
)
|
|
|
|
# Save to database
|
|
db.session.add(quotation)
|
|
db.session.commit()
|
|
|
|
current_app.logger.info(f'Quotation saved with ID: {quotation.id}')
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'quotation_id': quotation.id,
|
|
'message': 'Quotation saved successfully'
|
|
})
|
|
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
current_app.logger.error(f'Error saving quotation: {str(e)}', exc_info=True)
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Failed to save quotation',
|
|
'message': str(e)
|
|
}), 500 |