axione-elig-test/axione_api/api.py

import base64
import http.client
import sys
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from typing import TypedDict, Optional


def ptoRequest(ptoRef):
    ts = datetime.now(timezone.utc).isoformat()
    return f"""
        <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:ent="http://structureadresseftth.axione.fr/model/entreprise" xmlns:com="http://structureadresseftth.axione.fr/model/commun">
        <soapenv:Header/>
        <soapenv:Body>
            <ent:obtentionStructureAdresseDemandeSoap>
                <ent:entete versionWS="3.0" horodatageRequete="{ts}">
                    <com:operateurCommercial nom="AQUILENET" identifiant=""/>
                </ent:entete>
                <ent:referenceAdresse referenceHexacle="" identifiantImmeuble="" referencePTO="{ptoRef}" referenceBAN="">
                </ent:referenceAdresse>
            </ent:obtentionStructureAdresseDemandeSoap>
        </soapenv:Body>
        </soapenv:Envelope>
    """

def fantoirRequest(insee, rivoli, numVoie):
    ts = datetime.now(timezone.utc).isoformat()
    return f"""
        <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:ent="http://structureadresseftth.axione.fr/model/entreprise" xmlns:com="http://structureadresseftth.axione.fr/model/commun">
        <soapenv:Header/>
        <soapenv:Body>
            <ent:obtentionStructureAdresseDemandeSoap>
                <ent:entete versionWS="3.0" horodatageRequete="{ts}">
                    <com:operateurCommercial nom="AQUILENET" identifiant=""/>
                </ent:entete>
                <ent:referenceAdresse>
                        <com:referenceRivoli codeInsee="{insee}" codeRivoli="{rivoli}" numeroVoie="{numVoie}"/>
                </ent:referenceAdresse>
            </ent:obtentionStructureAdresseDemandeSoap>
        </soapenv:Body>
        </soapenv:Envelope>
    """

def query_axione_fantoir(cfg, insee, rivoli, numVoie):
    body = fantoirRequest(insee, rivoli, numVoie)
    return query_axione(cfg, body)


def query_axione_pto(cfg, ptoRef):
    body = ptoRequest(ptoRef)
    return query_axione(cfg, body)

def query_axione(cfg, body):
    # Note: the password should be the base64 of username:password.
    # Don't ask why.
    passwd = cfg.password
    headers = {
        "User-Agent": "aquilenet-elig-test/0.1",
        "Accept": "*/*",
        "Accept-Encoding": "identity",
        "Connection": "Keep-Alive",
        "Authorization": passwd,
    }
    respData = None
    if not cfg.debug:
        try:
            conn = http.client.HTTPSConnection(
                "ws-eligftth-val.axione.fr",
                443,
                source_address=(cfg.source_addr, 0),
                timeout=120,
            )
            conn.request("POST", "/v3/fai", body, headers=headers)
            response = conn.getresponse()
            respData = response.read()
        except Exception as e:
            print("Error while querying Axione: ", file=sys.stderr)
            print(str(e), file=sys.stderr)
            print("Query Body: ")
            print(body)
            print("Query Headers: ")
            print(str(headers))
            sys.exit(1)
        finally:
            conn.close()
    else:
        print("===================")
        print("Injecting dummy response for request: ")
        print("HEADERS: ")
        print(headers)
        print("BODY: ")
        print(body)
        with open("./fixtures/dummy-data-1.xml", "r") as f:
            dummyData = f.read()
        return dummyData
    return respData


class LigneResult(TypedDict):
    actif: str
    commercialisable: str
    existant: str
    raccordable: str
    rompu: str
    pbo: str
    pto: str


class EtageResult(TypedDict):
    reference: str
    nbLignesActives: str
    nbLignesExistantes: str
    nbLocauxFtth: str
    lignes: list[LigneResult]


class BatimentResult(TypedDict):
    etatBatiment: str
    identifiantImmeuble: str
    referenceBatiment: str
    etages: list[EtageResult]

class AxioneErreur(TypedDict):
    codeErreur: str
    libelleErreur: str

class AxioneResult(TypedDict):
    codeRetour: int
    axioneErreur: AxioneErreur
    batiments: list[BatimentResult]


def parse_response(resp_str) -> AxioneResult:
    root = ET.fromstring(resp_str)
    codeRetourXml = root.find(
        ".//{http://structureadresseftth.axione.fr/model/commun}codeRetour"
    )
    codeRetour = int(codeRetourXml.text.strip())
    axioneErreur = None
    parsedBatiments = []
    if codeRetour == 0:
      parsedBatiments = [
          parse_batiment(b)
          for b in root.findall(
              ".//{http://structureadresseftth.axione.fr/model/commun}batiment"
          )
      ]
    else:
      axioneErreur = AxioneErreur(
          codeErreur = root.find(
              ".//{http://structureadresseftth.axione.fr/model/commun}codeErreur"
          ).text.strip(),
          libelleErreur = root.find(
              ".//{http://structureadresseftth.axione.fr/model/commun}libelleErreur"
          ).text.strip()
      )
    return AxioneResult(
      codeRetour=codeRetour,
      axioneErreur=axioneErreur,
      batiments=parsedBatiments
    )

def parse_batiment(batiment) -> BatimentResult:
    etatBatiment = batiment.get("etatBatiment", None)
    identifiantImmeuble = batiment.get("identifiantImmeuble", None)
    referenceBatiment = batiment.get("referenceBatiment", None)
    etages = [
        parse_etage(e)
        for e in batiment.findall(
            ".//{http://structureadresseftth.axione.fr/model/commun}etage"
        )
    ]
    return {
        "etatBatiment": etatBatiment,
        "identifiantImmeuble": identifiantImmeuble,
        "referenceBatiment": referenceBatiment,
        "etages": etages,
    }


def parse_etage(etage) -> EtageResult:
    reference = etage.get("reference", None)
    nbLignesActives = etage.get("nombreLignesActives", None)
    nbLignesExistantes = etage.get("nombreLignesExistantes", None)
    nbLocauxFtth = etage.get("nombreLocauxFTTH", None)
    lignes: list[LigneResult] = []
    for ligne in etage.findall(
        ".//{http://structureadresseftth.axione.fr/model/commun}ligneFTTH"
    ):
        pl = parse_ligne(ligne)
        if pl is not None:
            lignes.append(pl)
    return {
        "reference": reference,
        "nbLignesActives": nbLignesActives,
        "nbLignesExistantes": nbLignesExistantes,
        "nbLocauxFtth": nbLocauxFtth,
        "lignes": lignes,
    }


def parse_ligne(ligne) -> Optional[LigneResult]:
    statut = ligne.find(
        ".//{http://structureadresseftth.axione.fr/model/commun}statutLigneFTTH"
    )
    if statut == None:
        # This line does not have any interesting data for us.
        return None
    prise = ligne.find(".//{http://structureadresseftth.axione.fr/model/commun}prise")
    refPto = ligne.find(
        ".//{http://structureadresseftth.axione.fr/model/commun}referencePTO"
    )
    actif = statut.get("actif", None)
    commercialisable = statut.get("commercialisable", None)
    existant = statut.get("existant", None)
    raccordable = statut.get("raccordable", None)
    rompu = statut.get("rompu", None)
    pbo = prise.get("referencePBO", None)
    pto = refPto.text.strip()
    return {
        "actif": actif,
        "commercialisable": commercialisable,
        "existant": existant,
        "raccordable": raccordable,
        "rompu": rompu,
        "pbo": pbo,
        "pto": pto,
    }