当前位置:   article > 正文

python里运行linux命令_python 代码里 加入linux命令

python 代码里 加入linux命令

使用版本:python3.9,centos7.8

python里提供了os模块,可以让我在代码中对linux发送命令,并获取返回值。

这里会讲到两种方法去使用os模块----os.popen和os.system

一.  os.system

os.system('命令字符串')。这个命令适用于一些你只需要知道结果是否成功的命令,因为它只会返回给你一个int类型的数字,0代表成功,非0代表失败。而且当你的命令字符串是一个子进程命令的话,他的返回值并不准确,他不会等待子进程在后台运行结束,只要子进程能够进入运行状态,他就会返回0。

二.  os.popen

在我看来,这个方法在使用子进程做并行或者在获取返回结果上,是很好用的,首先它一共有3个参数,

cmd--命令

mode--方式(默认值为‘r’,可选值为‘w’,一般场景中,r的使用比较普遍)

buffering--缓存值,默认为-1

该方法调用了subprocess里的方法,并最后帮助你关闭了资源,一条龙服务还是很人性化的。所以一般使用时,传一个命令即可,然后它提供了read(),readlines()给你去拿到返回值

即os.popen('cmd').read()

而且笔者认为它强大之处在于,当cmd是一个后台进程,即在命令最后加入&符号的进程,它依旧会等待进程结束并拿到返回值,我也用他帮我将原本需要5分钟+的程序中,加上使用子进程并行将时间缩短到20秒左右。

话不多说,分享实战经验,一个读取jenkins信息并写入时序数据库的流程:

获取相关信息,并发送请求方法:

  1. if action == 'jenkins':
  2. # 获取主机地址
  3. host = request.get_host()
  4. response_data = {'kind': 'JenkinsInfo.get', 'msg': 'jenkins任务构建信息入库', 'status': 200, 'param': ''}
  5. data = request.GET
  6. jenkins_name = data.get('jenkins_name', '')
  7. if jenkins_name in JENKINS.keys():
  8. jenkins_info = JENKINS[jenkins_name]
  9. else:
  10. response_data.update({
  11. 'param': '参数无效',
  12. 'status': 403
  13. })
  14. return JsonResponse(response_data, status=response_data['status'])
  15. # 连接jenkins
  16. server = self.to_jenkins(jenkins_info)
  17. all_jobs = server.get_jobs()
  18. job_name_list = [x.get('name') for x in all_jobs]
  19. cmdb_job = {}
  20. # 获取cmdb中的app_name
  21. cmdb_apps = AppBasisNew.objects.filter(**{'status': '已上线'}).values('app_name')
  22. app_name_list = [x.get('app_name') for x in cmdb_apps]
  23. # 判断cmdb中app是否在jenkins中
  24. env_list = ['archive-PRE-', 'PRE-', 'bak-PRE-', 'ARCHIVE-PRE-', 'BAK-PRE-']
  25. if app_name_list and job_name_list:
  26. for app_name in app_name_list:
  27. for env in env_list:
  28. if env+app_name in job_name_list:
  29. cmdb_job[env+app_name] = app_name
  30. if cmdb_job:
  31. job_str = str(cmdb_job).replace("{'", "").replace("'}", "").replace("', '", " ").replace("': '", "\|")
  32. res = os.popen(f"for job in {job_str}; do curl -G http://{host}/jenkins/pre/to_db?"f"jenkins_name={jenkins_name}\&job_name=$job & done").read()
  33. res_str = res.replace('}{', '}, {').join('[]')
  34. res_json = json.loads(res_str)
  35. response_data.update({
  36. 'param': res_json
  37. })
  38. else:
  39. response_data.update({
  40. 'param': 'jenkins与cmdb无匹配数据'
  41. })
  42. return JsonResponse(response_data, status=response_data['status'])

接受请求,返回相应:

  1. if action == 'to_db':
  2. data = request.GET
  3. jenkins_name = data['jenkins_name']
  4. job_name = data['job_name'].split('|')[0]
  5. project_name = data['job_name'].split('|')[1]
  6. server = self.to_jenkins(JENKINS[jenkins_name])
  7. build_info_list = []
  8. # 最新jenkins项目的构建时间
  9. last_timestamp = 0
  10. # 连接influxdb
  11. db_client = self.to_influxdb()
  12. try:
  13. job_info = server.get_job_info(job_name)
  14. except jenkins.JenkinsException:
  15. db_client.close()
  16. return JsonResponse({f'{jenkins_name}': f'{job_name} fail'})
  17. if job_info.get('builds'):
  18. # 获取该工程名每次构筑的时间:
  19. res = db_client.query(f"select max(build_time) from pre_job_info "
  20. f"where project_name::tag='{project_name}' and from_jenkins='{jenkins_name}'")
  21. last_time = 0
  22. if res:
  23. last_time = [x[u'max'] for x in res.get_points()][0]
  24. for job_build in range(1, len(job_info['builds'])+1):
  25. job_time = job_info['builds'][-job_build]['number']
  26. try:
  27. build_info = server.get_build_info(job_name, job_time)
  28. except jenkins.JenkinsException:
  29. db_client.close()
  30. return JsonResponse({f'{jenkins_name}': f'{job_name} fail'})
  31. if build_info.get('timestamp') // 1000 <= last_time:
  32. break
  33. if job_build == 1:
  34. last_timestamp = build_info.get('timestamp')//1000
  35. build_info_dict = ToDict().return_dic(build_info)
  36. # 控制台输出
  37. build_output = server.get_build_console_output(job_name, job_time)
  38. build_out_list = build_output.split('\n')
  39. build_agent_name = 'null'
  40. project_path = 'null'
  41. if jenkins_name == 'JENKINS_PRE1' or jenkins_name == 'JENKINS_PRE3':
  42. res_info = ''
  43. for i in build_out_list:
  44. if 'Running on' in i:
  45. res_info = i
  46. if res_info:
  47. build_agent_name = res_info.split()[-3]
  48. project_path = res_info.split()[-1]
  49. if jenkins_name == 'JENKINS_PRE2':
  50. project_path = build_out_list[1].split()[-1]
  51. build_agent_name = build_info_dict.get('builtOn')
  52. if job_info.get('lastStableBuild'):
  53. last_stable_build = job_info['lastStableBuild']['number']
  54. else:
  55. last_stable_build = 0
  56. if job_info.get('lastSuccessfulBuild'):
  57. last_successful_build = job_info['lastSuccessfulBuild']['number']
  58. else:
  59. last_successful_build = 0
  60. if build_info_dict.get('actions_lastBuiltRevision_branch_name'):
  61. build_branch_name = build_info_dict.get(
  62. 'actions_lastBuiltRevision_branch_name')
  63. code_version = build_info_dict.get(
  64. 'actions_lastBuiltRevision_branch_SHA1')
  65. else:
  66. build_branch_name = build_info_dict.get('changeSet_revisions_module', 'null')
  67. code_version = str(build_info_dict.get('changeSet_revisions_revision', 'null'))
  68. if build_info_dict.get('result') == 'SUCCESS':
  69. build_successful = True
  70. else:
  71. build_successful = False
  72. if build_info_dict.get('actions_parameters_name1') == 'Fabu_Type':
  73. fabu_type = build_info_dict.get('actions_parameters_value1')
  74. else:
  75. fabu_type = build_info_dict.get('actions_parameters_value2', 'all-in-one')
  76. res = {'measurement': 'pre_job_info',
  77. 'tags': {
  78. 'project_name': project_name,
  79. 'build_result': build_info_dict.get('result'),
  80. 'project_path': project_path
  81. },
  82. 'time': datetime.fromtimestamp(build_info_dict.get('timestamp') // 1000),
  83. 'fields': {
  84. 'build_agent_name': build_agent_name,
  85. 'build_branch_name': build_branch_name,
  86. 'build_causer': build_info_dict.get('actions_causes_userName'),
  87. 'build_exec_time': build_info_dict.get('duration') // 1000,
  88. 'build_measured_time': int(datetime.now().timestamp()),
  89. 'build_number': build_info_dict.get('id'),
  90. 'build_result': build_info_dict.get('result'),
  91. 'build_result_ordinal': build_info_dict.get('queueId'),
  92. 'build_scheduled_time': build_info_dict.get('estimatedDuration') // 1000,
  93. 'build_status_message': build_info_dict.get('building'),
  94. 'build_successful': build_successful,
  95. 'build_time': build_info_dict.get('timestamp') // 1000,
  96. 'last_stable_build': last_stable_build,
  97. 'last_successful_build': last_successful_build,
  98. 'project_build_health': job_info['healthReport'][0]['score'],
  99. 'project_name': project_name,
  100. 'project_path': project_path,
  101. 'Code_Version': code_version,
  102. 'Fabu_Type': fabu_type,
  103. 'from_jenkins': jenkins_name
  104. },
  105. }
  106. build_info_list.append(res)
  107. if build_info_list:
  108. db_client.write_points(build_info_list)
  109. else:
  110. db_client.close()
  111. return JsonResponse({f'{jenkins_name}': f'{job_name} has no new data'})
  112. # 数据入库验证
  113. new_res = db_client.query(f"select build_number from pre_job_info "
  114. f"where project_name::tag='{project_name}' and build_time={last_timestamp}")
  115. if new_res:
  116. db_client.close()
  117. return JsonResponse({f'{jenkins_name}': f'{job_name} success'})
  118. db_client.close()
  119. return JsonResponse({f'{jenkins_name}': f'{job_name} fail'})

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/从前慢现在也慢/article/detail/72359
推荐阅读
相关标签
  

闽ICP备14008679号