ebook_homebrew.rest.Upload.__init__() - Code Metrics - Inspection of "Merge pull request #70 from tubone24/heroku" - tubone24/ebook_homebrew - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( a56bd6...df426b )

by Yu

created 2019-04-30 08:04 UTC

ebook_homebrew.rest.Upload.init() A

↳ Parent: ebook_homebrew.rest

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	2
dl	0
loc	2
ccs	2
cts	2
cp	1
crap	1
rs	10
c	0
b	0
f	0

# -*- coding: utf-8 -*-
"""Provides Rest API interfaces
"""

import os
import base64
import glob
import json
import datetime
import tempfile
from collections import namedtuple
import responder
from marshmallow import Schema, fields

from .convert import Image2PDF
from .utils.logging import get_logger
from .__init__ import __version__

api = responder.API(
    title="Ebook-homebrew",
    debug=True,
    version=__version__,
    static_dir=os.path.join(os.path.dirname(os.path.abspath(__file__)), "static"),
    static_route="/static",
    openapi="3.0.2",
    docs_route="/docs",
    openapi_route="/schema.yml",
    description="Make PDF file taken in "
    "some image files such as "
    "jpeg, png and gif.",
    contact={
        "name": "tubone24",
        "url": "https://tubone-project24.xyz",
        "email": "[email protected]",
    },
    license={"name": "MIT", "url": "https://opensource.org/licenses/MIT"},
)

_logger = get_logger("RestAPI")


@api.schema("HealthCheck")
class HealthCheckSchema(Schema):
    status = fields.Str()


@api.schema("UploadImagesReq")
class UploadImagesReqSchema(Schema):
    contentType = fields.Str(required=True)
    images = fields.List(fields.Str(required=True))


@api.schema("UploadIdResp")
class UploadIdRespSchema(Schema):
    upload_id = fields.Str()


class Upload:
    """Upload_id Model"""

    def __init__(self, upload_id):
        self.upload_id = upload_id


@api.schema("ConvertReq")
class ConvertReqSchema(Schema):
    uploadId = fields.Str(required=True)
    contentType = fields.Str(required=True)


@api.schema("DownloadReq")
class DownloadReqSchema(Schema):
    uploadId = fields.Str(required=True)


api.add_route("/", static=True)


@api.route("/status")
def status(_, resp):
    """Health Check Response.
    ---
    get:
        description: Get Status
        responses:
            "200":
                description: OK
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/HealthCheck"
    """
    _logger.debug("health Check")
    Status = namedtuple("Status", ["status"])
    resp.media = HealthCheckSchema().dump(Status("ok")).data


@api.route("/data/upload")
async def upload_image_file(req, resp):
    """Upload Image files.
    ---
    post:
        summary: Base64 encoded Images

        requestBody:
            description: base64 encoded Images in images Array
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/UploadImagesReq"
        responses:
            "200":
                description: OK
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/UploadIdResp"
    """
    request = await req.media()
    data = UploadImagesReqSchema().load(request).data
    _logger.debug(data)
    content_type = data["contentType"]
    extension = convert_content_type_to_extension(content_type)
    images_b64 = data["images"]
    tmp_dir = tempfile.mkdtemp()
    write_image(images_b64, extension, tmp_dir)
    resp.media = UploadIdRespSchema().dump(Upload(str(tmp_dir))).data


@api.background.task
def write_image(images_b64, extension, tmp_dir):
    """Images write at tmp_dir

    This API is background task.

    Args:
     images_b64: Base64 encoded images list
     extension: Image extension
     tmp_dir: Temp directory writing images
    Returns:
        bool: If success return true.

    """
    for i, content in enumerate(images_b64):
        image = base64.b64decode(content)
        file_name = os.path.join(tmp_dir, str(i) + "." + extension)
        _logger.debug("file_name: {}".format(file_name))
        with open(file_name, "wb") as image_file:
            image_file.write(image)
    return True


@api.route("/convert/pdf")
async def convert_image_to_pdf(req, resp):
    """Convert Image files to PDF.
    ---
    post:
        summary: Upload Id witch get upload images and ContentType

        requestBody:
            description: Upload Id and ContentType
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/ConvertReq"
        responses:
            "200":
                description: OK
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/UploadIdResp"
    """
    request = await req.media()
    data = ConvertReqSchema().load(request).data
    _logger.debug(data)
    upload_id = data["uploadId"]
    content_type = data["contentType"]
    result_meta = os.path.join(upload_id, "result_meta.txt")
    if os.path.exists(result_meta):
        os.remove(result_meta)
    extension = convert_content_type_to_extension(content_type)
    file_list = sorted(
        glob.glob(os.path.join(upload_id, "*." + extension)), reverse=True
    )
    file_base, _ = os.path.splitext(os.path.basename(file_list[0]))
    digits = len(file_base)
    _logger.debug(file_list)
    convert_pdf(digits, extension, upload_id)
    resp.media = UploadIdRespSchema().dump(Upload(upload_id)).data


@api.background.task
def convert_pdf(digits, extension, upload_id):
    """Convert images to PDF

    This API is background task.

    Args:
     digits: file serial number digits
     extension: Image extension
     upload_id: Request ID
    Returns:
        bool: If success return true.

    """
    converter = Image2PDF(digits=digits, extension=extension, directory_path=upload_id)
    converter.make_pdf("result.pdf")
    with open(os.path.join(upload_id, "result_meta.txt"), "w") as result_txt:
        now = datetime.datetime.now()
        result = {
            "upload_id": upload_id,
            "digits": digits,
            "extension": extension,
            "datetime": now.strftime("%Y/%m/%d %H:%M:%S"),
        }
        result_txt.write(json.dumps(result))
    return True


@api.route("/convert/pdf/download")
async def download_result_pdf(req, resp):
    """Upload Image files.
    ---
    post:
        summary: Upload Id witch get upload images

        requestBody:
            description: Upload Id
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/UploadImagesReq"
        responses:
            "200":
                description: OK
                content:
                    application/pdf:
                        schema:
                            type: string
                            format: binary
            "404":
                description: FileNotFound
    """
    request = await req.media()
    data = DownloadReqSchema().load(request).data
    upload_id = data["uploadId"]
    result_meta = os.path.join(upload_id, "result_meta.txt")
    if os.path.exists(result_meta):
        with open(os.path.join(upload_id, "result.pdf"), "rb") as result_pdf:
            resp.headers["Content-Type"] = "application/pdf"
            resp.content = result_pdf.read()
    else:
        resp.status_code = api.status_codes.HTTP_404


def convert_content_type_to_extension(content_type):
    """Convert image extension to Content-Type

    Args:
     content_type: Content-Type
    Returns:
        str: extension

    """
    if content_type == "image/jpeg":
        extension = "jpg"
    elif content_type == "image/png":
        extension = "png"
    elif content_type == "image/gif":
        extension = "gif"
    else:
        extension = False

    return extension


1		# -- coding: utf-8 --
2	1	"""Provides Rest API interfaces
3		"""
4
5	1	import os
6	1	import base64
7	1	import glob
8	1	import json
9	1	import datetime
10	1	import tempfile
11	1	from collections import namedtuple
12	1	import responder
13	1	from marshmallow import Schema, fields
14
15	1	from .convert import Image2PDF
16	1	from .utils.logging import get_logger
17	1	from .__init__ import __version__
18
19	1	api = responder.API(
20		title="Ebook-homebrew",
21		debug=True,
22		version=__version__,
23		static_dir=os.path.join(os.path.dirname(os.path.abspath(__file__)), "static"),
24		static_route="/static",
25		openapi="3.0.2",
26		docs_route="/docs",
27		openapi_route="/schema.yml",
28		description="Make PDF file taken in "
29		"some image files such as "
30		"jpeg, png and gif.",
31		contact={
32		"name": "tubone24",
33		"url": "https://tubone-project24.xyz",
34		"email": "[email protected]",
35		},
36		license={"name": "MIT", "url": "https://opensource.org/licenses/MIT"},
37		)
38
39	1	_logger = get_logger("RestAPI")
40
41
42	1	@api.schema("HealthCheck")
43	1	class HealthCheckSchema(Schema):
44	1	status = fields.Str()
45
46
47	1	@api.schema("UploadImagesReq")
48	1	class UploadImagesReqSchema(Schema):
49	1	contentType = fields.Str(required=True)
50	1	images = fields.List(fields.Str(required=True))
51
52
53	1	@api.schema("UploadIdResp")
54	1	class UploadIdRespSchema(Schema):
55	1	upload_id = fields.Str()
56
57
58	1	class Upload:
59		"""Upload_id Model"""
60
61	1	def __init__(self, upload_id):
62	1	self.upload_id = upload_id
63
64
65	1	@api.schema("ConvertReq")
66	1	class ConvertReqSchema(Schema):
67	1	uploadId = fields.Str(required=True)
68	1	contentType = fields.Str(required=True)
69
70
71	1	@api.schema("DownloadReq")
72	1	class DownloadReqSchema(Schema):
73	1	uploadId = fields.Str(required=True)
74
75
76	1	api.add_route("/", static=True)
77
78
79	1	@api.route("/status")
80		def status(_, resp):
81		"""Health Check Response.
82		---
83		get:
84		description: Get Status
85		responses:
86		"200":
87		description: OK
88		content:
89		application/json:
90		schema:
91		$ref: "#/components/schemas/HealthCheck"
92		"""
93	1	_logger.debug("health Check")
94	1	Status = namedtuple("Status", ["status"])
95	1	resp.media = HealthCheckSchema().dump(Status("ok")).data
96
97
98	1	@api.route("/data/upload")
99		async def upload_image_file(req, resp):
100		"""Upload Image files.
101		---
102		post:
103		summary: Base64 encoded Images
104
105		requestBody:
106		description: base64 encoded Images in images Array
107		content:
108		application/json:
109		schema:
110		$ref: "#/components/schemas/UploadImagesReq"
111		responses:
112		"200":
113		description: OK
114		content:
115		application/json:
116		schema:
117		$ref: "#/components/schemas/UploadIdResp"
118		"""
119	1	request = await req.media()
120	1	data = UploadImagesReqSchema().load(request).data
121	1	_logger.debug(data)
122	1	content_type = data["contentType"]
123	1	extension = convert_content_type_to_extension(content_type)
124	1	images_b64 = data["images"]
125	1	tmp_dir = tempfile.mkdtemp()
126	1	write_image(images_b64, extension, tmp_dir)
127	1	resp.media = UploadIdRespSchema().dump(Upload(str(tmp_dir))).data
128
129
130	1	@api.background.task
131		def write_image(images_b64, extension, tmp_dir):
132		"""Images write at tmp_dir
133
134		This API is background task.
135
136		Args:
137		images_b64: Base64 encoded images list
138		extension: Image extension
139		tmp_dir: Temp directory writing images
140		Returns:
141		bool: If success return true.
142
143		"""
144	1	for i, content in enumerate(images_b64):
145	1	image = base64.b64decode(content)
146	1	file_name = os.path.join(tmp_dir, str(i) + "." + extension)
147	1	_logger.debug("file_name: {}".format(file_name))
148	1	with open(file_name, "wb") as image_file:
149	1	image_file.write(image)
150	1	return True
151
152
153	1	@api.route("/convert/pdf")
154		async def convert_image_to_pdf(req, resp):
155		"""Convert Image files to PDF.
156		---
157		post:
158		summary: Upload Id witch get upload images and ContentType
159
160		requestBody:
161		description: Upload Id and ContentType
162		content:
163		application/json:
164		schema:
165		$ref: "#/components/schemas/ConvertReq"
166		responses:
167		"200":
168		description: OK
169		content:
170		application/json:
171		schema:
172		$ref: "#/components/schemas/UploadIdResp"
173		"""
174	1	request = await req.media()
175	1	data = ConvertReqSchema().load(request).data
176	1	_logger.debug(data)
177	1	upload_id = data["uploadId"]
178	1	content_type = data["contentType"]
179	1	result_meta = os.path.join(upload_id, "result_meta.txt")
180	1	if os.path.exists(result_meta):
181	1	os.remove(result_meta)
182	1	extension = convert_content_type_to_extension(content_type)
183	1	file_list = sorted(
184		glob.glob(os.path.join(upload_id, "*." + extension)), reverse=True
185		)
186	1	file_base, _ = os.path.splitext(os.path.basename(file_list[0]))
187	1	digits = len(file_base)
188	1	_logger.debug(file_list)
189	1	convert_pdf(digits, extension, upload_id)
190	1	resp.media = UploadIdRespSchema().dump(Upload(upload_id)).data
191
192
193	1	@api.background.task
194		def convert_pdf(digits, extension, upload_id):
195		"""Convert images to PDF
196
197		This API is background task.
198
199		Args:
200		digits: file serial number digits
201		extension: Image extension
202		upload_id: Request ID
203		Returns:
204		bool: If success return true.
205
206		"""
207	1	converter = Image2PDF(digits=digits, extension=extension, directory_path=upload_id)
208	1	converter.make_pdf("result.pdf")
209	1	with open(os.path.join(upload_id, "result_meta.txt"), "w") as result_txt:
210	1	now = datetime.datetime.now()
211	1	result = {
212		"upload_id": upload_id,
213		"digits": digits,
214		"extension": extension,
215		"datetime": now.strftime("%Y/%m/%d %H:%M:%S"),
216		}
217	1	result_txt.write(json.dumps(result))
218	1	return True
219
220
221	1	@api.route("/convert/pdf/download")
222		async def download_result_pdf(req, resp):
223		"""Upload Image files.
224		---
225		post:
226		summary: Upload Id witch get upload images
227
228		requestBody:
229		description: Upload Id
230		content:
231		application/json:
232		schema:
233		$ref: "#/components/schemas/UploadImagesReq"
234		responses:
235		"200":
236		description: OK
237		content:
238		application/pdf:
239		schema:
240		type: string
241		format: binary
242		"404":
243		description: FileNotFound
244		"""
245	1	request = await req.media()
246	1	data = DownloadReqSchema().load(request).data
247	1	upload_id = data["uploadId"]
248	1	result_meta = os.path.join(upload_id, "result_meta.txt")
249	1	if os.path.exists(result_meta):
250	1	with open(os.path.join(upload_id, "result.pdf"), "rb") as result_pdf:
251	1	resp.headers["Content-Type"] = "application/pdf"
252	1	resp.content = result_pdf.read()
253		else:
254	1	resp.status_code = api.status_codes.HTTP_404
255
256
257	1	def convert_content_type_to_extension(content_type):
258		"""Convert image extension to Content-Type
259
260		Args:
261		content_type: Content-Type
262		Returns:
263		str: extension
264
265		"""
266	1	if content_type == "image/jpeg":
267	1	extension = "jpg"
268	1	elif content_type == "image/png":
269	1	extension = "png"
270	1	elif content_type == "image/gif":
271	1	extension = "gif"
272		else:
273	1	extension = False
274
275		return extension
276

tubone24 / ebook_homebrew

Push — master ( a56bd6...df426b )

ebook_homebrew.rest.Upload.__init__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

ebook_homebrew.rest.Upload.init() A