diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..392f4f6 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +__pycache__ +/elig-test.ini \ No newline at end of file diff --git a/address_finder/__init__.py b/address_finder/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/address_finder/api.py b/address_finder/api.py new file mode 100644 index 0000000..bcaa3b7 --- /dev/null +++ b/address_finder/api.py @@ -0,0 +1,71 @@ +import sqlite3 +import os +import json +from .model import Commune,FantoirVoie + +# DB with addresses info +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:" + +# 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 getCommunesFromNameOrZipJson(self, communeNameOrZip: str) -> list[Commune]: + con = sqlite3.connect(self.dbPath) + con.row_factory = sqlite3.Row + cur = con.cursor() + communes: list[Commune] = [] + 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}\"") + 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 getCommuneFantoirVoiesJson(self, communeInseeCode: str) -> list[FantoirVoie]: + + # Extract data from DB + con = sqlite3.connect(self.dbPath) + con.row_factory = sqlite3.Row + cur = con.cursor() + cur.execute(f"SELECT value from \"{DB_TABLE_FANTOIR_NAME}\" WHERE {DB_COL_FANTOIR_INSEE}=\"{DB_FANTOIR_INSEE_KEY_SUFFIX}{communeInseeCode}\"") + data_raw = cur.fetchone() + 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") + else: + 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 new file mode 100644 index 0000000..da84fb0 --- /dev/null +++ b/address_finder/model.py @@ -0,0 +1,17 @@ +from typing import TypedDict + +class Commune(TypedDict): + codeInsee: str + nom: str + codeZip: str + +class FantoirVoie(TypedDict): + id: str + dateAjout: int + libelle: list[str] + typeVoie:str + codeCommune: str + codeFantoir: str + cleRivoli: str + nomCommune: str + predecesseur: bool diff --git a/config/__init__.py b/config/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/axione_api/config.py b/config/config.py similarity index 58% rename from axione_api/config.py rename to config/config.py index 20cabc9..9473c92 100644 --- a/axione_api/config.py +++ b/config/config.py @@ -2,10 +2,11 @@ import configparser class Config: - def __init__(self, username, password, source_addr): + def __init__(self, username, password, source_addr, db_addresses_sqlite_path): self.username = username self.password = password self.source_addr = source_addr + self.db_addresses_sqlite_path = db_addresses_sqlite_path self.debug = False @@ -16,4 +17,5 @@ def parse_config(cfgPath): username = cfg.get("API", "username") passwd = cfg.get("API", "password") source_addr = cfg.get("API", "source_addr") - return Config(username, passwd, source_addr) + db_addresses_sqlite_path = cfg.get("ADDRESSES","db_addresses_sqlite_path") + return Config(username, passwd, source_addr,db_addresses_sqlite_path) diff --git a/elig-test.ini.sample b/elig-test.ini.sample index b4ae260..2f6c058 100644 --- a/elig-test.ini.sample +++ b/elig-test.ini.sample @@ -3,4 +3,6 @@ password = xxx # Whitelisted IP address from axione # to send the requests from. - source_addr = xxx.xxx.xxx.xxx \ No newline at end of file + source_addr = xxx.xxx.xxx.xxx +[ADDRESSES] + db_addresses_sqlite_path = path/to/db.sqlite \ No newline at end of file diff --git a/run-dev-server b/run-dev-server index 73af97d..07bf5f5 100755 --- a/run-dev-server +++ b/run-dev-server @@ -1,3 +1,3 @@ #!/usr/bin/env bash -DEBUG=true CONFIG=./elig-test.ini.sample FLASK_APP=webapp poetry run flask run --reload +DEBUG=true CONFIG=./elig-test.ini FLASK_APP=webapp poetry run flask run --reload diff --git a/webapp.py b/webapp.py index 905fcd3..f372305 100644 --- a/webapp.py +++ b/webapp.py @@ -1,8 +1,10 @@ import os from flask import Flask, render_template, request, escape +import json -from axione_api.config import parse_config +from config.config import parse_config from axione_api.api import query_axione_pto, parse_response +from address_finder.api import AddressFinder def load_config(): cfg_path = os.environ.get("CONFIG", "/etc/ftth-elig/conf.ini") @@ -19,6 +21,9 @@ def load_config(): return cfg cfg = load_config() + +addressFinder = AddressFinder(cfg.db_addresses_sqlite_path) + app = Flask(__name__) @app.route("/", methods=['GET']) @@ -30,3 +35,23 @@ def show_result(): pto = escape(request.form['pto']) result = parse_response(query_axione_pto(cfg, pto)) return render_template("result.html", pto=pto, result=result) + +@app.route("/addresses/communes", methods=['GET']) +def get_communes(): + to_search=request.args.get('s') + print(to_search) + communes=addressFinder.getCommunesFromNameOrZipJson(to_search) + response = app.response_class( + response=json.dumps(communes), + mimetype='application/json' + ) + return response + +@app.route("/addresses/fantoirvoies/", methods=['GET']) +def get_fantoir_voies(codeInsee): + fantoirVoies=addressFinder.getCommuneFantoirVoiesJson(codeInsee) + response = app.response_class( + response=json.dumps(fantoirVoies), + mimetype='application/json' + ) + return response \ No newline at end of file