mirror of
https://github.com/0-8-4/miui-auto-tasks.git
synced 2024-11-24 09:15:48 +00:00
ea361a0e04
* chore: captcha的参数拷贝，避免修改原参数 chore: 提升pylint分数，使代码更加规范 * fix: `@validator` cannot be applied to instance methods * deleted: .pylintrc * chore: no-else-return * chore: 优化代码，使其更符合PEP 8规范 * chore: 优化代码，使其更符合PEP 8规范 * chore: 优化代码，使其更符合PEP 8规范 * chore: 添加代码规范等级 * fix: fix dockerfile * modified: Dockerfile * chore: 更新docker ci
77 lines
2.1 KiB
Python
77 lines
2.1 KiB
Python
'''
|
|
Date: 2023-11-12 14:05:06
|
|
LastEditors: Night-stars-1 nujj1042633805@gmail.com
|
|
LastEditTime: 2023-11-18 00:32:53
|
|
'''
|
|
from typing import Any, Dict, Optional
|
|
|
|
import httpx
|
|
from onepush import notify
|
|
|
|
from .config import ConfigManager
|
|
from .logger import log
|
|
|
|
_conf = ConfigManager.data_obj
|
|
|
|
|
|
async def get(url: str,
|
|
*,
|
|
headers: Optional[Dict[str, str]] = None,
|
|
params: Optional[Dict[str, Any]] = None,
|
|
timeout: Optional[int] = 20,
|
|
**kwargs) -> httpx.Response:
|
|
"""
|
|
说明：
|
|
httpx的get请求封装
|
|
参数：
|
|
:param url: url
|
|
:param headers: 请求头
|
|
:param params: params
|
|
:param data: data
|
|
:param json: json
|
|
:param timeout: 超时时间
|
|
"""
|
|
async with httpx.AsyncClient() as client:
|
|
return await client.get(url,
|
|
headers=headers,
|
|
params=params,
|
|
timeout=timeout,
|
|
**kwargs)
|
|
|
|
|
|
async def post(url: str,
|
|
*,
|
|
headers: Optional[Dict[str, str]] = None,
|
|
params: Optional[Dict[str, Any]] = None,
|
|
timeout: Optional[int] = 20,
|
|
**kwargs) -> httpx.Response:
|
|
"""
|
|
说明：
|
|
httpx的post请求封装
|
|
参数：
|
|
:param url: url
|
|
:param headers: 请求头
|
|
:param params: params
|
|
:param data: data
|
|
:param json: json
|
|
:param timeout: 超时时间
|
|
"""
|
|
async with httpx.AsyncClient() as client:
|
|
return await client.post(url,
|
|
headers=headers,
|
|
params=params,
|
|
timeout=timeout,
|
|
**kwargs)
|
|
|
|
|
|
def notify_me(content=""):
|
|
"""
|
|
默认推送日志
|
|
"""
|
|
notifier = _conf.ONEPUSH.notifier
|
|
params = _conf.ONEPUSH.params
|
|
if not notifier or not params:
|
|
log.error('未配置推送或未正确配置推送')
|
|
return False
|
|
return notify(notifier, content=content, **params)
|