import xlsxwriter import io from flask import render_template, make_response, current_app, redirect, url_for, request, flash, abort, send_file from flask_login import login_required, current_user from . import bp from app.models import Agent, Project, Service, Capacity, Period, db, Company, AgentGrade, AgentStatus, AgentBap, \ Charge, Category, Label, AgentResponsability, AgentSkill, EmploymentType, ProjectStatus from app import db_mgr from app.auth.routes import role_required updated_message = "mis à jour" added_message = "ajouté" def form_manager(object_class, object_id_key, object_id, form_template, form_struct, urlfor_exist_dict, urlfor_done_dict): """ Generic form manager, holds etheir POST or GET method. Allows to modify existing object, or create new one. Form template has to deal with 'object_struct'. Class needs to have 'name' and 'id' attr. :param object_class: The object class we will deal with. Needs 'name' attr. :param object_id_key: The string to access get the object id from the form. :param object_id: Initial object id if any. :param form_template: Form template name. :param form_struct: Struct to pass to template for rendering. :param urlfor_exist_dict: Where to redirect if object already exist when creating new. :param urlfor_done_dict: Where do redirect when done. :return: """ if request.method == 'GET': if object_id: this_object = object_class.query.get(int(object_id)) else: this_object = object_class() object_struct = this_object.to_struct() return render_template(form_template, object_struct=object_struct, **form_struct) elif request.method == 'POST': try: object_id = request.form[object_id_key] except KeyError: object_id = None if object_id: # then update existing this_object = object_class.query.get(int(object_id)) done_string = updated_message + "." else: # check name doesnt exist yet existing_object = object_class.query.filter(object_class.name == request.form.get('name')).one_or_none() if existing_object: flash(f">{existing_object.name}< existe déjà.", 'warning') return redirect(url_for(**urlfor_exist_dict)) # or create from scratch this_object = object_class() done_string = added_message + "." # fill orm with form and write to db this_object.from_request(request) db.session.add(this_object) db.session.commit() # we're done flash(f"{this_object.name} (#{this_object.id}) " + done_string) urlfor_done_dict[object_id_key] = this_object.id return redirect(url_for(**urlfor_done_dict)) @bp.before_request def site_login(): try: if current_app.config['SITE_LOGIN'] \ and not current_user.is_authenticated: return redirect(url_for('auth.login')) except KeyError: # no such config, juste ignore pass @bp.before_request def catch_all_route(): current_app.logger.debug(f"{request.method} {request.path}") @bp.route('/') def index(): return render_template('index.html', subtitle="Page d'accueil") @bp.route('/services') @role_required('service') def services(): # get services list all_services = Service.query.order_by(Service.name).all() num_services = len(all_services) # pass to template return render_template('services.html', subtitle="Liste des services ({})".format(num_services), services=all_services) @bp.route('/projects') @role_required('project') def projects(): # get projects list all_projects = db_mgr.projects() num_projects = len(all_projects) # pass to template return render_template('projects.html', subtitle="Liste des projets ({})".format(num_projects), projects=all_projects) @bp.route('/project/') @role_required('project') def project(project_id): # TODO: am i the project manager ? this_project = Project.query.get(int(project_id)) charges_table = db_mgr.charges_by_project(project_id) return render_template('project.html', project=this_project.to_struct(), charges=charges_table, subtitle="{}".format(this_project.name)) @bp.route('/project/create', methods=('POST', 'GET')) @bp.route('/project//edit', methods=('POST', 'GET')) @role_required('service') def project_edit(project_id=None): _form_struct = {'categories': db_mgr.category_labels()} return form_manager(object_class=Project, object_id_key='project_id', object_id=project_id, form_template='project_form.html', form_struct=_form_struct, urlfor_exist_dict={'endpoint': 'main.project_edit'}, urlfor_done_dict={'endpoint': 'main.project', 'project_id': None}) @bp.route('/project//delete', methods=('POST', 'GET')) @role_required('admin') def project_delete(project_id=None): this_project = Project.query.get(int(project_id)) flash("Suppression du Projet pas encore implémentée", 'warning') # flash(f" {this_project.name} effacé") return redirect(url_for('main.project', project_id=this_project.id)) @bp.route('/projects/stats') @role_required('project') def projects_stats(): num_projects = len(Project.query.all()) this_categories = Category.query.all() return render_template('projects_stats.html', subtitle="Statistiques des projets ({})".format(num_projects), categories=this_categories) @bp.route('/agents') @role_required('project') def agents(): # get agents list all_agents = db_mgr.agents() num_agents = len(all_agents) # pass to template return render_template('agents.html', subtitle="Liste des agents ({})".format(num_agents), agents=all_agents) @bp.route('/agent/') @role_required('agent') def agent(agent_id): # TODO: am i the agent, the service or project manager , or the admin ? this_agent = Agent.query.get(int(agent_id)) charges_table = db_mgr.charges_by_agent_tabled(agent_id) return render_template('agent.html', agent=this_agent, charges=charges_table, subtitle=f"{this_agent.fullname}") @bp.route('/agent/create', methods=('POST', 'GET')) @bp.route('/agent//edit', methods=('POST', 'GET')) @role_required('service') def agent_edit(agent_id=None): _form_struct = {'companies': Company.query.all(), 'grades': AgentGrade.query.all(), 'statuses': AgentStatus.query.all(), 'baps': AgentBap.query.all()} return form_manager(object_class=Agent, object_id_key='agent_id', object_id=agent_id, form_template='agent_form.html', form_struct=_form_struct, urlfor_exist_dict={'endpoint': 'main.agent_edit'}, urlfor_done_dict={'endpoint': 'main.agent', 'agent_id': None}) @bp.route('/agent//delete', methods=('POST', 'GET')) @role_required('admin') def agent_delete(agent_id=None): this_agent = Agent.query.get(int(agent_id)) flash("Suppression de l'Agent pas encore implémentée", 'warning') # flash(f"Agent {this_agent.name} effacé") return redirect(url_for('main.agent', agent_id=this_agent.id)) @bp.route('/agents/stats') @role_required('project') def agents_stats(): num_agents = len(Agent.query.all()) all_periods = Period.query.order_by(Period.name).all() return render_template('agents_stats.html', subtitle="Statistiques des agents ({})".format(num_agents), periods=all_periods) @bp.route('/agents/responsabilities') @role_required('agent') def responsabilities(): num_resp = len(AgentResponsability.query.all()) all_resp = AgentResponsability.query.order_by(AgentResponsability.name).all() return render_template('responsabilities.html', subtitle="Liste des Responsabilités ({})".format(num_resp), responsabilities=all_resp) @bp.route('/responsability/') @role_required('agent') def responsability(responsability_id): this_responsability = AgentResponsability.query.get(int(responsability_id)) return render_template('responsability.html', subtitle=f"{this_responsability.name}", responsability=this_responsability) @bp.route('/responsability/create', methods=('POST', 'GET')) @bp.route('/responsability//edit', methods=('POST', 'GET')) @role_required('admin') def responsability_edit(responsability_id=None): return form_manager(object_class=AgentResponsability, object_id_key='responsability_id', object_id=responsability_id, form_template='responsability_form.html', form_struct={}, urlfor_exist_dict={'endpoint': 'main.responsability_edit'}, urlfor_done_dict={'endpoint': 'main.responsability', 'agent_id': None}) @bp.route('/responsability//delete', methods=('POST', 'GET')) @role_required('admin') def responsability_delete(responsability_id=None): this_responsability = AgentResponsability.query.get(int(responsability_id)) flash("Suppression de la Responsabilité pas encore implémentée", 'warning') # flash(f"Statut de Projet {this_status.name} effacé") return redirect(url_for('main.responsability', responsability_id=this_responsability.id)) @bp.route('/employments') @login_required def employments(): # get employments list all_employments = EmploymentType.query.order_by(EmploymentType.name).all() num_employments = len(all_employments) # pass to template return render_template('employments.html', subtitle="Liste des Emplois Types ({})".format(num_employments), employments=all_employments) @bp.route('/skills') @login_required def skills(): # get skills list all_skills = AgentSkill.query.order_by(AgentSkill.name).all() num_skills = len(all_skills) # pass to template return render_template('skills.html', subtitle="Liste des fonctions ({})".format(num_skills), skills=all_skills) @bp.route('/project_statuses') @login_required def project_statuses(): # get project_statuses list all_project_statuses = ProjectStatus.query.order_by(ProjectStatus.name).all() num_project_statuses = len(all_project_statuses) # pass to template return render_template('project_statuses.html', subtitle="Statuts de projets ({})".format(num_project_statuses), project_statuses=all_project_statuses) @bp.route('/project_status/') @role_required('admin') def project_status(status_id): # get label this_status = ProjectStatus.query.get(int(status_id)) # pass to template return render_template('project_status.html', subtitle=f"{this_status.name}", status=this_status) @bp.route('/project_status/create', methods=('POST', 'GET')) @bp.route('/project_status//edit', methods=('POST', 'GET')) @role_required('admin') def project_status_edit(status_id=None): return form_manager(object_class=ProjectStatus, object_id_key='status_id', object_id=status_id, form_template='project_status_form.html', form_struct={}, urlfor_exist_dict={'endpoint': 'main.project_status_edit'}, urlfor_done_dict={'endpoint': 'main.project_status', 'status_id': None}) @bp.route('/project_status//delete', methods=('POST', 'GET')) @role_required('admin') def project_status_delete(status_id=None): this_status = ProjectStatus.query.get(int(status_id)) flash("Suppression du Statut de Projet pas encore implémentée", 'warning') # flash(f"Statut de Projet {this_status.name} effacé") return redirect(url_for('main.project_status', status_id=this_status.id)) @bp.route('/capacities') @login_required def capacities(): # get capacities list all_capacities = Capacity.query.order_by(Capacity.name).all() num_capacities = len(all_capacities) # pass to template return render_template('capacities.html', subtitle="Liste des fonctions ({})".format(num_capacities), capacities=all_capacities) @bp.route('/capacity/') @role_required('admin') def capacity(capacity_id): # get capacity this_capacity = Capacity.query.get(int(capacity_id)) # pass to template return render_template('capacity.html', subtitle=f"{this_capacity.name}", capacity=this_capacity) @bp.route('/capacity/create', methods=('POST', 'GET')) @bp.route('/capacity//edit', methods=('POST', 'GET')) @role_required('admin') def capacity_edit(capacity_id=None): return form_manager(object_class=Capacity, object_id_key='capacity_id', object_id=capacity_id, form_template='capacity_form.html', form_struct={}, urlfor_exist_dict={'endpoint': 'main.capacity_edit'}, urlfor_done_dict={'endpoint': 'main.capacity', 'capacity_id': None}) @bp.route('/capacity//delete', methods=('POST', 'GET')) @role_required('admin') def capacity_delete(capacity_id=None): flash("Suppression du Capacity pas encore implémentée", 'warning') # this_label = Label.query.get(int(label_id)) # flash(f"Label {this_label.name} effacé") return redirect(url_for('main.capacities')) @bp.route('/labels') @login_required def labels(): # get labels list all_labels = Label.query.order_by(Label.name).all() num_labels = len(all_labels) # pass to template return render_template('labels.html', subtitle="Liste des labels ({})".format(num_labels), labels=all_labels) @bp.route('/label/') @role_required('admin') def label(label_id): # get label this_label = Label.query.get(int(label_id)) # pass to template return render_template('label.html', subtitle=f"{this_label.name}", label=this_label) @bp.route('/label/create', methods=('POST', 'GET')) @bp.route('/label//edit', methods=('POST', 'GET')) @role_required('admin') def label_edit(label_id=None): _form_struct = {'categories': Category.query.order_by(Category.name).all()} return form_manager(object_class=Label, object_id_key='label_id', object_id=label_id, form_template='label_form.html', form_struct=_form_struct, urlfor_exist_dict={'endpoint': 'main.label_edit'}, urlfor_done_dict={'endpoint': 'main.label', 'label_id': None}) @bp.route('/label//delete', methods=('POST', 'GET')) @role_required('admin') def label_delete(label_id=None): flash("Suppression du Label pas encore implémentée", 'warning') # this_label = Label.query.get(int(label_id)) # flash(f"Label {this_label.name} effacé") return redirect(url_for('main.labels')) @bp.route('/categories') @login_required def categories(): # get categories list all_categories = Category.query.order_by(Category.name).all() num_categories = len(all_categories) # pass to template return render_template('categories.html', subtitle="Liste des categories ({})".format(num_categories), categories=all_categories) @bp.route('/category/') @role_required('admin') def category(category_id): # get category this_category = Category.query.get(int(category_id)) # pass to template return render_template('category.html', subtitle=f"{this_category.name}", category=this_category) @bp.route('/category/create', methods=('POST', 'GET')) @bp.route('/category//edit', methods=('POST', 'GET')) @role_required('admin') def category_edit(category_id=None): _form_struct = {'labels': Label.query.order_by(Label.name).all()} return form_manager(object_class=Category, object_id_key='category_id', object_id=category_id, form_template='category_form.html', form_struct=_form_struct, urlfor_exist_dict={'endpoint': 'main.category_edit'}, urlfor_done_dict={'endpoint': 'main.category', 'category_id': None}) @bp.route('/category//delete', methods=('POST', 'GET')) @role_required('admin') def category_delete(category_id=None): flash("Suppression de la Catégorie pas encore implémentée", 'warning') # this_category = Category.query.get(int(category_id)) # flash(f"Category {this_category.name} effacé") return redirect(url_for('main.categories')) @bp.route('/periods') @login_required def periods(): # get categories list all_periods = Period.query.order_by(Period.name).all() num_periods = len(all_periods) # pass to template return render_template('periods.html', subtitle="Liste des périodes ({})".format(num_periods), periods=all_periods) @bp.route('/charge/add', methods=('POST', 'GET')) @role_required('service') def charge_add(): if request.method == 'GET': try: this_agent = Agent.query.get(int(request.args['agent_id'])) except KeyError: this_agent = None try: this_project = Project.query.get(int(request.args['project_id'])) except KeyError: this_project = None this_agents = Agent.query.order_by(Agent.firstname).all() this_projects = Project.query.order_by(Project.name).all() this_services = Service.query.order_by(Service.name).all() this_periods = Period.query.order_by(Period.id).all() this_capacities = Capacity.query.order_by(Capacity.name).all() return render_template('charge_form.html', subtitle="Affecter un agent", agent=this_agent, project=this_project, projects=this_projects, services=this_services, periods=this_periods, capacities=this_capacities, agents=this_agents) elif request.method == 'POST': this_agent = Agent.query.get(request.form.get('agent_id')) this_charge = Charge() this_charge.from_request(request) db.session.add(this_charge) db.session.commit() flash(f"Nouvelle charge pour l'agent {this_agent.fullname}") return redirect(url_for('main.agent', agent_id=this_agent.id)) # - - - - - - - - - - - - - - - - - - - - REST API - - - - - - - - - - - - - - - - - - - - def array_to_any(array, format='csv'): if format == 'csv': exporter = array_to_csv elif format == 'xls': exporter = array_to_xls else: abort(404) return exporter(array) def array_to_xls(array): output = io.BytesIO() # Create a workbook and add a worksheet. workbook = xlsxwriter.Workbook(output, {'in_memory': True}) worksheet = workbook.add_worksheet() # Iterate over the data and write it out row by row, col by col row = 0 for line in array: col = 0 for cell in line: worksheet.write(row, col, cell) col += 1 row += 1 workbook.close() output.seek(0) try: file_to_return = send_file(output, as_attachment=True, mimetype='application/vnd.ms-excel', attachment_filename='charge.xlsx') except FileNotFoundError: abort(404) return file_to_return def array_to_csv(array, sep=','): csv_table = [] for line in array: line = [cell.replace(sep, "-") for cell in line] line_string = sep.join(line) csv_table.append(line_string) resp = make_response("\n".join(csv_table)) resp.headers['Content-Type'] = 'text/plain;charset=utf8' return resp @bp.route('/charge/project///') @bp.route('/charge/project//') @role_required('project') def rest_charge_project(project_id, sort_key, any_format='csv'): return array_to_any(db_mgr.charges_by_project_stacked(project_id, sort_key), any_format) @bp.route('/charge/agent//') @bp.route('/charge/agent/') @role_required('service') def rest_charge_agent(agent_id, any_format='csv'): return array_to_any(db_mgr.charges_by_agent_stacked(agent_id), any_format) @bp.route('/charge/labels//') @bp.route('/charge/labels/') @role_required('project') def rest_labels_stats(category_id, any_format='csv'): return array_to_any(db_mgr.charges_by_labels_stacked(category_id), any_format) @bp.route('/charge/projects/') @bp.route('/charge/projects') @role_required('project') def rest_projects_stats(any_format='csv'): return array_to_any(db_mgr.charges_for_projects_stacked(), any_format) @bp.route('/count/agents/by_status//') @bp.route('/count/agents/by_status/') @role_required('project') def rest_agents_status_count(period_id, any_format='csv'): return array_to_any(db_mgr.count_agents_by_status(period_id), any_format)