编译函数，指代码生成数据流
(self, params, lang)
| 157 | return content.split(" ")[-1].strip() |
| 158 | |
| 159 | def compile(self, params, lang): |
| 160 | """ |
| 161 | 编译函数，指代码生成数据流 |
| 162 | """ |
| 163 | source_dir = params.source_dir |
| 164 | relpos = len(source_dir) + 1 |
| 165 | logger.info("开始编译项目 %s" % source_dir) |
| 166 | work_dir = params.work_dir |
| 167 | db_dir = os.path.join(work_dir, "db") |
| 168 | repo_id = params.repo_id |
| 169 | envs = os.environ |
| 170 | ZEUS_HOME = envs.get("ZEUS_HOME", None) |
| 171 | scm_revision = params["scm_revision"] |
| 172 | version = self.__get_zeus_version(params) |
| 173 | db_name = f"{repo_id}_{scm_revision}_{lang}_{version}" |
| 174 | db_path = os.path.join(db_dir, f"{db_name}.db") |
| 175 | inc = params["incr_scan"] |
| 176 | want_suffix = lang_map[lang] |
| 177 | logger.info("是否为增量编译: %s" % inc) |
| 178 | file_list = os.path.join(work_dir, "filelist.txt") |
| 179 | if not os.path.exists(db_dir): |
| 180 | if not os.path.exists(db_dir): |
| 181 | os.makedirs(db_dir) |
| 182 | if self.__download_database(params, db_name): |
| 183 | if os.path.exists(db_path): |
| 184 | return |
| 185 | if inc: |
| 186 | last_scm_revision = params["scm_last_revision"] |
| 187 | last_db_name = f"{repo_id}_{last_scm_revision}_{lang}_{version}" |
| 188 | last_db_path = os.path.join(db_dir, last_db_name + ".db") |
| 189 | logger.info(f"下载上个成功分析版本数据库{last_db_name}") |
| 190 | if not self.__download_database(params, last_db_name): |
| 191 | logger.info("下载全量数据库失败将重新生成，本次分析将只分析增量部分") |
| 192 | elif os.path.exists(last_db_path): |
| 193 | shutil.copyfile(last_db_path, db_path) |
| 194 | else: |
| 195 | logger.error("下载失败，本次将只分析增量文件，建议全量分析下项目。") |
| 196 | diffs = SCMMgr(params).get_scm_diff() |
| 197 | # 增量需要所有增量文件都重新生成，故这里不能过滤文件 |
| 198 | toscans = [diff.path.replace(os.sep, "/") for diff in diffs if diff.path.endswith(tuple(want_suffix))] |
| 199 | toscans = FilterPathUtil(params).get_include_files(toscans, relpos) |
| 200 | if not toscans: |
| 201 | return [] |
| 202 | with open(file_list, "w") as wf: |
| 203 | for toscan in toscans: |
| 204 | wf.writelines(toscan) |
| 205 | wf.write("\n") |
| 206 | inc_build_cmd = self.get_cmd( |
| 207 | "Zeus", |
| 208 | [ |
| 209 | "inc_compile", |
| 210 | "-l", |
| 211 | lang, |
| 212 | "-p", |
| 213 | db_name, |
| 214 | "-db", |
| 215 | db_dir, |
| 216 | "-s", |
nothing calls this directly
no test coverage detected