Aquilenet/Axione-IPE-Viewer
7
1
Fork 0
Code Issues 5 Pull requests Projects Releases Wiki Activity

fix netwo

Browse source
This commit is contained in:
Johan Le Baut 2023-03-06 15:53:31 +01:00
parent bbadc97f20
commit 93120d43b2
2 changed files with 27 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
webapp/main.py
View file

@ -22,6 +22,7 @@ class Config(TypedDict):
arcep_ipe_path: str
arcep_ipe_db_name: str
netwo_api_key: str
netwo_aquilenet_fixed_recurring_price: float
def parseConfig() -> Config:
@ -35,6 +36,9 @@ def parseConfig() -> Config:
"arcep_ipe_path": cfg.get("DB", "arcep_ipe_path"),
"arcep_ipe_db_name": cfg.get("DB", "arcep_ipe_db_name"),
"netwo_api_key": cfg.get("NETWO", "api_key"),
"netwo_aquilenet_fixed_recurring_price": float(
cfg.get("NETWO", "aquilenet_fixed_recurring_price") or 0.00
),
}
@ -44,7 +48,9 @@ cfg: Config = parseConfig()
axione = Axione(cfg.get("axione_ipe_path"), cfg.get("axione_ipe_db_name"))
arcep = Arcep(cfg.get("arcep_ipe_path"), cfg.get("arcep_ipe_db_name"))
liazo = Liazo()
netwo = Netwo(cfg.get("netwo_api_key"))
netwo = Netwo(
cfg.get("netwo_api_key"), cfg.get("netwo_aquilenet_fixed_recurring_price")
)
elig_api_routes = EligibilityApiRoutes(app, axione, netwo)
elig_api_routes.add_routes()
elig_api_exceptions = FlaskExceptions(app)

27
webapp/netwo/netwo.py
View file

@ -9,6 +9,7 @@ from eligibility_api.elig_api_exceptions import NetwoApiErrorException
from ipe_fetcher import FAIEligibilityStatus
NETWO_DEPLOYED_STATUS = "Deployed"
TVA_INCREASE_COEFF = 1.2
class NetwooEligibility(TypedDict):
@ -25,12 +26,13 @@ class NetwooEligibility(TypedDict):
class Netwo:
def __init__(self, netwo_api_key: str):
def __init__(self, netwo_api_key: str, aquilenet_fixed_recurring_price: float):
self.netwo_api_headers = {
"x-actor-slug": "aquilenet",
"x-api-key": netwo_api_key,
"Accept": "application/json",
}
self.aquilenet_fixed_recurring_price = aquilenet_fixed_recurring_price
def get_netwo_imb_coordinates(self, ref_imb: str) -> dict:
"""
@ -45,15 +47,14 @@ class Netwo:
headers=self.netwo_api_headers,
)
status_code = response.status_code
if status_code != 200:
if status_code != 200 or not response.json():
raise NetwoApiErrorException(
f"Could not GET netwo imb ref {ref_imb}", status_code
)
imb_payload = response.json()
return imb_payload
@staticmethod
def _filter_netwo_raw_elig_results(raw_elig: dict, search_ftto: bool) -> list:
def _filter_netwo_raw_elig_results(self, raw_elig: dict, search_ftto: bool) -> list:
filtered_elig = []
inf_search = ["ftth"]
if search_ftto:
@ -72,6 +73,14 @@ class Netwo:
access_fee = offer.get("access_fee") or 0.00
recurring_price = offer.get("recurring_price") or 0.00
commitment_duration = offer.get("commitment_duration") or 0
access_fee_ttc = round(
access_fee * TVA_INCREASE_COEFF,
)
total_recurring_price_ttc = round(
recurring_price * TVA_INCREASE_COEFF
+ self.aquilenet_fixed_recurring_price,
2,
)
filtered_elig.append(
{
"entity_id": entity_id,
@ -81,15 +90,19 @@ class Netwo:
"infrastructure_type": inf_type,
"debit": debit,
"access_fee": access_fee,
"access_fee_ttc": access_fee_ttc,
"recurring_price": recurring_price,
"total_recurring_price_ttc": total_recurring_price_ttc,
"commitment_duration": commitment_duration,
"per_month_price_one_year": round(
access_fee / 12 + recurring_price, 2
"per_month_price_one_year_ttc": round(
access_fee_ttc / 12 + total_recurring_price_ttc, 2
),
}
)
sort_elig = sorted(
filtered_elig, key=lambda x: x["per_month_price_one_year"], reverse=False
filtered_elig,
key=lambda x: x["per_month_price_one_year_ttc"],
reverse=False,
)
return sort_elig