This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
devicehub-teal/ereuse_devicehub/resources/documents/documents.py

336 lines
12 KiB
Python

import csv
import datetime
import enum
import uuid
from collections import OrderedDict
from io import StringIO
from typing import Callable, Iterable, Tuple
import boltons
import flask
import flask_weasyprint
import teal.marshmallow
from boltons import urlutils
from flask import make_response, g, request
from flask.json import jsonify
from teal.cache import cache
from teal.resource import Resource, View
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.action import models as evs
from ereuse_devicehub.resources.device import models as devs
from ereuse_devicehub.resources.deliverynote.models import Deliverynote
from ereuse_devicehub.resources.device.views import DeviceView
from ereuse_devicehub.resources.documents.device_row import DeviceRow, StockRow, ActionRow
from ereuse_devicehub.resources.lot import LotView
from ereuse_devicehub.resources.lot.models import Lot
from ereuse_devicehub.resources.hash_reports import insert_hash, ReportHash, verify_hash
class Format(enum.Enum):
HTML = 'HTML'
PDF = 'PDF'
class DocumentView(DeviceView):
class FindArgs(DeviceView.FindArgs):
format = teal.marshmallow.EnumField(Format, missing=None)
def get(self, id):
"""Get a collection of resources or a specific one.
---
parameters:
- name: id
in: path
description: The identifier of the resource.
type: string
required: false
responses:
200:
description: Return the collection or the specific one.
"""
args = self.QUERY_PARSER.parse(self.find_args,
flask.request,
locations=('querystring',))
if id:
# todo we assume we can pass both device id and action id
# for certificates... how is it going to end up being?
try:
id = uuid.UUID(id)
except ValueError:
try:
id = int(id)
except ValueError:
raise teal.marshmallow.ValidationError('Document must be an ID or UUID.')
else:
query = devs.Device.query.filter_by(id=id)
else:
query = evs.Action.query.filter_by(id=id)
else:
flask.current_app.auth.requires_auth(lambda: None)() # todo not nice
query = self.query(args)
type = urlutils.URL(flask.request.url).path_parts[-2]
if type == 'erasures':
template = self.erasure(query)
if args.get('format') == Format.PDF:
res = flask_weasyprint.render_pdf(
flask_weasyprint.HTML(string=template), download_filename='{}.pdf'.format(type)
)
insert_hash(res.data)
else:
res = flask.make_response(template)
return res
@staticmethod
def erasure(query: db.Query):
def erasures():
for model in query:
if isinstance(model, devs.Computer):
for erasure in model.privacy:
yield erasure
elif isinstance(model, devs.DataStorage):
erasure = model.privacy
if erasure:
yield erasure
else:
assert isinstance(model, evs.EraseBasic)
yield model
url_pdf = boltons.urlutils.URL(flask.request.url)
url_pdf.query_params['format'] = 'PDF'
url_web = boltons.urlutils.URL(flask.request.url)
url_web.query_params['format'] = 'HTML'
params = {
'title': 'Erasure Certificate',
'erasures': tuple(erasures()),
'url_pdf': url_pdf.to_text(),
'url_web': url_web.to_text()
}
return flask.render_template('documents/erasure.html', **params)
class DevicesDocumentView(DeviceView):
@cache(datetime.timedelta(minutes=1))
def find(self, args: dict):
query = (x for x in self.query(args) if x.owner_id == g.user.id)
return self.generate_post_csv(query)
def generate_post_csv(self, query):
"""Get device query and put information in csv format."""
data = StringIO()
cw = csv.writer(data, delimiter=';', lineterminator="\n", quotechar='"')
first = True
document_ids = self.get_documents_id()
for device in query:
d = DeviceRow(device, document_ids)
if first:
cw.writerow(d.keys())
first = False
cw.writerow(d.values())
bfile = data.getvalue().encode('utf-8')
output = make_response(bfile)
insert_hash(bfile)
output.headers['Content-Disposition'] = 'attachment; filename=export.csv'
output.headers['Content-type'] = 'text/csv'
return output
def get_documents_id(self):
# documentIds = {dev_id: document_id, ...}
deliverys = Deliverynote.query.all()
documentIds = {x.id: d.document_id for d in deliverys for x in d.lot.devices}
return documentIds
class ActionsDocumentView(DeviceView):
@cache(datetime.timedelta(minutes=1))
def find(self, args: dict):
query = (x for x in self.query(args) if x.owner_id == g.user.id)
return self.generate_post_csv(query)
def generate_post_csv(self, query):
"""Get device query and put information in csv format."""
data = StringIO()
cw = csv.writer(data, delimiter=';', lineterminator="\n", quotechar='"')
first = True
for device in query:
for allocate in device.get_metrics():
d = ActionRow(allocate)
if first:
cw.writerow(d.keys())
first = False
cw.writerow(d.values())
bfile = data.getvalue().encode('utf-8')
output = make_response(bfile)
insert_hash(bfile)
output.headers['Content-Disposition'] = 'attachment; filename=actions_export.csv'
output.headers['Content-type'] = 'text/csv'
return output
class LotsDocumentView(LotView):
def find(self, args: dict):
query = (x for x in self.query(args) if x.owner_id == g.user.id)
return self.generate_lots_csv(query)
def generate_lots_csv(self, query):
"""Get lot query and put information in csv format."""
data = StringIO()
cw = csv.writer(data)
first = True
for lot in query:
l = LotRow(lot)
if first:
cw.writerow(l.keys())
first = False
cw.writerow(l.values())
bfile = data.getvalue().encode('utf-8')
output = make_response(bfile)
insert_hash(bfile)
output.headers['Content-Disposition'] = 'attachment; filename=lots-info.csv'
output.headers['Content-type'] = 'text/csv'
return output
class LotRow(OrderedDict):
def __init__(self, lot: Lot) -> None:
super().__init__()
self.lot = lot
# General information about lot
self['Id'] = lot.id.hex
self['Name'] = lot.name
self['Registered in'] = format(lot.created, '%c')
try:
self['Description'] = lot.description
except:
self['Description'] = ''
class StockDocumentView(DeviceView):
# @cache(datetime.timedelta(minutes=1))
def find(self, args: dict):
query = (x for x in self.query(args) if x.owner_id == g.user.id)
return self.generate_post_csv(query)
def generate_post_csv(self, query):
"""Get device query and put information in csv format."""
data = StringIO()
cw = csv.writer(data)
first = True
for device in query:
d = StockRow(device)
if first:
cw.writerow(d.keys())
first = False
cw.writerow(d.values())
bfile = data.getvalue().encode('utf-8')
output = make_response(bfile)
insert_hash(bfile)
output.headers['Content-Disposition'] = 'attachment; filename=devices-stock.csv'
output.headers['Content-type'] = 'text/csv'
return output
class CheckView(View):
model = ReportHash
def get(self):
qry = dict(request.values)
hash3 = qry.get('hash')
result = False
if hash3 and ReportHash.query.filter_by(hash3=hash3).count():
result = True
return jsonify(result)
class StampsView(View):
"""
This view render one public ans static page for see the links for to do the check
of one csv file
"""
def get_url_path(self):
url = urlutils.URL(request.url)
url.normalize()
url.path_parts = url.path_parts[:-2] + ['check', '']
return url.to_text()
def get(self):
result = ('', '')
return flask.render_template('documents/stamp.html', rq_url=self.get_url_path(),
result=result)
def post(self):
result = ('', '')
if 'docUpload' in request.files:
file_check = request.files['docUpload']
result = ('Bad', 'Sorry, this file has not been produced by this website')
if file_check.mimetype in ['text/csv', 'application/pdf']:
if verify_hash(file_check):
result = ('Ok', 'Yes, this file has been produced by this website')
return flask.render_template('documents/stamp.html', rq_url=self.get_url_path(),
result=result)
class DocumentDef(Resource):
__type__ = 'Document'
SCHEMA = None
VIEW = None # We do not want to create default / documents endpoint
AUTH = False
def __init__(self, app,
import_name=__name__,
static_folder='static',
static_url_path=None,
template_folder='templates',
url_prefix=None,
subdomain=None,
url_defaults=None,
root_path=None,
cli_commands: Iterable[Tuple[Callable, str or None]] = tuple()):
super().__init__(app, import_name, static_folder, static_url_path, template_folder,
url_prefix, subdomain, url_defaults, root_path, cli_commands)
d = {'id': None}
get = {'GET'}
view = DocumentView.as_view('main', definition=self, auth=app.auth)
# TODO @cayop This two lines never pass
if self.AUTH:
view = app.auth.requires_auth(view)
self.add_url_rule('/erasures/', defaults=d, view_func=view, methods=get)
self.add_url_rule('/erasures/<{}:{}>'.format(self.ID_CONVERTER.value, self.ID_NAME),
view_func=view, methods=get)
devices_view = DevicesDocumentView.as_view('devicesDocumentView',
definition=self,
auth=app.auth)
devices_view = app.auth.requires_auth(devices_view)
stock_view = StockDocumentView.as_view('stockDocumentView', definition=self)
stock_view = app.auth.requires_auth(stock_view)
self.add_url_rule('/devices/', defaults=d, view_func=devices_view, methods=get)
lots_view = LotsDocumentView.as_view('lotsDocumentView', definition=self)
lots_view = app.auth.requires_auth(lots_view)
self.add_url_rule('/lots/', defaults=d, view_func=lots_view, methods=get)
stock_view = StockDocumentView.as_view('stockDocumentView', definition=self, auth=app.auth)
stock_view = app.auth.requires_auth(stock_view)
self.add_url_rule('/stock/', defaults=d, view_func=stock_view, methods=get)
check_view = CheckView.as_view('CheckView', definition=self, auth=app.auth)
self.add_url_rule('/check/', defaults={}, view_func=check_view, methods=get)
stamps_view = StampsView.as_view('StampsView', definition=self, auth=app.auth)
self.add_url_rule('/stamps/', defaults={}, view_func=stamps_view, methods={'GET', 'POST'})
actions_view = ActionsDocumentView.as_view('ActionsDocumentView',
definition=self,
auth=app.auth)
actions_view = app.auth.requires_auth(actions_view)
self.add_url_rule('/actions/', defaults=d, view_func=actions_view, methods=get)