python调用subprocess模块实现命令行操作控制SVN的方法
更新时间：2022年09月08日 10:52:43 作者：Logintern09
这篇文章主要介绍了使用python的subprocess模块实现对SVN的相关操作，通过设置GitSvn类，在该类下自定义执行SVN常规操作的方法，需要的朋友跟随小编一起看看吧
使用python的subprocess模块实现对SVN的相关操作。
设置GitSvn类，在该类下自定义执行SVN常规操作的方法。
SVN的常规操作包括：
（1）获取SVN当前版本，通过方法get_version()实现；
（2）下载SVN指定仓库，通过方法download()实现，实际是通过调用SVN的命令行操作指令svn checkout实现的下载整个仓库功能；
（3）获取SVN某个仓库下的所有文件列表，通过方法search_dir()实现，实际是通过调用SVN的命令行操作指令svn list实现的获取仓库文件列表功能；
（4）在SVN指定位置创建新的文件夹，通过方法mkdir_command()实现，实际是通过调用SVN的命令行操作指令svn mkdir实现的创建文件夹或者目录功能；
（5）将本地仓库文件添加到SVN仓库，通过方法upload_file()实现，实际是通过调用SVN的命令行操作指令svn add实现的添加文件功能；
（6）将本地仓库已添加的文件提交SVN指定仓库，通过方法commit_command()实现，实际是通过调用SVN的命令行操作指令svn commit实现的提交文件功能；
（7）删除SVN仓库的目录，通过方法delete_url()实现，实际是通过调用SVN的命令行操作指令svn delete实现的删除目录功能；
（8）锁定SVN仓库的文件，通过方法lock_file()实现，实际是通过调用SVN的命令行操作指令svn lock实现的锁定文件功能；
（9）将SVN仓库的文件解除锁定，通过方法unlock_file()实现，实际是通过调用SVN的命令行操作指令svn unlock实现的解除文件锁定功能；
（10）查看SVN仓库文件当前状态，通过方法check_status()实现，实际是通过调用SVN的命令行操作指令svn status实现的查看文件状态功能；
（11）更新SVN仓库的某个文件，通过方法update_file()实现，实际是通过调用SVN的命令行操作指令svn up实现的更新文件功能；
GitSvn类定义如下：
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Author: Logintern09
import os
import time
import subprocess
from log_manage import logger
rq = time.strftime('%Y%m%d', time.localtime(time.time()))
file_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "temp_files")
tempfile = os.path.join(file_path, rq + '.log')
class GitSvn(object):
def __init__(self, bill_addr):
self.bill_addr = bill_addr
def get_version(self):
cmd = "svn info %s" % self.bill_addr
logger.info(cmd)
output = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(result, error) = output.communicate()
result = result.decode(encoding="gbk")
logger.info(result)
error = error.decode(encoding="gbk")
logger.error(error)
with open(tempfile, "w") as f:
f.write(result)
if error.strip() != "":
if error.startswith("svn: E170013:"):
res = "网络异常，暂时连接不上系统，请检查网络或其他配置！"
raise SystemError(res)
def download(self, dir_path):
with open(tempfile, "r") as f:
result = f.readlines()
for line in result:
if line.startswith("Revision"):
laster_version = int(line.split(":")[-1].strip())
logger.info(line)
break
cur_path = os.path.dirname(os.path.realpath(__file__))
id_path = os.path.join(cur_path, default_sys_name)
cmd = "svn checkout %s %s -r r%s" % (dir_path, id_path, laster_version)
logger.info(cmd)
output = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(result, error) = output.communicate()
result = result.decode(encoding="gbk")
error = error.decode(encoding="gbk")
logger.info(result)
logger.error(error)
if error.strip() != "":
if error.startswith("svn: E170013:"):
res = "网络异常，暂时连接不上系统，请检查网络或其他配置！"
raise SystemError(res)
if (error.startswith("svn: E155004:")) or (error.startswith("svn: E155037:")):
cur_path = os.path.dirname(os.path.realpath(__file__))
file_path = os.path.join(cur_path, default_sys_name)
cmd = "svn cleanup"
logger.info(cmd)
output = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
cwd=file_path)
(result, error) = output.communicate()
result = result.decode(encoding="gbk")
error = error.decode(encoding="gbk")
logger.info(result)
logger.error(error)
def search_dir(self):
cmd = "svn list %s" % self.bill_addr
logger.info(cmd)
output = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(result, error) = output.communicate()
result = result.decode(encoding="gbk")
error = error.decode(encoding="gbk")
logger.info(result)
logger.error(error)
with open(tempfile, "w") as f:
f.write(result)
if error.strip() != "":
if error.startswith("svn: E170013:"):
res = "网络异常，暂时连接不上系统，请检查网络或其他配置！"
raise SystemError(res)
def mkdir_command(self, id_num):
cmd = "svn mkdir %s" % id_num
logger.info(cmd)
cur_path = os.path.dirname(os.path.realpath(__file__))
file_path = os.path.join(cur_path, default_sys_name)
output = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
cwd=file_path)
(result, error) = output.communicate()
result = result.decode(encoding="gbk")
error = error.decode(encoding="gbk")
logger.info(result)
logger.error(error)
if error.strip() != "":
if error.startswith("svn: E170013:"):
res = "网络异常，暂时连接不上系统，请检查网络或其他配置！"
raise SystemError(res)
def upload_file(self, file_path):
cmd = "svn add %s --force" % file_path
root_path, file_name = os.path.split(file_path)
logger.info(cmd)
output = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
cwd=root_path)
(result, error) = output.communicate()
result = result.decode(encoding="gbk")
error = error.decode(encoding="gbk")
logger.info(result)
logger.error(error)
if error.strip() != "":
if error.startswith("svn: E170013:"):
res = "网络异常，暂时连接不上系统，请检查网络或其他配置！"
raise SystemError(res)
def commit_command(self, file):
# 执行了锁定的用户执行了提交操作（提交操作将自动解锁）
if os.path.isfile(file):
file_path, file_name = os.path.split(file)
else:
file_path = file
cmd = "svn commit %s -m 'update_files'" % file
logger.info(cmd)
output = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
cwd=file_path)
(result, error) = output.communicate()
result = result.decode(encoding="gbk")
error = error.decode(encoding="gbk")
logger.info(result)
logger.error(error)
if error.strip() != "":
if error.startswith("svn: E170013:"):
res = "网络异常，暂时连接不上系统，请检查网络或其他配置！"
raise SystemError(res)
elif error.startswith("svn: E730053:"):
res = "数据上传失败，请重新提交数据！！！"
raise SystemError(res)
else:
res = "数据上传失败，请重新提交数据！！！"
raise SystemError(res)
def delete_url(self, url):
cmd = "svn delete %s" % url
logger.info(cmd)
output = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(result, error) = output.communicate()
result = result.decode(encoding="gbk")
error = error.decode(encoding="gbk")
logger.info(result)
logger.error(error)
if error.strip() != "":
if error.startswith("svn: E170013:"):
res = "网络异常，暂时连接不上系统，请检查网络或其他配置！"
raise SystemError(res)
def lock_file(self, file_name):
cur_path = os.path.dirname(os.path.realpath(__file__))
id_path = os.path.join(cur_path, default_sys_name)
file_path = os.path.join(id_path, file_name)
cmd = "svn lock %s" % file_path
logger.info(cmd)
output = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(result, error) = output.communicate()
result = result.decode(encoding="gbk")
error = error.decode(encoding="gbk")
logger.info(result)
logger.error(error)
if error.strip() != "":
if error.startswith("svn: E170013:"):
res = "网络异常，暂时连接不上系统，请检查网络或其他配置！"
raise SystemError(res)
elif error.startswith("svn: warning: W160042:"):
res = "系统资源已被其他用户锁定，请稍后重试！"
raise SystemError(res)
def unlock_file(self, file_name):
# 不使用--force 参数 可以解锁被自己锁定的文集 即普通的release lock
cur_path = os.path.dirname(os.path.realpath(__file__))
id_path = os.path.join(cur_path, default_sys_name)
file_path = os.path.join(id_path, file_name)
cmd = "svn unlock %s --force" % file_path
logger.info(cmd)
output = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(result, error) = output.communicate()
result = result.decode(encoding="gbk")
error = error.decode(encoding="gbk")
logger.info(result)
logger.error(error)
if error.strip() != "":
if error.startswith("svn: E170013:"):
res = "网络异常，暂时连接不上系统，请检查网络或其他配置！"
raise SystemError(res)
def check_status(self, file_name):
# ?：不在svn的控制中；M：内容被修改；C：发生冲突；A：预定加入到版本库；K：被锁定
cur_path = os.path.dirname(os.path.realpath(__file__))
id_path = os.path.join(cur_path, default_sys_name)
file_path = os.path.join(id_path, file_name)
cmd = "svn status -v %s" % file_path
logger.info(cmd)
output = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(result, error) = output.communicate()
result = result.decode(encoding="gbk")
error = error.decode(encoding="gbk")
logger.info(result)
logger.error(error)
# 获取状态结果解析
status_list = result.split(" ")
status_flag_list = []
for i in status_list:
if i.isalpha():
status_flag_list.append(i)
else:
break
if result.strip() != "":
if error.startswith("svn: E170013:"):
res = "网络异常，暂时连接不上系统，请检查网络或其他配置！"
raise SystemError(res)
elif "C" in status_flag_list:
return "C"
elif "K" in status_flag_list:
return "K"
def update_file(self, file_name):
cur_path = os.path.dirname(os.path.realpath(__file__))
id_path = os.path.join(cur_path, default_sys_name)
file_path = os.path.join(id_path, file_name)
cmd = "svn up %s" % file_path
logger.info(cmd)
output = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(result, error) = output.communicate()
result = result.decode(encoding="gbk")
error = error.decode(encoding="gbk")
logger.info(result)
logger.error(error)
with open(tempfile, "w") as f:
f.write(result)
with open(tempfile, "r") as f:
result = f.readlines()
count = -1
for line in result:
count += 1
if (line.strip() != "") and (count == 1):
if line.startswith("C"):
res = "更新系统资源时发现存在冲突，请稍后重试！"
raise SystemError(res)
if error.strip() != "":
if error.startswith("svn: E170013:"):
res = "网络异常，暂时连接不上系统，请检查网络或其他配置！"
raise SystemError(res)
elif (error.startswith("svn: E155037:")) or (error.startswith("svn: E155004:")):
cur_path = os.path.dirname(os.path.realpath(__file__))
file_path = os.path.join(cur_path, default_sys_name)
cmd = "svn cleanup"
logger.info(cmd)
output = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
cwd=file_path)
(result, error) = output.communicate()
result = result.decode(encoding="gbk")
error = error.decode(encoding="gbk")
logger.info(result)
logger.error(error)
上述类GitSvn的应用实例如下：
if __name__ == '__main__':
default_url = svn:*** # 用户本地SVN客户端的URL地址
git_class = GitSvn(default_url)
git_class.get_version()
git_class.download()
# 验证查看目录文件列表功能
git_class.search_dir()
# 验证删除目录功能
cur_path = os.path.dirname(os.path.realpath(__file__))
file_path = os.path.join(cur_path, default_sys_name)
id_path = os.path.join(file_path, 'SCR202202100002')
git_class.delete_url(id_path)
# 验证文件加锁功能
git_class.lock_file(file_name="单号总表.xlsx")
# 验证文件解锁功能
git_class.unlock_file(file_name="单号总表.xlsx")
# 检查文件状态
git_class.check_status(file_name="单号总表.xlsx")
# 验证创建目录功能
git_class.mkdir_command("SCR202203280001")
# 验证提交文件功能
cur_path = os.path.dirname(os.path.realpath(__file__))
sys_path = os.path.join(cur_path, default_sys_name)
target_dir = os.path.join(sys_path, "SCR202203280001")
git_class.upload_file(target_dir)
git_class.commit_command(target_dir)
到此这篇关于python调用subprocess模块实现命令行操作控制SVN的文章就介绍到这了,更多相关python subprocess模块内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
这篇文章主要介绍了Python调用C语言开发的共享库方法实例,本文同时给出了C语言和Python调用简单实例,需要的朋友可以参考下2015-03-03
本篇文章主要介绍了Sublime开发python程序的示例代码，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧2018-01-01
Python3实现将文件树中所有文件和子目录归档到tar压缩文件的方法
这篇文章主要介绍了Python3实现将文件树中所有文件和子目录归档到tar压缩文件的方法,涉及Python3使用tarfile模块实现tar压缩文件的技巧,需要的朋友可以参考下2015-05-05
这篇文章主要介绍了python中 * 的用法详解，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2019-07-07
事件处理，是 GUI 程序中不可或缺的重要组成部分，相比来说，控件只是组成一台机器的零部件。本文我们将对 Tkinter 中的事件处理机制做详细的介绍，需要的可以参考一下2022-01-01
这篇文章主要为大家详细介绍了python微信跳一跳系列之棋子定位之像素遍历，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2018-02-02
下面小编就为大家分享一篇python生成不重复随机数和对list乱序的解决方法，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2018-04-04
今天我们先讲一下编写python脚本处理json的核心功能，有些散乱，后期在进行整体脚本的编写,对python json字典关系相关知识感兴趣的朋友一起看看吧2022-06-06
这篇文章主要为大家详细介绍了python实现碑帖图片横向拼接，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2021-03-03
这篇文章主要介绍了python爬虫入门教程之点点美女图片爬虫代码分享,本文以采集抓取点点网美女图片为例,需要的朋友可以参考下2014-09-09
最新评论