-
时间处理方法：
1. 获取当前系统时间，以%Y-%m-%d %H:%M:%S方式进行格式化，精确到秒：
def current_datetime(fmt="%Y-%m-%d %H:%M:%S"):
"""
:return: 2024-08-31 10:41:45
"""
return datetime.datetime.now().strftime(fmt)2. 获取当前日期
def get_today():
"""
:return: 当前时间 年-月-日
"""
my_today = datetime.date.today()
return str(my_today)3. 获取相对系统时间，以%Y-%m-%d %H:%M:%S方式进行格式化，精确到秒，delta为正代表未来时间，为负代表过去时间，为0代表现在时间
def get_time_str(delta=0, fmt="%Y-%m-%d %H:%M:%S"):
"""
:return: 2024-08-31 10:41:45
"""
my_time = int(time.time()) + delta * 24 * 3600
return time.strftime(fmt, time.localtime(my_time))4. 打印带时间戳的日志，调用上方get_time_str方
def printf(my_str):
print(f'{[{get_time_str(0)}]}{my_str}')
5. 获取linux时间戳，单位为ms, delta为正代表未来时间，为负代表过去时间，为0代表现在时间
def get_delay_linux_time(delta=0):
"""
:return: 1724986889058
"""
my_linux_time = int(time.time()*1000) + delta * 24 * 3600
return my_linux_time
6. 将指定格式比如 dt=2023-12-28 15:34:34 转换为linux时间戳，delta为正代表未来时间，为负代表过去时间，为0代表现在时间
def datetime_to_timestamp(dt, fmt="%Y-%m-%d %H:%M:%S", delta=0):
"""
:return: 1724985705
"""
return int(time.mktime(time.strptime(dt, fmt))) + delta * 24 * 3600-
文件操作：
1. 创建目录
def make_dir(my_dir_path):
if not os.path.exists(my_dir_path):
os.mkdir(my_dir_path)2. 读取excel
def read_excel(filename, sheet_index=0):
wb = load_workbook(filename)
ws = wb.worksheets[sheet_index]
# ws = wb.active
data = []
for d in ws.iter_rows(max_row=ws.max_row, max_col=ws.max_column):
data.append([x.value for x in d])
return data3. 写入excel
def write_to_excel(filename, data):
wb = Workbook()
ws = wb.active
if isinstance(data, (list, tuple)):
for row in range(0, len(data)):
ws.append(data[row])
wb.save(filename)4. 读取csv文件
def read_csv(my_csv_path):
my_list = []
with open(my_csv_path) as my_csv_file:
my_reader = csv.reader(my_csv_file)
for my_row in list(my_reader):
# print(''.join(my_row))
my_param_list = ','.join(my_row).split(',')
my_list.append(my_param_list)
return my_list5. 写入json文件
def write_json(json_path, my_json):
with open(json_path, 'w', encoding='utf-8') as f:
json.dump(my_json, f, ensure_ascii=False)
6. 读取json文件
def read_json_file(json_path):
with open(json_path, 'r', encoding='utf-8') as f:
return json.load(f)7. 追加写入txt文件
def write_txt(my_content, my_file_path):
with open(my_file_path, 'a', encoding='utf-8') as f: # 使用 'a' 模式
f.write(my_content + '\n')
print(f'{my_content}')
8. 压缩zip文件
def compress_files(src_file_path, zip_file_full_path):
# 要压缩的文件夹路径
folder_path = src_file_path
# 压缩文件保存路径及名称
zip_filename = zip_file_full_path
# 使用shutil模块的make_archive()函数压缩文件夹
shutil.make_archive(zip_filename, 'zip', folder_path)
print('文件夹已成功压缩为', zip_filename + '.zip')
http请求
1. 发送post请求，如果是静态数据请求，可以设置max_retries值，如果是异步长时间任务执行，max_retries慎用
def send_post_request(my_url, my_header, my_data,max_retries=0):
retry_count = 0 # 当前重试次数
while retry_count < max_retries:
try:
my_response = requests.post(my_url, data=json.dumps(my_data), headers=my_header, timeout=15).text
my_dict = json.loads(my_response)
return my_dict # 请求成功，跳出循环
except requests.exceptions.Timeout:
retry_count += 1
print(f"正在进行第 {retry_count} 次重试...")
except requests.exceptions.RequestException as e:
print(f"请求异常：{e}")
return None # 可根据需求选择继续重试或者直接结束
if retry_count >= max_retries:
print("达到最大重试次数，请求失败。")
return None
# else:
# print("请求成功，响应内容：", response_text)
2. 发送get请求，如果是静态数据请求，可以设置max_retries值，如果是异步长时间任务执行，max_retries慎用
def send_get_request(my_url, my_header,max_retries=0):
retry_count = 0 # 当前重试次数
while retry_count < max_retries:
try:
my_response = requests.get(my_url, headers=my_header, timeout=15).text
my_dict = json.loads(my_response)
return my_dict # 请求成功，跳出循环
except requests.exceptions.Timeout:
retry_count += 1
print(f"正在进行第 {retry_count} 次重试...")
except requests.exceptions.RequestException as e:
print(f"请求异常：{e}")
return None # 可根据需求选择继续重试或者直接结束
if retry_count >= max_retries:
print("达到最大重试次数，请求失败。")
return None
路径获取，兼容linux和windows
# 如果当前脚本文件test.py存在于工程目录：my_project/app/test.py
# 且存在同级目录data：my_project/app/data/
self.base_dir = os.path.abspath(os.path.dirname(__file__)) # 获取文件绝对路径
self.root_dir = self.base_dir.split('app')[0] # 获取工程根路径
self.my_data_dir = os.path.join(self.root_dir, 'data') # 获取同级data路径