#!/usr/bin/python3
# -*- coding: utf-8 -*-
########################################################################
#
# This file is part of python module <pySPC>.
# Copyright (C) 2013-2020 R. Marty
# (renaud.marty@developpement-durable.gouv.fr)
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program (see COPYING.txt).
# If not, see <http://www.gnu.org/licenses/>.
#
########################################################################
"""
Webservice - PHyC - Client et requêtes
"""
from datetime import timedelta as td
import os
import os.path
import ssl
from pyspc.core.config import Config
import pyspc.core.exception as _exception
from ._check import Check
from ._get import Get
from ._request import Request
from ._session import Session
DEFAULT_CFG_FILENAME = os.path.join(os.path.dirname(__file__), 'phyc.txt')
"""Fichier interne dédié à la PHyC"""
HOSTNAMES = {}
"""Dictionnaire des hôtes des documents en ligne"""
if os.path.exists(DEFAULT_CFG_FILENAME):
config = Config(filename=DEFAULT_CFG_FILENAME)
config.read()
HOSTNAMES.update({s: config[s].get('hostname', '') for s in config})
[docs]
class PHyC(Check, Get, Request, Session):
"""
Structure du client SOAP accédant aux données de la PHyC
Attributes
----------
hostname : str
Hôte du webservice
proxies : None, dict
Dictionnaire des proxys {'protocol': 'proxy'}
username : str
Identifiant du contact
password : str
Mot de passe du contact
client : suds.client.Client
Client de connexion à la PHyC
session : None, str
Identifiant de la session PHyC
"""
[docs]
def __init__(self, hostname=None, username='',
password='', proxies=None, warning=False): # nosec
"""
Initialisation de l'instance du webservice PHyC.
Initialisation du client SOAP
Parameters
----------
hostname : str
Hôte du webservice
proxies : None, dict
Dictionnaire des proxys {'protocol': 'proxy'}
username : str
Identifiant du contact
password : str
Mot de passe du contact
warning : bool
Afficher les avertissements (proxy)
"""
import suds.client as _sclient
super().__init__()
if proxies is not None and warning:
_exception.Warning(
__name__,
"Le webservice PHyC du projet HYDRO3 est uniquement "
"accessible depuis le RIE. Il n'est donc pas nécessaire de "
"définir le proxy. Il est défini ainsi: proxies={}")
self.proxies = {}
if hostname is None and 'phyc' in HOSTNAMES:
self.hostname = HOSTNAMES['phyc']
else:
_exception.check_str(hostname)
self.hostname = hostname
os.environ['http_proxy'] = self.proxies.get('http', '')
os.environ['https_proxy'] = self.proxies.get('https', '')
# Création du client
try:
self.client = _sclient.Client(self.hostname)
except Exception as e:
self.client = None
if isinstance(e.args[0], ssl.SSLCertVerificationError):
try:
ssl._create_default_https_context = \
ssl._create_unverified_context
self.client = _sclient.Client(self.hostname)
except Exception:
raise ValueError(
f"Le client est mal configuré: {e.args[0]}") from e
_exception.Warning(
None,
"Connection PHyC SANS vérification SSL. "
"Veuillez vérifier vos certificats.")
else:
raise ValueError(
f"Le client est mal configuré: {e.args[0]}") from e
# Autres informations
self.session = None
self.username = username
self.password = password
# https://stackoverflow.com/questions/35569042/ssl-certificate-verify-failed-with-python3
# export PYTHONHTTPSVERIFY=0
# ok avec py37
# import ssl
# ssl._create_default_https_context = ssl._create_unverified_context
# version plus récentes
# ssl.SSLContext.verify_mode = ssl.VerifyMode.CERT_OPTIONAL
# import urllib.request, ssl
# def urllib_get_2018():
# # Using a protected member like this is not any more fragile
# # than extending the class and using it. I would use it.
# url = 'https://localhost:6667/my-endpoint'
# ssl._create_default_https_context = ssl._create_unverified_context
# with urllib.request.urlopen(url = url) as f:
# print(f.read().decode('utf-8'))
# def urllib_get_2022():
# # Finally! Able to use the publice API. Happy happy!
# url = 'https://localhost:6667/my-endpoint'
# scontext = ssl.SSLContext(ssl.PROTOCOL_TLS)
# scontext.verify_mode = ssl.VerifyMode.CERT_NONE
# with urllib.request.urlopen(url = url, context=scontext) as f:
# print(f.read().decode('utf-8'))
# https://stackoverflow.com/questions/37327368/bypass-ssl-when-im-using-suds-for-consume-web-service
def __str__(self):
"""Afficher les méta-données de l'instance PHyC."""
text = """
*************************************
*********** WEBSERVICE PHyC *********
*************************************
* HOSTNAME = {hostname}
* PROXIES = {proxies}
* USERNAME = {username}
* SESSION = {session}
*************************************
"""
return text.format(**vars(self))
@staticmethod
def _define_subperiods(start, end, size):
"""Définir une liste de <size> sous-périodes entre start et end."""
delta = end - start
subperiods = []
for s in range(size):
fdt = s * (delta.total_seconds() / size) * td(seconds=1) + start
ldt = (s+1) * (delta.total_seconds() / size) * td(seconds=1) + \
start
subperiods.append([fdt, ldt])
return subperiods