缘起：公司每次迭代都会统一拉起分支，每个领域有各自的微服务，微服务内部按功能和使用方等维度划分多个不同的子模块，例如对外的spi接口层，spi实现层，SDK层，对内的api接口层、实现层，业务实现的bootstarp层，web层，service层等等；在编译时就会出现以下问题：
- 循环依赖 2：依赖项命名规则不规范，导致无法不知道应该编译那个项目的那个模块 3：依赖模块超多，纯手工编译需要根据依赖报错信息来指定编译依赖的模块，耗时太大
基于此，我实现了一个自动依赖识别和构建的程序，它可以分析目录下所有的项目，解析出它们对应的模块，当在编译时依赖模块找不到时，自动切换到指定项目中进行相应 模块的编译
功能：
- 自动遍历目录下所有的工程，检查是否是maven工程，解析maven是否支持profile的配置，生成项目模块的映射关系
- 指定编写某个工程或某个工程的子模块，会自动发现maven仓库中没有的依赖模块，然后进行依赖模块的编译和deploy
#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
import json
import os
import subprocess
import xml.etree.ElementTree as ET
from xml.etree.ElementTree import ParseError
ns = {'ns': 'http://maven.apache.org/POM/4.0.0'}
print("hello world")
force_no_cache=False
offline = False
param_profile = 'test'
param_version = '0.0.1-SNAPSHOT'
# param_setting = None
param_setting = "/root/.m2/settings.xml"
param_exclusions = {
'com.tianjiaguo.site:test-parent' : '!com.tianjiaguo.site:testsdk'
}
def check_pom_plugin(pom_path, plugin_artifact):
global ns
try:
tree = ET.parse(pom_path)
root = tree.getroot()
extensions = root.find('.//ns:build/ns:extensions', ns)
if extensions:
for ext in extensions.findall('ns:extension', ns):
artifact_id = ext.find('ns:artifactId', ns)
if artifact_id is not None and artifact_id.text == plugin_artifact:
return True
plugins = root.find('.//ns:build/ns:plugins', ns)
if plugins:
for plugin in plugins.findall('ns:plugin', ns):
artifact_id = plugin.find('ns:artifactId', ns)
if artifact_id is not None and artifact_id.text == plugin_artifact:
return True
return False
except Exception as e:
print(f"解析错误：{str(e)}")
return False
def check_pom_profile(pom_path, profile_id):
global ns
try:
tree = ET.parse(pom_path)
root = tree.getroot()
profiles = root.find('.//ns:profiles', ns)
if profiles:
for profile in profiles.findall('ns:profile', ns):
pid = profile.find('ns:id', ns)
if pid is not None and pid.text == profile_id:
return True
return False
except Exception as e:
print(f"解析错误：{str(e)}")
return False
def extract_artifacts(error_line: str):
if not error_line.startswith('[ERROR] Failed to execute goal on project '):
return None
# 找到关键部分
start = error_line.find("The following artifacts could not be resolved: ")
if start != -1:
end = -1
end_strs = [
# ': Cannot access ',
# ': Failure to find ',
# ': Could not find ',
]
for end_str in end_strs:
end = error_line.find(end_str)
if end != -1:
break
if end == -1:
error_line_toend = error_line[start + len("The following artifacts could not be resolved: "):]
if error_line_toend.find(": ") == -1:
return None
artifacts_str = error_line_toend.split(": ")[0]
artifacts_list = [artifact.strip() for artifact in artifacts_str.split(", ")]
return artifacts_list
if start != -1 and end != -1:
# 提取并分割字符串
artifacts_str = error_line[start + len("The following artifacts could not be resolved: "):end]
artifacts_list = [artifact.strip() for artifact in artifacts_str.split(", ")]
return artifacts_list
else:
m_pattern = [
r"Could not find artifact ([^ ]+)",
r"Failure to find ([^ ]+)",
]
for pattern in m_pattern:
matches = re.findall(pattern, error_line)
if len(matches) > 0:
# 提取依赖包信息
artifacts = []
for match in matches:
artifact = match
artifacts.append(artifact)
return artifacts if len(artifacts) > 0 else None
print(f'依赖提取失败，文本行是： {error_line}')
return None
def get_depend_module_name(artifact:str):
f_d_info = artifact.split(":")
failed_dep = f"{f_d_info[0]}:{f_d_info[1]}"
return failed_dep
import re
def find_all_dependency(base_dir, item):
global ns
global param_profile
global param_setting
work_dir = os.path.join(base_dir, item)
if os.path.isfile(os.path.join(work_dir, 'g_no_compile__')):
print(f"目录不获取依赖信息 {work_dir}")
return {}
if not os.path.isfile(os.path.join(work_dir, 'pom.xml')):
print(f"目录不是Maven工程 {work_dir}")
return {}
os.chdir(work_dir)
if item == 'maventest':
print(f'检查对外三包测试工程 {item}')
if item == 'fiverify':
print(f'检查对外三包测试工程 {item}')
project_validate = check_pom_plugin(os.path.join(work_dir, "pom.xml"), "fi-common-effective-profile")
has_profile = check_pom_profile(os.path.join(work_dir, 'pom.xml'), param_profile)
if project_validate:
c1 = f" -f effective-profile-pom.xml "
else:
c1 = ""
if has_profile:
c2 = f" -P{param_profile} "
else:
c2 = ""
if param_setting:
s1 = f" -s {param_setting} "
else:
s1 = ""
try:
command = f"set -e && set -x && cd {work_dir} && mvn help:effective-pom -Doutput=effective-pom.xml {c2} {s1}"
subprocess.run(command, shell=True, check=True, cwd=work_dir)
tree = ET.parse(os.path.join(work_dir, 'effective-pom.xml'))
root = tree.getroot()
project_infos = {} # cur_module_info ++
project_roots = root.findall('.//ns:project', ns)
module_step = -1 # 判断是否根pom，这个在某些项目上有问题
if len(project_roots) == 0:
project_roots.append(root)
for prj_root in project_roots:
module_step += 1
group_id = prj_root.find('ns:groupId', ns).text
artifact_id = prj_root.find('ns:artifactId', ns).text
version = prj_root.find('ns:version', ns).text
if version != param_version:
print(f"请检查模块的版本号 {item} {group_id}:{artifact_id}:{version}")
module_name = f'{group_id}:{artifact_id}'
if has_profile:
cur_module_info = { module_name: { "dir": item, "profile": param_profile, "validate": project_validate, "step": module_step, } }
else:
cur_module_info = { module_name: { "dir": item, "profile": None, "validate": project_validate, "step": module_step, } }
project_infos.update(cur_module_info)
return project_infos
# for module in root.findall('.//ns:modules/ns:module', ns):
# module_name = module.text.strip()
# module_pom = os.path.join(work_dir, module_name, 'pom.xml')
# if os.path.exists(module_pom):
# module_tree = ET.parse(module_pom)
# module_root = module_tree.getroot()
# group_id = module_root.find('ns:groupId', ns)
# if group_id is None:
# group_id = root.find('ns:groupId', ns)
# artifact_id = module_root.find('ns:artifactId', ns)
# if group_id is not None and artifact_id is not None:
# # module.append(f'{group_id.text}:{artifact_id.text}')
# # "com.tianjiaguo.third-c:test": { "dir": "maventest", "profile": None|test, "validate": None|True|False, }
# module_name = f'{group_id.text}:{artifact_id.text}'
# cur_module_info = { module_name: { "dir": item, "profile": param_profile, "validate": project_validate, } }
# project_infos.update(cur_module_info)
# return project_infos
except (ParseError, subprocess.CalledProcessError) as e:
print(f"获取子模块 {item} 信息失败：{str(e)}")
return {}
# command = f"mvn {c2} validate {s1} && mvn {c1} {c2} dependency:tree {s1} |tee dependency-v1.log" # 替换为你要执行的命令
# process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
# project_infos = {} # cur_module_info ++
# while True:
# output = process.stdout.readline()
# if output == '' and process.poll() is not None:
# break
# if output:
# out_line = output.strip()
# print(out_line)
# match = re.search(r'<([^>]+)>', out_line)
# if match:
# # 获取到模块编译的开始
# module_name = match.group(1).strip()
# print(f'开始处理模块 {module_name} 的依赖信息')
# # "com.tianjiaguo.third-package:test": { "dir": "maventest", "profile": None|test, "validate": None|True|False, }
# cur_module_info = { module_name: { "dir": item, "profile": param_profile, "validate": project_validate, } }
# project_infos.update(cur_module_info)
# return project_infos
# def save_dict_to_json(dictionary, file_path):
# """将字典序列化为JSON并保存到本地文件"""
# with open(file_path, 'w', encoding='utf-8') as file:
# json.dump(dictionary, file, ensure_ascii=False, indent=4)
# def load_dict_from_json(file_path):
# """从本地JSON文件中加载并还原为字典"""
# with open(file_path, 'r', encoding='utf-8') as file:
# return json.load(file)
def find_all_dependency_with_cache(base_dir, item):
work_dir = os.path.join(base_dir, item)
dependency_info_file = os.path.join(work_dir, 'dependency_info.json')
if not force_no_cache and os.path.isfile(dependency_info_file):
with open(dependency_info_file, 'r', encoding='utf-8') as file:
return json.load(file)
else:
dependencies = find_all_dependency(base_dir=base_dir, item=item)
if dependencies and len(dependencies) > 0:
with open(dependency_info_file, 'w', encoding='utf-8') as file:
json.dump(dependencies, file, ensure_ascii=False, indent=4 )
return dependencies
def find_all_dependencys(work_dir: str):
project_infos = {} # "com.tianjiaguo.site:test": { "dir": "maventest", "profile": None, "validate": None, }
os.chdir(work_dir)
for item in os.listdir(work_dir):
item_path = os.path.join(work_dir, item)
if os.path.isdir(item_path):
os.chdir(item_path)
ds = find_all_dependency_with_cache(work_dir, item)
project_infos.update(ds)
return project_infos
def find_all_dependencys_with_cache(work_dir: str):
project_infos = {} # "com.tianjiaguo.site:test": { "dir": "maventest", "profile": None, "validate": None, }
os.chdir(work_dir)
dependency_info_file = os.path.join(work_dir, 'dependency_info.json')
if not force_no_cache and os.path.isfile(dependency_info_file):
with open(dependency_info_file, 'r', encoding='utf-8') as file:
return json.load(file)
else:
dependencies = find_all_dependencys(work_dir=work_dir)
if dependencies and len(dependencies) > 0:
with open(dependency_info_file, 'w', encoding='utf-8') as file:
json.dump(dependencies, file, ensure_ascii=False, indent=4 )
return dependencies
if __name__ == "__main__":
# work_dir="/mnt/g/workspace/test"
work_dir="/mnt/g/workspace/0.0.1-SNAPSHOT"
os.chdir(work_dir)
project_infos = find_all_dependencys_with_cache(work_dir=work_dir)
# root_project_infos = []
# for project_info_key, project_info in project_infos.items():
# project_step = project_info['step']
# if project_step == 0:
# root_project_infos.append({
# 'project_artifact' : project_info_key,
# 'project_info' : project_info,
# })
# print(f'failed_dep_infos.append("{project_info_key}:pom:{param_version}" ) # {project_info["dir"]}')
failed_dep_infos = []
success_deps = {
"com.tianjiaguo.site:test-models":True,
}
for dep_key, dep_info in project_infos.items():
if dep_key.endswith(('-sdk')):
failed_dep_infos.append(f'{dep_key}:pom:{param_version}')
for dep_key, dep_info in project_infos.items():
if dep_key.endswith(('-api')):
failed_dep_infos.append(f'{dep_key}:pom:{param_version}')
###################################################################
# ...
# failed_dep_infos.append("com.tianjiaguo.site:test:pom:0.0.1-SNAPSHOT" ) # maventest
# failed_dep_infos.append("com.tianjiaguo.site:test-parent:pom:0.0.1-SNAPSHOT" ) # fi-pom ---------------
# failed_dep_infos.append("com.tianjiaguo.site:test-pom-parent:pom:0.0.1-SNAPSHOT" ) # fi-pom ---------------
while len(failed_dep_infos) > 0 :
failed_dep_info = failed_dep_infos.pop()
failed_dep = get_depend_module_name(failed_dep_info)
if failed_dep in success_deps:
print(f"构建 {failed_dep} 成功")
continue
else:
print(f"构建 {failed_dep}")
if failed_dep not in project_infos:
print(f"依赖 {failed_dep} 找不到，请检查本地工程是否齐全")
exit(1)
project_info = project_infos[failed_dep]
project_dir = project_info['dir']
project_profile = project_info['profile']
project_validate = project_info['validate']
project_step = project_info['step']
project_path = f"{work_dir}/{project_dir}"
os.chdir(project_path)
# pom.xml
if project_validate:
cmd_pom_file = f" -f effective-profile-pom.xml "
else:
cmd_pom_file = ""
# profile
if project_profile:
cmd_profile = f" -P{project_profile} "
else:
cmd_profile = ""
# pl
if failed_dep in param_exclusions:
exclude_pl = param_exclusions[failed_dep]
else:
exclude_pl = None
if project_step == 0:
cmd_pl = ""
if exclude_pl:
cmd_pl = f" -pl '{exclude_pl}' "
else:
if exclude_pl:
cmd_pl = f" -pl '{failed_dep},{exclude_pl}' "
else:
cmd_pl = f" -pl '{failed_dep}' "
# offline
if offline:
cmd_offline = " -o "
else:
cmd_offline = ""
# setting
if param_setting:
cmd_setting = f" -s {param_setting} "
else:
cmd_setting = ""
# ###################################################################
cmd_validate = f" && mvn {cmd_profile} validate {cmd_setting}"
cmd_clean_install = f" && mvn {cmd_pom_file} {cmd_profile} clean install -Dmaven.test.skip=true {cmd_pl} {cmd_setting} {cmd_offline} "
if 'com.tianjiaguo.site:test' == failed_dep:
cmd_dploy_pl_pom = ""
cmd_deploy_root_pom = ""
else:
cmd_dploy_pl_pom = f" && mvn {cmd_pom_file} {cmd_profile} deploy -Dmaven.test.skip=true {cmd_pl} {cmd_setting} {cmd_offline} "
cmd_deploy_root_pom = f" && mvn {cmd_pom_file} {cmd_profile} deploy -Dmaven.test.skip=true -N {cmd_setting} {cmd_offline} "
command = f"set -e && set -x && cd {project_path} {cmd_validate} {cmd_clean_install} {cmd_dploy_pl_pom} {cmd_deploy_root_pom} " # 替换为你要执行的命令
process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
is_dep_err = False
while True:
output = process.stdout.readline()
if output == '' and process.poll() is not None:
break
if output:
out_line = output.strip()
print(out_line)
artifacts_list = extract_artifacts(out_line)
if artifacts_list and len(artifacts_list) > 0:
is_dep_err = True
failed_dep_infos.append(failed_dep_info)
for artifact in artifacts_list:
print(f"获取依赖 {artifact} 失败")
failed_dep_infos.append(artifact)
if process.returncode != 0:# {os.path.join({work_dir},{project_dir})}
print(f"执行 {failed_dep} 编译程序失败，执行依赖构建脚本，脚本：{command}")
# subprocess.Popen("ls .", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
if not is_dep_err:
print(f'执行 {failed_dep} 失败，需要处理编译问题')
exit(0)
else:
print(f"执行 {failed_dep} 编译程序成功")
success_deps[failed_dep] = True
if len(failed_dep_infos) > 0:
print(f'继续编译...')