# coding: utf-8
import re
import os
import random
import datetime
import poste
from utils import encode_file, decode_file
[docs]class Indirizzo(object):
"""
Un indirizzo
"""
def __init__(self, dug='', toponimo='', numero_civico='', esponente=''):
"""
Tutti gli elementi
:param suds_client: the soap client
:param dug: via/viale/...
:param toponimo: la via, il viale... es.: Europa
:param numero_civico: ovvio...
:param esponente: ???
"""
self.dug = dug
self.toponimo = toponimo
self.numero_civico = numero_civico
self.esponente = esponente
def validate(self, nominativo):
indirizzo = " ".join([self.dug, self.toponimo,
self.numero_civico, self.esponente])
if len(indirizzo) > 44:
msg = u"dug + indirizzo + numero + esponente non deve superare 44 chars (%s in %r)"
raise poste.RecipientValidationError(msg % (indirizzo, nominativo))
[docs] def get_instance(self, suds_client):
"""
get an instance from web service, suitable for further processing
"""
creation_name = '{http://ComunicazioniElettroniche.XOL}Indirizzo'
indirizzo = suds_client.factory.create(creation_name)
indirizzo._DUG = self.dug
indirizzo._Toponimo = self.toponimo
indirizzo._NumeroCivico = self.numero_civico
indirizzo._Esponente = self.esponente
return indirizzo
def __repr__(self):
return "%s %s %s" % (self.dug, self.toponimo, self.numero_civico)
[docs]class Nominativo(poste.RecipientTOL):
"""
Un destinatario o un mittente deve avere definito un Indirizzo
"""
def __init__(self, dug='', toponimo='', numero_civico='',
esponente='', zona='', tipo_indirizzo='', stato='',
telefono='', forza_destinazione='false',
casella_postale='', ufficio_postale='', provincia='',
complemento_nominativo='', ragione_sociale='',
cognome='', nome='', frazione='', citta='',
complemento_indirizzo='', cap='', validate=True):
"""
L'oggetto delle poste necessita l'Indirizzo come oggetto separato, noi no...
:param suds_client: the soap client
:param dug: via/viale/...
:param toponimo: la via, il viale... es.: Europa
:param numero_civico: ovvio...
:param esponente: ???
:param nome:
:param cognome:
:param zona:
:param tipo_indirizzo: NORMALE | CASELLA POSTALE
:param stato:
:param telefono:
:param forza_destinazione:
:param casella_postale:
:param ufficio_postale:
:param provincia:
:param complemento_nominativo:
:param ragione_sociale:
:param frazione:
:param citta:
:param complemento:
:param cap: codice avviamento postale
"""
self.zona = zona
self.tipo_indirizzo = tipo_indirizzo
self.stato = stato
self.telefono = telefono
self.forza_destinazione = forza_destinazione
self.casella_postale = casella_postale
self.ufficio_postale = ufficio_postale
self.provincia = provincia
self.complemento_nominativo = complemento_nominativo
self.ragione_sociale = ragione_sociale
self.cognome = cognome
self.nome = nome
self.frazione = frazione
self.citta = citta
self.complemento_indirizzo = complemento_indirizzo
self.cap = cap
self._Indirizzo = Indirizzo(dug, toponimo, numero_civico, esponente)
if validate:
self.validate()
[docs] def get_instance(self, suds_client):
"""
get an instance from web service, suitable for further processing
"""
creation_name = '{http://ComunicazioniElettroniche.XOL}Nominativo'
nominativo = suds_client.factory.create(creation_name)
nominativo._Zona = self.zona
nominativo._TipoIndirizzo = self.tipo_indirizzo
nominativo._Stato = self.stato
nominativo._Telefono = self.telefono
nominativo._ForzaDestinazione = self.forza_destinazione
nominativo._CasellaPostale = self.casella_postale
nominativo._UfficioPostale = self.ufficio_postale
nominativo._Provincia = self.provincia
nominativo._ComplementoNominativo = self.complemento_nominativo
nominativo._RagioneSociale = self.ragione_sociale
nominativo._Cognome = self.cognome
nominativo._Nome = self.nome
nominativo._Frazione = self.frazione
nominativo._Citta = self.citta
nominativo._ComplementoIndirizzo = self.complemento_indirizzo
nominativo._CAP = self.cap
nominativo.Indirizzo = self._Indirizzo.get_instance(suds_client)
return nominativo
[docs] def validate(self):
"Check that all parameters fit what Poste requires"
# implementa pag. 53 di "Specifiche interfaccia"
# if not re.search('^[0-9]*$', self.telefono):
# msg = "Il telefono può contenere solo caratteri numerici (%s)" % dest_name
# raise poste.RecipientValidationError(msg)
self.check_length('telefono', 18)
self.check_length('cap', 10)
self.check_length('casella_postale', 44)
self.check_length('citta', 44)
self.check_length('complemento_indirizzo', 44)
self.check_length('complemento_nominativo', 44)
self.check_length('frazione', 44)
self.check_length('nome', 44)
self.check_length('provincia', 4)
self.check_length('stato', 44)
self.check_length('ufficio_postale', 50)
self.check_length('zona', 44)
self.check_length(['nome', 'cognome'], 44)
self.check_length(['cap', 'citta', 'provincia'], 44)
self.check_length(['frazione', 'complemento_indirizzo'], 44)
self._Indirizzo.validate(self)
if not self.stato:
msg = u"Il campo 'stato' è richiesto (%r)"
raise poste.RecipientValidationError(msg % self)
if self.tipo_indirizzo and not self.tipo_indirizzo in ('NORMALE', 'CASELLA POSTALE'):
msg = u"Tipo indirizzo deve essere 'NORMALE' o 'CASELLA POSTALE' non '%s'"
raise poste.RecipientValidationError(msg % self.tipo_indirizzo)
# if not self.cognome and (self.nome and not self.ragione_sociale):
# msg = u"'Cognome' deve essere presente se manca 'nome' ed è valorizzata " + \
# u"'Ragione sociale' (%r)" % (self)
# raise poste.RecipientValidationError(msg)
# if not self.cognome and (self.nome and not self.ragione_sociale):
# msg = u"Il cognome è necessario se manca ragione sociale %s"
# raise poste.RecipientValidationError(msg % self)
# if not self.nome and (self.cognome and not self.ragione_sociale):
# msg = u"Il nome è necessario se manca ragione sociale %s"
# raise poste.RecipientValidationError(msg % self)
def check_length(self, attrs, max_chars):
if not isinstance(attrs, list):
attrs = [attrs,]
value = " ".join(getattr(self, at, '') or '' for at in attrs)
if len(value) > max_chars:
msg = (u"Il campo %s puo' contenere al piu' %d caratteri, " +
"'%s' li supera nell'istanza %r") % (attrs, max_chars, value, self)
raise poste.RecipientValidationError(msg)
def __repr__(self):
return "<Nominativo %s %s %s>" % (self.nome, self.cognome, self.ragione_sociale)
[docs]class Destinatario(Nominativo):
"""
Un destinatario
"""
def __init__(self, *args, **kwargs):
"""Stessi parametri di Destinatario ed in più id_destinatario
:param id_destinatario: id che identifica il destinatario lato client
"""
try:
self.id_destinatario = kwargs.pop('id_destinatario')
except KeyError:
self.id_destinatario = str(random.random())[2:]
self.nominativo = Nominativo(**kwargs)
[docs] def get_instance(self, suds_client):
"""
get an instance from web service, suitable for further processing
"""
creation_name = '{http://ComunicazioniElettroniche.XOL}Destinatario'
destinatario = suds_client.factory.create(creation_name)
destinatario.Nominativo = self.nominativo.get_instance(suds_client)
destinatario._IdDestinatario = self.id_destinatario
return destinatario
def __repr__(self):
return "<Destinatario %r %s>" % (self.nominativo, self.id_destinatario)
[docs]class Mittente(poste.RecipientTOL):
"""
Un mittente
"""
creation_name = '{http://ComunicazioniElettroniche.XOL}Mittente'
def __init__(self, *args, **kwargs):
"""Stessi parametri di Destinatario ed in più invio_dati_mittente
:param invia_stampa: boolean
"""
try:
self.invia_stampa = kwargs.pop('invia_stampa')
except KeyError:
self.invia_stampa = None
self.nominativo = Nominativo(**kwargs)
[docs] def get_instance(self, suds_client):
"""
get an instance from web service, suitable for further processing
"""
creation_name = '{http://ComunicazioniElettroniche.XOL}Mittente'
mittente = suds_client.factory.create(creation_name)
mittente.Nominativo = self.nominativo.get_instance(suds_client)
# mittente._invia_stampa = self.invia_stampa
return mittente
[docs]class Opzioni(object):
"Not implemeted by Poste Italiane as of this writing"
NotImplemented
[docs]class OpzioniDiStampa(object):
"Not implemeted by Poste Italiane as of this writing"
NotImplemented
[docs]class Documento(object):
"""
Documento da spedire
"""
def __init__(self, filename):
self.filename = filename
self.tipo_documento = os.path.splitext(filename)[1].lstrip('.')
self.immagine, self.md5 = poste.encode_file(filename)
[docs] def get_instance(self, suds_client):
"""
get an instance from web service, suitable for further processing
"""
creation_name = '{http://ComunicazioniElettroniche.XOL}Documento'
documento = suds_client.factory.create(creation_name)
documento.Firmatari = '' # Not implemente by Poste Italiane
documento.Immagine = self.immagine
documento.MD5 = self.md5
documento._TipoDocumento = self.tipo_documento
return documento
class DatiRicevuta(object):
def __init__(self, nominativo):
"""Dati per ricevuta di ritorno
:param nominativo: un oggetto :class:`Nominativo`
"""
if isinstance(nominativo, (list, tuple, set)):
raise NotImplementedError("Le ricevute di ritorno multiple non sono implementate " +
"Da questa libreria")
self.nominativo = nominativo
def get_instance(self, suds_client):
"""
get an instance from web service, suitable for further processing
"""
creation_name = 'DatiRicevuta'
dati_ricevuta = suds_client.factory.create(creation_name)
# FIXME non imp
if self.nominativo:
dati_ricevuta.Nominativo = [self.nominativo.get_instance(suds_client)]
return dati_ricevuta
return None
[docs]class ArrayOfDestinatario(object):
"""Array di destinatari. Es::
thunder = poste.Destinatario(\*\*THUNDER_ADDR)
test_addr = poste.Destinatario(\*\*THUNDER_TEST)
arr = poste.ArrayOfDestinatario((thunder, test_addr))
"""
def __init__(self, destinatari):
if not isinstance(destinatari, (list, tuple, set)):
destinatari = [destinatari,]
assert isinstance(destinatari[0], Destinatario), "destinatari deve essere un Destinatario non %r" % destinatari[0]
self.destinatari = destinatari
[docs] def get_instance(self, suds_client):
"""
get an instance from web service, suitable for further processing
"""
creation_name = 'ArrayOfDestinatario'
dest_array = suds_client.factory.create(creation_name)
for dest in self.destinatari:
dest_array.Destinatario.append(dest.get_instance(suds_client))
return dest_array
def __len__(self):
return len(self.destinatari)
[docs]class ROLSubmit(object):
"""
Raccomandata OnLine. Oggetto che riunisce tutte le informazioni
necessarie all'invio di una raccomandata. Può essere creato
esplicitamente o implicitamente da una operazione di invio
:meth:`poste.RolClient.invia`.
Quando un ROL viene inviato, viene valorizzata la variabile result con
un oggetto che conserva tutte le informazioni ricevute dal servizio
remoto, in particolare guid e id_richiesta necessari per fare il
tracking della raccomandata con l'help desk delle poste e per richiedere ulteriori
informazioni come la valorizzazione, la conferma o l'annullamento.
"""
#: La risposta ottenuta dal server quando questo viene inviato. Valorizzato solo
#: se il rol è stato inviato
result = None
#: GuidUser: identificativo univoco del servizio. Utile con Help Desk, es: ROL201211000000142
guid = None
#: id_richiesta: necessario per richiedere valorizzazione o annullamento/conferma
id_richiesta = None
def __init__(self, documento, destinatari, mittente=None,
data_lettera=None, ricevuta_ritorno=None):
"""
Oggetto contenente informazioni relative all'invio
:param documento: a :class:`Documento` or a filename
:param destinatari: un :class:`Destinatario` o una lista/tupla/set di
destinatari
:param ricevuta_ritorno: Lista di nominativi cui inviare ricevuta di ritorno
"""
if not isinstance(documento, Documento):
documento = Documento(documento)
self.documento = documento
self.destinatari = ArrayOfDestinatario(destinatari)
if mittente:
assert isinstance(mittente, Mittente)
self.mittente = mittente
self.data_lettera = data_lettera or datetime.date.today()
self.ricevuta_ritorno = poste.DatiRicevuta(ricevuta_ritorno)
[docs] def get_instance(self, suds_client):
"""
get an instance from web service, suitable for further processing
"""
creation_name = 'ROLSubmit'
rol_submit = suds_client.factory.create(creation_name)
rol_submit.Mittente = self.mittente.get_instance(suds_client)
rol_submit.Destinatari = self.destinatari.get_instance(suds_client)
rol_submit.NumeroDestinatari = len(self.destinatari)
rol_submit.Documento = self.documento.get_instance(suds_client)
if self.ricevuta_ritorno:
rol_submit.DatiRicevuta = self.ricevuta_ritorno.get_instance(suds_client)
return rol_submit
@property
[docs] def guid(self):
"""Ritona il guid (identificativo del server) di un invio di questo rol
Ritorna None se non è stato fatto alcun invio
"""
try:
return self.result.IDRichiesta
except KeyError:
return None
@property
[docs] def guid_utente(self):
"""Ritona il guid (identificativo del server) di un invio di questo rol
Ritorna None se non è stato fatto alcun invio
"""
try:
return self.result.GuidUtente
except KeyError:
return None
@property
[docs] def id_richiesta(self):
"""Ritona il guid (identificativo del server) di un invio di questo rol
Ritorna None se non è stato fatto alcun invio
"""
try:
return self.result.IDRichiesta
except KeyError:
return None
class LOLSubmit(ROLSubmit):
def get_instance(self, suds_client):
"""
get an instance from web service, suitable for further processing
"""
creation_name = 'LOLSubmit'
lol_submit = suds_client.factory.create(creation_name)
lol_submit.Mittente = self.mittente.get_instance(suds_client)
lol_submit.Destinatari = self.destinatari.get_instance(suds_client)
lol_submit.NumeroDestinatari = len(self.destinatari)
lol_submit.Documento = self.documento.get_instance(suds_client)
return lol_submit
[docs]class Richiesta(object):
"""
Oggetto richiesta, necessario per alcune operazioni come la valorizzazione
e la richieta di informazioni su un invio
"""
def __init__(self, guid=None, id_richiesta=None, guid_utente=None, xol_submit=None):
"""Oggetto Richiesta. Guid ed id_richiesta, necessari dal servizio
online, possono essere desunte dal xol_submit, se e solo se questo è
già stato inviato
:param xol_submit: :class:`poste.ROLSubmit`
:param mode: dcs o valorizza
:param guid: a GuidUser to retrieve information on a submitted rol
:param id_richiesta: an id_richiesta relative to a submitted rol
"""
self.guid = guid or id_richiesta
self.guid_utente = guid_utente
self.xol_submit = xol_submit
if xol_submit:
assert xol_submit.result, "questo xol_submit non e' ancora stato inviato"
self.guid = xol_submit.result.IDRichiesta
self.guid_utente = xol_submit.result.GuidUtente
else:
assert self.guid, "Il guid unico (alias id_request) e' richiesto"
[docs] def get_instance(self, suds_client):
"""
get an instance from web service, suitable for further processing
"""
creation_name = '{http://ComunicazioniElettroniche.XOL}Richiesta'
richiesta = suds_client.factory.create(creation_name)
richiesta.IDRichiesta = self.guid
if self.guid_utente:
richiesta.GuidUtente = self.guid_utente
return richiesta
@property
def id_request(self):
return self.guid