首页
AI导航
美图
服务
付费
树洞
留言
云主机
推荐
邻居
更多
我的书单
我的足迹
罗盘时钟
圈小猫
工作打分
本站统计
版本历史
推荐
txt阅读器
主机监控
M商城
网址导航
在线工具
证件照制作
Search
1
docker和docker-compose一键安装脚本
824 阅读
2
docker下运行grafana和grafana Image Renderer
664 阅读
3
grafana的Dashboard面板添加阈值报警
632 阅读
4
WooCommerce对接第三方支付插件开发
503 阅读
5
基于docker的部署fecmall开源电商系统
442 阅读
ChatGPT
虚拟化
数据库
运维
基础知识
监控预警
数据展示
运维工具
web安全
系统服务
开发
python
php
java
shell
go
html5
项目
博客
电商
工具
娱乐
影视
读书
读书笔记
综合
VPS报告
规范文档
知识总结
经验分享
关于本站
登录
Search
标签搜索
python
django
电商平台
运维工具
Joe主题
docker
zabbix
蓝鲸智云
运维
监控
typecho
grafana
wordpress
运维知识
mysql
php
elk
nginx
web安全
VPS测试
IT不难
累计撰写
245
篇文章
累计收到
209
条评论
首页
栏目
ChatGPT
虚拟化
数据库
运维
基础知识
监控预警
数据展示
运维工具
web安全
系统服务
开发
python
php
java
shell
go
html5
项目
博客
电商
工具
娱乐
影视
读书
读书笔记
综合
VPS报告
规范文档
知识总结
经验分享
关于本站
页面
美图
服务
留言
邻居
我的足迹
本站统计
版本历史
推荐
M商城
网址导航
搜索到
21
篇与
的结果
2023-03-20
利用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}
2023年03月20日
2 阅读
0 评论
0 点赞
2023-02-21
Django后台展示(二)
说明{callout color="#f0ad4e"}需要通过后台直接周期任务运行的结果日志,这样就不需要每次登录服务器。{/callout}{card-default label="效果" width="80%"}{/card-default}代码接口代码def getlog(request): ''' 日志查看 ''' if request.method == 'GET': logpath = request.GET.get('path', default='') content = '无日志' if os.path.exists(logpath): with open(logpath, 'r') as f: content = f.read() return render(request, 'logview.html', {'logpath': logpath, 'content': content})模板文件{% block content %} <h2>{{ logpath }}</h2> <div style="background-color: black;color: azure" > <span style="font-size: 10px"><pre>{{ content }} </pre> </div> {% endblock %}
2023年02月21日
14 阅读
0 评论
0 点赞
2023-02-21
Django后台展示(一)
说明{callout color="#f0ad4e"}Django用的比较多,最近自己的taskmonitor项目添加了采集爬虫相关的内容。总结一下,其中的代码片段。{/callout}{card-default label="效果" width="85%"}{/card-default}代码片段改变状态颜色 @admin.display(description='调度状态', ordering='status') def status_colored(self, obj): colors = { 0: 'red', 1: 'green' } return format_html( '<p style="color:{};">{}</p>', colors[obj.status], STATUS_CHOICE[obj.status], )重写保存函数 #重写保存函数 def save_model(self, request, obj, form, change): #新增默认不启动计划任务,测试后调整调度状态 if change: if len(obj.crond.split(' ')) != 5: obj.status = 0 else: obj.status = 0 #保存 super(SpiderTaskAdmin, self).save_model(request, obj, form, change)任务调度开关 def schedule_switch(self, request, queryset): """ 任务调度开关 """ for obj in queryset: c_len= len(obj.crond.strip().split()) if c_len == 5: if obj.status == 1: obj.status = 0 res = remove_celery_task(obj.id) else: print(obj.status) obj.status = 1 res = add_celery_task(obj.id) if res: obj.save()自定义操作操作列设置 # 定义一些操作示例 @admin.display(description='手动操作', ordering='sname') def operate(self, obj): #容器名称 con_name = 'spider_{}_1'.format(obj.sname) #获取容器 containers = get_containers({"name":con_name}) #操作容器 if containers: uparas = '{}-正在运行...'.format(con_name) up_btn = "<button onclick='self.parent.app.$msgbox(\"{u_d}\")' class='el-button el-button--info el-button--small'>启动</button>" dparas = "'{}','{}'".format(obj.sname, '0') down_btn = "<button onclick='javascript:spidermanager({d_d})' class='el-button el-button--danger el-button--small'>停止</button>" else: uparas = "'{}','{}'".format(obj.sname, '1') up_btn = "<button onclick='javascript:spidermanager({u_d})' class='el-button el-button--success el-button--small'>启动</button>" dparas = '{}-不存在'.format(con_name) down_btn = "<button onclick='self.parent.app.$msgbox(\"{d_d}\")' class='el-button el-button--info el-button--small'>停止</button>" # 查看日志 if obj.logpath: log_data = {'name':'{}|日志'.format(con_name), 'icon': 'fas fa-user-tie', 'url':'/spider/getlog/?path={}'.format(obj.logpath)} log_btn = "<button onclick='self.parent.app.openTab({l_d})' class='el-button el-button--primary el-button--small'>最新日志</button>" else: log_data = '{}-最新日志无'.format(con_name) log_btn = "<button onclick='self.parent.app.$msgbox(\"{l_d}\")' class='el-button el-button--info el-button--small'>最新日志</button>" #拼接并返回 html_str = "<div>" + up_btn + down_btn + log_btn + "</div>" return format_html(html_str, u_d=uparas, d_d=dparas, l_d=log_data)自定义js class Media: js = ('js/common.js',)js代码//spidermanager函数传入cname和opera两个参数 //cname 容器名 //opera 启动、关闭 function spidermanager(sname, opera){ var url = "/spider/manager/?sname="+sname+"&opera="+String(opera); var Http = new XMLHttpRequest(); Http.open("GET", url, false); Http.send(null); console.log(Http.responseText); }接口代码def manager(request): ''' 容器操作 ''' if request.method == 'GET': opera = request.GET.get('opera', default='2') sname = request.GET.get('sname', default='') if not sname or opera == '2': messages.error(request, '参数非法') else: con_name = 'spider_{}_1'.format(sname) containers = get_containers({"name":con_name}) if containers: if opera == '1': messages.info(request, '容器:{} |已运行...'.format(con_name)) else: for container in containers: container.stop() messages.success(request, '容器:{} |成功停止...'.format(con_name)) else: if opera == '1': sync_start_process.delay(sname) messages.success(request, '容器:{} |成功启动...'.format(con_name)) else: messages.info(request, '容器:{} |未运行...'.format(con_name)) return HttpResponse('ok')
2023年02月21日
13 阅读
0 评论
0 点赞
2022-11-08
子进程并发执行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
2022年11月08日
27 阅读
0 评论
0 点赞
2022-10-25
docker下pip安装软件包超时
前言{callout color="#f0ad4e"}今天做服务迁移,在新系统上构建docker镜像老是失败,开始以为是pip源的问题。替换了所有的源都超时。更换网络模式host后解决。{/callout}{card-describe title="解决办法"}version: "3" networks: snet: services: backend: build: context: . dockerfile: ./docker/Dockerfile network: host volumes: - ./backend:/myapp - ./frontend:/frontend - /var/log/django/:/data/log/django environment: - "SET_CONTAINER_TIMEZONE=true" - "TZ=Asia/Shanghai" #command: sh /myapp/run.sh command: sh /myapp/run-test.sh restart: unless-stopped extra_hosts: - "mysqlserver:172.16.2.155" #- "mysqlserver:172.16.4.10" ports: - "8013:8013" networks: - snet{/card-describe}
2022年10月25日
23 阅读
0 评论
1 点赞
2022-10-19
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)
2022年10月19日
24 阅读
2 评论
1 点赞
2022-09-21
Django框架打印日志配置
setting.py{message type="success" content="主项目配置文件设置"/}# 日志配置 LOGGING = { 'version': 1, #使用的python内置的logging模块,那么python可能会对它进行升级,所以需要写一个版本号,目前就是1版本 'disable_existing_loggers': False, #是否去掉目前项目中其他地方中以及使用的日志功能,但是将来我们可能会引入第三方的模块,里面可能内置了日志功能,所以尽量不要关闭。 'formatters': { #日志记录格式 'standard': { #levelname等级,asctime记录时间,module表示日志发生的文件名称,lineno行号,message错误信息 'format': '%(levelname)s %(asctime)s %(module)s %(lineno)d %(message)s' }, 'simple': { 'format': '%(levelname)s %(module)s %(lineno)d %(message)s' }, }, 'filters': { #过滤器:可以对日志进行输出时的过滤用的 'require_debug_true': { #在debug=True下产生的一些日志信息,要不要记录日志,需要的话就在handlers中加上这个过滤器,不需要就不加 '()': 'django.utils.log.RequireDebugTrue', }, 'require_debug_false': { #和上面相反 '()': 'django.utils.log.RequireDebugFalse', }, }, 'handlers': { #日志处理方式,日志实例,向哪里输出 'console': { #在控制台输出时的实例 'level': 'DEBUG', #日志等级;debug是最低等级,那么只要比它高等级的信息都会被记录 'filters': ['require_debug_true'], #在debug=True下才会打印在控制台 'class': 'logging.StreamHandler', #使用的python的logging模块中的StreamHandler来进行输出 'formatter': 'simple' }, 'file': { 'level': 'INFO', 'class': 'logging.handlers.RotatingFileHandler', # 日志位置,日志文件名,日志保存目录必须手动创建 'filename': os.path.join(os.path.dirname('/var/log/promon/'), "promon.log"), #注意,你的文件应该有读写权限。 # 日志文件的最大值,这里我们设置300M 'maxBytes': 300 * 1024 * 1024, # B # 日志文件的数量,设置最大日志数量为10 'backupCount': 10, # 日志格式:详细格式 'formatter': 'standard', 'encoding': 'utf-8', # 设置默认编码,否则打印出来汉字乱码 }, }, # 日志对象 'loggers': { 'django': { #和django结合起来使用,将django中之前的日志输出内容的时候,按照我们的日志配置进行输出, 'handlers': ['console', 'file'], #将来项目上线,把console去掉 'propagate': True, #冒泡:是否将日志信息记录冒泡给其他的日志处理系统,工作中都是True,不然django这个日志系统捕获到日志信息之后,其他模块中可能也有日志记录功能的模块,就获取不到这个日志信息了 }, } }引用#日志 import logging logger = logging.getLogger('django')打印日志logger.warn('warn') logger.warn('info') logger.warn('error')
2022年09月21日
28 阅读
0 评论
1 点赞
2022-08-12
百度分享链接可用性批量检查
前言{callout color="#f0ad4e"}前一段时间收集了一些百度分享的资源,在整理的过程中过发现好多都失效了。写一个模块,自动检查是否可用。不可用的打一个标记,然后就不处理了。{/callout}相关文章: wordpress直接通过数据库导出文章标题、分类信息到xls表格实现{message type="success" content="Django 添加自定义动作,并编写检查函数"/}class ResourceAdmin(admin.ModelAdmin): """ 管理类 """ ... #自定义动作 actions = ['check_url'] def check_url(self, request, queryset): """ 检查分享链接有效性,并更新状态 """ for obj in queryset: tmp_li = obj.rurl.split('"') if len(tmp_li) > 11: if not check_url_available(tmp_li[7]): obj.rstatus = 2 obj.save() ...{message type="success" content="检查可用性函数"/}import re import time import random from feapder import Request def check_url_available(link): """ 检查百度分享是否失效 """ if not re.match(r'^https?:/{2}\w.+$', link): return False res = Request(link, render=False).get_response() count = 0 for k in res.bs4().find_all('div', class_='share-error-left'): count += 1 if count > 0: return False time.sleep(random.randint(1, 5)) return True
2022年08月12日
25 阅读
2 评论
1 点赞
2022-07-19
利用feapder解析知乎文章标题和url
前言{callout color="#f0ad4e"}想获取某个博主下的文章列表,通过浏览器查看页面源码,然后保存到文件内。然后通过feader的Selector模块将标题和url解析出来。{/callout}python脚本内容from feapder.network.selector import Selector with open('a.html', 'r') as f: text = f.read() selector = Selector(text) r_list = selector.xpath('//div[@class="List-item"]') for r in r_list: title = r.xpath('./div/div/h2/span/a/text()').extract_first() url = r.xpath('./div/div/h2/span/a/@href').extract_first() aurl = "https:{}".format(url) print("{},{}".format(title,aurl))
2022年07月19日
41 阅读
0 评论
1 点赞
2022-07-18
通过Django和Vue.js快速构建项目
前言{card-describe title="项目特点"}vueJS MVVM框架数据双向绑定单文件组件清晰的生命周期学习曲线平滑{/card-describe}{message type="success" content="M(Django)+C(Django)+MVVM(VueJS)=M+MVVM+C=MMVVCM"/}构建过程创建Django项目django-admin startproject ulb_manager python3 manage.py startapp backendvue创建一个前端项目vue-init webpack frontend # 构建 npm install npm run build配置Django使用Django的通用视图 TemplateView#配置 根 urls.py (即 ulb_manager/urls.py) 添加两行 url(r'^$', TemplateView.as_view(template_name="index.html")), url(r'^api/', include('backend.urls', namespace='api'))配置Django项目的模板搜索路径#打开 settings.py (ulb_manager/settings.py),找到TEMPLATES配置项,修改如下: 'DIRS': ['frontend/dist'],配置静态文件搜索路径#打开 settings.py (ulb_manager/settings.py),找到 STATICFILES_DIRS 配置项,配置如下: # Add for vuejs STATICFILES_DIRS = [ os.path.join(BASE_DIR, "frontend/dist/static"), ]开发环境{callout color="#f0ad4e"}使用VueJS的开发环境脱离了Django环境,访问Django的API,出现了跨域问题,有两种解决办法a.在VueJS层上做转发 (proxyTable)b.在Django层注入 header{/callout}#django 端配置 pip3 install django-cors-headers #修改配置文件 vim task_webserver/settings.py ... # 解决跨域问题,注意与common.CommonMiddleware的顺序 'corsheaders.middleware.CorsMiddleware', 'django.middleware.common.CommonMiddleware', # 增加跨域忽略 CORS_ALLOW_CREDENTIALS = True #线上设为False
2022年07月18日
37 阅读
0 评论
4 点赞
1
2
3