京东商智_Python和Shell脚本中执行CMD和HTTP命令

发表于 | 分类于

京东商智_Python和Shell脚本中执行CMD和HTTP命令

1.背景

当我们无法远程登录到目标节点时，我们需要将想要执行的逻辑写成代码并打包上传到目标节点，然后指定代码包，最常见的就是Shell脚本和Python脚本。CMD命令是所有操作系统都支持的语言，也就不需要安装编译器、执行器，整个操作系统都是CMD命令的上下文环境。Shell脚本本身就是一批CMD命令组成的，直接使用sh命令即可执行；Python脚本中利用Python语法特性可以方便快捷实现很多逻辑，同时又经常需要调用CMD命令与操作系统交互实现指定目的，所以在Python脚本中熟练使用CMD命令进行交互非常重要。

2.shell脚本执行CMD使用案例

如下脚本业务逻辑为获取ClickHouse集群节点列表，然后在每个节点上执行计数SQL，查看每个分片副本之间的数据差异：

1
2
3
4
5
6
7
ips=(`curl -u rw_sz_merchant_flow_slave:xD0t7iu6Qva7KMcGyK-_ 'http://ckpub202.olap.jd.com:2000/?' -d "select host_name from (select shard_num,concat(host_name,':',(case when port=9600 then '8623' when port=9700 then '8723' when port=9800 then '8823' when port=9900 then '8923' else '8123' end)) as host_name from system.clusters where cluster='LF02_CK_Pub_202' order by shard_num asc) "`);
for i in "${ips[@]}"
do echo "=====$i =========";
curl -u rw_sz_merchant_flow_slave:xD0t7iu6Qva7KMcGyK-_ 'http://'$i'/?' -d "select count(1) from sz.app_jdr_traffic_sz_heatmap_click_mvp_i_d_d where dt = '2023-12-31'"
#curl -u rw_sz_merchant_flow_slave:xD0t7iu6Qva7KMcGyK-_ 'http://'$i'/?' -d "drop table sz.app_jdr_traffic_sz_heatmap_click_mvp_i_d_d_d"
echo "=========$res==============="
done

使用sh命令即可运行此脚本：

1
sh ./{脚本所在目录}/{脚本名}.sh

3.Python脚本执行CMD使用案例

3.1 Python脚本通过subprocess.Popen()函数执行CMD命令

Python在subprocess模块中封装好了Popen对象用于创建和管理子进程，我们常用该对象来执行CMD命令。另外在数据开发脚本中常在subprocess.Popen()外再封装一些对返回结果的标准处理流程，作为工具方法对外提供使用，如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
# -*- coding: utf-8 -*-
import os
import threading
import subprocess

def run_shell_cmd(self, shellcmd, encoding='utf8'):
    res = subprocess.Popen(shellcmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    results = []
    while True:
        line = res.stdout.readline().decode(encoding).strip()
        if line == '' and res.poll() is not None:
            break
        else:
            results.append(line)
            self.log.info(line)
    ReturnCode = res.returncode
    return [ReturnCode, '\n'.join(results)]

  
# run_shell_cmd_err 执行脚本，丢弃标准流，返回错误流
# 标准输出指定为dev/null，标准流的日志通过log正常输出
# return结果只包含日志内容
def run_shell_cmd_err(self, shellcmd, encoding='utf8'):
    devnull = open(os.devnull, "wb")
    res = subprocess.Popen(shellcmd, shell=True, stdout=devnull, stderr= subprocess.PIPE)
    results = []
    while True:
        line = res.stderr.readline().decode(encoding).strip()
        if line == '' and res.poll() is not None:
            break
        else:
            results.append(line)
            self.log.info(line)
    ReturnCode = res.returncode
    return [ReturnCode, '\n'.join(results)]

  
# run_shell_cmd_out 执行脚本，标准流和错误流都在log输出一份，另外标准流在out也输出一份
# return结果中只包含脚本执行结果
def run_shell_cmd_out(self, shellcmd, encoding='utf8'):
    res = subprocess.Popen(shellcmd, shell=True, stdout= subprocess.PIPE, stderr= subprocess.PIPE)
    task = threading.Thread(target=self.__process_output, args=(res, encoding)) 
    task.daemon = True
    task.start()
    results = []
    while True:
        line = res.stdout.readline().decode(encoding).strip()
        if line == '' and res.poll() is not None:
            break
        else:
            results.append(line)
            self.log.info(line)
            print(line)
    ReturnCode = res.returncode
    return [ReturnCode, '\n'.join(results)]

3.2 Python脚本通过CMD命令连接ClickHouse集群执行SQL并获取解析结果

如下脚本实现根据ClickHouse分布式表名获取到分布式表建表语句，然后获取SQL执行结果解析出对应本地表名：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
"""
# 根据源分布式表名获取源本地表名
"""
def get_local_table_name(self):
    sql = """SELECT create_table_query FROM system.tables WHERE name = '%s' AND database = '%s';""" % (self.ck_table, self.ck_database)
    cmd = """curl -u %s:%s 'http://%s:%s/?' -d "%s" """ % (self.ck_user, self.ck_password, self.ck_host, self.ck_port, sql)
    p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)
    table_create_sql, _ = p.communicate()
    print("分布式表建表语句如下：")
    print(table_create_sql.decode("utf8").replace("\\'", "'"))
    pattern = r"Distributed\('\w+', '\w+', '(\w+)',"
    match = re.search(pattern, result.decode("utf8").replace("\\'", "'"))
    if match:
        # 从分组中得到第三个参数，即匹配括号内第三个'(\w+)'
        local_table_name = match.group(1)
        print("获取到源表本地表名: " + local_table_name)
        return local_table_name
    else:
        print("获取源表本地表名失败.")
        sys.exit(1)

3.3 subprocess.Popen()函数执行结果输出方式p.stdout与p.communicate()与p.wait()的区别

Popen.stdout

创建Popen实例时，如果设置了参数stdout=subprocess.PIPE，那么Popen.stdout将是一个指向子进程标准输出的文件类似对象。你可以直接使用它的read()或readline()等方法来读取子进程的输出。但是，这种方式可能会导致死锁，因为它不处理子进程的标准错误输出。如果子进程产生了足够多的错误输出填满系统的管道缓冲区，它可能会因为没有读取标准错误输出而被挂起。

示例：

1
2
3
4
import subprocess

p = subprocess.Popen(['command'], stdout=subprocess.PIPE)
output = p.stdout.read()

Popen.communicate()

Popen.communicate()方法是与子进程交互的推荐方式。它允许你发送数据给子进程的标准输入，并且读取标准输出和标准错误输出，直到子进程完成。communicate()方法会等待子进程结束，并返回一个元组，通常为(stdout_data, stderr_data)。调用communicate()还可以帮助避免死锁，因为它会同时读取标准输出和标准错误流，这在处理可能同时生成标准输出和标准错误输出的子进程时非常重要。

示例：

1
2
3
4
import subprocess

p = subprocess.Popen(['command'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout_data, stderr_data = p.communicate()

Popen.wait()

p.wait() 方法同样会阻塞当前进程，直到子进程结束，然后它会返回子进程的退出码，返回值为0表示子进程成功结束。通常用于不关心子进程的输出，只想要知道子进程何时结束和成功与否的场合。

1
2
3
4
import subprocess 

p = subprocess.Popen(['command'], stdout=subprocess.PIPE)
returncode = p.wait()

总结

1）p.stdout直接读取子进程的标准输出流；而p.communicate()会读取子进程的标准输出和标准错误输出，从而避免了潜在的死锁问题。

2）p.stdout.read()可能需要多次调用来获取所有输出，特别是对于阻塞调用；而p.communicate()只需调用一次，它会等待直到有所有输出产生。

3）p.communicate()提供了一种发送输入和处理输出的同步方式，而p.stdout需要结合其他手动处理错误输出或额外的多线程/多进程逻辑。

3.4 subprocess.Popen()函数入参stderr=subprocess.PIPE与stderr=subprocess.STDOUT的区别

stderr=subprocess.PIPE

当stderr入参为stderr=subprocess.PIPE时，创建了一个新的管道，Python程序可以通过这个管道读取子进程的标准错误输出，stdout入参stderr=subprocess.PIPE同理。

示例：

1
2
3
4
5
6
import subprocess

p = subprocess.Popen(["command"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output, errput = p.communicate()
print(output.decode("utf8"))
print(errput.decode("utf8"))

stderr=subprocess.STDOUT

stderr入参设置为stderr=subprocess.STDOUT表示子进程的标准错误输出被重定向到子进程的标准输出。使用该设置可以合并标准输出流和标准错误流。在该设置的情况下，不能独立捕获标准输出和标准错误，它们将会在同一个数据流中混杂在一起。这种方式在不关心错误信息和输出之间区别，只想获取所有输出信息时很有用。

示例：

1
2
3
4
5
import subprocess

p = subprocess.Popen(["command"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
output, _ = p.communicate()
print(output.decode("utf8"))

4.shell脚本执行HTTP方法

curl（Command Line URL viewer）是一个用来请求web服务器的控制台命令，用来模拟客户端向服务器发送http请求，获取响应消息。作用与postman等图形界面工具一样，如果熟练的话，完全可以取代Postman这一类的图形界面工具。

初学者教程与用法指南

不带有任何参数时，curl就是发出GET请求。

1
$ curl https://www.example.com

上面命令向www.example.com发出GET请求，服务器返回的内容会在命令行输出。

-A

-A参数指定客户端的用户代理标头，即User-Agent。curl 的默认用户代理字符串是curl/[version]。

1
$ curl -A 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/76.0.3809.100 Safari/537.36' https://google.com

上面命令将User-Agent改成 Chrome 浏览器。

1
$ curl -A '' https://google.com

上面命令会移除User-Agent标头。

也可以通过-H参数直接指定标头，更改User-Agent。

1
$ curl -H 'User-Agent: php/1.0' https://google.com

-b

-b参数用来向服务器发送 Cookie。

1
$ curl -b 'foo=bar' https://google.com

上面命令会生成一个标头Cookie: foo=bar，向服务器发送一个名为foo、值为bar的 Cookie。

1
$ curl -b 'foo1=bar;foo2=bar2' https://google.com

上面命令发送两个 Cookie。

1
$ curl -b cookies.txt https://www.google.com

上面命令读取本地文件cookies.txt，里面是服务器设置的 Cookie（参见-c参数），将其发送到服务器。

-c

-c参数将服务器设置的 Cookie 写入一个文件。

1
$ curl -c cookies.txt https://www.google.com

上面命令将服务器的 HTTP 回应所设置 Cookie 写入文本文件cookies.txt。

-d

-d参数用于发送 POST 请求的数据体。

1
2
3
$ curl -d'login=emma＆password=123'-X POST https://google.com/login
# 或者
$ curl -d 'login=emma' -d 'password=123' -X POST  https://google.com/login

使用-d参数以后，HTTP 请求会自动加上标头Content-Type : application/x-www-form-urlencoded。并且会自动将请求转为 POST 方法，因此可以省略-X POST。

-d参数可以读取本地文本文件的数据，向服务器发送。

1
$ curl -d '@data.txt' https://google.com/login

上面命令读取data.txt文件的内容，作为数据体向服务器发送。

–data-urlencode

--data-urlencode参数等同于-d，发送 POST 请求的数据体，区别在于会自动将发送的数据进行 URL 编码。

1
$ curl --data-urlencode 'comment=hello world' https://google.com/login

上面代码中，发送的数据hello world之间有一个空格，需要进行 URL 编码。

-e

-e参数用来设置 HTTP 的标头Referer，表示请求的来源。

1
curl -e 'https://google.com?q=example' https://www.example.com

上面命令将Referer标头设为https://google.com?q=example。

-H参数可以通过直接添加标头Referer，达到同样效果。

1
curl -H 'Referer: https://google.com?q=example' https://www.example.com

-F

-F参数用来向服务器上传二进制文件。

1
$ curl -F 'file=@photo.png' https://google.com/profile

上面命令会给 HTTP 请求加上标头Content-Type: multipart/form-data，然后将文件photo.png作为file字段上传。

-F参数可以指定 MIME 类型。

1
$ curl -F 'file=@photo.png;type=image/png' https://google.com/profile

上面命令指定 MIME 类型为image/png，否则 curl 会把 MIME 类型设为application/octet-stream。

-F参数也可以指定文件名。

1
$ curl -F 'file=@photo.png;filename=me.png' https://google.com/profile

上面命令中，原始文件名为photo.png，但是服务器接收到的文件名为me.png。

-G

-G参数用来构造 URL 的查询字符串。

1
$ curl -G -d 'q=kitties' -d 'count=20' https://google.com/search

上面命令会发出一个 GET 请求，实际请求的 URL 为https://google.com/search?q=kitties&count=20。如果省略--G，会发出一个 POST 请求。

如果数据需要 URL 编码，可以结合--data--urlencode参数。

1
$ curl -G --data-urlencode 'comment=hello world' https://www.example.com

-H

-H参数添加 HTTP 请求的标头。

1
$ curl -H 'Accept-Language: en-US' https://google.com

上面命令添加 HTTP 标头Accept-Language: en-US。

1
$ curl -H 'Accept-Language: en-US' -H 'Secret-Message: xyzzy' https://google.com

上面命令添加两个 HTTP 标头。

1
$ curl -d '{"login": "emma", "pass": "123"}' -H 'Content-Type: application/json' https://google.com/login

上面命令添加 HTTP 请求的标头是Content-Type: application/json，然后用-d参数发送 JSON 数据。

-i

-i参数打印出服务器回应的 HTTP 标头。

1
$ curl -i https://www.example.com

上面命令收到服务器回应后，先输出服务器回应的标头，然后空一行，再输出网页的源码。

-I

-I参数向服务器发出 HEAD 请求，然会将服务器返回的 HTTP 标头打印出来。

1
$ curl -I https://www.example.com

上面命令输出服务器对 HEAD 请求的回应。

--head参数等同于-I。

1
$ curl --head https://www.example.com

-k

-k参数指定跳过 SSL 检测。

1
$ curl -k https://www.example.com

上面命令不会检查服务器的 SSL 证书是否正确。

-L

-L参数会让 HTTP 请求跟随服务器的重定向。curl 默认不跟随重定向。

1
$ curl -L -d 'tweet=hi' https://api.twitter.com/tweet

–limit-rate

--limit-rate用来限制 HTTP 请求和回应的带宽，模拟慢网速的环境。

1
$ curl --limit-rate 200k https://google.com

上面命令将带宽限制在每秒 200K 字节。

-o

-o参数将服务器的回应保存成文件，等同于wget命令。

1
$ curl -o example.html https://www.example.com

上面命令将www.example.com保存成example.html。

-O

-O参数将服务器回应保存成文件，并将 URL 的最后部分当作文件名。

1
$ curl -O https://www.example.com/foo/bar.html

上面命令将服务器回应保存成文件，文件名为bar.html。

-s

-s参数将不输出错误和进度信息。

1
$ curl -s https://www.example.com

上面命令一旦发生错误，不会显示错误信息。不发生错误的话，会正常显示运行结果。

如果想让 curl 不产生任何输出，可以使用下面的命令。

1
$ curl -s -o /dev/null https://google.com

-S

-S参数指定只输出错误信息，通常与-s一起使用。

1
$ curl -s -o /dev/null https://google.com

上面命令没有任何输出，除非发生错误。

-u

-u参数用来设置服务器认证的用户名和密码。

1
$ curl -u 'bob:12345' https://google.com/login

上面命令设置用户名为bob，密码为12345，然后将其转为 HTTP 标头Authorization: Basic Ym9iOjEyMzQ1。

curl 能够识别 URL 里面的用户名和密码。

1
$ curl https://bob:12345@google.com/login

上面命令能够识别 URL 里面的用户名和密码，将其转为上个例子里面的 HTTP 标头。

1
$ curl -u 'bob' https://google.com/login

上面命令只设置了用户名，执行后，curl 会提示用户输入密码。

-v

-v参数输出通信的整个过程，用于调试。

1
$ curl -v https://www.example.com

--trace参数也可以用于调试，还会输出原始的二进制数据。

1
$ curl --trace - https://www.example.com

-x

-x参数指定 HTTP 请求的代理。

1
$ curl -x socks5://james:cats@myproxy.com:8080 https://www.example.com

上面命令指定 HTTP 请求通过myproxy.com:8080的 socks5 代理发出。

如果没有指定代理协议，默认为 HTTP。

1
$ curl -x james:cats@myproxy.com:8080 https://www.example.com

上面命令中，请求的代理使用 HTTP 协议。

-X

-X参数指定HTTP请求的方法。

1
$ curl -X POST https://www.example.com

上面命令对https://www.example.com发出 POST 请求。

5.Python脚本执行HTTP使用案例

python脚本中常用urllib.request.urlopen()函数来在脚本运行节点上提交http请求，如下是一个使用python脚本来请求ClickHouse集群的http请求示例，该http请求的作用给ClickHouse集群发送一条select或者insert语句，并得到执行结果。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import time
import os
import datetime
import subprocess
from urllib.request import urlopen,Request
import sys
from urllib import parse
import traceback

cmd = """ curl -u biz_ge:_GUDy-6fUteJTqlIwRRS 'http://ckpub97.olap.jd.com:2000/?' -d """
url = """http://ckpub97.olap.jd.com:2000/?user=biz_ge&password=_GUDy-6fUteJTqlIwRRS&query="""
sql1 = """SELECT * from pc_online.app_ge_d0103t04_plus_good_v2_new_all_imck2ck_4_tmp_d limit 100"""
sql2 = """insert into pc_online.app_ge_d0103t04_plus_good_v2_new_all_imck2ck_4_tmp_d (lvl_code,dt,cate_1,shop,tp,jdr_sch_trade_deal_ord_parent__ord_dis_qtty_trade_deal_snapshot_95discount_flagIplus_highfavor_sku,jdr_sch_user_deal_ord_sku_cnt_plus_user_deal_snapshot_95discount_flagIplus_highfavor_sku,jdr_sch_user_deal_ord_sku_cnt_plus_user_deal_snapshot_plus_user_promotion_flagIplus_highfavor_sku,jdr_sch_trade_deal_ord_parent__ord_dis_qtty_trade_deal_snapshot_95discount_flag,jdr_sch_user_deal_ord_sku_cnt_plus_user_deal_snapshot_95discount_flag,jdr_sch_user_deal_ord_sku_cnt_plus_user_deal_snapshot_plus_user_promotion_flag) SELECT
  '708bbb0f-a198-3867-8138-f4be58d21516' as lvl_code,
  '2024-04-09' ckdt,
   toUInt64OrZero(item_first_cate_cd) cate_1,
  toUInt64OrZero(shop_id) shop,
  'BY_WEEK_ACCU' tp,
  uniq(if(t.plus_score_flag = '1', t.discount_deal_parent_ord_qtty, null)) jdr_sch_trade_deal_ord_parent__ord_dis_qtty_trade_deal_snapshot_95discount_flagIplus_highfavor_sku,
  uniq(if(t.plus_score_flag = '1', t.discount_deal_user_qtty, null)) jdr_sch_user_deal_ord_sku_cnt_plus_user_deal_snapshot_95discount_flagIplus_highfavor_sku,
  uniq(if(t.plus_score_flag = '1', t.plus_deal_user_qtty, null)) jdr_sch_user_deal_ord_sku_cnt_plus_user_deal_snapshot_plus_user_promotion_flagIplus_highfavor_sku,
  uniq(t.discount_deal_parent_ord_qtty) jdr_sch_trade_deal_ord_parent__ord_dis_qtty_trade_deal_snapshot_95discount_flag,
  uniq(t.discount_deal_user_qtty) jdr_sch_user_deal_ord_sku_cnt_plus_user_deal_snapshot_95discount_flag,
  uniq(t.plus_deal_user_qtty) jdr_sch_user_deal_ord_sku_cnt_plus_user_deal_snapshot_plus_user_promotion_flag
FROM pc_online.app_ge_d0103t04_plus_good_v2_new_all t -- 定义驱动 PLUS专享商品交易明细表-copy
WHERE (
  (t.dt >= '2024-04-08')
  AND (t.dt <= '2024-04-09')
)
GROUP BY
   toUInt64OrZero(item_first_cate_cd),
  toUInt64OrZero(shop_id) settings max_execution_time = 600, max_memory_usage = 60000000000, max_bytes_before_external_group_by = 30000000000"""
try:
  encoded_sql = parse.quote(url + sql2.replace("\n", " "))
  req = Request(encoded_sql, data=parse.urlencode({}).encode('utf-8'))
  req.add_header('Content-Type', 'application/x-www-form-urlencoded; charset=utf-8')
  response = urlopen(req)
  print("正常执行完毕：" + response.read().decode('utf-8'))
except Exception as e:
  print ("data exception: ", e)
  traceback.print_exc()
  sys.exit(1) #完成

注意：

1.使用空格替换掉换行符，http请求中不能有换行。

2.Request.add_header可以添加http请求头。

3.response.read()可以获取到http响应内容字节流；response.status可以获取到200、404等响应状态码；response.headers可以获取到Content-Type等响应头。