利用FastAPI快速开发服务模块 前言FastAPI 是一个用于构建 Web API 的现代 Python 框架{card-describe title="特点"}快速高效自动文档生成数据校验多种输入输出格式支持安全认证轻量级易学习{/card-describe}目录结构/fastapi |-- app | |-- commands ----- 放置一些命令行 | | `-- __init__.py | |-- exceptions ----- 自定义的异常类 | | |-- __init__.py | | `-- exception.py | |-- http ----- http目录 | | |-- api ----- api控制器目录 | | | |-- __init__.py | | | |-- auth.py ----- 登录认证api的控制器 | | | |-- demo.py | | | `-- users.py | | |-- middleware ----- 放置自定义中间件 | | | `-- __init__.py | | |-- __init__.py | | `-- deps.py ----- 依赖 | |-- jobs ----- 调度任务 | | |-- __init__.py | | `-- demo_job.py | |-- models ----- 模型目录 | | |-- __init__.py | | |-- base_model.py ----- 定义模型的基类 | | `-- user.py | |-- providers ----- 核心服务提供者 | | |-- __init__.py | | |-- app_provider.py ----- 注册应用的全局事件、中间件等 | | |-- database.py ----- 数据库连接 | | |-- handle_exception.py ----- 异常处理器 | | |-- logging_provider.py ----- 集成loguru日志系统 | | `-- route_provider.py ----- 注册路由文件routes/* | |-- schemas ----- 数据模型，负责请求和响应资源数据的定义和格式转换 | | |-- __init__.py | | `-- user.py | |-- services ----- 服务层，业务逻辑层 | | |-- auth ----- 认证相关服务 | | | |-- __init__.py | | | |-- grant.py ----- 认证核心类 | | | |-- hashing.py | | | |-- jwt_helper.py | | | |-- oauth2_schema.py | | | `-- random_code_verifier.py | | `-- __init__.py | |-- support ----- 公共方法 | | |-- __init__.py | | `-- helper.py | `-- __init__.py |-- bootstrap ----- 启动项 | |-- __init__.py | |-- application.py ----- 创建app实例 | `-- scheduler.py ----- 创建调度器实例 |-- config ----- 配置目录 | |-- auth.py ----- 认证-JWT配置 | |-- config.py ----- app配置 | |-- database.py ----- 数据库配置 | `-- logging.py ----- 日志配置 |-- database | `-- migrations ----- 初始化SQL | `-- 2022_09_07_create_users_table.sql |-- routes ----- 路由目录 | |-- __init__.py | `-- api.py ----- api路由 |-- storage | `-- logs ----- 日志目录 |-- README.md |-- main.py ----- app/api启动入口 |-- requirements.txt `-- scheduler.py ----- 调度任务启动入口{card-describe title="集成的模块"}日志系统：集成 loguru，一个优雅、简洁的日志库异常处理：定义认证异常类，注册 Exception Handler路由注册：路由集中注册，按模块划分为不同的文件，代码层次结构清晰系统配置：基于 pydantic.BaseSettings，使用 .env 文件设置环境变量。配置文件按功能模块划分，默认定义了app基础配置、数据库配置、日志配置、认证配置数据库：基于 peewee，一个轻量级的Python ORM框架中间件：默认注册了全局CORS中间件JWT认证：易于扩展新的认证方式。测试登录认证请先执行初始化的SQL：database/migrations/*.sql（验证码的存储和校验方法请自行实现）调度任务，基于 APScheduler 调度任务框架，（定时任务与api是分开启动的）{/card-describe}运行# 执行初始化SQL： # 导入database/migrations/2022_09_07_create_users_table.sql # 主程序 python main.py # 任务调度器 python scheduler.py优化自定义首页from fastapi.responses import PlainTextResponse # include_in_schema排除生成文档 @app.get("/", include_in_schema=False, response_class=PlainTextResponse) async def root(): return 'hello'关闭模块{message type="success" content="修改路由注册文件,直接注销相关模块即可"/} routes/api.py自定义说明文档#静态文件 app.mount("/static", StaticFiles(directory="static"), name="static") description = """ ## 说明 - 生成支付订单 """ app.openapi()["info"] = { "title": "abpay api", "version": "1.0.0", "description": description, "contact": { "name": "xwzy", "email": "1940728253@qq.com", "url": "https://me.itbunan.xyz", }, "license": { "name": "Apache 2.0", "url": "https://www.apache.org/licenses/LICENSE-2.0.html", }, "x-logo": { "url": "/static/img/logo.png" } }{card-default label="效果" width="75%"}{/card-default}

使用chatgpt快速生成python代码画架构图 前言写技术文章，有时候需要画图说明。自己画图比较费时费力，还不好看。最近chatgpt大火，利用chatgpt生成python代码绘图。模块Graphviz{message type="success" content="Graphviz 是一个开源的图形可视化工具，它可以自动生成各种类型的图形，并支持多种输出格式（如 PNG、SVG、PDF 等）。"/}diagrams{message type="success" content="diagrams 是一个 Python 库，用于创建各种系统架构图和流程图。"/}安装## 安装graphviz ## mac brew install graphviz ## 验证 dot -v ##安装模块 pip3 install diagrams graphviz绘图利用chatgpt生成架构图{card-describe title="描述想法"}我想画一个网络流程图，流程如下客户端发送请求到A站，A站将请求发送到核心系统，核心系统将请求发送到B站，B站将请求发送到上游通道，上游通道将结果返回到B站，B站将结果返回给核心系统，核心系统将结果返回A站，A站将结果在客户端展示{/card-describe}{card-default label="chatgpt描述" width="90%"}{/card-default}生成的代码from diagrams import Cluster, Diagram from diagrams.aws.compute import EC2 from diagrams.aws.database import RDS from diagrams.aws.network import ELB from diagrams.onprem.client import Client from diagrams.onprem.compute import Server with Diagram("支付流程", show=False, outformat="png"): client = Client("客户端") lb_a = ELB("A站插件") sg_a = Server("核心系统") lb_b = ELB("B站转发模块") upc = Server("上游") client >> lb_a >> sg_a >> lb_b >> upc >> lb_b >> sg_a >> lb_a >> client{card-default label="结果" width="90%"}{/card-default}参考python3使用diagrams生成架构图

直播推流管理系统prtmp的开发过程总结 前言去年通过docker部署rtmp服务并进行网络压力测试，今年我们的业务终于用到了直播流服务。自己写了一个小系统用来从上游拉取直播流并分发到阿里云或者腾讯的直播服务。然后供小程序调用。特此记录整个配置过程。{card-default label="直播流管理" width="75%"}{/card-default}直播服务{message type="success" content="一般需要准备2个子域名，一个用来拉流，一个用来推流。可以多准备一个子域名，给自己写的流管理系统使用。"/}阿里云{callout color="#f0ad4e"}刚开始，用户不多，流量也很小，可以使用按量计费模式。{/callout}{card-default label="直播" width="75%"}{/card-default}自建直播服务{message type="success" content="直接用docker启动"/}参考: centos7部署rtmp服务并进行压力测试直播流管理系统{message type="success" content="初始功能比较简单，用到了flask+js+celery，用docker方式启动。"/}{card-describe title="主要文件"}main.py 视图，接口等函数utils.py 辅助函数{/card-describe}主要接口{message type="success" content="查询接口，推流接口,推流播放链接生成"/}推流url生成def ali_push_url(appName, streamName): """ 阿里云直播服务推流url """ #推流 push_domain = 'push.mytest.com' push_key = 'ZtBxxxxxxxMEKW' #过期时间 expire_time = 86400 time_stamp = int(time.time()) + expire_time #推流url pstr = '/{}/{}-{}-0-0-{}'.format(appName, streamName, time_stamp, push_key) pmd5 = md5_sign(pstr) purl= 'rtmp://{}/{}/{}?auth_key={}-0-0-{}'.format(push_domain, appName, streamName, time_stamp, pmd5) return purl直播服务urldef ali_live_url(appName, streamName): """ 阿里云直播服务播放url """ resp = {} #播放 play_domain = 'live.mytest.com' play_key = 'ulwxxxxxxxxxOm' #过期时间 expire_time = 86400 time_stamp = int(time.time()) + expire_time #播流url rstr = '/{}/{}-{}-0-0-{}'.format(appName, streamName, time_stamp, play_key) fstr = '/{}/{}.flv-{}-0-0-{}'.format(appName, streamName, time_stamp, play_key) hstr = '/{}/{}.m3u8-{}-0-0-{}'.format(appName, streamName, time_stamp, play_key) rmd5 = md5_sign(rstr) fmd5 = md5_sign(fstr) hmd5 = md5_sign(hstr) resp['rtmp_url'] = 'rtmp://{}/{}/{}?auth_key={}-0-0-{}'.format(play_domain, appName, streamName, time_stamp, rmd5) resp['flv_url'] = 'http://{}/{}/{}.flv?auth_key={}-0-0-{}'.format(play_domain, appName, streamName, time_stamp, fmd5) resp['hls_url'] = 'http://{}/{}/{}.m3u8?auth_key={}-0-0-{}'.format(play_domain, appName, streamName, time_stamp, hmd5) return resp推流任务函数@celery.task def push_rtmp_task(sid, source, target): """ 推流任务函数 """ print(source) print(target) # 视频源输入参数 input_args = ['-i', source] # 推流输出参数 output_args = ['-vcodec', 'libx264', '-acodec', 'aac', '-f', 'flv', target] # 合并参数 command = ['ffmpeg'] + input_args + output_args # 执行命令 with subprocess.Popen(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) as process: process.wait() # 更新推流信息 living_info = load_json('./data/living.json') if sid in living_info.keys(): living_info.pop(sid) save_living_info(living_info)前端功能{message type="success" content="通过js做数据交互"/}手动推流 //手动推流 function ManPushRtmp(dom) { var mymessage = confirm("确认手动推送此直播流到" + $(dom).attr("ptype") + "？" + $(dom).attr("url")); if (mymessage == true) { $('#myModal').modal('show'); $.ajax({ url : '/api/rtmppush?url=' + $(dom).attr("url") + '&utype=' + $(dom).attr("utype") + '&sid=' + $(dom).attr("sid") + '&mid=' + $(dom).attr("mid") + '&lid=' + $(dom).attr("lid") + '&ptype=' + $(dom).attr("ptype"), type : 'get', success : function(data) { $('#myModal').modal('hide'); if (data.code == 1000){ $("#child_table").bootstrapTable('refresh', data.data); alert(data.msg); } else { alert("推流失败! " + data.msg) } }, error : function(data){ $('#myModal').modal('hide'); alert("接口异常! " + data.msg) } }); } };停止推流 //停止推流 function ManStopPush(dom) { var mymessage = confirm("确认手动停止此直播流？" + $(dom).attr("stream_id")); if (mymessage == true) { $('#myModal').modal('show'); $.ajax({ url : '/api/stoppush?stream_id=' + $(dom).attr("stream_id"), type : 'get', success : function(data) { $('#myModal').modal('hide'); if (data.code == 1000){ alert(data.msg); location.reload(); } else { alert("失败! " + data.msg) } }, error : function(data){ $('#myModal').modal('hide'); alert("接口异常!" + data.msg) } }); } };异步任务{message type="success" content="通过celery实现异步任务执行"/}celery实例化from celery import Celery # celery配置 app.config['CELERY_BROKER_URL'] = 'redis://redis:6379/0' app.config['CELERY_RESULT_BACKEND'] = 'redis://redis:6379/0' #celery实例化 celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL']) celery.conf.update(app.config)caddy代理prmtp.mytest.com { tls admin@mytest.com encode gzip log { output file /opt/logs/access.log } header / { Strict-Transport-Security "max-age=31536000;includeSubdomains;preload" } #访问认证 #密码:123456 basicauth /* { admin $2a$14$DIjtbTxbUSZHfHJUrjuU9.45SlrcwICIXNVSwVxehsnHhTXBBN Nsi } ## HTTP 代理配置, 后端服务端口 reverse_proxy http://backend:5000 }启动#手动启动 python main.py celery -A main.celery worker -l info # 项目启动 docker-compose up -dFAQjs传递url参数需要转码{message type="success" content="使用encodeURIComponent函数"/}

centos7升级python3.9版本 前言{callout color="#f0ad4e"}系统自带的版本比较老了，好多代码跑不动。需要升级版本，记录升级过程{/callout}获取最新包官方下载:https://www.python.org/downloads/安装# 下载 wget https://www.python.org/ftp/python/3.9.15/Python-3.9.15.tar.xz # 解压 tar xvf Python-3.9.15.tar.xz # 编译 cd Python-3.9.15/ ./configure --prefix=/usr/local/python39 make && make install # 创建软连接 ln -s /usr/local/python39/bin/python3.9 /usr/bin/python39 ln -s /usr/local/python39/bin/pip3.9 /usr/local/bin/pip39

子进程并发执行python代码 说明{callout color="#f0ad4e"}一个url处理函数非常耗时，导致nginx在等待过程中超时。改成并发处理，问题解决。记录下代码片段。{/callout}代码# 调用过程 for obj in queryset: res1 = parse_domains(obj.ilink) if 'href_domains' in res1: threads = [] for url in res1['href_domains']: domain = url.split('/')[2] #排除已存在的网站 if domain in exclude_urls: continue t = threading.Thread(target=processing_url_func, args=(url,)) threads.append(t) for p in threads: p.start() obj.status = 0 obj.save() #暂停4秒 time.sleep(4)# 子进程函数 def processing_url_func(url): """ 处理url """ #获取网站信息 res2 = get_web_info(url) if res2: #print(res2) iLi = iLinks.objects.create(title=res2[0],note=res2[1],home=url) iLi.save()# 获取url信息 def get_web_info(url): session = sessions() req_header = {'User-Agent':'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36'} print(url) try: tag = 0 resp = session.get(url, headers=req_header, allow_redirects=True, verify=False) except ReadTimeout as e: tag = 1 print(e) except ConnectionError as e: tag = 2 print(e) except RequestException as e: tag =3 print(e) if tag > 0: return False soup = BeautifulSoup(resp.text, 'html.parser') title = '' note = '' if soup.title: title = _filter_emoji(soup.title.string) keywords = soup.find(attrs={"name":"keywords"}) if keywords: note = _filter_emoji(keywords['content']) if title and note: return [title, note] else: return False

python生成随机字符串 说明{message type="success" content="生成随机字符串，并保存到xls文件中"/}import random import string import xlwt import datetime from time import strftime #输入生成个数 sums = input("请输入生成个数:") #检查参数 if not sums.isdigit(): print("输入非法") exit(0) #创建workbook 设置编码 workbook = xlwt.Workbook(encoding='utf-8') #创建一个worksheet worksheet =workbook.add_sheet('worksheet') #第一列宽度 worksheet.col(1).width = 3333 #输出 for i in range(int(sums)): print('{}...'.format(i)) val1 = ''.join(random.sample(string.digits, 7)) val2 = ''.join(random.sample(string.digits, 8)) val = '2022-{}-{}'.format(val1, val2) worksheet.write(i, 0, label=val) #文件名 data_s = datetime.datetime.now().strftime("%Y-%m-%d-%H-%M-%S") file_s = '随机订单号-{}-{}.xls'.format(sums, data_s) #保存 workbook.save(file_s)

python开发-基本使用总结 前言{callout color="#f0ad4e"}总结python使用过程中的知识点。{/callout}语法字典嵌套赋值# 方法一： d={} d.setdefault('prov', {}).setdefault('city', {})['down'] = "wang" print(d) class MagicDict(dict): def __getitem__(self, item): try: return dict.__getitem__(self, item) except KeyError: value = self[item] = type(self)() return value # 方法二： d = MagicDict() d["a"]["b"]["c"] = 10 d[1][2][3] = 5 d[2][2]['test'] = 3 print (d) # 方法三： from collections import defaultdict d = defaultdict(dict) d['hell']['good']= 'xiaorui.cc' print (d)模块语法检查,清理多余importpip3 install pyflakes pyflakes xxx.py其它调用百度api{message type="error" content="最多处理1993个字符"/}json转换报错{alert type="info"}Object of type 'int64' is not JSON serializable{/alert}{message type="success" content="numpy.int64 转换成 int 类型"/}

博客优化-利用tinify进行图片无损压缩 前言{callout color="#f0ad4e"}博客搭建有一段时间了，随着内容的增多，感觉在打开的时候加载时间越来越长。所以就想优化一下，特此记录优化过程。{/callout}分析{callout color="#f0ad4e"} PageSpeed 是我经常用的用来分析页面加载时间比较喜欢的一个用具。 使用也很简单，输入域名点击分析即可。{/callout}{card-default label="分析结果" width="90%"}{/card-default}{message type="error" content="优化建议第一项就是图片大小。而博主上传的图片也没有经过压缩过。是应该优化一下了。"/}图片压缩脚本{callout color="#f0ad4e"}参考网上的压缩脚本，改了一下，更适合自己使用。利用tinify模块做图片压缩效果很好，就是每月有500次使用限制。目前博主的内容不多，完全够用。tinify的key申请地址 {/callout}{message type="success" content="压缩脚本内容，"/}隐藏内容，请前往内页查看详情{message type="success" content="运行脚本，自动记录断点，中断后重新运行即可。"/}python3 img_cop.py{card-default label="运行过程" width="75%"}{/card-default}压缩效果{message type="success" content="压缩效果很明显，肉眼也看不出区别"/}{card-default label="大小对比" width="75%"}{/card-default}访问效果对比{message type="success" content="再次运行PageSpeed，对比效果。由9s降到了4s"/}{card-default label="对比效果" width="90%"}{/card-default}

wordpress利用python直接操作数据库添加用户和商品评论 前言{callout color="#f0ad4e"}利用python直接操作wordpress数据库，批量添加用户，评论等。{/callout}{card-default label="批量添加评论页面" width="75%"}{/card-default}{card-describe title="主要模块"}模块说明flaskweb框架faker随机账号信息生成pymysql数据库操作{/card-describe}主函数{message type="success" content="main.py"/}# -*- coding: utf-8 -*- """主函数""" import argparse import sys from utils import * from server import * #程序入口 if __name__ == "__main__": parser = argparse.ArgumentParser(description="wp辅助工具") parser.add_argument("-u","--adduser", action="store_true", help="批量添加用户") parser.add_argument("-s","--server", action="store_true", help="启动web服务") if not sys.argv[1:]: sys.argv.extend(['-h']) args = parser.parse_args() if args.adduser: add_users(3700) if args.server: app.run(debug=True, host='0.0.0.0', port=3060) exit(){message type="success" content="server.py"/}# -*- coding: utf-8 -*- """ Created on 2022-05-23 21:16:39 --------- @summary: --------- @author: wxzy """ from flask import Flask, render_template, session, redirect, url_for, flash from flask_wtf import FlaskForm from wtforms import StringField, SubmitField, TextAreaField from wtforms.validators import DataRequired from flask_bootstrap import Bootstrap from utils import * class CommentsForm(FlaskForm): """ 评论表单类 """ product_name = StringField("商品名:", validators=[DataRequired()]) comment = TextAreaField("评论列表:", validators=[DataRequired()], render_kw={"style": "height: 500px"}) submit = SubmitField("提交") app = Flask(__name__) app.config["SECRET_KEY"] = '79537d00f4834892986f09a100aa1aaf' bootstrap = Bootstrap(app) @app.route('/', methods=["GET", "POST"]) def index(): """ 添加表单页面 """ product_name = None form = CommentsForm() #提交评论 if form.validate_on_submit(): if session.get("product_name") == form.product_name.data and session.get("comment") == form.comment.data : flash("本商品已经提交过一次相同评论,请不要重复提交~!") else: #添加评论 res = addcomment(form.product_name.data, form.comment.data) if res[0] == 1: flash("此名称查询到{}个对应商品，请输入商品全称!".format(res[1])) elif res[0] == 2: flash("此商品名未查询到对应商品, 请检查商品名输入是否正确!") else: flash("恭喜你你，成功添加{}条评论!".format(res[1])) # 更新session 字段 session["product_name"] = form.product_name.data session["comment"] = form.comment.data return redirect(url_for("index")) return render_template("index.html",form=form)数据库操作批量添加用户def add_users(num): """ 批量添加用户 """ user_str = '' email_array = ['@163.com', '@gmail.com', '@126.com', '@microsoft.com', '@mail.com', '@fastmail.com'] fake = Faker('en_US') #打开数据库 conn = pymysql.connect(host=HOST, user=USER, password=PASSWORD, db=DB, port=PORT, charset='utf8') cur = conn.cursor() for i in range(num): i = i+1 user = '{}{}'.format(fake.first_name(), random.randint(10, 999)) email = '{}{}'.format(user, random.choice(email_array)) registered = datetime.strftime(datetime.now(),'%Y-%m-%d %H:%M:%S') sql = "insert into wp_users(user_login, user_pass, user_nicename, user_email, user_registered, user_status, display_name) values(%s, %s, %s, %s, %s, %s, %s);" #密码123456 #user_activation_key 无意义 cur.execute(sql, (user, '$P$BSBdHomZc571IfTLJhf3PhxaN3x7dn0', user, email, registered, 0, user)) conn.commit() print('新增第{}个用户:{}'.format(i, email)) user_str = user_str + email + '\n' #关闭数据库 cur.close() conn.close() #保存结果到文件 with open('/tmp/user{}.txt'.format(datetime.strftime(datetime.now(),'%Y-%m-%d-%H-%M-%S')), 'a+') as w: w.write(user_str)批量添加评论def addcomment(product, comments): """ 商品添加评论函数 """ num = 0 #打开数据库 conn = pymysql.connect(host=HOST, user=USER, password=PASSWORD, db=DB, port=PORT, charset='utf8') cur = conn.cursor() #查询商品 sql = "select ID from wp_posts where post_title like '%{}%'".format(product) cur.execute(sql) ids = cur.fetchall() if len(ids) == 0: res = [2, 0] elif len(ids) > 1: res = [1, len(ids)] else: #变量 comment_post_ID = ids[0][0] comment_array = comments.splitlines() # ip_array ips = os.popen("cat /data/log/nginx-proxy/access.log | awk '{print $1}' | sort | uniq").readlines() # author sql = "select ID,user_nicename,user_email,user_registered from wp_users where id not in (select user_id from wp_comments group by user_id having count(user_id) > 4) and id not in (select user_id from wp_comments where comment_post_ID = {}) order by RAND() limit {};".format(comment_post_ID, len(comment_array)) cur.execute(sql) authors = cur.fetchall() for comment in comment_array: if len(comment) == 0: continue #时间 date = date_gmt = datetime.strftime(datetime.now(),'%Y-%m-%d %H:%M:%S') #插入数据 try: sql = "insert into wp_comments(comment_post_ID, comment_author, comment_author_email, comment_author_IP, comment_date, comment_date_gmt, comment_content, comment_agent, comment_type, user_id) values(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s);" cur.execute(sql, (comment_post_ID, authors[num][1], authors[num][2], random.choice(ips).strip(), date, date_gmt, comment, random.choice(AGENTS), 'review', authors[num][0])) last_id = cur.lastrowid sql = "INSERT INTO `wp_commentmeta` (`comment_id`, `meta_key`, `meta_value`) VALUES (%s, 'rating', %s);" cur.execute(sql, (last_id, random.choices(population=[3, 4, 5], weights=[0.1, 0.2, 0.7]).pop())) sql = "UPDATE wp_posts SET comment_count = comment_count + 1 WHERE ID = '{}';".format(comment_post_ID) cur.execute(sql) sql = "UPDATE wp_postmeta SET meta_value = meta_value + 1 WHERE post_id = '{}' AND meta_key = '_wc_review_count';".format(comment_post_ID) cur.execute(sql) conn.commit() except: conn.rollback() #评论加1 num = num + 1 res = [0, num] #关闭链接 cur.close() conn.close() #返回结果 return res

利用python脚本自动更新腾讯云安全组防火墙白名单 {card-default label="安全组" width="75%"}{/card-default}前言{callout color="#f0ad4e"}经常在家里，公司等地方远程管理腾讯云的vps，本来想讲固定IP添加到防火墙白名单。可是家里的IP经常变化，于是写了个脚本。调用腾讯云的api更新防火墙白名单IP地址。{/callout}脚本内容update_while_ip.pyimport requests import json from tencentcloud.common import credential from tencentcloud.common.profile.client_profile import ClientProfile from tencentcloud.common.profile.http_profile import HttpProfile from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException from tencentcloud.vpc.v20170312 import vpc_client, models import argparse from setting import * import datetime, os def AddGroupRole(config, sourceip): try: cred = credential.Credential(config['SecretId'], config['SecretKey']) httpProfile = HttpProfile() httpProfile.endpoint = "vpc.tencentcloudapi.com" clientProfile = ClientProfile() clientProfile.httpProfile = httpProfile client = vpc_client.VpcClient(cred, config['Region'], clientProfile) #req = models.CreateSecurityGroupPoliciesRequest() req = models.ModifySecurityGroupPoliciesRequest() params = { "SecurityGroupPolicySet": { "Ingress": [ { "Protocol": "ALL", "CidrBlock": sourceip, "Action": "ACCEPT", "PolicyDescription": "mac当前公网IP" }, { "Protocol": "ALL", "CidrBlock": '1.2.3.44', "Action": "ACCEPT", "PolicyDescription": "堡垒机" }, ] }, "SecurityGroupId": config['SecurityGroupId'] } req.from_json_string(json.dumps(params)) #resp = client.CreateSecurityGroupPolicies(req) resp = client.ModifySecurityGroupPolicies(req) print('{}:{}'.format(config['description'], resp.to_json_string())) except TencentCloudSDKException as err: print(err) def GetCompanyOldIp(): ''' 读旧IP ''' if not os.path.exists(IPFILE): return '0.0.0.0' with open(IPFILE, 'r') as fr: oldip = fr.read().strip() return oldip def WriteIp(sip): ''' 写新IP ''' with open(IPFILE, 'w') as fw: fw.write(sip) def UpdateWhiteIP(tag): ''' 更新IP白名单函数 ''' OldIp = GetCompanyOldIp() NewIp = requests.get(url="https://ifconfig.me/ip").text if tag == 'f': for item in TCONFIG: AddGroupRole(item, NewIp) if NewIp != OldIp: print('{}======公网ip更新:{}--->{}'.format(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), OldIp, NewIp)) if tag == 'all': for item in TCONFIG: AddGroupRole(item, NewIp) if tag == 'm': AddGroupRole(TCONFIG[0], NewIp) if tag == 'a': AddGroupRole(TCONFIG[1], NewIp) #保存IP WriteIp(NewIp) if __name__ == '__main__': parser = argparse.ArgumentParser(description="更新信任IP白名单") parser.add_argument("-u","--update", help="m:我的腾讯云，a:公司, all:全部", type=str, choices=['m', 'a', 'f', 'all'], default='all') args = parser.parse_args() if args.update in ['m', 'a', 'f', 'all']: UpdateWhiteIP(args.update) else: print('参数非法') exit()配置文件 setting.py#配置信息列表 TCONFIG = [ { "description":"我的腾讯云主机", "SecretId": "AKIDGPL2VxxxxxxxxxxxxxxxY6XyITuCJ", #腾讯云API接口密钥 "SecretKey": "EKG6oOsDxxxxxxxxxxxxxxGJ8mHfy", #secretkey "Region": "ap-shanghai", #vps 安全组所在地区 "SecurityGroupId": "sg-3baexxx" #安全组IP }, ] #保存IP文件 IPFILE='/tmp/ip.txt'配置别名vim .bashrc alias uip='python3 /opt/project/sys/tencent/update_while_ip.py'手动更新uip -u f