import io import os import re from datetime import datetime from flask import (Blueprint, request, jsonify, send_file, redirect, url_for, flash, render_template, make_response, current_app, session, after_this_request) from io import BytesIO from werkzeug.security import generate_password_hash from .models import (GISMapServer, Folder, Service, Layer, Profile, Notification, BillingInfo, SubscriptionPlan, UserSettings, Company, Plan, Invoice, Invitation, BlockDWGData) from .forms import * from .extensions import db, oauth from flask_login import current_user, login_user, logout_user, login_required from urllib.parse import urlparse, urljoin from .tasks import download_layer_task from .decorators import role_required import logging import sys from .weather import get_weather, get_forecast, deg_to_cardinal from .weather_icons import weather_icons import uuid from werkzeug.utils import secure_filename import pdfkit import pandas as pd from itsdangerous import URLSafeTimedSerializer from .email import send_email from app import bcrypt, cache from .invoice_utils import generate_invoice, calculate_sales_tax from fuzzywuzzy import fuzz, process from .siteTrackerCalls import get_next_distribution_wo, parse_distribution_prefix from .bulkUploadGen import process_file logging.getLogger().addHandler(logging.StreamHandler(sys.stdout)) main = Blueprint('main', __name__) google = oauth.create_client('google') UPLOAD_FOLDER = '/home/ubuntu/webapp/app/uploads/' current_app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER # Ensure the folder exists if not os.path.exists(UPLOAD_FOLDER): os.makedirs(UPLOAD_FOLDER) @main.route('/assign_role', methods=['GET', 'POST']) @login_required @role_required('admin','Super Admin') # Only admin can assign roles def assign_role(): users = User.query.all() form = RoleAssignForm() if request.method == 'POST': # Check if the form is the role assignment form and if it is valid print("request.form", request.form) if form.validate_on_submit(): user = form.user.data role = form.role.data if role not in user.roles: user.roles.append(role) db.session.commit() flash('Role assigned successfully.', 'success') else: flash('User already has this role.', 'warning') return redirect(url_for('main.admin_dashboard')) return redirect(url_for('main.admin_dashboard')) # Route to edit a user @main.route('/edit_user/', methods=['GET', 'POST']) @login_required @role_required('Super Admin') def edit_user(user_id): user = User.query.get_or_404(user_id) form = EditUserForm(obj=user) if form.validate_on_submit(): user.first_name = form.first_name.data user.last_name = form.last_name.data user.email = form.email.data user.company_id = form.company_name.data user.is_active = form.is_active.data user.roles = form.roles.data db.session.commit() return redirect(url_for('main.admin_dashboard')) return redirect(url_for('main.admin_dashboard', user_id=user_id)) # Route to delete a user @main.route('/delete_user/', methods=['POST']) @login_required @role_required('Super Admin') def delete_user(user_id): user = User.query.get_or_404(user_id) db.session.delete(user) db.session.commit() flash('User deleted successfully', 'success') return redirect(url_for('main.admin_dashboard')) @main.route('/add_plan', methods=['GET', 'POST']) @login_required @role_required('Super Admin') def add_plan(): form = EditPlanForm() if request.method == 'POST': if form.validate_on_submit(): plan = Plan( name=form.name.data, description=form.description.data, cost=form.cost.data, usage_limit=form.usage_limit.data ) db.session.add(plan) db.session.commit() flash('New plan added successfully', 'success') return redirect(url_for('main.admin_dashboard')) # Change 'view_plans' to your actual view route for plans return redirect(url_for('main.admin_dashboard')) # Route to edit a plan @main.route('/edit_plan/', methods=['GET', 'POST']) @login_required @role_required('Super Admin') def edit_plan(plan_id): plan = Plan.query.get_or_404(plan_id) form = EditPlanForm(obj=plan) if form.validate_on_submit(): form.populate_obj(plan) db.session.commit() flash('Plan updated successfully', 'success') return redirect(url_for('plan_management')) return render_template('edit_plan.html', form=form, plan=plan) @main.route('/toggle_plan_status/', methods=['GET']) @login_required def toggle_plan_status(plan_id): plan = Plan.query.get_or_404(plan_id) if plan.status.lower() == 'active': plan.status = 'Inactive' else: plan.status = 'Active' db.session.commit() flash(f'Plan status updated to {plan.status}.', 'success') return redirect(url_for('main.company_dashboard')) @main.route('/delete_company/', methods=['POST']) @login_required @role_required('Super Admin') def delete_company(company_id): company = Company.query.get_or_404(company_id) db.session.delete(company) db.session.commit() flash('Company deleted successfully.', 'success') return redirect(url_for('main.admin_dashboard')) @main.route('/edit_company/', methods=['GET', 'POST']) @login_required @role_required('Super Admin') def edit_company(company_id): company = Company.query.get_or_404(company_id) companyForm = UpdateCompanyAddressForm(obj=company) if companyForm.validate_on_submit(): company.name = companyForm.name.data company.street_address = companyForm.street_address.data company.city = companyForm.city.data company.state = companyForm.state.data company.zip_code = companyForm.zip_code.data company.country = companyForm.country.data db.session.commit() flash('Company updated successfully.', 'success') return redirect(url_for('main.admin_dashboard')) # Pre-fill form with current address if request.method == 'GET': companyForm.street_address.data = company.street_address companyForm.city.data = company.city companyForm.state.data = company.state companyForm.zip_code.data = company.zip_code companyForm.country.data = company.country return render_template('edit_company.html', form=companyForm, company=company, title='Edit Company') @main.route('/update_company_address/', methods=['GET', 'POST']) @login_required def update_company_address(company_id): company = Company.query.get_or_404(company_id) current_app.logger.info('Loading html...') # Ensure the current user is a company admin if not current_user.has_role('Company Admin') or current_user.company_id != company_id: current_app.logger.info('Permission garnted') flash('You do not have permission to update this address.', 'error') return redirect(url_for('main.index')) form = UpdateCompanyAddressForm() # Pre-fill form with current address # if request.method == 'GET': # current_app.logger.info("GET METHOD") # try: # form.street_address.data = company.street_address # form.city.data = company.city # form.state.data = company.state # form.zip_code.data = company.zip_code # form.country.data = company.country # except: # current_app.logger.info('For some reason the form didnt initialize') # pass # else: # current_app.logger.info("POST METHOD") # if form.validate_on_submit(): # company.street_address = form.street_address.data # company.city = form.city.data # company.state = form.state.data # company.zip_code = form.zip_code.data # company.country = form.country.data # db.session.commit() # flash('Company address updated successfully.', 'success') # return redirect(url_for('main.company_dashboard', company_id=company.id)) # else: # current_app.logger.info("POST BUT FAIL TO VALIDATE METHOD") # form.street_address.data = company.street_address # form.city.data = company.city # form.state.data = company.state # form.zip_code.data = company.zip_code # form.country.data = company.country current_app.logger.info('Got it') return render_template('update_company_address.html', form=form, company=company) # Route to delete a plan @main.route('/delete_plan/', methods=['POST']) @login_required @role_required('Super Admin') def delete_plan(plan_id): plan = Plan.query.get_or_404(plan_id) db.session.delete(plan) db.session.commit() flash('Plan deleted successfully', 'success') return redirect(url_for('plan_management')) @main.route('/admin', methods=['GET', 'POST']) @login_required @role_required('Super Admin') @cache.cached(timeout=60) def admin_dashboard(): users = User.query.all() plans = Plan.query.all() companies = Company.query.all() invoices = Invoice.query.all() form = RoleAssignForm() pform = EditPlanForm() cform = AddCompanyForm() eform = UpdateCompanyAddressForm() print([(company.id, company.name) for company in companies]) # Populate the company choices for the SelectField pform.company_id.choices = [(company.id, company.name) for company in companies] if request.method == 'POST': if 'add_role' in request.form and form.validate_on_submit(): user = form.user.data role = form.role.data if role not in user.roles: user.roles.append(role) db.session.commit() flash('Role assigned successfully.', 'success') else: flash('User already has this role.', 'warning') return redirect(url_for('main.admin_dashboard')) elif 'add_plan' in request.form and pform.validate_on_submit(): # Check if the form is the plan creation form and if it is valid print('add plan worked') plan = Plan( name=pform.name.data, description=pform.description.data, cost=pform.cost.data, company_id=pform.company_id.data, usage_limit=pform.usage_limit.data ) db.session.add(plan) db.session.commit() flash('New plan added successfully', 'success') return redirect(url_for('main.admin_dashboard')) elif 'add_company' in request.form and cform.validate_on_submit(): print('add company worked') company = Company( name=cform.name.data ) db.session.add(company) db.session.commit() flash('New company added successfully', 'success') return redirect(url_for('main.admin_dashboard')) else: # Print form errors print('we got a problem') for field, errors in form.errors.items(): for error in errors: print(f"Error in {field}: {error}") flash(f"Error in {field}: {error}", 'success') print('Companies added: ' + str(companies)) return render_template('admin.html', users=users, plans=plans, company=companies, form=form, pform=pform, cform=cform, eform=eform, invoices=invoices, title="Admin Panel") def create_notification(user_id, message): notification = Notification(user_id=user_id, message=message) db.session.add(notification) db.session.commit() def get_unread_notifications(user_id): return Notification.query.filter_by(user_id=user_id, status='unread').all() @main.route('/company') @login_required @role_required('Company Admin', 'Super Admin') @cache.cached(timeout=60) def company_dashboard(): # Fetch company data, users, roles, notifications, etc. company = Company.query.filter_by(name=current_user.company_name).first() users = User.query.filter_by(company_name=current_user.company_name).all() plans = Plan.query.all() # Adjust as necessary to get relevant plans invoices = Invoice.query.filter_by(company_id=company.id).all() notifications = Notification.query.filter_by(user_id=current_user.id, status='unread').all() title = current_user.company_name + ' Dashboard' for user in users: user.role_names = [role.name for role in user.roles] return render_template('company.html', company=company, users=users, plans=plans, invoices=invoices, notifications=notifications, title=title) @main.route('/company/users', methods=['GET']) @login_required def view_users(): if not current_user.company_id: flash('You are not associated with any company.', 'error') return redirect(url_for('index')) users = User.query.filter_by(company_id=current_user.company_id).all() return users @main.route('/generate_invoice//', methods=['GET']) @login_required def generate_invoice_route(user_id, plan_id): invoice = generate_invoice(user_id, plan_id) flash('Invoice generated successfully.', 'success') return redirect(url_for('main.view_invoice', invoice_id=invoice.id)) # @main.route('/invoices', methods=['GET']) # @login_required # def list_invoices(): # if current_user.is_super_admin: # invoices = Invoice.query.all() # elif current_user.is_company_admin: # Invoice.query.filter( # (Invoice.user_id == current_user.id) | # (Invoice.company_id == current_user.company_id) # ).all() # else: # invoices = Invoice.query.filter(Invoice.user_id == current_user.id.all()) # return render_template('invoices.html', invoices=invoices) @main.route('/invoice/', methods=['GET']) @login_required def view_invoice(invoice_id): invoice = Invoice.query.get_or_404(invoice_id) if not current_user.has_role('Super Admin'): flash('You do not have permission to view this invoice.', 'error') return redirect(url_for('main.list_invoices')) return render_template('invoice.html', invoice=invoice) @main.route('/invoice//download/pdf', methods=['GET']) @login_required def download_invoice_pdf(invoice_id): try: # Fetch the invoice from the database invoice = Invoice.query.get_or_404(invoice_id) # Render the template with the invoice rendered = render_template('invoice_pdf.html', invoice=invoice) # Path to wkhtmltopdf executable path_wkhtmltopdf = '/usr/bin/wkhtmltopdf' # Adjust this path if necessary config = pdfkit.configuration(wkhtmltopdf=path_wkhtmltopdf) # Generate the PDF pdf = pdfkit.from_string(rendered, False, configuration=config) # Prepare the response response = make_response(pdf) response.headers['Content-Type'] = 'application/pdf' response.headers['Content-Disposition'] = f'attachment; filename=invoice_{invoice_id}.pdf' return response except Exception as e: current_app.logger.error(f"Error generating PDF: {e}") return str(e), 500 @main.route('/invoice//download/excel', methods=['GET']) @login_required def download_invoice_excel(invoice_id): invoice = Invoice.query.get_or_404(invoice_id) df = pd.DataFrame([{ #'Customer Id': invoice.customer_id, 'Order Date': invoice.order_date, 'Shipped Electronically To': invoice.shipped_electronically_to, 'Plan': invoice.plan.name, 'Plan Description': invoice.plan_description, 'Price': invoice.price, 'Qty': invoice.qty, 'Extended Price': invoice.extended_price, 'Subtotal': invoice.subtotal, 'Sales Tax': invoice.sales_tax, 'Total': invoice.total, 'Balance Due': invoice.balance_due }]) output = io.BytesIO() with pd.ExcelWriter(output, engine='xlsxwriter') as writer: df.to_excel(writer, index=False, sheet_name='Invoice') output.seek(0) return send_file( output, download_name=f'invoice_{invoice_id}.xlsx', as_attachment=True, mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' ) @main.route('/create_invoice', methods=['POST']) @login_required def create_invoice(): print('welcome to invoicing') print(current_user.first_name) print(current_user.email) if not current_user.has_role('Super Admin'): flash('You do not have permission to perform this action.', 'error') return redirect(url_for('main.index')) print('you are a super admin') billing_period_str = request.form.get('billing_period') billing_period = datetime.strptime(billing_period_str, '%Y-%m-%d') company_id = request.form.get('company') plan_id = request.form.get('plan') user_id = request.form.get('user') # Optional field from the form print(company_id) company = Company.query.get_or_404(company_id) print(company.name) plan = Plan.query.get_or_404(plan_id) # Find the company admin company_admin = None print(str(company.users)) for user in company.users: print(f'Checking user: {user.first_name} {user.last_name}, Roles: {[role.name for role in user.roles]}') if user.has_role('Company Admin'): print(f'Found company admin: {user.first_name} {user.last_name}') company_admin = user break recipient = company_admin.email if company_admin else None if user_id: user = User.query.get_or_404(user_id) recipient = user.email else: user_id = company_admin.id # Calculate prices and taxes price = plan.cost qty = 1 extended_price = price * qty subtotal = extended_price sales_tax = calculate_sales_tax(company.zip_code, subtotal) # Implement this function total = subtotal + sales_tax balance_due = total invoice = Invoice( company_id=company.id, user_id=user_id if user_id else None, order_date=billing_period, shipped_electronically_to=recipient, plan_id=plan.id, plan_description=plan.description, price=price, qty=qty, extended_price=extended_price, subtotal=subtotal, sales_tax=sales_tax, total=total, balance_due=balance_due, is_paid=False ) db.session.add(invoice) db.session.commit() flash('Invoice created successfully.', 'success') return redirect(url_for('main.admin_dashboard')) @main.route('/user//update_role', methods=['POST']) @login_required def update_user_role(user_id): user = User.query.get_or_404(user_id) new_role = request.form.get('role') if new_role not in ['User', 'Admin']: flash('Invalid role selected.', 'error') return redirect(url_for('main.company_dashboard')) # Clear existing roles user.roles.clear() # Assign new role role = Role.query.filter_by(name=new_role).first() if role: user.roles.append(role) db.session.commit() flash('User role updated successfully.', 'success') else: flash('Role not found.', 'error') return redirect(url_for('main.company_dashboard')) @main.route('/invite_user', methods=['GET', 'POST']) @login_required def invite_user(): if request.method == 'POST': first_name = request.form.get('first_name') last_name = request.form.get('last_name') email = request.form.get('email') role = request.form.get('role') if role not in ['User', 'Admin']: flash('Invalid role selected.', 'error') return redirect(url_for('main.invite_user')) # Check if an invitation with this email already exists existing_invitation = Invitation.query.filter_by(email=email).first() if existing_invitation: flash('An invitation has already been sent to this email.', 'error') return redirect(url_for('main.invite_user')) token = URLSafeTimedSerializer(current_app.config['SECRET_KEY']).dumps(email, salt='invite-salt') invitation = Invitation( company_id=current_user.company_name, # Assuming current_user.company_name is the company ID email=email, first_name=first_name, last_name=last_name, role=role, token=token ) db.session.add(invitation) db.session.commit() invite_link = url_for('main.accept_invite', token=token, _external=True) decline_link = url_for('main.decline_invite', token=token, _external=True) send_email( to=email, subject='You are invited to join our company', template='invite_email', first_name=first_name, last_name=last_name, role=role, invite_link=invite_link, decline_link=decline_link ) notification = Notification( user_id=current_user.id, message=f'Invitation sent to {email} for {role} role.' ) db.session.add(notification) db.session.commit() flash('Invitation sent successfully.', 'success') return redirect(url_for('main.company_dashboard')) return redirect(url_for('main.company_dashboard')) @main.route('/accept_invite/', methods=['GET', 'POST']) def accept_invite(token): try: email = URLSafeTimedSerializer(current_app.config['SECRET_KEY']).loads(token, salt='invite-salt', max_age=3600) except: flash('The invitation link is invalid or has expired.', 'error') return redirect(url_for('main.index')) invitation = Invitation.query.filter_by(token=token, email=email).first_or_404() if invitation.status == 'accepted': flash('This invitation has already been accepted.', 'error') return redirect(url_for('main.index')) if request.method == 'POST': password = request.form.get('password') password_hash = bcrypt.generate_password_hash(password).decode('utf-8') # Create the user record user = User( first_name=invitation.first_name, last_name=invitation.last_name, email=invitation.email, company_name=invitation.company_id, password_hash=password_hash, _is_active=True ) role = Role.query.filter_by(name=invitation.role).first() if role: user.roles.append(role) db.session.add(user) invitation.status = 'accepted' db.session.commit() flash('You have successfully joined the company.', 'success') return redirect(url_for('main.login')) return render_template('set_password.html', token=token) @main.route('/decline_invite/', methods=['GET']) def decline_invite(token): try: email = URLSafeTimedSerializer(current_app.config['SECRET_KEY']).loads(token, salt='invite-salt', max_age=3600) except: flash('The invitation link is invalid or has expired.', 'error') return redirect(url_for('main.register')) invitation = Invitation.query.filter_by(token=token, email=email).first_or_404() if invitation.status == 'declined': flash('This invitation has already been declined.', 'error') return redirect(url_for('main.register')) invitation.status = 'declined' db.session.commit() flash('You have declined the invitation.', 'info') return redirect(url_for('main.register')) #----------------------------------------------------------------------------------------------------------------- @main.route('/api/get_user_settings', methods=['GET']) @login_required def get_user_settings(): # Fetch user settings from the database based on the current user user_settings = UserSettings.query.filter_by(user_id=current_user.id).first() if user_settings: settings_data = { 'theme_layout': user_settings.theme_layout, 'page_layout': user_settings.page_layout, 'sidebar_type': user_settings.sidebar_type, 'layout': user_settings.layout, 'direction_layout': user_settings.direction_layout, 'color_theme_layout': user_settings.color_theme_layout, 'card_layout': user_settings.card_layout } else: # Default settings if none are found for the user settings_data = { 'theme_layout': 'light-layout', 'page_layout': 'vertical-layout', 'sidebar_type': 'full-sidebar', 'layout': 'full-layout', 'direction_layout': 'ltr-layout', 'color_theme_layout': 'blue-theme-layout', 'card_layout': 'card-without-border' } return jsonify(settings_data) #------------------------------------------------------------------------------------------------------------------ @main.route('/') @main.route('/index') @login_required @cache.cached(timeout=60) def index(): user_profile = current_user.profile if current_user.profile else Profile() if user_profile.city is not None: city = current_user.profile.city # Get city from user's profile weather_data = get_weather(city) forecast_data = get_forecast(city) # Convert current temperature to integer weather_data['main']['temp'] = int(weather_data['main']['temp']) # Extract 8-hour forecast for today, showing every 2 hours today_forecast = [] for i in range(0, 5): today_forecast.append({ 'time': forecast_data['list'][i]['dt_txt'], 'temp': int(forecast_data['list'][i]['main']['temp']), # Convert forecast temperature to integer 'clouds': forecast_data['list'][i]['clouds']['all'], 'icon': weather_icons.get(forecast_data['list'][i]['weather'][0]['main'], "ti ti-cloud") }) # Extract daily forecast from the 5-day forecast data weekly_forecast = [] for i in range(0, len(forecast_data['list']), 8): weekly_forecast.append({ 'date': forecast_data['list'][i]['dt_txt'], 'temp': int(forecast_data['list'][i]['main']['temp']), 'clouds': forecast_data['list'][i]['clouds']['all'], 'icon': weather_icons.get(forecast_data['list'][i]['weather'][0]['main'], "ti ti-cloud") }) wind_deg = weather_data['wind']['deg'] weather_data['wind']['cardinal'] = deg_to_cardinal(wind_deg) weather_data['icon'] = weather_icons.get(weather_data['weather'][0]['main'], "ti ti-cloud") return render_template('index.html', weather_data=weather_data, today_forecast=today_forecast, weekly_forecast=weekly_forecast, title="Home") return render_template('index.html', title="Home") def is_safe_url(target): ref_url = urlparse(request.host_url) test_url = urlparse(urljoin(request.host_url, target)) return test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc @main.route('/register', methods=['GET', 'POST']) def register(): if current_user.is_authenticated: return redirect(url_for('main.index')) form = RegistrationForm() if form.validate_on_submit(): user = User( first_name=form.first_name.data, last_name=form.last_name.data, email=form.email.data, company_name=form.company_name.data, password_hash=generate_password_hash(form.password.data) ) db.session.add(user) db.session.commit() profile = Profile( user_id=user.id, street_address=form.street_address.data, city=form.city.data, state=form.state.data, phone_number=form.phone_number.data, location=form.location.data, currency=form.currency.data ) db.session.add(profile) notification = Notification( user_id=user.id, notification_email=form.email.data ) db.session.add(notification) billing_info = BillingInfo( user_id=user.id, business_name=form.company_name.data, business_address=form.street_address.data, first_name=form.first_name.data, last_name=form.last_name.data ) db.session.add(billing_info) subscription_plan = SubscriptionPlan( user_id=user.id, plan_name='Free', sites_per_month=0 ) db.session.add(subscription_plan) db.session.commit() if form.profile_picture.data: picture_path = os.path.join(main.root_path, 'static/profile_pics', f'{user.id}.jpg') form.profile_picture.data.save(picture_path) profile.profile_picture = f'{user.id}.jpg' db.session.commit() flash('Congratulations, you are now a registered user!', 'success') return redirect(url_for('main.login')) else: for field, errors in form.errors.items(): for error in errors: flash(f"Error in {getattr(form, field).label.text}: {error}", 'danger') return render_template('register.html', title='Register', form=form) @main.route('/login', methods=['GET', 'POST']) def login(): if current_user.is_authenticated: return redirect(url_for('main.index')) form = LoginForm() if form.validate_on_submit(): logging.debug('Form validated successfully') user = User.query.filter_by(email=form.email.data).first() if user is None: logging.debug('No user found with that email') flash('Invalid email or password', 'danger') return redirect(url_for('main.login')) if not user.check_password(form.password.data): logging.debug('Password check failed') flash('Invalid email or password', 'danger') return redirect(url_for('main.login')) logging.debug('Logging in user') login_user(user, remember=form.remember_me.data) # Set a cookie if remember_me is checked if form.remember_me.data: logging.debug('Setting remember_me cookie') resp = make_response(redirect(url_for('main.index'))) resp.set_cookie('remember_email', form.email.data, max_age=30 * 24 * 60 * 60) # 30 days return resp next_page = request.args.get('next') if not next_page or not is_safe_url(next_page): next_page = url_for('main.index') logging.debug('Redirecting to next page or dashboard') flash('Logged in successfully.', 'success') return redirect(next_page) logging.debug('Form did not validate') # Get the remembered email from the cookie if it exists remembered_email = request.cookies.get('remember_email') if remembered_email: form.email.data = remembered_email return render_template('login.html', title='Sign In', form=form) @main.route('/login/google') def login_google(): google = oauth.create_client('google') redirect_uri = url_for('main.google_callback', _external=True) nonce = str(uuid.uuid4()) # Generate a nonce session['nonce'] = nonce # Store the nonce in the session return google.authorize_redirect(redirect_uri, nonce=nonce) @main.route('/auth/google/callback') def google_callback(): google = oauth.create_client('google') token = google.authorize_access_token() nonce = session.pop('nonce', None) # Retrieve and remove the nonce from the session if not nonce: flash("Nonce not found. Please try logging in again.", "error") return redirect(url_for('main.login')) userinfo = google.parse_id_token(token, nonce) # Use the nonce when parsing the ID token if userinfo: user = User.query.filter_by(email=userinfo['email']).first() if not user: user = User( google_id=str(userinfo['sub']), # Ensure ID is stored as a string first_name=userinfo['given_name'], last_name=userinfo['family_name'], email=userinfo['email'], profile_picture=userinfo['picture'] ) # Assign a default role to the user default_role = Role.query.filter_by(name='user').first() if default_role: user.roles.append(default_role) db.session.add(user) db.session.commit() try: login_user(user) except Exception as e: print(e) return redirect(url_for('main.index')) flash("User email not available or not verified by Google.", "error") return redirect(url_for('main.login')) @main.route('/logout') def logout(): logout_user() flash('You have been logged out.', 'info') return redirect(url_for('main.login')) @main.route('/api/save_settings', methods=['POST']) @login_required def save_settings(): data = request.json user_settings = UserSettings.query.filter_by(user_id=current_user.id).first() if not user_settings: user_settings = UserSettings(user_id=current_user.id) user_settings.theme_layout = data.get('theme_layout', user_settings.theme_layout) user_settings.page_layout = data.get('page_layout', user_settings.page_layout) user_settings.sidebar_type = data.get('sidebar_type', user_settings.sidebar_type) user_settings.layout = data.get('layout', user_settings.layout) user_settings.direction_layout = data.get('direction_layout', user_settings.direction_layout) user_settings.color_theme_layout = data.get('color_theme_layout', user_settings.color_theme_layout) user_settings.card_layout = data.get('card_layout', user_settings.card_layout) db.session.add(user_settings) db.session.commit() return jsonify({'success': True, 'message': 'Settings saved successfully'}) @main.route('/load_settings', methods=['GET']) @login_required def load_settings(): settings = UserSettings.query.filter_by(user_id=current_user.id).first() if settings: return jsonify({ 'theme_layout': settings.theme_layout, 'direction_layout': settings.direction_layout, 'color_theme_layout': settings.color_theme_layout, 'page_layout': settings.page_layout, 'layout': settings.layout, 'sidebar_type': settings.sidebar_type, 'card_layout': settings.card_layout }), 200 return jsonify({"message": "No settings found"}), 404 @main.route('/profile', methods=['GET', 'POST']) @login_required def profile(): if current_user.profile is None: current_user.profile = Profile() profile_form = ProfileForm(obj=current_user.profile) change_password_form = ChangePasswordForm() if current_user.notifications is None: current_user.notifications = Notification() notification_form = NotificationForm(obj=current_user) if current_user.billing_info is None: current_user.billing_info = BillingInfo() billing_info_form = BillingInfoForm(obj=current_user) empty_form = EmptyForm() if request.method == 'POST': if 'update_profile' in request.form and profile_form.validate_on_submit(): current_user.first_name = profile_form.first_name.data current_user.last_name = profile_form.last_name.data current_user.email = profile_form.email.data current_user.company_name = profile_form.company_name.data current_user.profile.street_address = profile_form.street_address.data current_user.profile.city = profile_form.city.data current_user.profile.state = profile_form.state.data current_user.profile.phone_number = profile_form.phone_number.data current_user.profile.location = profile_form.location.data current_user.profile.currency = profile_form.currency.data db.session.commit() flash('Your profile has been updated.', 'success') return redirect(url_for('main.profile')) elif 'change_password' in request.form and change_password_form.validate_on_submit(): if current_user.check_password(change_password_form.current_password.data): current_user.set_password(change_password_form.new_password.data) db.session.commit() flash('Your password has been updated.', 'success') return redirect(url_for('main.profile')) else: flash('Current password is incorrect.', 'danger') elif 'update_notifications' in request.form and notification_form.validate_on_submit(): current_user.notifications.notification_email = notification_form.notification_email.data current_user.notifications.newsletter_notifications = notification_form.newsletter_notifications.data current_user.notifications.invoice_notifications = notification_form.invoice_notifications.data current_user.notifications.delivery_notifications = notification_form.delivery_notifications.data current_user.notifications.email_notifications = notification_form.email_notifications.data db.session.commit() flash('Notification preferences have been updated.', 'success') return redirect(url_for('main.profile')) elif 'update_billing_info' in request.form and billing_info_form.validate_on_submit(): current_user.billing_info.business_name = billing_info_form.business_name.data current_user.billing_info.business_address = billing_info_form.business_address.data current_user.billing_info.first_name = billing_info_form.first_name.data current_user.billing_info.last_name = billing_info_form.last_name.data current_user.billing_info.business_sector = billing_info_form.business_sector.data current_user.billing_info.country = billing_info_form.country.data db.session.commit() flash('Billing information has been updated.', 'success') return redirect(url_for('main.profile')) elif request.method == 'GET': profile_form.first_name.data = current_user.first_name profile_form.last_name.data = current_user.last_name profile_form.email.data = current_user.email if current_user.company_name is not None: profile_form.company_name.data = current_user.company_name if current_user.profile is not None: profile_form.street_address.data = current_user.profile.street_address profile_form.city.data = current_user.profile.city profile_form.state.data = current_user.profile.state profile_form.phone_number.data = current_user.profile.phone_number profile_form.location.data = current_user.profile.location profile_form.currency.data = current_user.profile.currency if current_user.notifications is not None: notification_form.notification_email.data = current_user.notifications.notification_email notification_form.newsletter_notifications.data = current_user.notifications.newsletter_notifications notification_form.invoice_notifications.data = current_user.notifications.invoice_notifications notification_form.delivery_notifications.data = current_user.notifications.delivery_notifications notification_form.email_notifications.data = current_user.notifications.email_notifications return render_template('profile.html', profile_form=profile_form, change_password_form=change_password_form, notification_form=notification_form, billing_info_form=billing_info_form, empty_form=empty_form) @main.route('/change_password', methods=['POST']) @login_required def change_password(): form = ChangePasswordForm() if form.validate_on_submit(): if current_user.check_password(form.current_password.data): current_user.set_password(form.new_password.data) db.session.commit() flash('Your password has been updated.', 'success') else: flash('Current password is incorrect.', 'danger') return redirect(url_for('main.profile')) @main.route('/upload_profile_picture', methods=['POST']) @login_required def upload_profile_picture(): if 'profile_picture' not in request.files: flash('No file part', 'danger') return redirect(url_for('main.profile')) file = request.files['profile_picture'] if file.filename == '': flash('No selected file', 'danger') return redirect(url_for('main.profile')) if file: filename = secure_filename(file.filename) picture_path = os.path.join(main.root_path, 'static/profile_pics', filename) file.save(picture_path) if current_user.profile is None: current_user.profile = Profile() current_user.profile.profile_picture = filename db.session.commit() flash('Your profile picture has been updated!', 'success') return redirect(url_for('main.profile')) @main.route('/reset_profile_picture', methods=['POST']) @login_required def reset_profile_picture(): if current_user.profile.profile_picture: picture_path = os.path.join(main.root_path, 'static/profile_pics', current_user.profile.profile_picture) if os.path.exists(picture_path): os.remove(picture_path) current_user.profile.profile_picture = None db.session.commit() flash('Your profile picture has been reset!', 'success') return redirect(url_for('main.profile')) def get_layer_data(): layers = Layer.query.all() layer_data = [] for layer in layers: if layer.latitude is not None and layer.longitude is not None: layer_data.append({ "type": "Feature", "geometry": { "type": "Point", "coordinates": [layer.longitude, layer.latitude] }, "properties": { "name": layer.name, "description": layer.service.name, # Assuming you want to show the service name "link": url_for('main.layer_details', layer_id=layer.id) # Assuming you have a layer detail route } }) return layer_data @main.route('/get_layers') @login_required def get_layers(): layer_data = get_layer_data() return jsonify({"type": "FeatureCollection", "features": layer_data}) @main.route('/get_feature_layers') @login_required def get_feature_layers(): layers = Layer.query.all() layer_data = [] for layer in layers: service = Service.query.get(layer.service_id) map_server = GISMapServer.query.get(service.map_server_id) if service and map_server: layer_url = f"{map_server.url}/rest/services/{service.name}/MapServer/{layer.id}" layer_data.append({ "url": layer_url, "name": layer.name, "description": service.name # Assuming the description should be the service name }) return jsonify(layer_data) @main.route('/preview_layer///') @login_required def preview_layer(map_server_id, service_id, layer_id): layer_id += 1 # Fetch the map server URL map_server = GISMapServer.query.get(map_server_id) if not map_server: return jsonify({'error': 'Invalid map server'}), 400 # Fetch the service service = Service.query.get(service_id) if not service or service.map_server_id != map_server_id: return jsonify({'error': 'Invalid service'}), 400 # Fetch the layer layer = Layer.query.get(layer_id) if not layer or layer.service_id != service_id: return jsonify({'error': 'Invalid layer'}), 400 # Construct the layer URL layer_url = f"{map_server.url}/rest/services/{service.name}/{service.type}/{layer.id - 1}" # Fetch the extent information from ArcGIS server response = requests.get(f"{layer_url}?f=pjson") data = response.json() if 'extent' not in data: return jsonify({'error': 'Invalid layer data from ArcGIS'}), 400 extent = data['extent'] current_app.logger.info(f"Preview Layer called with map_server_id: {map_server_id}, service_id: {service_id}, layer_id: {layer_id}") current_app.logger.info(f"Preview Layer url: {layer_url}") return jsonify({ 'layer_url': layer_url, 'extent': extent }) @main.route('/layers', methods=['GET']) @login_required def layers(): arcgis_url = request.args.get('arcgis_url') folder_name = request.args.get('folder_name', '') service_name = request.args.get('service_name', '') layer_id = request.args.get('layer_id', '') if layer_id: url = f'{arcgis_url}/rest/services/{service_name}/{layer_id}?f=pjson' elif service_name: if folder_name: url = f'{arcgis_url}/rest/services/{folder_name}/{service_name}/MapServer?f=pjson' else: url = f'{arcgis_url}/rest/services/{service_name}/MapServer?f=pjson' elif folder_name: url = f'{arcgis_url}/rest/services/{folder_name}?f=pjson' else: url = f'{arcgis_url}/rest/services?f=pjson' current_app.logger.info(f"Fetching URL: {url}") response = requests.get(url) data = response.json() current_app.logger.info(f"Data received from ArcGIS: {data}") # Save GISMapServer if not exists map_server = GISMapServer.query.filter_by(url=arcgis_url).first() if not map_server: map_server = GISMapServer(url=arcgis_url) db.session.add(map_server) db.session.commit() # Process folders if 'folders' in data and data['folders']: for folder in data['folders']: existing_folder = Folder.query.filter_by(name=folder, map_server_id=map_server.id).first() if not existing_folder: new_folder = Folder(name=folder, map_server_id=map_server.id) db.session.add(new_folder) db.session.commit() current_app.logger.info(f"Processing folder: {folder}") # Process services if 'services' in data and data['services']: if folder_name: parent_folder = Folder.query.filter_by(name=folder_name, map_server_id=map_server.id).first() for service in data['services']: existing_service = Service.query.filter_by(name=service['name'], folder_id=parent_folder.id).first() if not existing_service: new_service = Service(name=service['name'], type=service['type'], folder_id=parent_folder.id) db.session.add(new_service) db.session.commit() current_app.logger.info(f"Processing service: {service['name']} of type {service['type']}") else: for service in data['services']: existing_service = Service.query.filter_by(name=service['name'], map_server_id=map_server.id).first() if not existing_service: new_service = Service(name=service['name'], type=service['type'], map_server_id=map_server.id) db.session.add(new_service) db.session.commit() current_app.logger.info(f"Processing service: {service['name']} of type {service['type']}") # Render the appropriate template if 'folders' in data and data['folders']: return render_template('mapsearch.html', folders=data['folders'], services=data.get('services', []), arcgis_url=arcgis_url, title="Folders & Services") elif 'services' in data and data['services']: return render_template('services.html', services=data['services'], arcgis_url=arcgis_url, title="Services") elif 'layers' in data and data['layers']: parent_service = Service.query.filter_by(name=service_name, map_server_id=map_server.id).first() if not parent_service: raise ValueError(f"Service '{service_name}' not found in database for map server '{arcgis_url}'") layers = [] for layer in data['layers']: existing_layer = Layer.query.filter_by(name=layer['name'], service_id=parent_service.id).first() if not existing_layer: lat, lon, extent, spatial_reference = fetch_coordinates(url) new_layer = Layer(name=layer['name'], service_id=parent_service.id, latitude=lat, longitude=lon) db.session.add(new_layer) db.session.commit() current_app.logger.info(f"Processing layer: {layer['name']}") # Append map_server_id, service_id, and layer_id to each layer layer['map_server_id'] = map_server.id layer['service_id'] = parent_service.id layers.append(layer) return render_template('layers.html', layers=layers, arcgis_url=arcgis_url,service_name=service_name, title="Layers") else: return render_template('error.html', error="No results found for this endpoint.", arcgis_url=arcgis_url, title="Error") from pyproj import Transformer, CRS import requests def fetch_coordinates(service_url): try: response = requests.get(service_url) data = response.json() extent = data.get("fullExtent", {}) spatial_reference = extent.get("spatialReference", {}).get("latestWkid", None) if "xmin" in extent and "ymin" in extent and "xmax" in extent and "ymax" in extent and spatial_reference: transformer = Transformer.from_crs(f"epsg:{spatial_reference}", "epsg:4326", always_xy=True) x_center = (extent["xmin"] + extent["xmax"]) / 2 y_center = (extent["ymin"] + extent["ymax"]) / 2 lon, lat = transformer.transform(x_center, y_center) return lat, lon, extent, spatial_reference else: return None, None, None, None except requests.RequestException as e: current_app.logger.error(f"Error fetching coordinates: {e}") return None, None, None, None @main.route('/layers', methods=['POST']) @login_required def add_layer(): service_id = request.form['service_id'] service = Service.query.get(service_id) if service: for layer in service.layers: layer_url = f"{service.url}/{layer.id}" lat, lon = fetch_coordinates(layer_url) if lat is not None and lon is not None: layer.latitude = lat layer.longitude = lon db.session.commit() else: current_app.logger.warning(f"Could not fetch coordinates for layer {layer.name}") return redirect(url_for('main.viewmap')) @main.route('/download_layer/') @login_required def download_layer(layer_name): arcgis_url = request.args.get('arcgis_url') top_layer_name = request.args.get('top_layer_name') if not arcgis_url or not top_layer_name or not layer_name: return jsonify({"error": "Missing required parameters"}), 400 task = download_layer_task.apply_async(args=[arcgis_url, top_layer_name, layer_name]) return jsonify({"task_id": task.id}), 202 @main.route('/download_layer/status/') @login_required def download_layer_status(task_id): task = download_layer_task.AsyncResult(task_id) if task.state == 'PENDING': response = { 'state': task.state, 'current': 0, 'total': 1, 'status': 'Pending...' } elif task.state != 'FAILURE': response = { 'state': task.state, 'current': task.info.get('current', 0), 'total': task.info.get('total', 1), 'status': task.info.get('status', '') } if 'result' in task.info: response['result'] = task.info['result'] else: response = { 'state': task.state, 'current': 1, 'total': 1, 'status': str(task.info) } return jsonify(response) @main.route('/download_layer/result/') @login_required def download_layer_result(task_id): task = download_layer_task.AsyncResult(task_id) if task.state == 'SUCCESS': return send_file(BytesIO(task.result), download_name=f"layer_{task_id}.zip", as_attachment=True) else: return jsonify({"error": "Task not completed or failed"}), 400 @main.route('/layer_details/') @login_required def layer_details(layer_id): arcgis_url = request.args.get('arcgis_url', '') top_layer_name = request.args.get('top_layer_name', '') current_app.logger.info( f"Received request for layer details with arcgis_url={arcgis_url}, layer_id={layer_id}, top_layer_name={top_layer_name}") url = f"{arcgis_url}/rest/services/{top_layer_name}/MapServer/{layer_id}?f=json" current_app.logger.info(f"Fetching layer details from URL: {url}") response = requests.get(url) data = response.json() current_app.logger.info(f"Layer details received: {data}") if 'error' in data: current_app.logger.error(f"ARCGIS Error: {data['error']}") return f"Error fetching layer details: {data['error']['message']}" return render_template('layer_details.html', layer=data, arcgis_url=arcgis_url, top_layer_name=top_layer_name) @main.route('/navigate_up') @login_required def navigate_up(): arcgis_url = request.args.get('arcgis_url', '') top_layer_name = request.args.get('top_layer_name', '') layer_id = request.args.get('layer_id', '') # Logic to navigate up a layer or to the previous layer # This might involve removing the last part of the URL or layer structure # Adjust the logic as per your application needs return redirect(url_for('main.layers', arcgis_url=arcgis_url, top_layer_name=top_layer_name, layer_id=layer_id)) @main.route('/addmap') @login_required def addmap(): return render_template('addmap.html') @main.route('/projectload') @login_required def projectload(): return render_template('newbulkupload.html') @main.route('/projectload', methods=['POST']) @login_required def handle_projectload(): file = request.files.get('data_file') cabinet_name = request.form.get('cabinet_name') cabinet_type = request.form.get('cabinet_type') round_value = request.form.get("round") or "0" if not file: return "File upload is required", 400 upload_folder = os.path.join(current_app.root_path, 'uploads') os.makedirs(upload_folder, exist_ok=True) input_path = os.path.join(upload_folder, f"{uuid.uuid4()}_{file.filename}") file.save(input_path) project_meta = {} next_wo_number = None if cabinet_type == "existing": next_wo_number, project_meta = get_next_distribution_wo(cabinet_name) if not next_wo_number or not project_meta: flash( "❌ Could not find Ring Number or related Distribution WOs for this cabinet. " "Please double-check the cabinet name.", "danger" ) return redirect(url_for("main.projectload")) elif cabinet_type == "new": from app.siteTrackerCalls import parse_distribution_prefix # Read input and extract county from the first address import pandas as pd df = pd.read_csv(input_path) if input_path.endswith('.csv') else pd.read_excel(input_path) first_address = str(df["Attribute_complete_address"].iloc[0]) if "Attribute_complete_address" in df.columns else "" county_name = "" for county in [ "Pinellas", "Hillsborough", "Polk", "Sarasota", "Manatee", "Charlotte", "Lee", "Collier", "Bay", "Walton", "Okaloosa", "Santa Rosa", "Escambia", "Hernando", "Citrus", "Sumter", "Marion", "Osceola", "St. Lucie", "Indian River", "Martin", "Palm Beach", "Broward", "Miami-Dade", "Seminole", "Volusia", "Brevard" ]: if county.lower() in first_address.lower(): county_name = county break # Load market order info import json json_path = os.path.join(os.path.dirname(__file__), 'market_orders.json') with open(json_path) as f: market_data = json.load(f) market_order = next((m for m in market_data if county_name in m.get("counties", [])), None) if market_order: project_meta = { "MO": market_order.get("market_order_name", "Unknown"), "County": county_name, "Distribution_Area": request.form.get("distribution_area") or "Unknown", "Engineering_Vendor": market_order.get("engineering_vendor", "Unknown"), "Construction_Vendor": market_order.get("construction_vendor", "Unknown"), "Round": round_value } else: project_meta = { "MO": request.form.get("mo", "Unknown"), "County": county_name or request.form.get("county", "Unknown"), "Distribution_Area": request.form.get("distribution_area", "Unknown"), "Engineering_Vendor": request.form.get("engineering_vendor", "Unknown"), "Construction_Vendor": request.form.get("construction_vendor", "Unknown"), "Round": round_value } # Generate the next WO number prefix, base_num = parse_distribution_prefix(cabinet_name) next_wo_number = f"{prefix}{base_num:03d}" else: return "Please select if the cabinet is NEW or EXISTING", 400 base, ext = os.path.splitext(input_path) output_path = f"{base}_output.xlsx" from app.bulkUploadGen import process_file process_file(input_path, output_path, next_wo_number, project_meta) if not os.path.exists(output_path): return f"Output file not found: {output_path}", 500 @after_this_request def cleanup(response): try: for fname in os.listdir(upload_folder): fpath = os.path.join(upload_folder, fname) if os.path.isfile(fpath): os.remove(fpath) except Exception as e: current_app.logger.warning(f"Cleanup failed: {e}") return response if not os.path.exists(output_path): return f"Output file not found: {output_path}", 500 return send_file(output_path, as_attachment=True) @main.route('/viewmap') @login_required def viewmap(): try: map_servers = GISMapServer.query.options( db.joinedload(GISMapServer.folders).joinedload(Folder.services).joinedload(Service.layers), db.joinedload(GISMapServer.services).joinedload(Service.layers) ).all() current_app.logger.info(f'Map Servers: {map_servers}') return render_template('viewmap.html', map_servers=map_servers) except Exception as e: current_app.logger.error(f'Error retrieving map servers: {e}') return render_template('error.html', error=str(e)) @main.route('/gis_map') @login_required def gis_map(): return render_template('maps.html') ####################################### # # DWG Builder Routes Start # ####################################### import traceback # Route to handle the file upload @main.route('/upload_file', methods=['GET', 'POST']) def upload_file(): if request.method == 'POST': # Retrieve company_id from the form company_id = request.form.get('company_id') # Check if the file is part of the request if 'file' not in request.files: flash('No file part') return redirect(request.url) file = request.files['file'] # If the user doesn't select a file if file.filename == '': flash('No selected file') return redirect(request.url) if file: # Save the file filename = secure_filename(file.filename) file_path = os.path.join(current_app.config['UPLOAD_FOLDER'], filename) file.save(file_path) # Parse the file and insert into the database #try: # Function to clean and normalize headers def clean_header(value): if isinstance(value, str): return value.strip().lower() return value # Function to check if the row is a valid header row def is_valid_header(row): row = row[:3].apply(clean_header) return ("manufacture" in row[0] and "model" in row[1] and "dwg name" in row[2]) # Function to extract equipment type from header import pandas as pd import pandas as pd def extract_equipment_type_from_header(header_cell): if isinstance(header_cell, str): return header_cell.split()[0] # Extract the first word as equipment type (can customize this) return None def parse_and_insert(file_path, company_id): print(f"Starting to process file: {file_path}") # Load the Excel file xls = pd.ExcelFile(file_path) all_tables = [] # Store extracted data from all tables # Loop through each sheet in the Excel file for sheet_name in xls.sheet_names: print(f"Processing sheet: {sheet_name}") df = pd.read_excel(xls, sheet_name=sheet_name, header=None) if df.empty or df.shape[0] == 0: print(f"Skipping empty sheet: {sheet_name}") continue # Skip empty sheets # We only need to inspect the first row (row 0) for headers try: first_row = df.iloc[0] except IndexError: print(f"Sheet {sheet_name} does not have enough rows. Skipping...") continue # Skip sheets with no data print(f"Inspecting first row for table headers: {first_row.values}") # Iterate over columns in groups of 3, starting at column 1 col = 0 while col <= len(first_row) - 3: # Check if we have valid headers in the current group of three columns group = first_row[col:col + 3].apply(lambda x: x.lower() if isinstance(x, str) else None) print(f"Checking group in columns {col}-{col + 2}: {group.values}") # Handle both 'model' and 'sub-folder' as valid for the 'Model' column in the database if ("manufacture" in group.iloc[0] and ("model" in group.iloc[1] or "sub-folder" in group.iloc[1]) and "dwg name" in group.iloc[2]): # Extract the equipment type from the first header cell equipment_type = extract_equipment_type_from_header(first_row[col]) if equipment_type: # Extract the table from these columns, starting from the second row table = df.iloc[1:, col:col + 3] table.columns = ['Manufacturer', 'Model or Sub-Folder', 'Block DWG Name'] table['Equipment Type'] = equipment_type # Remove rows where 'Model or Sub-Folder' column is blank or NaN table = table[ table['Model or Sub-Folder'].notna() & (table['Model or Sub-Folder'] != '')] # Rename 'Model or Sub-Folder' to 'Model' to match your database structure table.rename(columns={'Model or Sub-Folder': 'Model'}, inplace=True) # Print the table for debugging print( f"Extracted table from sheet {sheet_name}, columns {col}-{col + 2} with equipment type {equipment_type}:\n{table}") # Append the extracted table to the list of all tables all_tables.append(table) # Move to the next group of columns (skip gaps between tables) col += 4 # Combine all tables into a single DataFrame if all_tables: combined_df = pd.concat(all_tables, ignore_index=True) print(f"Combined DataFrame:\n{combined_df}") # Insert the DataFrame into the BlockDWGData database using SQLAlchemy for _, row in combined_df.iterrows(): # Check if the record already exists existing_record = BlockDWGData.query.filter_by( manufacturer=row['Manufacturer'], model=row['Model'], block_dwg_name=row['Block DWG Name'], equipment_type=row['Equipment Type'], company_id=company_id ).first() if not existing_record: # Add the new record to the session if not already in the database new_entry = BlockDWGData( manufacturer=row['Manufacturer'], model=row['Model'], block_dwg_name=row['Block DWG Name'], equipment_type=row['Equipment Type'], company_id=company_id ) db.session.add(new_entry) # Commit all changes to the database db.session.commit() print("Data inserted successfully into the database.") else: print("No valid tables found in the uploaded file.") raise ValueError("No valid tables found.") # Call the parsing and insertion function with company_id parse_and_insert(file_path, company_id) flash("File uploaded and data inserted successfully!") #except Exception as e: db.session.rollback() # Rollback if there's an error #flash(f"Error processing file: {str(e)}") return redirect(url_for('main.upload_file')) else: # Handle GET request to display the form companies = Company.query.all() # Query the companies from the database return render_template('upload_blocks.html', companies=companies) # Pass companies to the form @main.route('/search-block', methods=['GET', 'POST']) def search_block(): if request.method == 'POST': search_query = request.form.get('model') equipment_type = request.form.get('equipment_type') company_id = request.form.get('company_id') # Fetch all models for the specified equipment type and company all_models = BlockDWGData.query.filter_by(equipment_type=equipment_type, company_id=company_id).all() model_list = [entry.model for entry in all_models] # Perform fuzzy matching best_match = process.extractOne(search_query, model_list) if best_match: matched_entry = BlockDWGData.query.filter_by(model=best_match[0], equipment_type=equipment_type, company_id=company_id).first() return render_template('search_results.html', result=matched_entry) else: flash('No match found.') # Fetch available companies for the form companies = Company.query.all() return render_template('block_search.html', companies=companies) # API endpoint to query BlockDWGData table and perform fuzzy search on the model @main.route('/api/search_block_data', methods=['GET']) def search_block_data(): query_model = request.args.get('model') if not query_model: return jsonify({'error': 'Model parameter is required'}), 400 # Fetch all data from BlockDWGData table all_data = BlockDWGData.query.all() # Prepare list for matching results matching_results = [] # Perform fuzzy search using fuzzywuzzy for row in all_data: similarity = fuzz.ratio(row.model.lower(), query_model.lower()) # Compare lowercase for case-insensitivity print(f"Comparing '{row.model}' with '{query_model}': Similarity = {similarity}") # Debugging output if similarity >= 70: # Define a threshold for matching, e.g., 70% matching_results.append({ 'id': row.id, 'manufacturer': row.manufacturer, 'model': row.model, 'block_dwg_name': row.block_dwg_name, 'equipment_type': row.equipment_type, 'company_id': row.company_id, 'similarity': similarity }) # Return the matching results in JSON format print(f"Matching results: {matching_results}") # Debugging output return jsonify(matching_results) @main.route('/api/get_all_block_data', methods=['GET']) def get_all_block_data(): # Fetch all data from BlockDWGData table all_data = BlockDWGData.query.all() # Convert data to a readable format for debugging data = [{ 'id': row.id, 'manufacturer': row.manufacturer, 'model': row.model, 'block_dwg_name': row.block_dwg_name, 'equipment_type': row.equipment_type, 'company_id': row.company_id } for row in all_data] return jsonify(data)