-
Notifications
You must be signed in to change notification settings - Fork 4
/
general.py
149 lines (125 loc) · 5.74 KB
/
general.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
import os
from enum import IntEnum
from typing import Optional, Tuple, Union
import click
from requests.exceptions import BaseHTTPError, HTTPError
from .base import _click_exception_with_exit_code
from .local import _check_local_plantuml, print_local_check_info
from .remote import _check_remote_plantuml, print_remote_check_info
from ..models.base import PlantumlType, Plantuml, PlantumlResourceType
from ..models.local import LocalPlantuml, LocalPlantumlExecuteError
from ..models.remote import RemotePlantuml
from ..utils import load_text_file, linear_process, save_binary_file, auto_decode
def print_double_check_info(local_ok: bool, local: LocalPlantuml,
remote_ok: bool, remote: RemotePlantuml) -> None:
"""
Check if remote and local plantuml is found and okay
:param local_ok: local plantuml object initialize success or not
:param local: local plantuml object or raised exception when initialize
:param remote_ok: remote plantuml object initialize success or not
:param remote: remote plantuml object or raised exception when initialize
"""
_local_ok = _check_local_plantuml(local_ok, local)
_remote_ok = _check_remote_plantuml(remote_ok, remote)
if not _local_ok and not _remote_ok:
raise _click_exception_with_exit_code('PlantumlNotFound', 'Neither local nor remote plantuml is found.', -1)
class PlantumlCheckType(IntEnum):
LOCAL = PlantumlType.LOCAL.value
REMOTE = PlantumlType.REMOTE.value
BOTH = 3
def print_check_info(check_type: PlantumlCheckType,
local_ok: bool, local: LocalPlantuml,
remote_ok: bool, remote: RemotePlantuml) -> None:
"""
Check for all the situations of plantuml
:param check_type: type of checking process (BOTH, LOCAL and REMOTE)
:param local_ok: local plantuml object initialize success or not
:param local: local plantuml object or raised exception when initialize
:param remote_ok: remote plantuml object initialize success or not
:param remote: remote plantuml object or raised exception when initialize
"""
if check_type == PlantumlCheckType.BOTH:
print_double_check_info(local_ok, local, remote_ok, remote)
elif check_type == PlantumlCheckType.LOCAL:
print_local_check_info(local_ok, local)
elif check_type == PlantumlCheckType.REMOTE:
print_remote_check_info(remote_ok, remote)
else:
# nothing to check, maybe warnings can be placed here.
pass
def print_text_graph(plantuml: Plantuml, sources: Tuple[str], concurrency: int): # noqa
"""
Print text graph of source codes
:param plantuml: plantuml object
:param sources: source code files
:param concurrency: concurrency when running this
"""
_error_count = 0
def _process_text(src: str):
try:
return True, plantuml.dump_txt(load_text_file(src))
except (LocalPlantumlExecuteError, OSError, BaseHTTPError, HTTPError) as e:
return False, e
def _print_text(src: str, ret: Tuple[bool, Union[str, LocalPlantumlExecuteError]]):
_success, _data = ret
if _success:
click.secho(f'{src}: ', fg='green')
click.echo(_data)
else:
nonlocal _error_count
if isinstance(_data, LocalPlantumlExecuteError):
click.secho(f'{src}: [error with exitcode {_data.exitcode}]', fg='red')
click.secho(_data.stderr, fg='red')
else:
if hasattr(_data, 'response'):
response = getattr(_data, 'response')
if hasattr(response, 'status_code'):
code = getattr(response, 'status_code')
else:
code = None
else:
response, code = None, None
if code:
click.secho(f'{src}: [{type(_data).__name__} {code}]', fg='red')
else:
click.secho(f'{src}: [{type(_data).__name__}]', fg='red')
click.secho(str(_data), fg='red')
if response is not None and code and response.content:
click.secho(auto_decode(response.content), fg='red')
click.secho('', fg='red')
_error_count += 1
linear_process(
items=sources,
process=lambda i, src: _process_text(src),
post_process=lambda i, src, ret: _print_text(src, ret),
concurrency=concurrency,
)
if _error_count > 0:
raise _click_exception_with_exit_code(
name='TextGraphError',
message=f'{_error_count} error(s) found when generating text graph.',
exitcode=-2,
)
def process_plantuml(plantuml: Plantuml, sources: Tuple[str],
outputs: Tuple[str], output_dir: Optional[str],
type_: PlantumlResourceType, concurrency: int):
if outputs and len(outputs) != len(sources):
raise ValueError(f'Amount of output file(s) should be {len(sources)}, but {len(outputs)} found.')
def _output_filename(index: int):
if outputs:
name = outputs[index]
else:
_, _filename = os.path.split(sources[index])
_name, _ = os.path.splitext(_filename)
name = f'{_name}.{type_.name.lower()}'
return os.path.join(output_dir or os.curdir, name)
def _process_code(index: int):
return plantuml.dump_binary(type_, load_text_file(sources[index]))
def _post_process_data(index: int, ret: bytes):
save_binary_file(_output_filename(index), ret)
linear_process(
items=sources,
process=lambda i, src: _process_code(i),
post_process=lambda i, src, ret: _post_process_data(i, ret),
concurrency=concurrency,
)