348 lines
12 KiB
Python
348 lines
12 KiB
Python
# The COPYRIGHT file at the top level of this repository contains the full
|
|
# copyright notices and license terms.
|
|
from io import BytesIO
|
|
from trytond.pool import PoolMeta, Pool
|
|
from trytond.pyson import Eval, Not
|
|
from trytond.model import ModelView, fields, Workflow
|
|
from trytond.config import config
|
|
from trytond.transaction import Transaction
|
|
from email.header import Header
|
|
from trytond.sendmail import sendmail_transactional
|
|
from trytond.report import Report
|
|
from cryptography.fernet import Fernet
|
|
from email.mime.multipart import MIMEMultipart
|
|
from email.mime.text import MIMEText
|
|
from trytond.wizard import Wizard, StateView, Button, StateTransition
|
|
import logging
|
|
import re
|
|
|
|
try:
|
|
import qrcode
|
|
except ImportError:
|
|
qrcode = None
|
|
try:
|
|
import html2text
|
|
except ImportError:
|
|
html2text = None
|
|
|
|
__all__ = ['ShipmentOut', 'SendLink', 'SendLinkStart']
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def format_emails(email):
|
|
if not email:
|
|
return []
|
|
values = re.split(r'[,;]', email)
|
|
return [v.strip() for v in values]
|
|
|
|
|
|
class ShipmentWebLinkMixin(object):
|
|
|
|
encrypted_url = fields.Function(fields.Char('Encrypted URL'),
|
|
'get_encrypted_url')
|
|
encrypted_id = fields.Function(fields.Char('Encrypted id'),
|
|
'get_encrypted_id')
|
|
attachments = fields.Function(fields.Many2Many('ir.attachment',
|
|
None, None, 'Attachments'), 'get_attachments')
|
|
encrypted_url_qr = fields.Function(
|
|
fields.Binary('Encrypted URL QR'),
|
|
'get_encrypted_url_qr')
|
|
|
|
@classmethod
|
|
def __setup__(cls):
|
|
super(ShipmentWebLinkMixin, cls).__setup__()
|
|
cls._error_messages.update({
|
|
'missing_fernet_key': 'Missing Fernet key configuration',
|
|
'wrong_state_send_link':
|
|
'The action "Send link" must be done from draft state.',
|
|
'missing_link_email':
|
|
'Must define an Email address to send link of Shipment "%s".'
|
|
})
|
|
cls.state.selection.extend([('mail_sent', 'Mail sent')])
|
|
cls._buttons['draft']['invisible'] = ~Eval('state').in_(
|
|
['waiting', 'cancelled', 'mail_sent'])
|
|
|
|
@classmethod
|
|
def get_custom_title(self, report_name, record, title):
|
|
return None
|
|
|
|
def get_attachments(self, name):
|
|
pool = Pool()
|
|
Attachment = pool.get('ir.attachment')
|
|
return Attachment.search([('resource', '=', str(self))])
|
|
|
|
@classmethod
|
|
def get_email(cls, report, record, languages):
|
|
"Return email.mime and title from the report execution"
|
|
pool = Pool()
|
|
ActionReport = pool.get('ir.action.report')
|
|
report_id = None
|
|
if isinstance(report, Report):
|
|
Report_ = report
|
|
else:
|
|
if isinstance(report, ActionReport):
|
|
report_name = report.report_name
|
|
report_id = report.id
|
|
else:
|
|
report_name = report
|
|
Report_ = pool.get(report_name, type='report')
|
|
converter = None
|
|
title = None
|
|
msg = MIMEMultipart('alternative')
|
|
msg.add_header('Content-Language', ', '.join(
|
|
l.code for l in languages))
|
|
for language in languages:
|
|
with Transaction().set_context(language=language.code):
|
|
ext, content, _, title = Report_.execute(
|
|
[record.id], {
|
|
'action_id': report_id,
|
|
'language': language,
|
|
})
|
|
if ext == 'html' and html2text:
|
|
if not converter:
|
|
converter = html2text.HTML2Text()
|
|
part = MIMEText(
|
|
converter.handle(content), 'plain', _charset='utf-8')
|
|
part.add_header('Content-Language', language.code)
|
|
msg.attach(part)
|
|
part = MIMEText(content, ext, _charset='utf-8')
|
|
part.add_header('Content-Language', language.code)
|
|
msg.attach(part)
|
|
custom_title = cls.get_custom_title(report_name, record, title)
|
|
return msg, custom_title or title
|
|
|
|
@classmethod
|
|
@ModelView.button
|
|
@Workflow.transition('mail_sent')
|
|
def send_link(cls, shipments, length=8, from_=None, email=None,
|
|
send_mail=True, raise_error=True):
|
|
pool = Pool()
|
|
User = pool.get('res.user')
|
|
Move = pool.get('stock.move')
|
|
|
|
if not shipments:
|
|
cls.raise_user_error('wrong_state_send_link')
|
|
moves = [m for shipment in shipments
|
|
if shipment.state == 'received' for m in shipment.incoming_moves]
|
|
if moves:
|
|
with Transaction().set_context(check_origin=False,
|
|
check_shipment=False):
|
|
Move.cancel(moves)
|
|
Move.draft(moves)
|
|
if not send_mail:
|
|
return
|
|
|
|
if from_ is None:
|
|
from_ = config.get('email', 'from')
|
|
for shipment in shipments:
|
|
shipment_email = email or shipment.get_to_email()
|
|
shipment.encrypted_url = shipment.get_encrypted_url()
|
|
user = User(Transaction().user)
|
|
if not shipment_email:
|
|
if raise_error:
|
|
cls.raise_user_error('missing_link_email',
|
|
shipment.rec_name)
|
|
else:
|
|
logger.info("Missing address for '%s' to send email",
|
|
shipment.warehouse.rec_name)
|
|
return
|
|
msg, title = shipment.get_email_shipment_flask(user)
|
|
msg['From'] = from_
|
|
msg['To'] = shipment_email
|
|
msg['Subject'] = Header(title, 'utf-8')
|
|
sendmail_transactional(from_, format_emails(shipment_email), msg)
|
|
|
|
def get_encrypted_id(self, name=None):
|
|
fernet = self.get_fernet()
|
|
encrypted_id = fernet.encrypt(str(self.id).encode())
|
|
return encrypted_id.decode()
|
|
|
|
def get_encrypted_url(self, name=None, report=False):
|
|
pool = Pool()
|
|
Config = pool.get('stock.configuration')
|
|
|
|
config = Config(1)
|
|
_type = 'shipment_in/'
|
|
if self.__name__ == 'stock.shipment.out':
|
|
_type = 'shipment_out/'
|
|
elif self.__name__ == 'stock.shipment.internal':
|
|
_type = 'shipment_internal/'
|
|
if report:
|
|
return config.shipment_link_url + _type + 'R' + self.encrypted_id
|
|
else:
|
|
return config.shipment_link_url + _type + self.encrypted_id
|
|
|
|
def get_languages(self, user):
|
|
pool = Pool()
|
|
Lang = pool.get('ir.lang')
|
|
if user.language:
|
|
languages = [user.language]
|
|
else:
|
|
languages = Lang.search([
|
|
('code', '=', Transaction().language),
|
|
])
|
|
return languages
|
|
|
|
@classmethod
|
|
def get_fernet(cls):
|
|
fernet_key = config.get('cryptography_shipment', 'fernet_key')
|
|
if not fernet_key:
|
|
cls.raise_user_error('missing_fernet_key')
|
|
else:
|
|
return Fernet(fernet_key)
|
|
|
|
def get_to_email(self):
|
|
return self.warehouse.address.party.email
|
|
|
|
@classmethod
|
|
def get_weblink_report_file(cls, shipments):
|
|
raise NotImplementedError()
|
|
|
|
def get_encrypted_url_qr(self, name=None):
|
|
if qrcode is None:
|
|
return None
|
|
with BytesIO() as output:
|
|
qr = qrcode.make(self.get_encrypted_url(report=True))
|
|
qr.save(output, 'PNG')
|
|
return output.getvalue()
|
|
|
|
|
|
class ShipmentOut(ShipmentWebLinkMixin, metaclass=PoolMeta):
|
|
__name__ = 'stock.shipment.out'
|
|
|
|
@classmethod
|
|
def __setup__(cls):
|
|
super(ShipmentOut, cls).__setup__()
|
|
cls._buttons['wait']['invisible'] = ~Eval('state').in_(
|
|
['assigned', 'draft', 'mail_sent'])
|
|
cls._transitions |= set((
|
|
('draft', 'mail_sent'),
|
|
('mail_sent', 'waiting'),
|
|
('mail_sent', 'draft'),
|
|
('waiting', 'mail_sent'),
|
|
))
|
|
|
|
def get_email_shipment_flask(self, user):
|
|
return self.get_email(
|
|
'stock.shipment.out.web_link', self, self.get_languages(user))
|
|
|
|
@classmethod
|
|
def get_weblink_report_file(cls, shipments):
|
|
ReportShipmentOut = Pool().get('stock.shipment.out.delivery_note',
|
|
type='report')
|
|
|
|
return ReportShipmentOut.execute(list(map(int, shipments)), {})
|
|
|
|
|
|
class ShipmentIn(ShipmentWebLinkMixin, metaclass=PoolMeta):
|
|
__name__ = 'stock.shipment.in'
|
|
|
|
@classmethod
|
|
def __setup__(cls):
|
|
super(ShipmentIn, cls).__setup__()
|
|
cls._buttons['receive']['invisible'] = ~Eval('state').in_(
|
|
['draft', 'mail_sent'])
|
|
cls._transitions |= set((
|
|
('draft', 'mail_sent'),
|
|
('mail_sent', 'received'),
|
|
('mail_sent', 'draft'),
|
|
('received', 'mail_sent'),
|
|
('received', 'draft'),
|
|
))
|
|
|
|
def get_email_shipment_flask(self, user):
|
|
return self.get_email(
|
|
'stock.shipment.in.web_link', self, self.get_languages(user))
|
|
|
|
@classmethod
|
|
def send_link(cls, shipments, length=8, from_=None, email=None,
|
|
send_mail=True, raise_error=True):
|
|
Move = Pool().get('stock.move')
|
|
|
|
moves = [m for shipment in shipments
|
|
if shipment.state == 'received' for m in shipment.inventory_moves]
|
|
if moves:
|
|
with Transaction().set_context(check_origin=False,
|
|
check_shipment=False):
|
|
Move.cancel(moves)
|
|
Move.delete(moves)
|
|
|
|
super(ShipmentIn, cls).send_link(shipments, length=length, from_=from_,
|
|
email=email, send_mail=send_mail, raise_error=raise_error)
|
|
|
|
@classmethod
|
|
def get_weblink_report_file(cls, shipments):
|
|
ReportShipmentIn = Pool().get('stock.shipment.in.delivery_receipt',
|
|
type='report')
|
|
|
|
return ReportShipmentIn.execute(list(map(int, shipments)), {})
|
|
|
|
|
|
class ShipmentInternal(ShipmentWebLinkMixin, metaclass=PoolMeta):
|
|
__name__ = 'stock.shipment.internal'
|
|
|
|
@classmethod
|
|
def __setup__(cls):
|
|
super().__setup__()
|
|
|
|
cls._buttons['wait']['invisible'] = Not(Eval('state').in_(
|
|
['assigned', 'draft', 'mail_sent']))
|
|
cls._transitions |= set((
|
|
('draft', 'mail_sent'),
|
|
('mail_sent', 'waiting'),
|
|
('mail_sent', 'draft'),
|
|
('waiting', 'mail_sent')))
|
|
|
|
def get_email_shipment_flask(self, user):
|
|
return self.get_email(
|
|
'stock.shipment.internal.web_link', self, self.get_languages(user))
|
|
|
|
def get_to_email(self):
|
|
return self.company.party.email
|
|
|
|
@classmethod
|
|
def get_weblink_report_file(cls, shipments):
|
|
ReportShipmentInternal = Pool().get('stock.shipment.internal.report',
|
|
type='report')
|
|
return ReportShipmentInternal.execute(list(map(int, shipments)), {})
|
|
|
|
|
|
class SendLinkStart(ModelView):
|
|
"""Shipment Send Link Start"""
|
|
__name__ = 'stock.shipment.send_link.start'
|
|
|
|
email = fields.Char('Email', required=True)
|
|
|
|
|
|
class SendLink(Wizard):
|
|
"""Shipment Send Link"""
|
|
__name__ = 'stock.shipment.send_link'
|
|
|
|
start = StateView('stock.shipment.send_link.start',
|
|
'stock_shipment_web_link.send_link_start_view_form', [
|
|
Button('Cancel', 'end', 'tryton-cancel'),
|
|
Button('Send link', 'sent_link', 'tryton-ok', default=True),
|
|
])
|
|
sent_link = StateTransition()
|
|
|
|
def default_start(self, fields):
|
|
pool = Pool()
|
|
Shipment = pool.get(Transaction().context['active_model'])
|
|
|
|
context = Transaction().context
|
|
shipment = Shipment(context['active_id'])
|
|
if shipment.state != 'draft':
|
|
Shipment.raise_user_error('wrong_state_send_link')
|
|
return {
|
|
'email': shipment.get_to_email()
|
|
}
|
|
|
|
def transition_sent_link(self):
|
|
pool = Pool()
|
|
Shipment = pool.get(Transaction().context['active_model'])
|
|
|
|
context = Transaction().context
|
|
shipments = Shipment.browse(context['active_ids'])
|
|
Shipment.send_link(shipments, email=self.start.email)
|
|
return 'end'
|