OCA reporting engine fork for dev and update.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

191 lines
7.3 KiB

# -*- coding: utf-8 -*-
# © 2015 Antiun Ingenieria S.L. - Antonio Espinosa
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import base64
from contextlib import closing
import os
import subprocess
import tempfile
import time
from openerp import models, api, _
from openerp.exceptions import Warning, AccessError
from openerp.tools.safe_eval import safe_eval
import logging
_logger = logging.getLogger(__name__)
def _normalize_filepath(path):
path = path or ''
path = path.strip()
if not os.path.isabs(path):
me = os.path.dirname(__file__)
path = '{}/../static/certificate/'.format(me) + path
path = os.path.normpath(path)
return path if os.path.exists(path) else False
class Report(models.Model):
_inherit = 'report'
def _certificate_get(self, cr, uid, ids, report, context=None):
if report.report_type != 'qweb-pdf':
_logger.info(
"Can only sign qweb-pdf reports, this one is '%s' type",
report.report_type)
return False
m_cert = self.pool['report.certificate']
company_id = self.pool['res.users']._get_company(cr, uid)
certificate_ids = m_cert.search(cr, uid, [
('company_id', '=', company_id),
('model_id', '=', report.model)], context=context)
if not certificate_ids:
_logger.info(
"No PDF certificate found for report '%s'",
report.report_name)
return False
for cert in m_cert.browse(cr, uid, certificate_ids, context=context):
# Check allow only one document
if cert.allow_only_one and len(ids) > 1:
_logger.info(
"Certificate '%s' allows only one document, "
"but printing %d documents",
cert.name, len(ids))
continue
# Check domain
if cert.domain:
m_model = self.pool[cert.model_id.model]
domain = [('id', 'in', tuple(ids))]
domain = domain + safe_eval(cert.domain)
doc_ids = m_model.search(cr, uid, domain, context=context)
if not doc_ids:
_logger.info(
"Certificate '%s' domain not satisfied", cert.name)
continue
# Certificate match!
return cert
return False
def _attach_filename_get(self, cr, uid, ids, certificate, context=None):
if len(ids) != 1:
return False
obj = self.pool[certificate.model_id.model].browse(cr, uid, ids[0])
filename = safe_eval(certificate.attachment, {
'object': obj,
'time': time
})
return filename
def _attach_signed_read(self, cr, uid, ids, certificate, context=None):
if len(ids) != 1:
return False
filename = self._attach_filename_get(
cr, uid, ids, certificate, context=context)
if not filename:
return False
signed = False
m_attachment = self.pool['ir.attachment']
attach_ids = m_attachment.search(cr, uid, [
('datas_fname', '=', filename),
('res_model', '=', certificate.model_id.model),
('res_id', '=', ids[0])
])
if attach_ids:
signed = m_attachment.browse(cr, uid, attach_ids[0]).datas
signed = base64.decodestring(signed)
return signed
def _attach_signed_write(self, cr, uid, ids, certificate, signed,
context=None):
if len(ids) != 1:
return False
filename = self._attach_filename_get(
cr, uid, ids, certificate, context=context)
if not filename:
return False
m_attachment = self.pool['ir.attachment']
try:
attach_id = m_attachment.create(cr, uid, {
'name': filename,
'datas': base64.encodestring(signed),
'datas_fname': filename,
'res_model': certificate.model_id.model,
'res_id': ids[0],
})
except AccessError:
raise Warning(
_('Saving signed report (PDF):') + '\n',
_('You do not have enought access rights to save attachments'))
else:
_logger.info(
"The signed PDF document '%s' is now saved in the database",
filename)
return attach_id
def _signer_bin(self, opts):
me = os.path.dirname(__file__)
return 'java -jar {}/../static/jar/jPdfSign.jar '.format(me) + opts
def pdf_sign(self, pdf, certificate):
pdfsigned = pdf + '.signed.pdf'
p12 = _normalize_filepath(certificate.path)
passwd = _normalize_filepath(certificate.password_file)
if not (p12 and passwd):
raise Warning(
_('Signing report (PDF)'),
_('Certificate or password file not found'))
signer_opts = '"%s" "%s" "%s" "%s"' % (p12, pdf, pdfsigned, passwd)
signer = self._signer_bin(signer_opts)
process = subprocess.Popen(
signer, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
out, err = process.communicate()
if process.returncode:
raise Warning(
_('Signing report (PDF)'),
_('jPdfSign failed (error code: %s). '
'Message: %s') % (str(process.returncode), err))
return pdfsigned
@api.v7
def get_pdf(self, cr, uid, ids, report_name, html=None, data=None,
context=None):
signed_content = False
report = self._get_report_from_name(cr, uid, report_name)
certificate = self._certificate_get(
cr, uid, ids, report, context=context)
if certificate and certificate.attachment:
signed_content = self._attach_signed_read(
cr, uid, ids, certificate, context=context)
if signed_content:
_logger.info("The signed PDF document '%s/%s' was loaded from "
"the database", report_name, ids)
return signed_content
content = super(Report, self).get_pdf(
cr, uid, ids, report_name, html=html, data=data,
context=context)
if certificate:
# Creating temporary origin PDF
pdf_fd, pdf = tempfile.mkstemp(
suffix='.pdf', prefix='report.tmp.')
with closing(os.fdopen(pdf_fd, 'w')) as pf:
pf.write(content)
_logger.info(
"Signing PDF document '%s/%s' with certificate '%s'",
report_name, ids, certificate.name)
signed = self.pdf_sign(pdf, certificate)
# Read signed PDF
if os.path.exists(signed):
with open(signed, 'rb') as pf:
content = pf.read()
# Manual cleanup of the temporary files
for fname in (pdf, signed):
try:
os.unlink(fname)
except (OSError, IOError):
_logger.error('Error when trying to remove file %s', fname)
if certificate.attachment:
self._attach_signed_write(
cr, uid, ids, certificate, content, context=context)
return content