• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3
4#
5# Copyright (c) 2023 Huawei Device Co., Ltd.
6# Licensed under the Apache License, Version 2.0 (the "License");
7# you may not use this file except in compliance with the License.
8# You may obtain a copy of the License at
9#
10#     http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing, software
13# distributed under the License is distributed on an "AS IS" BASIS,
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15# See the License for the specific language governing permissions and
16# limitations under the License.
17#
18
19import sys
20import re
21import os
22
23from containers.colors import Colors
24from helper.noInstance import NoInstance
25from resources.global_var import STATUS_FILE
26from util.io_util import IoUtil
27from exceptions.ohos_exception import OHOSException
28
29
30class LogLevel():
31    INFO = 0
32    WARNING = 1
33    ERROR = 2
34    DEBUG = 3
35
36
37class LogUtil(metaclass=NoInstance):
38
39    @staticmethod
40    def hb_info(msg, mode='normal'):
41        level = 'info'
42        if mode == 'silent':
43            for line in str(msg).splitlines():
44                sys.stdout.write('\033[K')
45                sys.stdout.write(
46                    '\r' + (LogUtil.message(level, line)).strip('\n'))
47                sys.stdout.flush()
48        elif mode == 'normal':
49            level = 'info'
50            for line in str(msg).splitlines():
51                sys.stdout.write(LogUtil.message(level, line))
52                sys.stdout.flush()
53
54    @staticmethod
55    def hb_warning(msg):
56        level = 'warning'
57        for line in str(msg).splitlines():
58            sys.stderr.write(LogUtil.message(level, line))
59            sys.stderr.flush()
60
61    @staticmethod
62    def hb_error(msg):
63        level = 'error'
64        sys.stderr.write('\n')
65        for line in str(msg).splitlines():
66            sys.stderr.write(LogUtil.message(level, line))
67            sys.stderr.flush()
68
69    @staticmethod
70    def message(level, msg):
71        if isinstance(msg, str) and not msg.endswith('\n'):
72            msg += '\n'
73        if level == 'error':
74            msg = msg.replace('error:', f'{Colors.ERROR}error{Colors.END}:')
75            return f'{Colors.ERROR}[OHOS {level.upper()}]{Colors.END} {msg}'
76        elif level == 'info':
77            return f'[OHOS {level.upper()}] {msg}'
78        else:
79            return f'{Colors.WARNING}[OHOS {level.upper()}]{Colors.END} {msg}'
80
81    @staticmethod
82    def write_log(log_path, msg, level):
83        os.makedirs(os.path.dirname(log_path), exist_ok=True)
84        sys.stderr.write('\n')
85        with open(log_path, 'at', encoding='utf-8') as log_file:
86            for line in str(msg).splitlines():
87                sys.stderr.write(LogUtil.message(level, line))
88                sys.stderr.flush()
89                log_file.write(LogUtil.message(level, line))
90
91    @staticmethod
92    def analyze_build_error(error_log, status_code_prefix):
93        with open(error_log, 'rt', encoding='utf-8') as log_file:
94            data = log_file.read()
95            status_file = IoUtil.read_json_file(STATUS_FILE)
96            choices = []
97            status_map = {}
98            for status_code, status in status_file.items():
99                if not status_code.startswith(status_code_prefix):
100                    continue
101                if isinstance(status, dict) and status.get('pattern'):
102                    choices.append(status['pattern'])
103                    status_map[status['pattern']] = status.get('code')
104            best_match = None
105            best_ratio = 0
106            for choice in choices:
107                pattern = re.compile(choice, re.DOTALL)
108                match = pattern.search(data)
109                if not match:
110                    continue
111                ratio = len(match.group()) / len(data)
112                if ratio > best_ratio:
113                    best_ratio = ratio
114                    best_match = choice
115            return_status_code = status_map.get(
116                best_match) if best_match else f'{status_code_prefix}000'
117        return return_status_code
118
119    @staticmethod
120    def get_gn_failed_log(log_path):
121        error_log = os.path.join(os.path.dirname(log_path), 'error.log')
122        is_gn_failed = False
123        with open(log_path, 'rt', encoding='utf-8') as log_file:
124            lines = log_file.readlines()
125        error_lines = []
126        for i, line in enumerate(lines):
127            if line.startswith('ERROR at'):
128                error_lines.extend(lines[i: i + 50])
129                is_gn_failed = True
130                break
131        for log in error_lines[:50]:
132            LogUtil.hb_error(log)
133            with open(error_log, 'at', encoding='utf-8') as log_file:
134                log_file.write(log + '\n')
135        if is_gn_failed:
136            return_status_code = LogUtil.analyze_build_error(error_log, '3')
137            raise OHOSException(
138                'GN Failed! Please check error in {}, and for more build information in {}'.format(
139                    error_log, log_path), return_status_code)
140
141    @staticmethod
142    def get_ninja_failed_log(log_path):
143        error_log = os.path.join(os.path.dirname(log_path), 'error.log')
144        is_ninja_failed = False
145        with open(log_path, 'rt', encoding='utf-8') as log_file:
146            data = log_file.read()
147        failed_pattern = re.compile(r'(ninja: error:.*?)\n', re.DOTALL)
148        failed_log = failed_pattern.findall(data)
149        if failed_log:
150            is_ninja_failed = True
151        for log in failed_log:
152            LogUtil.hb_error(log)
153            with open(error_log, 'at', encoding='utf-8') as log_file:
154                log_file.write(log)
155        if is_ninja_failed:
156            return_status_code = LogUtil.analyze_build_error(error_log, '4')
157            raise OHOSException(
158                'NINJA Failed! Please check error in {}, and for more build information in {}'.format(
159                    error_log, log_path), return_status_code)
160
161    @staticmethod
162    def get_compiler_failed_log(log_path):
163        error_log = os.path.join(os.path.dirname(log_path), 'error.log')
164        is_compiler_failed = False
165        with open(log_path, 'rt', encoding='utf-8') as log_file:
166            data = log_file.read()
167        failed_pattern = re.compile(
168            r'(\[\d+/\d+\].*?)(?=\[\d+/\d+\]|'
169            'ninja: build stopped)', re.DOTALL)
170        failed_log = failed_pattern.findall(data)
171        if failed_log:
172            is_compiler_failed = True
173        for log in failed_log:
174            if 'FAILED:' in log:
175                LogUtil.hb_error(log)
176                with open(error_log, 'at', encoding='utf-8') as log_file:
177                    log_file.write(log)
178        if is_compiler_failed:
179            return_status_code = LogUtil.analyze_build_error(error_log, '4')
180            raise OHOSException(
181                'COMPILE Failed! Please check error in {}, and for more build information in {}'.format(
182                    error_log, log_path), return_status_code)
183
184    @staticmethod
185    def get_failed_log(log_path):
186        last_error_log = os.path.join(os.path.dirname(log_path), 'error.log')
187        if os.path.exists(last_error_log):
188            mtime = os.stat(last_error_log).st_mtime
189            os.rename(
190                last_error_log, '{}/error.{}.log'.format(os.path.dirname(last_error_log), mtime))
191        LogUtil.get_gn_failed_log(log_path)
192        LogUtil.get_ninja_failed_log(log_path)
193        LogUtil.get_compiler_failed_log(log_path)
194        raise OHOSException(
195            'BUILD Failed! Please check build log for more information: {}'.format(log_path))
196