mirror of
https://github.com/0-8-4/miui-auto-tasks.git
synced 2024-12-04 20:27:24 +00:00
6ade926505
* add missing dependency: apscheduler fix some bugs in Dockerfile * 🔧 自动更新requirements * fix: 解决某些情况下不能正确生成、读取配置文件 update: 将python版本升级到3.12，迁移pydantic v2依赖 修改Dockerfile * 🔧 自动更新requirements * Pylint * Hadolint - DL3042 https://github.com/hadolint/hadolint/wiki/DL3042 --------- Co-authored-by: JaHIY <jaklsy@gmail.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: Night-stars-1 <99261160+Night-stars-1@users.noreply.github.com>
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)
|