#!/usr/bin/python
# -*- coding: utf-8 -*-
# Hive Appier Framework
# Copyright (c) 2008-2024 Hive Solutions Lda.
#
# This file is part of Hive Appier Framework.
#
# Hive Appier Framework is free software: you can redistribute it and/or modify
# it under the terms of the Apache License as published by the Apache
# Foundation, either version 2.0 of the License, or (at your option) any
# later version.
#
# Hive Appier Framework 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
# Apache License for more details.
#
# You should have received a copy of the Apache License along with
# Hive Appier Framework. If not, see .
__author__ = "João Magalhães "
""" The author(s) of the module """
__copyright__ = "Copyright (c) 2008-2024 Hive Solutions Lda."
""" The copyright for the module """
__license__ = "Apache License, Version 2.0"
""" The license for the module """
import csv
import uuid
from . import model
from . import legacy
from . import typesf
from . import exceptions
def serialize(obj):
if isinstance(obj, model.Model):
return obj.model
if isinstance(obj, typesf.AbstractType):
return obj.json_v()
if type(obj) == type(None):
return ""
return legacy.UNICODE(obj)
def serialize_csv(
items, encoding="utf-8", errors="strict", delimiter=";", strict=False
):
# verifies if the strict mode is active and there're no items defined
# if that's the case an operational error is raised, otherwise an in
# case the items are not provided the default (empty string) is returned
if strict and not items:
raise exceptions.OperationalError(
message="Empty items object provided, no keys available"
)
if not items:
return str()
# builds the encoder taking into account the provided encoding string
# value, this encoder will be used to encode each of the partial values
# that is going to be set in the target CSV buffer
encoder = build_encoder(encoding, errors=errors)
# retrieves the first element and uses it to determine if the current
# sequence to be serialized is map or sequence based
first = items[0]
is_map = type(first) == dict
# retrieves the various keys from the first element of the provided sequence
# of items then runs the eager operation (list loading) and sorts the provided
# keys according to the default sorting order defined for the sequence, note
# that in case the sequence is not map based the first element is ignored
keys = first.keys() if is_map else first
keys = legacy.eager(keys)
if is_map:
keys.sort()
else:
items = items[1:]
# constructs the first row (names/keys row) using the gathered sequence of keys
# and encoding them using the currently build encoder
keys_row = [encoder(key) if type(key) == legacy.UNICODE else key for key in keys]
# creates the new string buffer and uses it as the basis for the construction of
# the CSV writer object, writing then the already build first row
buffer = legacy.StringIO()
writer = csv.writer(buffer, delimiter=delimiter)
writer.writerow(keys_row)
# iterates over the complete set of items to serialize each of it's attribute values
# using the order defined in the keys sequence that has been defined, note that
# this behavior is not applied to sequence based serialization (more linear)
for item in items:
# creates the list that is going to support the complete set of encoded
# values (according to the defined encoding) and then start the base row
# sequence with an empty value of the passed value (for map based values)
row_e = []
row = [] if is_map else item
# iterates over the complete set of values in the map and serializes each
# of them, adding then them to the base row sequence
for key in keys if is_map else ():
value = item[key]
value = serialize(value)
row.append(value)
# iterates over the complete set of row values to "sanitize" them by encoding
# them in case they are defined as unicode based values
for value in row:
is_unicode = isinstance(value, legacy.UNICODE)
if is_unicode:
value = encoder(value)
row_e.append(value)
# writes the encoded based row sequence to the writer in order to output the
# values to the output based stream
writer.writerow(row_e)
# retrieves the buffer string value as the resulting value and returns it to the
# caller method as the final result for the CSV serialization
result = buffer.getvalue()
return result
def serialize_ics(items, encoding="utf-8", errors="strict"):
encoder = build_encoder(encoding, errors=errors)
buffer = legacy.StringIO()
buffer.write("BEGIN:VCALENDAR\r\n")
buffer.write("METHOD:PUBLISH\r\n")
buffer.write("X-WR-TIMEZONE:America/Los_Angeles\r\n")
buffer.write("CALSCALE:GREGORIAN\r\n")
buffer.write("VERSION:2.0\r\n")
buffer.write("PRODID:-//PUC Calendar// v2.0//EN)\r\n")
for item in items:
start = item["start"]
end = item["end"]
description = item["description"]
location = item["location"]
timezone = item.get("timezone", "Etc/GMT")
_uuid = item.get("uuid", None)
_uuid = _uuid or str(uuid.uuid4())
start = encoder(start)
end = encoder(end)
description = encoder(description)
location = encoder(location)
timezone = encoder(timezone)
_uuid = encoder(_uuid)
buffer.write("BEGIN:VEVENT\r\n")
buffer.write("UID:%s\r\n" % _uuid)
buffer.write("TZID:%s\r\n" % timezone)
buffer.write("DTSTART:%s\r\n" % start)
buffer.write("DTEND:%s\r\n" % end)
buffer.write("DTSTAMP:%s\r\n" % start)
buffer.write("SUMMARY:%s\r\n" % description)
buffer.write("LOCATION:%s\r\n" % location)
buffer.write("END:VEVENT\r\n")
buffer.write("END:VCALENDAR\r\n")
result = buffer.getvalue()
return result
def build_encoder(encoding, errors="strict"):
if legacy.PYTHON_3:
return lambda v: v
else:
return lambda v: v if v == None else v.encode(encoding, errors=errors)