248 lines
8.9 KiB
Python
248 lines
8.9 KiB
Python
# coding: utf-8
|
|
"""
|
|
This module handles the configuration management for Flatisfy.
|
|
|
|
It loads the default configuration, then overloads it with the provided config
|
|
file and then overloads it with command-line options.
|
|
"""
|
|
from __future__ import absolute_import, print_function, unicode_literals
|
|
from builtins import str
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
import traceback
|
|
|
|
import appdirs
|
|
|
|
from flatisfy import tools
|
|
|
|
|
|
# Default configuration
|
|
DEFAULT_CONFIG = {
|
|
# Constraints to match
|
|
"constraints": {
|
|
"type": None, # RENT, SALE, SHARING
|
|
"house_types": [], # List of house types, must be in APART, HOUSE,
|
|
# PARKING, LAND, OTHER or UNKNOWN
|
|
"postal_codes": [], # List of postal codes
|
|
"area": (None, None), # (min, max) in m^2
|
|
"cost": (None, None), # (min, max) in currency unit
|
|
"rooms": (None, None), # (min, max)
|
|
"bedrooms": (None, None), # (min, max)
|
|
"time_to": {} # Dict mapping names to {"gps": [lat, lng],
|
|
# "time": (min, max) }
|
|
# Time is in seconds
|
|
},
|
|
# Navitia API key
|
|
"navitia_api_key": None,
|
|
# Number of filtering passes to run
|
|
"passes": 3,
|
|
# Maximum number of entries to fetch
|
|
"max_entries": None,
|
|
# Directory in wich data will be put. ``None`` is XDG default location.
|
|
"data_directory": None,
|
|
# Path to the modules directory containing all Weboob modules. ``None`` if
|
|
# ``weboob_modules`` package is pip-installed, and you want to use
|
|
# ``pkgresource`` to automatically find it.
|
|
"modules_path": None,
|
|
# SQLAlchemy URI to the database to use
|
|
"database": None,
|
|
# Path to the Whoosh search index file. Use ``None`` to put it in
|
|
# ``data_directory``.
|
|
"search_index": None,
|
|
# Web app port
|
|
"port": 8080,
|
|
# Web app host to listen on
|
|
"host": "127.0.0.1",
|
|
# Web server to use to serve the webapp (see Bottle deployment doc)
|
|
"webserver": None,
|
|
# List of Weboob backends to use (default to any backend available)
|
|
"backends": None,
|
|
}
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
def validate_config(config):
|
|
"""
|
|
Check that the config passed as argument is a valid configuration.
|
|
|
|
:param config: A config dictionary to fetch.
|
|
:return: ``True`` if the configuration is valid, ``False`` otherwise.
|
|
"""
|
|
def _check_constraints_bounds(bounds):
|
|
"""
|
|
Check the bounds for numeric constraints.
|
|
"""
|
|
assert len(bounds) == 2
|
|
assert all(
|
|
x is None or
|
|
(
|
|
isinstance(x, (float, int)) and
|
|
x >= 0
|
|
)
|
|
for x in bounds
|
|
)
|
|
if bounds[0] is not None and bounds[1] is not None:
|
|
assert bounds[1] > bounds[0]
|
|
|
|
try:
|
|
# Note: The traceback fetching code only handle single line asserts.
|
|
# Then, we disable line-too-long pylint check and E501 flake8 checks
|
|
# and use long lines whenever needed, in order to have the full assert
|
|
# message in the log output.
|
|
# pylint: disable=locally-disabled,line-too-long
|
|
assert "type" in config["constraints"]
|
|
assert isinstance(config["constraints"]["type"], str)
|
|
assert config["constraints"]["type"].upper() in ["RENT",
|
|
"SALE", "SHARING"]
|
|
|
|
assert "house_types" in config["constraints"]
|
|
assert config["constraints"]["house_types"]
|
|
for house_type in config["constraints"]["house_types"]:
|
|
assert house_type.upper() in ["APART", "HOUSE", "PARKING", "LAND",
|
|
"OTHER", "UNKNOWN"]
|
|
|
|
assert "postal_codes" in config["constraints"]
|
|
assert config["constraints"]["postal_codes"]
|
|
|
|
assert "area" in config["constraints"]
|
|
_check_constraints_bounds(config["constraints"]["area"])
|
|
|
|
assert "cost" in config["constraints"]
|
|
_check_constraints_bounds(config["constraints"]["cost"])
|
|
|
|
assert "rooms" in config["constraints"]
|
|
_check_constraints_bounds(config["constraints"]["rooms"])
|
|
|
|
assert "bedrooms" in config["constraints"]
|
|
_check_constraints_bounds(config["constraints"]["bedrooms"])
|
|
|
|
assert "time_to" in config["constraints"]
|
|
assert isinstance(config["constraints"]["time_to"], dict)
|
|
for name, item in config["constraints"]["time_to"].items():
|
|
assert isinstance(name, str)
|
|
assert "gps" in item
|
|
assert isinstance(item["gps"], list)
|
|
assert len(item["gps"]) == 2
|
|
assert "time" in item
|
|
_check_constraints_bounds(item["time"])
|
|
|
|
assert config["passes"] in [0, 1, 2, 3]
|
|
assert config["max_entries"] is None or (isinstance(config["max_entries"], int) and config["max_entries"] > 0) # noqa: E501
|
|
|
|
assert config["data_directory"] is None or isinstance(config["data_directory"], str) # noqa: E501
|
|
assert os.path.isdir(config["data_directory"])
|
|
assert isinstance(config["search_index"], str)
|
|
assert config["modules_path"] is None or isinstance(config["modules_path"], str) # noqa: E501
|
|
|
|
assert config["database"] is None or isinstance(config["database"], str) # noqa: E501
|
|
|
|
assert isinstance(config["port"], int)
|
|
assert isinstance(config["host"], str)
|
|
assert config["webserver"] is None or isinstance(config["webserver"], str) # noqa: E501
|
|
assert config["backends"] is None or isinstance(config["backends"], list) # noqa: E501
|
|
|
|
return True
|
|
except (AssertionError, KeyError):
|
|
_, _, exc_traceback = sys.exc_info()
|
|
return traceback.extract_tb(exc_traceback)[-1][-1]
|
|
|
|
|
|
def load_config(args=None):
|
|
"""
|
|
Load the configuration from file.
|
|
|
|
:param args: An argparse args structure.
|
|
:return: The loaded config dict.
|
|
"""
|
|
LOGGER.info("Initializing configuration...")
|
|
# Default configuration
|
|
config_data = DEFAULT_CONFIG.copy()
|
|
|
|
# Load config from specified JSON
|
|
if args and getattr(args, "config", None):
|
|
LOGGER.debug("Loading configuration from %s.", args.config)
|
|
try:
|
|
with open(args.config, "r") as fh:
|
|
config_data.update(json.load(fh))
|
|
except (IOError, ValueError) as exc:
|
|
LOGGER.error(
|
|
"Unable to load configuration from file, "
|
|
"using default configuration: %s.",
|
|
exc
|
|
)
|
|
|
|
# Overload config with arguments
|
|
if args and getattr(args, "passes", None) is not None:
|
|
LOGGER.debug(
|
|
"Overloading number of passes from CLI arguments: %d.",
|
|
args.passes
|
|
)
|
|
config_data["passes"] = args.passes
|
|
if args and getattr(args, "max_entries", None) is not None:
|
|
LOGGER.debug(
|
|
"Overloading maximum number of entries from CLI arguments: %d.",
|
|
args.max_entries
|
|
)
|
|
config_data["max_entries"] = args.max_entries
|
|
if args and getattr(args, "port", None) is not None:
|
|
LOGGER.debug("Overloading web app port: %d.", args.port)
|
|
config_data["port"] = args.port
|
|
if args and getattr(args, "host", None) is not None:
|
|
LOGGER.debug("Overloading web app host: %s.", args.host)
|
|
config_data["host"] = str(args.host)
|
|
|
|
# Handle data_directory option
|
|
if args and getattr(args, "data_dir", None) is not None:
|
|
LOGGER.debug("Overloading data directory from CLI arguments.")
|
|
config_data["data_directory"] = args.data_dir
|
|
elif config_data["data_directory"] is None:
|
|
config_data["data_directory"] = appdirs.user_data_dir(
|
|
"flatisfy",
|
|
"flatisfy"
|
|
)
|
|
LOGGER.debug("Using default XDG data directory: %s.",
|
|
config_data["data_directory"])
|
|
|
|
if not os.path.isdir(config_data["data_directory"]):
|
|
LOGGER.info("Creating data directory according to config: %s",
|
|
config_data["data_directory"])
|
|
os.mkdir(config_data["data_directory"])
|
|
|
|
if config_data["database"] is None:
|
|
config_data["database"] = "sqlite:///" + os.path.join(
|
|
config_data["data_directory"],
|
|
"flatisfy.db"
|
|
)
|
|
|
|
if config_data["search_index"] is None:
|
|
config_data["search_index"] = os.path.join(
|
|
config_data["data_directory"],
|
|
"search_index"
|
|
)
|
|
|
|
config_validation = validate_config(config_data)
|
|
if config_validation is True:
|
|
LOGGER.info("Config has been fully initialized.")
|
|
return config_data
|
|
LOGGER.error("Error in configuration: %s.", config_validation)
|
|
return None
|
|
|
|
|
|
def init_config(output=None):
|
|
"""
|
|
Initialize an empty configuration file.
|
|
|
|
:param output: File to output content to. Defaults to ``stdin``.
|
|
"""
|
|
config_data = DEFAULT_CONFIG.copy()
|
|
|
|
if output and output != "-":
|
|
with open(output, "w") as fh:
|
|
fh.write(tools.pretty_json(config_data))
|
|
else:
|
|
print(tools.pretty_json(config_data))
|