ebook_homebrew.rest.convert_image_to_pdf() - Code Metrics - tubone24/ebook_homebrew - Measure and Improve Code Quality continuously with Scrutinizer

ebook_homebrew.rest.convert_image_to_pdf() A
last analyzed 2021-03-23 19:11 UTC

↳ Parent: ebook_homebrew.rest

Complexity

Conditions

Size

Total Lines	64
Code Lines	30

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	25
CRAP Score	4

Importance

Changes

Metric	Value
cc	4
eloc	30
nop	2
dl	0
loc	64
ccs	25
cts	25
cp	1
crap	4
rs	9.16
c	0
b	0
f	0

How to fix Long Method

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

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

from .convert import Image2PDF
from .utils.logging import get_logger
from .models.rest_models import (
    UploadModel,
    ErrorModel,
    FileNotFoundModel,
    StatusModel,
    ListUploadFiles,
)
from .rdb import UploadedFile
from .__init__ import __version__

api = responder.API(
    title="Ebook-homebrew",
    debug=True,
    cors=True,
    cors_params={
        "allow_origins": ["*"],
        "allow_methods": ["GET", "POST"],
        "allow_headers": ["*"],
    },
    version=__version__,
    static_dir=os.path.join(os.path.dirname(os.path.abspath(__file__)), "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"},
)

upload_file = UploadedFile()

_logger = get_logger("RestAPI")


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


@api.schema("ListUploadFiles")
class ListUploadFilesSchema(Schema):
    fileList = fields.List(fields.Dict(fields.Str()))


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


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


@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.schema("ErrorResp")
class ErrorRespSchema(Schema):
    error = fields.Str()
    errorDate = fields.Date()


@api.schema("FileNotFoundResp")
class FileNotFoundRespSchema(Schema):
    reason = fields.Str()
    errorDate = fields.Date()


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")
    resp.media = HealthCheckSchema().dump(StatusModel("ok", __version__)).data


@api.route("/data/upload/list")
def list_upload_files(_, resp):
    """Responce File List.
    ---
    get:
        description: Get file list
        responses:
            "200":
                description: OK
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/ListUploadFiles"
    """
    _logger.debug("List File")
    file_list = upload_file.get_all_uploaded_file()
    resp.media = ListUploadFilesSchema().dump(ListUploadFiles(file_list)).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"
            "400":
                description: BadRequest
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/ErrorResp"
    """
    request = await req.media()
    try:
        data = UploadImagesReqSchema(strict=True).load(request).data
    except ValidationError as error:
        resp.status_code = api.status_codes.HTTP_400
        resp.media = ErrorRespSchema().dump(ErrorModel(error)).data
        return
    _logger.debug(data)
    content_type = data["contentType"]
    images_b64 = data["images"]
    tmp_dir = tempfile.mkdtemp()
    write_image(images_b64, content_type, tmp_dir)
    resp.media = UploadIdRespSchema().dump(UploadModel(tmp_dir)).data


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

    This API is background task.

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

    """
    global file_name
    last_index = 0
    extension = convert_content_type_to_extension(content_type)
    for i, content in enumerate(images_b64):
        image = base64.b64decode(content.split(",")[-1])
        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)
            last_index = i
    upload_file.add_uploaded_file(
        name=file_name,
        file_path=str(tmp_dir),
        file_type=content_type,
        last_index=last_index,
    )
    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"
            "400":
                description: BadRequest
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/ErrorResp"
            "404":
                description: UploadIdNotFound
                content:
                    application/json:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/FileNotFoundResp"
    """
    request = await req.media()
    try:
        data = ConvertReqSchema(strict=True).load(request).data
    except ValidationError as error:
        resp.status_code = api.status_codes.HTTP_400
        resp.media = ErrorRespSchema().dump(ErrorModel(error)).data
        return
    _logger.debug(data)
    upload_id = data["uploadId"]
    content_type = data["contentType"]
    if not os.path.isdir(upload_id):
        resp.status_code = api.status_codes.HTTP_404
        resp.media = (
            FileNotFoundRespSchema()
            .dump(FileNotFoundModel("upload_id is NotFound"))
            .data
        )
        return
    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(UploadModel(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/DownloadReq"
        responses:
            "200":
                description: OK
                content:
                    application/pdf:
                        schema:
                            type: string
                            format: binary
            "400":
                description: BadRequest
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/ErrorResp"
            "404":
                description: ResultFileNotFound
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/FileNotFoundResp"
    """
    request = await req.media()
    try:
        data = DownloadReqSchema(strict=True).load(request).data
    except ValidationError as error:
        resp.status_code = api.status_codes.HTTP_400
        resp.media = ErrorRespSchema().dump(ErrorModel(error)).data
        return
    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
        resp.media = (
            FileNotFoundRespSchema().dump(FileNotFoundModel("resultFile NotFound")).data
        )


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	import responder
12	1	from marshmallow import Schema, fields, ValidationError
13
14	1	from .convert import Image2PDF
15	1	from .utils.logging import get_logger
16	1	from .models.rest_models import (
17		UploadModel,
18		ErrorModel,
19		FileNotFoundModel,
20		StatusModel,
21		ListUploadFiles,
22		)
23	1	from .rdb import UploadedFile
24	1	from .__init__ import __version__
25
26	1	api = responder.API(
27		title="Ebook-homebrew",
28		debug=True,
29		cors=True,
30		cors_params={
31		"allow_origins": ["*"],
32		"allow_methods": ["GET", "POST"],
33		"allow_headers": ["*"],
34		},
35		version=__version__,
36		static_dir=os.path.join(os.path.dirname(os.path.abspath(__file__)), "static"),
37		openapi="3.0.2",
38		docs_route="/docs",
39		openapi_route="/schema.yml",
40		description="Make PDF file taken in "
41		"some image files such as "
42		"jpeg, png and gif.",
43		contact={
44		"name": "tubone24",
45		"url": "https://tubone-project24.xyz",
46		"email": "[email protected]",
47		},
48		license={"name": "MIT", "url": "https://opensource.org/licenses/MIT"},
49		)
50
51	1	upload_file = UploadedFile()
52
53	1	_logger = get_logger("RestAPI")
54
55
56	1	@api.schema("HealthCheck")
57	1	class HealthCheckSchema(Schema):
58	1	status = fields.Str()
59	1	version = fields.Str()
60
61
62	1	@api.schema("ListUploadFiles")
63	1	class ListUploadFilesSchema(Schema):
64	1	fileList = fields.List(fields.Dict(fields.Str()))
65
66
67	1	@api.schema("UploadImagesReq")
68	1	class UploadImagesReqSchema(Schema):
69	1	contentType = fields.Str(required=True)
70	1	images = fields.List(fields.Str(required=True), required=True)
71
72
73	1	@api.schema("UploadIdResp")
74	1	class UploadIdRespSchema(Schema):
75	1	upload_id = fields.Str()
76	1	release_date = fields.Date()
77
78
79	1	@api.schema("ConvertReq")
80	1	class ConvertReqSchema(Schema):
81	1	uploadId = fields.Str(required=True)
82	1	contentType = fields.Str(required=True)
83
84
85	1	@api.schema("DownloadReq")
86	1	class DownloadReqSchema(Schema):
87	1	uploadId = fields.Str(required=True)
88
89
90	1	@api.schema("ErrorResp")
91	1	class ErrorRespSchema(Schema):
92	1	error = fields.Str()
93	1	errorDate = fields.Date()
94
95
96	1	@api.schema("FileNotFoundResp")
97	1	class FileNotFoundRespSchema(Schema):
98	1	reason = fields.Str()
99	1	errorDate = fields.Date()
100
101
102	1	api.add_route("/", static=True)
103
104
105	1	@api.route("/status")
106		def status(_, resp):
107		"""Health Check Response.
108		---
109		get:
110		description: Get Status
111		responses:
112		"200":
113		description: OK
114		content:
115		application/json:
116		schema:
117		$ref: "#/components/schemas/HealthCheck"
118		"""
119	1	_logger.debug("health Check")
120	1	resp.media = HealthCheckSchema().dump(StatusModel("ok", __version__)).data
121
122
123	1	@api.route("/data/upload/list")
124		def list_upload_files(_, resp):
125		"""Responce File List.
126		---
127		get:
128		description: Get file list
129		responses:
130		"200":
131		description: OK
132		content:
133		application/json:
134		schema:
135		$ref: "#/components/schemas/ListUploadFiles"
136		"""
137		_logger.debug("List File")
138		file_list = upload_file.get_all_uploaded_file()
139		resp.media = ListUploadFilesSchema().dump(ListUploadFiles(file_list)).data
140
141
142	1	@api.route("/data/upload")
143		async def upload_image_file(req, resp):
144		"""Upload Image files.
145		---
146		post:
147		summary: Base64 encoded Images
148
149		requestBody:
150		description: base64 encoded Images in images Array
151		content:
152		application/json:
153		schema:
154		$ref: "#/components/schemas/UploadImagesReq"
155		responses:
156		"200":
157		description: OK
158		content:
159		application/json:
160		schema:
161		$ref: "#/components/schemas/UploadIdResp"
162		"400":
163		description: BadRequest
164		content:
165		application/json:
166		schema:
167		$ref: "#/components/schemas/ErrorResp"
168		"""
169	1	request = await req.media()
170	1	try:
171	1	data = UploadImagesReqSchema(strict=True).load(request).data
172	1	except ValidationError as error:
173	1	resp.status_code = api.status_codes.HTTP_400
174	1	resp.media = ErrorRespSchema().dump(ErrorModel(error)).data
175	1	return
176	1	_logger.debug(data)
177	1	content_type = data["contentType"]
178	1	images_b64 = data["images"]
179	1	tmp_dir = tempfile.mkdtemp()
180	1	write_image(images_b64, content_type, tmp_dir)
181	1	resp.media = UploadIdRespSchema().dump(UploadModel(tmp_dir)).data
182
183
184	1	@api.background.task
185		def write_image(images_b64, content_type, tmp_dir):
186		"""Images write at tmp_dir
187
188		This API is background task.
189
190		Args:
191		images_b64: Base64 encoded images list
192		content_type: Image ContentType
193		tmp_dir: Temp directory writing images
194		Returns:
195		bool: If success return true.
196
197		"""
198		global file_name
199	1	last_index = 0
200	1	extension = convert_content_type_to_extension(content_type)
201	1	for i, content in enumerate(images_b64):
202	1	image = base64.b64decode(content.split(",")[-1])
203	1	file_name = os.path.join(tmp_dir, str(i) + "." + extension)
204	1	_logger.debug("file_name: {}".format(file_name))
205	1	with open(file_name, "wb") as image_file:
206	1	image_file.write(image)
207	1	last_index = i
208	1	upload_file.add_uploaded_file(
209		name=file_name,
210		file_path=str(tmp_dir),
211		file_type=content_type,
212		last_index=last_index,
213		)
214	1	return True
215
216
217	1	@api.route("/convert/pdf")
218		async def convert_image_to_pdf(req, resp):
219		"""Convert Image files to PDF.
220		---
221		post:
222		summary: Upload Id witch get upload images and ContentType
223
224		requestBody:
225		description: Upload Id and ContentType
226		content:
227		application/json:
228		schema:
229		$ref: "#/components/schemas/ConvertReq"
230		responses:
231		"200":
232		description: OK
233		content:
234		application/json:
235		schema:
236		$ref: "#/components/schemas/UploadIdResp"
237		"400":
238		description: BadRequest
239		content:
240		application/json:
241		schema:
242		$ref: "#/components/schemas/ErrorResp"
243		"404":
244		description: UploadIdNotFound
245		content:
246		application/json:
247		application/json:
248		schema:
249		$ref: "#/components/schemas/FileNotFoundResp"
250		"""
251	1	request = await req.media()
252	1	try:
253	1	data = ConvertReqSchema(strict=True).load(request).data
254	1	except ValidationError as error:
255	1	resp.status_code = api.status_codes.HTTP_400
256	1	resp.media = ErrorRespSchema().dump(ErrorModel(error)).data
257	1	return
258	1	_logger.debug(data)
259	1	upload_id = data["uploadId"]
260	1	content_type = data["contentType"]
261	1	if not os.path.isdir(upload_id):
262	1	resp.status_code = api.status_codes.HTTP_404
263	1	resp.media = (
264		FileNotFoundRespSchema()
265		.dump(FileNotFoundModel("upload_id is NotFound"))
266		.data
267		)
268	1	return
269	1	result_meta = os.path.join(upload_id, "result_meta.txt")
270	1	if os.path.exists(result_meta):
271	1	os.remove(result_meta)
272	1	extension = convert_content_type_to_extension(content_type)
273	1	file_list = sorted(
274		glob.glob(os.path.join(upload_id, "*." + extension)), reverse=True
275		)
276	1	file_base, _ = os.path.splitext(os.path.basename(file_list[0]))
277	1	digits = len(file_base)
278	1	_logger.debug(file_list)
279	1	convert_pdf(digits, extension, upload_id)
280	1	resp.media = UploadIdRespSchema().dump(UploadModel(upload_id)).data
281
282
283	1	@api.background.task
284		def convert_pdf(digits, extension, upload_id):
285		"""Convert images to PDF
286
287		This API is background task.
288
289		Args:
290		digits: file serial number digits
291		extension: Image extension
292		upload_id: Request ID
293		Returns:
294		bool: If success return true.
295
296		"""
297	1	converter = Image2PDF(digits=digits, extension=extension, directory_path=upload_id)
298	1	converter.make_pdf("result.pdf")
299	1	with open(os.path.join(upload_id, "result_meta.txt"), "w") as result_txt:
300	1	now = datetime.datetime.now()
301	1	result = {
302		"upload_id": upload_id,
303		"digits": digits,
304		"extension": extension,
305		"datetime": now.strftime("%Y/%m/%d %H:%M:%S"),
306		}
307	1	result_txt.write(json.dumps(result))
308	1	return True
309
310
311	1	@api.route("/convert/pdf/download")
312		async def download_result_pdf(req, resp):
313		"""Upload Image files.
314		---
315		post:
316		summary: Upload Id witch get upload images
317
318		requestBody:
319		description: Upload Id
320		content:
321		application/json:
322		schema:
323		$ref: "#/components/schemas/DownloadReq"
324		responses:
325		"200":
326		description: OK
327		content:
328		application/pdf:
329		schema:
330		type: string
331		format: binary
332		"400":
333		description: BadRequest
334		content:
335		application/json:
336		schema:
337		$ref: "#/components/schemas/ErrorResp"
338		"404":
339		description: ResultFileNotFound
340		content:
341		application/json:
342		schema:
343		$ref: "#/components/schemas/FileNotFoundResp"
344		"""
345	1	request = await req.media()
346	1	try:
347	1	data = DownloadReqSchema(strict=True).load(request).data
348	1	except ValidationError as error:
349	1	resp.status_code = api.status_codes.HTTP_400
350	1	resp.media = ErrorRespSchema().dump(ErrorModel(error)).data
351	1	return
352	1	upload_id = data["uploadId"]
353	1	result_meta = os.path.join(upload_id, "result_meta.txt")
354	1	if os.path.exists(result_meta):
355	1	with open(os.path.join(upload_id, "result.pdf"), "rb") as result_pdf:
356	1	resp.headers["Content-Type"] = "application/pdf"
357	1	resp.content = result_pdf.read()
358		else:
359	1	resp.status_code = api.status_codes.HTTP_404
360	1	resp.media = (
361		FileNotFoundRespSchema().dump(FileNotFoundModel("resultFile NotFound")).data
362		)
363
364
365	1	def convert_content_type_to_extension(content_type):
366		"""Convert image extension to Content-Type
367
368		Args:
369		content_type: Content-Type
370		Returns:
371		str: extension
372
373		"""
374	1	if content_type == "image/jpeg":
375	1	extension = "jpg"
376	1	elif content_type == "image/png":
377	1	extension = "png"
378	1	elif content_type == "image/gif":
379	1	extension = "gif"
380		else:
381	1	extension = False
382
383		return extension
384

tubone24 / ebook_homebrew

ebook_homebrew.rest.convert_image_to_pdf() A last analyzed 2021-03-23 19:11 UTC

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

ebook_homebrew.rest.convert_image_to_pdf() A
last analyzed 2021-03-23 19:11 UTC