【实现curl请求与python脚本的转化】

实现curl请求与python脚本的转化

• 成果
• 实现思路
• 使用的技术
• 后端处理逻辑
• 主运行函数
• 工具相关函数
• 核心业务函数
• 效果呈现
• • 转化get请求
  • post请求
  • put请求
• 完整代码

成果

什么都不用说,先上效果截图

上图是实现的最终效果,是不是觉得很cool

实现思路

	1.实现一个可视化页面来接收用户输入curl请求2.将curl请求传给后端进行数据处理3.将处理好的数据再返回给前端页面

使用的技术

	pywebioyapf

后端处理逻辑

1.通过浏览器F12复制curl请求
2.拿到curl请求内容,针对curl请求去做关键数据的转化
3.自定义一套python脚本模板
4.动态替换python脚本模板中的动态参数

主运行函数

def run():with use_scope("index", clear=True):put_row([put_text("测试常用调试工具").style(title_style)])show_curl_2_python()

工具相关函数

def clear_case(file_name):if os.path.exists(file_name):os.remove(file_name)def write_to_file(file_name, data):with open(file_name, mode="w", encoding="utf-8") as f:f.write(data)def read_file(file_name):if os.path.exists(file_name):with open(file_name, mode="r", encoding="utf-8") as f:data_str = f.read()# print(data_str)return data_strelse:return "请输入数据,再进行操作"

核心业务函数

@use_scope("request", clear=True)
def show_curl_2_python():put_row([None,put_column([put_html("
"),put_text("复制curl请求: 打开浏览器  -->  F12-->  选中请求  -->  copy  -->  Copy as cURL(bash)").style(title_style),put_text("请输入curl请求:"),put_button("转化", onclick=show_curl_to_python_result),put_textarea(name="curl_data", rows=40, value="", code={'mode': "curl",'theme': 'darcula'})], size="50px 50px 50px 50px 100%"),None,put_column([put_html("
"),None,None,put_text("转化为python脚本后:"),put_textarea(name="python_data", rows=40, value="", code={'mode': "python",'theme': 'darcula'})], size="50px 50px 50px 50px 100%"),None],size="20px 48% 4% 48% 20px")@use_scope("request", clear=False)
def show_curl_to_python_result():curl_to_python()pin.pin_update("python_data", value=read_file(files["py_template"]))def curl_to_python():try:curl_data = pin.pin.curl_data# print(curl_data)curl_re_expression = re.compile("curl(.*?)--compressed", re.S) or re.compile("curl(.*?)--insecure", re.S)curl_result_list = re.findall(curl_re_expression, curl_data)# print(curl_result_list)if not curl_result_list:for file_name in files.values():clear_case(file_name)if curl_result_list:curl_result = curl_result_list[0]url_re_expression = re.compile("(http.*?)'", re.S)url_result_list = re.findall(url_re_expression, curl_result)url = url_result_list[0]headers_re_expression = re.compile("-H '(.*?)'", re.S)headers_result_list = re.findall(headers_re_expression, curl_result)# print(headers_result_list)headers = {}for header_str in headers_result_list:if ":" in header_str:key = header_str.replace(" ", "").split(":")[0]value = header_str.replace(" ", "").split(":")[1]headers[key] = value# print(headers)method_re_expression = re.compile("-X '(.*?)'", re.S)# print(method_re_expression)method_result_list = re.findall(method_re_expression, curl_result)# print(method_result_list)content_type_re_expression = re.compile("-H 'Content-Type: (.*?)'", re.S)content_type_result_list = re.findall(content_type_re_expression, curl_result)# print(content_type_result_list)data = Nonemethod = Noneif not method_result_list and not content_type_result_list:method = "get"if method_result_list:method = str(method_result_list[0]).lower()if content_type_result_list:# print(method)if "application" in content_type_result_list[0] or "multipart" in content_type_result_list[0]:if not method:method = "post"# print(method)if "application" in content_type_result_list[0]:data_re_expression = re.compile("--data-raw '(.*?)'", re.S)data_result_list = re.findall(data_re_expression, curl_result)# print(data_result_list)if data_result_list:data = json.loads(data_result_list[0])if "multipart" in content_type_result_list[0]:data_re_expression = re.compile("--data-raw \$'(.*?)'", re.S)data_result_list = re.findall(data_re_expression, curl_result)if data_result_list:data = data_result_list[0]py_json_template = """import requestsheaders = {{headers}}json_data = {{data}}response = requests.{{method}}('{{url}}', headers=headers, json=json_data, verify=False)print(response.text)"""py_template = """import requestsheaders = {{headers}}data = '{{data}}'
response = requests.{{method}}('{{url}}', headers=headers, data=str(data).encode("utf-8"), verify=False)print(response.text)"""if data and (isinstance(data, dict) or isinstance(data, list)):py_template = py_json_template.replace("{{headers}}", str(headers)).replace("{{data}}", str(data)).replace("{{method}}", method).replace("{{url}}", url)elif data and isinstance(data, str):py_template = py_template.replace("{{headers}}", str(headers)).replace("{{data}}", data).replace("{{method}}", method).replace("{{url}}", url)else:# print("我执行了吗")py_template = str(py_template).replace("{{headers}}", str(headers)).replace("data = '{{data}}'", "").replace("{{method}}", method).replace("{{url}}", url).replace('data=str(data).encode("utf-8"), ', "")py_template = str(FormatCode(py_template)[0])space_re_expression = re.compile("(:\n\s*?)'", re.S)space_result_list = re.findall(space_re_expression, py_template)# print(space_result_list)for space in space_result_list:# print(space)py_template = py_template.replace(space, ": ")# print(py_template)write_to_file("py_template.py", py_template)# print("我执行到底了")except:write_to_file("py_template.py", "curl请求格式错误,请重新复制")

效果呈现

使用的接口地址
http://httpbin.org/

转化get请求

curl

curl 'http://httpbin.org/get' \-H 'Accept-Language: zh-CN,zh;q=0.9' \-H 'Cache-Control: no-cache' \-H 'Connection: keep-alive' \-H 'Pragma: no-cache' \-H 'Referer: http://httpbin.org/' \-H 'User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36' \-H 'accept: application/json' \--compressed \--insecure

python

import requestsheaders = {'Accept-Language': 'zh-CN,zh;q=0.9','Cache-Control': 'no-cache','Connection': 'keep-alive','Pragma': 'no-cache','Referer': 'http','User-Agent': 'Mozilla/5.0(WindowsNT10.0;Win64;x64)AppleWebKit/537.36(KHTML,likeGecko)Chrome/107.0.0.0Safari/537.36','accept': 'application/json'
}response = requests.get('http://httpbin.org/get',headers=headers,verify=False)print(response.text)

post请求

curl

curl 'http://httpbin.org/post' \-X 'POST' \-H 'Accept-Language: zh-CN,zh;q=0.9' \-H 'Cache-Control: no-cache' \-H 'Connection: keep-alive' \-H 'Content-Length: 0' \-H 'Origin: http://httpbin.org' \-H 'Pragma: no-cache' \-H 'Referer: http://httpbin.org/' \-H 'User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36' \-H 'accept: application/json' \--compressed \--insecure

python

import requestsheaders = {'Accept-Language': 'zh-CN,zh;q=0.9','Cache-Control': 'no-cache','Connection': 'keep-alive','Content-Length': '0','Origin': 'http','Pragma': 'no-cache','Referer': 'http','User-Agent': 'Mozilla/5.0(WindowsNT10.0;Win64;x64)AppleWebKit/537.36(KHTML,likeGecko)Chrome/107.0.0.0Safari/537.36','accept': 'application/json'
}response = requests.post('http://httpbin.org/post',headers=headers,verify=False)print(response.text)

put请求

curl

curl 'http://httpbin.org/put' \-X 'PUT' \-H 'Accept-Language: zh-CN,zh;q=0.9' \-H 'Cache-Control: no-cache' \-H 'Connection: keep-alive' \-H 'Content-Length: 0' \-H 'Origin: http://httpbin.org' \-H 'Pragma: no-cache' \-H 'Referer: http://httpbin.org/' \-H 'User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36' \-H 'accept: application/json' \--compressed \--insecure

python

import requestsheaders = {'Accept-Language': 'zh-CN,zh;q=0.9','Cache-Control': 'no-cache','Connection': 'keep-alive','Content-Length': '0','Origin': 'http','Pragma': 'no-cache','Referer': 'http','User-Agent': 'Mozilla/5.0(WindowsNT10.0;Win64;x64)AppleWebKit/537.36(KHTML,likeGecko)Chrome/107.0.0.0Safari/537.36','accept': 'application/json'
}response = requests.put('http://httpbin.org/put',headers=headers,verify=False)print(response.text)

完整代码

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Copyright ©  版权所有
# @Time    : 2022/11/16 12:01
# @Author  : 胡浩浩
# @Email   : hu_haohao@venusgroup.com.cn
# @File    : curl2python.py
# @IDE     : PyCharm
# @description :
import json
import os
import refrom pywebio import *
from pywebio.output import *
from screeninfo import get_monitors
from pywebio.pin import put_textarea
from pywebio.platform.tornado_http import start_server
from ruamel import yaml
from yapf.yapflib.yapf_api import FormatCodecss = """
.container {margin-top: 0;max-width: 100%;
}
.CodeMirror {font-size: 13px;width: {{width}};overflow: scroll;
}
"""title_style = """
color: red;
"""
code_mirror_width = str(get_monitors()[0].width / 2)css = css.replace("{{width}}", code_mirror_width + "px")
config(title="curl2python", theme="minty", css_style=css)
files = {"py_template": "py_template.py",
}def clear_case(file_name):if os.path.exists(file_name):os.remove(file_name)def write_to_file(file_name, data):with open(file_name, mode="w", encoding="utf-8") as f:f.write(data)def read_file(file_name):if os.path.exists(file_name):with open(file_name, mode="r", encoding="utf-8") as f:data_str = f.read()# print(data_str)return data_strelse:return "请输入数据,再进行操作"@use_scope("request", clear=True)
def show_curl_2_python():put_row([put_column([put_html("
"),put_text("复制curl请求: 打开浏览器  -->  F12-->  选中请求  -->  copy  -->  Copy as cURL(bash)").style(title_style),put_text("请输入curl请求:"),put_button("转化", onclick=show_curl_to_python_result),put_textarea(name="curl_data", rows=40, value="", code={'mode': "curl",'theme': 'darcula'})], size="50px 50px 50px 50px 100%"),None,put_column([put_html("
"),None,None,put_text("转化为python脚本后:"),put_textarea(name="python_data", rows=40, value="", code={'mode': "python",'theme': 'darcula'})], size="50px 50px 50px 50px 100%"),None],size="48% 4% 48% 20px")@use_scope("request", clear=False)
def show_curl_to_python_result():curl_to_python()pin.pin_update("python_data", value=read_file(files["py_template"]))def curl_to_python():try:curl_data = pin.pin.curl_data# print(curl_data)curl_re_expression = re.compile("curl(.*?)--compressed", re.S) or re.compile("curl(.*?)--insecure", re.S)curl_result_list = re.findall(curl_re_expression, curl_data)# print(curl_result_list)if not curl_result_list:for file_name in files.values():clear_case(file_name)if curl_result_list:curl_result = curl_result_list[0]url_re_expression = re.compile("(http.*?)'", re.S)url_result_list = re.findall(url_re_expression, curl_result)url = url_result_list[0]headers_re_expression = re.compile("-H '(.*?)'", re.S)headers_result_list = re.findall(headers_re_expression, curl_result)# print(headers_result_list)headers = {}for header_str in headers_result_list:if ":" in header_str:key = header_str.replace(" ", "").split(":")[0]value = header_str.replace(" ", "").split(":")[1]headers[key] = value# print(headers)method_re_expression = re.compile("-X '(.*?)'", re.S)# print(method_re_expression)method_result_list = re.findall(method_re_expression, curl_result)# print(method_result_list)content_type_re_expression = re.compile("-H 'Content-Type: (.*?)'", re.S)content_type_result_list = re.findall(content_type_re_expression, curl_result)# print(content_type_result_list)data = Nonemethod = Noneif not method_result_list and not content_type_result_list:method = "get"if method_result_list:method = str(method_result_list[0]).lower()if content_type_result_list:# print(method)if "application" in content_type_result_list[0] or "multipart" in content_type_result_list[0]:if not method:method = "post"# print(method)if "application" in content_type_result_list[0]:data_re_expression = re.compile("--data-raw '(.*?)'", re.S)data_result_list = re.findall(data_re_expression, curl_result)# print(data_result_list)if data_result_list:data = json.loads(data_result_list[0])if "multipart" in content_type_result_list[0]:data_re_expression = re.compile("--data-raw \$'(.*?)'", re.S)data_result_list = re.findall(data_re_expression, curl_result)if data_result_list:data = data_result_list[0]py_json_template = """import requestsheaders = {{headers}}json_data = {{data}}response = requests.{{method}}('{{url}}', headers=headers, json=json_data, verify=False)print(response.text)"""py_template = """import requestsheaders = {{headers}}data = '{{data}}'
response = requests.{{method}}('{{url}}', headers=headers, data=str(data).encode("utf-8"), verify=False)print(response.text)"""if data and (isinstance(data, dict) or isinstance(data, list)):py_template = py_json_template.replace("{{headers}}", str(headers)).replace("{{data}}", str(data)).replace("{{method}}", method).replace("{{url}}", url)elif data and isinstance(data, str):py_template = py_template.replace("{{headers}}", str(headers)).replace("{{data}}", data).replace("{{method}}", method).replace("{{url}}", url)else:# print("我执行了吗")py_template = str(py_template).replace("{{headers}}", str(headers)).replace("data = '{{data}}'", "").replace("{{method}}", method).replace("{{url}}", url).replace('data=str(data).encode("utf-8"), ', "")py_template = str(FormatCode(py_template)[0])space_re_expression = re.compile("(:\n\s*?)'", re.S)space_result_list = re.findall(space_re_expression, py_template)# print(space_result_list)for space in space_result_list:# print(space)py_template = py_template.replace(space, ": ")# print(py_template)write_to_file("py_template.py", py_template)# print("我执行到底了")except:write_to_file("py_template.py", "curl请求格式错误,请重新复制")def run():with use_scope("index", clear=True):put_row([put_text("测试常用调试工具").style(title_style)])show_curl_2_python()if __name__ == '__main__':for file_name in files.values():clear_case(file_name)start_server(run, port=8897)

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！