Skip to content
Snippets Groups Projects

Feature/comp 276 publisher v2

Merged Remco Tukker requested to merge feature/COMP-276_publisher_v2 into develop
2 files
+ 121
2
Compare changes
  • Side-by-side
  • Inline
Files
2
+ 112
0
 
from decimal import Decimal
 
 
from sqlalchemy import delete, select
 
 
from compendium_v2.db import db
 
from compendium_v2.db.model import BudgetEntry, ChargingStructure, ECProject, FeeType, FundingSource, InstitutionURLs,\
 
NrenStaff, ParentOrganization, Policy, SubOrganization, TrafficVolume
 
from compendium_v2.db.survey_model import ResponseStatus, SurveyResponse
 
 
 
def map_2023(year, nren, answers):
 
for table_class in [BudgetEntry, ChargingStructure, ECProject, FundingSource, InstitutionURLs,
 
NrenStaff, ParentOrganization, Policy, SubOrganization, TrafficVolume]:
 
db.session.execute(delete(table_class).where(table_class.year == 2023))
 
 
answers = answers["data"]
 
budget = answers.get("budget")
 
if budget:
 
db.session.add(BudgetEntry(nren_id=nren.id, nren=nren, year=year, budget=Decimal(budget)))
 
 
funding_source = answers.get("income_sources")
 
if funding_source:
 
db.session.add(FundingSource(
 
nren_id=nren.id, nren=nren, year=year,
 
client_institutions=Decimal(funding_source.get("client_institutions", 0)),
 
european_funding=Decimal(funding_source.get("european_funding", 0)),
 
gov_public_bodies=Decimal(funding_source.get("gov_public_bodies", 0)),
 
commercial=Decimal(funding_source.get("commercial", 0)),
 
other=Decimal(funding_source.get("other", 0))
 
))
 
 
charging = answers.get("charging_mechanism")
 
if charging:
 
db.session.add(ChargingStructure(nren_id=nren.id, nren=nren, year=year, fee_type=FeeType[charging]))
 
 
staff_roles = answers.get("staff_roles", {})
 
staff_employment_type = answers.get("staff_employment_type", {})
 
if staff_roles or staff_employment_type:
 
db.session.add(NrenStaff(
 
nren_id=nren.id, nren=nren, year=year,
 
permanent_fte=Decimal(staff_employment_type.get("permanent_fte", 0)),
 
subcontracted_fte=Decimal(staff_employment_type.get("subcontracted_fte", 0)),
 
technical_fte=Decimal(staff_roles.get("technical_fte", 0)),
 
non_technical_fte=Decimal(staff_roles.get("nontechnical_fte", 0))
 
))
 
 
parent = answers.get("parent_organization_name")
 
if parent:
 
db.session.add(ParentOrganization(nren_id=nren.id, nren=nren, year=year, organization=parent))
 
 
subs = answers.get("suborganization_details")
 
if subs:
 
for sub in subs:
 
db.session.add(SubOrganization(
 
nren_id=nren.id, nren=nren, year=year,
 
organization=sub.get("suborganization_name"),
 
role=sub.get("suborganization_role") # TODO handle 'other' option properly
 
))
 
 
ec_projects = answers.get("ec_project_names")
 
if ec_projects:
 
for ec_project in ec_projects:
 
db.session.add(ECProject(nren_id=nren.id, nren=nren, year=year, project=ec_project.get("ec_project_name")))
 
 
strategy = answers.get("corporate_strategy_url", "")
 
policies = answers.get("policies", {})
 
if strategy or policies:
 
db.session.add(Policy(
 
nren_id=nren.id, nren=nren, year=year,
 
strategic_plan=strategy,
 
environmental=policies.get("environmental_policy", {}).get("url", ""),
 
equal_opportunity=policies.get("equal_opportunity_policy", {}).get("url", ""),
 
connectivity=policies.get("connectivity_policy", {}).get("url", ""),
 
acceptable_use=policies.get("acceptable_use_policy", {}).get("url", ""),
 
privacy_notice=policies.get("privacy_notice", {}).get("url", ""),
 
data_protection=policies.get("data_protection_contact", {}).get("url", "")
 
# TODO gender_equality_policy missing?
 
))
 
 
traffic_estimate = answers.get("traffic_estimate")
 
if traffic_estimate:
 
db.session.add(TrafficVolume(
 
nren_id=nren.id, nren=nren, year=year,
 
strategic_plan=strategy,
 
to_customers=Decimal(traffic_estimate.get("to_customers")),
 
from_customers=Decimal(traffic_estimate.get("from_customers")),
 
to_external=Decimal(traffic_estimate.get("to_external")),
 
from_external=Decimal(traffic_estimate.get("from_external")),
 
))
 
 
institution_urls = answers.get("connected_sites_lists")
 
if institution_urls:
 
db.session.add(InstitutionURLs(
 
nren_id=nren.id, nren=nren, year=year,
 
urls=institution_urls,
 
))
 
 
 
def publish(year):
 
responses = db.session.scalars(
 
select(SurveyResponse).where(SurveyResponse.survey_year == year)
 
.where(SurveyResponse.status == ResponseStatus.completed)
 
).unique()
 
 
question_mapping = {
 
2023: map_2023
 
}
 
 
mapping_function = question_mapping[year]
 
 
for response in responses:
 
mapping_function(year, response.nren, response.answers)
Loading