diff --git a/.gitignore b/.gitignore index 392f4f6..88d0713 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,3 @@ __pycache__ -/elig-test.ini \ No newline at end of file +/elig-test.ini +.vscode \ No newline at end of file diff --git a/README.md b/README.md index d276094..d26f959 100644 --- a/README.md +++ b/README.md @@ -45,3 +45,9 @@ curl -v http://127.0.0.1:5000/addresses/communes?s=29530 # Chercher les voies d'une commune via son code insee curl -v http://127.0.0.1:5000/addresses/fantoirvoies/29036 ``` + +### Jeux de données + +#### Insee + +Importez ce CSV https://www.data.gouv.fr/fr/datasets/base-officielle-des-codes-postaux/ dans la base de données diff --git a/address_finder/api.py b/address_finder/api.py index b40d90d..212c62e 100644 --- a/address_finder/api.py +++ b/address_finder/api.py @@ -1,79 +1,110 @@ import sqlite3 import sys import json -from .model import Commune,FantoirVoie +from .model import Commune, FantoirVoie +import re # DB with addresses info -DB_ADDRESSES_PATH_ENV="DB_ADDRESSES_PATH" -DB_ADDRESSES_DEFAULT_PATH="/etc/fantoir.sqlite" +DB_ADDRESSES_PATH_ENV = "DB_ADDRESSES_PATH" +DB_ADDRESSES_DEFAULT_PATH = "/etc/fantoir.sqlite" # Table for insee codes -DB_TABLE_INSEE_NAME="insee" -DB_COL_COMMUNE_INSEE="Code_commune_INSEE" -DB_COL_COMMUNE_NAME="Nom_commune" -DB_COL_COMMUNE_POSTE="Code_postal" - -# Table for Fantoir voies (code Rivoli) -DB_TABLE_FANTOIR_NAME="keyv" -DB_COL_FANTOIR_INSEE="key" -DB_FANTOIR_INSEE_KEY_SUFFIX="keyv:" +DB_TABLE_INSEE_NAME = "insee" +DB_COL_COMMUNE_INSEE = "Code_commune_INSEE" +DB_COL_COMMUNE_NAME = "Nom_commune" +DB_COL_COMMUNE_POSTE = "Code_postal" # Utility to find an address class AddressFinder: - def __init__(self, db_addresses_sqlite_path: str): - self.dbPath = db_addresses_sqlite_path - print("DB addresses Path : " + self.dbPath) + def __init__(self, db_insee_communes_sqlite_path: str, db_fantoir_voies_sqlite_path: str): + self.dbCommunesPath = db_insee_communes_sqlite_path + self.dbFantoirPath = db_fantoir_voies_sqlite_path + print("DB insee communes Path : " + self.dbCommunesPath) + print("DB Fantoir voies Path : " + self.dbFantoirPath) - def getCommunesFromNameOrZip(self, communeNameOrZip: str) -> list[Commune]: - con = sqlite3.connect(self.dbPath) + def getCommunesFromNameOrZip(self, communeNameOrZip: str, limit: int = None) -> list[Commune]: + con = sqlite3.connect(self.dbCommunesPath) con.row_factory = sqlite3.Row cur = con.cursor() - communes: list[Commune] = [] - try: - if communeNameOrZip is None: - cur.execute(f"SELECT * from \"{DB_TABLE_INSEE_NAME}\"") - else: - cur.execute(f"SELECT * from \"{DB_TABLE_INSEE_NAME}\" WHERE {DB_COL_COMMUNE_NAME}=\"{communeNameOrZip}\" COLLATE nocase OR {DB_COL_COMMUNE_POSTE}=\"{communeNameOrZip}\"") - except sqlite3.OperationalError as err: - print("Error querying DB : {0}".format(err), file=sys.stderr) - return [] - rows = [dict(row) for row in cur.fetchall()] - con.close() - for row in rows: - commune=Commune( - codeInsee=row[DB_COL_COMMUNE_INSEE], - nom=row[DB_COL_COMMUNE_NAME], - codeZip=row[DB_COL_COMMUNE_POSTE]) - communes.append(commune) - return communes - def getCommuneFantoirVoies(self, communeInseeCode: str) -> list[FantoirVoie]: - + # Check if a search limit is specified, make sure it is an integer + select_limit = "" + if limit is not None: + try: + select_limit = f"LIMIT {int(limit)}" + except ValueError: + print("Error, limit arg not a valid int: ", limit) + + try: + # If no search parameter, select all + if communeNameOrZip is None: + cur.execute( + f"SELECT * from \"{DB_TABLE_INSEE_NAME}\" {select_limit}") + else: + communeSearch = communeNameOrZip + zipSearch = communeNameOrZip + searchOpertor = "OR" + + # Allow search zip and commune at the same time, in the format "29530 PLO" + regexCommuneAndZip = r"[0-9]{5} .+" + if re.match(regexCommuneAndZip, communeNameOrZip): + splitSearch = communeNameOrZip.split(' ') + zipSearch = splitSearch[0] + communeSearch = ' '.join(splitSearch[1:]) + searchOpertor = "AND" + cur.execute( + f"SELECT * from \"{DB_TABLE_INSEE_NAME}\" WHERE {DB_COL_COMMUNE_NAME} LIKE \"%{communeSearch}%\" COLLATE nocase {searchOpertor} {DB_COL_COMMUNE_POSTE} LIKE \"{zipSearch}%\" {select_limit}") + except sqlite3.OperationalError as err: + print("Error querying DB : {0}".format(err), file=sys.stderr) + return [] + + communesMap = dict() + for row in cur.fetchall(): + row_obj = dict(row) + commune = Commune( + codeInsee=row_obj[DB_COL_COMMUNE_INSEE], + nom=row_obj[DB_COL_COMMUNE_NAME], + codeZip=row_obj[DB_COL_COMMUNE_POSTE]) + # This way we avoid duplicates in DB + communesMap[commune["codeInsee"]] = commune + + con.close() + return list(communesMap.values()) + + def getCommuneFantoirVoies(self, communeInseeCode: str, voieSearch: str = None, limit: int = None) -> list[FantoirVoie]: + # Extract data from DB - con = sqlite3.connect(self.dbPath) + con = sqlite3.connect(self.dbFantoirPath) con.row_factory = sqlite3.Row cur = con.cursor() + if voieSearch is None: + voieSearch='' + + # Check if a search limit is specified, make sure it is an integer + select_limit = "" + if limit is not None: + try: + select_limit = f"LIMIT {int(limit)}" + except ValueError: + print("Error, limit arg not a valid int: ", limit) + try: - cur.execute(f"SELECT value from \"{DB_TABLE_FANTOIR_NAME}\" WHERE {DB_COL_FANTOIR_INSEE}=\"{DB_FANTOIR_INSEE_KEY_SUFFIX}{communeInseeCode}\"") + cur.execute( + f"SELECT trim(libelle), rivoli_with_key from fantoir WHERE full_insee=\"{communeInseeCode}\" AND libelle like \"%{voieSearch}%\" {select_limit}") except sqlite3.OperationalError as err: - print("Error querying DB : {0}".format(err), file=sys.stderr) - return [] - data_raw = cur.fetchone() + print("Error querying DB : {0}".format(err), file=sys.stderr) + return [] + data_raw = cur.fetchall() con.close() - ## Get JSON payload - fantoir_dict = [] # Check if data where found if data_raw is not None: - data = dict(data_raw) - # Extract the data behind "value" which is a JSON structure - data_dict=json.loads(data.get("value")) - # In extracted JSON data, the interesting payload is behind "value" key - fantoir_dict = data_dict.get("value") + fantoir_dict = dict(data_raw) else: - print("Did not found any data matching Insee code " + str(communeInseeCode)) - + print("Did not found any data matching Insee code " + + str(communeInseeCode)) + # Return the json dump return fantoir_dict diff --git a/address_finder/model.py b/address_finder/model.py index da84fb0..b1a9218 100644 --- a/address_finder/model.py +++ b/address_finder/model.py @@ -6,12 +6,5 @@ class Commune(TypedDict): codeZip: str class FantoirVoie(TypedDict): - id: str - dateAjout: int - libelle: list[str] - typeVoie:str - codeCommune: str - codeFantoir: str - cleRivoli: str - nomCommune: str - predecesseur: bool + libelle: str + rivoli_with_key: int diff --git a/axione_api/api.py b/axione_api/api.py index 492844f..496a422 100644 --- a/axione_api/api.py +++ b/axione_api/api.py @@ -23,14 +23,37 @@ def ptoRequest(ptoRef): """ +def fantoirRequest(insee, rivoli, numVoie): + ts = datetime.now(timezone.utc).isoformat() + return f""" + + + + + + + + + + + + + + """ + +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 = base64.b64encode(f"{cfg.username}:{cfg.password}".encode("utf8")).decode( - "utf8" - ) + passwd = cfg.password headers = { "User-Agent": "aquilenet-elig-test/0.1", "Accept": "*/*", @@ -38,7 +61,7 @@ def query_axione_pto(cfg, ptoRef): "Connection": "Keep-Alive", "Authorization": passwd, } - resp = None + respData = None if not cfg.debug: try: conn = http.client.HTTPSConnection( @@ -67,11 +90,10 @@ def query_axione_pto(cfg, ptoRef): print(headers) print("BODY: ") print(body) - print("===================") with open("./fixtures/dummy-data-1.xml", "r") as f: dummyData = f.read() return dummyData - return resp + return respData class LigneResult(TypedDict): @@ -98,17 +120,45 @@ class BatimentResult(TypedDict): referenceBatiment: str etages: list[EtageResult] +class AxioneErreur(TypedDict): + codeErreur: str + libelleErreur: str -def parse_response(resp_str) -> list[BatimentResult]: +class AxioneResult(TypedDict): + codeRetour: int + axioneErreur: AxioneErreur + batiments: list[BatimentResult] + + +def parse_response(resp_str) -> AxioneResult: root = ET.fromstring(resp_str) - parsedBatiments = [ - parse_batiment(b) - for b in root.findall( - ".//{http://structureadresseftth.axione.fr/model/commun}batiment" - ) - ] - return parsedBatiments - + 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) diff --git a/config/config.py b/config/config.py index 9473c92..f55d4d0 100644 --- a/config/config.py +++ b/config/config.py @@ -2,11 +2,12 @@ import configparser class Config: - def __init__(self, username, password, source_addr, db_addresses_sqlite_path): + def __init__(self, username, password, source_addr, db_insee_communes_sqlite_path, db_fantoir_voies_sqlite_path): self.username = username self.password = password self.source_addr = source_addr - self.db_addresses_sqlite_path = db_addresses_sqlite_path + self.db_insee_communes_sqlite_path = db_insee_communes_sqlite_path + self.db_fantoir_voies_sqlite_path = db_fantoir_voies_sqlite_path self.debug = False @@ -17,5 +18,6 @@ def parse_config(cfgPath): username = cfg.get("API", "username") passwd = cfg.get("API", "password") source_addr = cfg.get("API", "source_addr") - db_addresses_sqlite_path = cfg.get("ADDRESSES","db_addresses_sqlite_path") - return Config(username, passwd, source_addr,db_addresses_sqlite_path) + db_insee_communes_sqlite_path = cfg.get("ADDRESSES","db_insee_communes_sqlite_path") + db_fantoir_voies_sqlite_path = cfg.get("ADDRESSES","db_fantoir_voies_sqlite_path") + return Config(username, passwd, source_addr,db_insee_communes_sqlite_path, db_fantoir_voies_sqlite_path) diff --git a/elig-test.ini.sample b/elig-test.ini.sample index 2f6c058..a468be5 100644 --- a/elig-test.ini.sample +++ b/elig-test.ini.sample @@ -5,4 +5,5 @@ # to send the requests from. source_addr = xxx.xxx.xxx.xxx [ADDRESSES] - db_addresses_sqlite_path = path/to/db.sqlite \ No newline at end of file + db_insee_communes_sqlite_path = path/to/db.sqlite + db_fantoir_voies_sqlite_path = path/to/db.sqlite \ No newline at end of file diff --git a/fixtures/dummy-data-2.xml b/fixtures/dummy-data-2.xml new file mode 100644 index 0000000..a8b8789 --- /dev/null +++ b/fixtures/dummy-data-2.xml @@ -0,0 +1,46 @@ + + + + + + + + + 0 + + BEFO + true + + + + + + + + + + + + + + + BF-ISS1-0110 + + + + + + + + + + PME + OC + + + + + + + + \ No newline at end of file diff --git a/fixtures/dummy-data-3.xml b/fixtures/dummy-data-3.xml new file mode 100644 index 0000000..f83388c --- /dev/null +++ b/fixtures/dummy-data-3.xml @@ -0,0 +1,36 @@ + + + + + + + + + 1 + I01 + Code Rivoli introuvable, manquant ou incomplet + + + + + + + + + + + + + + + + PME + OI + + + + + + + + \ No newline at end of file diff --git a/run-dev-server b/run-dev-server index 07bf5f5..4827992 100755 --- a/run-dev-server +++ b/run-dev-server @@ -1,3 +1,3 @@ #!/usr/bin/env bash -DEBUG=true CONFIG=./elig-test.ini FLASK_APP=webapp poetry run flask run --reload +DEBUG=true TEMPLATES_AUTO_RELOAD=true FLASK_ENV=development CONFIG=./elig-test.ini FLASK_APP=webapp poetry run flask run --reload diff --git a/scripts/import-laposte-insee.sh b/scripts/import-laposte-insee.sh new file mode 100755 index 0000000..88c8b28 --- /dev/null +++ b/scripts/import-laposte-insee.sh @@ -0,0 +1,29 @@ +#!/usr/bin/env bash +set -eau -o pipefail + +if [ "$#" -ne 2 ]; then + echo "Usage: import-laposte-insee.sh path-to-laposte-insee-CSV fantoir-sqlite-db-path" + echo "" + echo "ERROR: Missing laposte CSV." + echo "You can download it at https://www.data.gouv.fr/fr/datasets/base-officielle-des-codes-postaux/" + exit 1 +fi + +tmpDir=$(mktemp -d) +clean_tmp () { + rm -r "${tmpDir}" +} +trap clean_tmp EXIT +tmpSql="${tmpDir}"/import-laposte-hexasmal.sql + +echo "Importing laposte/insee hexasmal data into the fantoir db." + +cat >"${tmpSql}" < + - + + + + + + + + Aquilenet: Éligibilité FTTH -

Aquilenet

-
-

Test d'Éligibilité FTTH Aquilenet

-
- - -
+

+ AQUILENET +

+
+ +
+ +
+
+ +

Test d'éligibilité par PTO

+
+
+ + + + Où trouver mon numéro de PTO ? +
+
+ PTO (Point de terminaison optique) est un numéro unique que vous pouvez trouver sur le boîtier de + raccordement de la fibre. + C'est un petit boîtier blanc installé dans la maison +
+ Responsive image +
+
+
+ +
+ +
+ +
+ +
+

Test d'éligibilité par adresse

+
+
+ + +
+
+
+ + +
+
+ + +
+
+
+ +
+
+ +
+
+
+ + + diff --git a/templates/result.html b/templates/result.html index 5db72a8..57a2385 100644 --- a/templates/result.html +++ b/templates/result.html @@ -3,7 +3,19 @@ - + + + + + + + Aquilenet: Éligibilité FTTH