1.os

文件目录

import os# 创建文件夹
os.mkdir(dir)
# 判断文件是否存在
os.path.exists(path)
# 列出文件夹下文件列表 
os.listdir(dir)""" 常用 """
# 当前文件相对路径
os.getcwd()# 当前文件绝对路径
os.path.abspath(__file__)# 当前文件所在的的文件夹 （常用）
os.path.dirname(__file__)# 添加文件路径用sys
sys.path.append(os.path.abspath(os.pardir))   # 添加父目录print '***获取上级目录***'
print os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
print os.path.abspath(os.path.dirname(os.getcwd()))
print os.path.abspath(os.path.join(os.getcwd(), ".."))print '***获取上上级目录***'
print os.path.abspath(os.path.join(os.getcwd(), "../.."))# 删除文件       
os.remove(my_file)
# 删除文件夹     
os.rmdir(path)（需为空目录）
# 递归删除目录
os.removedirs(path)   
# 修改当前工作目录
os.chdir(path)

os.sep 跨平台路径

python是跨平台的。在Windows上，文件的路径分隔符是'\'，在Linux上是'/'

使用 os.sep 的话，就不用考虑这个了，os.sep根据你所处的平台，自动采用相应的分隔符号。
在替换路径字符串或者合并字符串的时候有用。

2.sys

查看python版本

import sys
print(sys.version) # python 版本print(sys.executable) # python 路径

添加python环境路径

import sys# 添加路径
sys.path.append("../src/")# 添加父目录
sys.path.append(os.path.abspath(os.pardir))

3.time / datetime

from datetime import datetime, timedeltatimedelta
time_step = 5 * np.timedelta64(1, 'm')
timedelta转为float形式时间。
（1）time_d_float = time_d.total_seconds()
（2）两个timedelta形式相除：
time_d_min = time_d / datetime.timedelta(minutes=1)
time_d_ms  = time_d / datetime.timedelta(milliseconds=1)
DatetimeIndex

datetime <---> str

# datetime ---> str
# datetime(2022, 12, 28, 21, 50)  --> 2022-12-28 21:50:00
val_dt.strftime("%Y-%m-%d %H:%M:%S")# str ---> datetime
# 2022-12-28 21:50:00  --> datetime(2022, 12, 28, 21, 50) 
datetime.strptime(val_str, "%Y-%m-%d %H:%M:%S")

datetime <---> unix_timestamp

# datetime ---> unix_timestamp 
# datetime(2022, 12, 28, 21, 50) --> 1672235400 
方法一：
time.mktime(val_datetime.timetuple())
方法二：
int(val_datetime.strftime('%s'))# unix_timestamp ---> datetime
# 1672235400 --> datetime(2022, 12, 28, 21, 50)
datetime.fromtimestamp(val_unixTimestamp)

4.json

json/dict 相互转换

loads()：将json数据转化成dict数据

dumps()：将dict数据转化成json数据

load()：读取json文件数据，转成dict数据

dump()：将dict数据转化成json数据后写入json文件

# 写json文件
json_str = json.dumps(input_dict)
with open('test_data.json', 'w') as json_file:json_file.write(json_str)# 读json文件
with open('./data/test_data.json', encoding='utf-8') as f:line = f.readline()           input_dict = json.loads(line) 

5.正则表达式-[re]

常见表达式
\d   匹配所有的数字
\D   匹配所有，但是数字除外
\s   空格
\S   匹配所有但是空格除外
\w   匹配所有的字母
\W   匹配所有但是字母除外
.    任意除换行符 \n
.   表示点符号，斜杠本身是转义字符常见的表达式举例：
{1,3}  表示数字1到3范围
？     匹配0个或者1个结果
$      匹配字符串的结尾部分
^      匹配字符串的开始部分
|      匹配左右表达式任意一个
[]     字符集任意范围,例如[A-Z]表示大写A到Z
{x}    计算一共找到x的数量一些空格符：
\n     换行
\s     空格
\t     一个tab，制表符
\e     转换字符串，转义
\f     form feed
\r     return\"     "

1.贪婪匹配和非贪婪匹配

“?” 来控制正则贪婪和非贪婪匹配的情况。加？为贪婪匹配，匹配最早符合要求的字符串。多用在如果是标点符号结尾等情况，非贪婪匹配可能会匹配到很长的字符串。

2.提取指定内容

多用在日志/文本分析。

可以使用(.+?)来提取。

import re
str = 'name: 123liyue'
re.findall(r"name: (.+?)liyue")  # 结果为list, 表示提取出的所有结果。

3.小括号()

正则表达式中的小括号的作用是对字符进行分组，并保存匹配的文本。与位于小括号之间的模式匹配的内容都会被捕获。

4.或

1.竖线 |

竖线 | ，表示或，具有最低优先级，例如：python|Python 可以匹配 python 或 Python。

2.中括号[]

[abc] ，表示匹配abc中的任一个字符
[a-z] ，表示匹配a-z中的任一字符
[0-9] ，表示匹配0-9中的任一数字

注意：小括号表示分组，中括号不表示分组。[rack|Rack|机柜] 这样是错误的，表达的是任何一个字符！这里应该用小括号（rack|Rack|机柜]）

举例：

.提取所有的数字

方法一：直接提取所有的数字

方法二：剔除所有非数字

    return re.sub(r'\D', "", str)

参考：python最全的常用正则表达式大全——包括校验数字、字符、一些特殊的需求等等_python正则表达式 数字字母和特殊符号-CSDN博客

6.日志(logging)

相比print，logging模块可以通过改变level来控制一些语句是否被输出，方便调试。

使用：

import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)logger.info('Start reading database')
# read database here
records = {'john': 55, 'tom': 66}
logger.debug('Records: %s', records)
logger.info('Updating records ...')
# update records here
logger.info('Finish updating records')# 运行结果：
# INFO:__main__:Start reading database
# INFO:__main__:Updating records ...
# INFO:__main__:Finish updating records

1.logging模块的日志级别

日志等级（level）	描述
DEBUG	最详细的日志信息，典型应用场景是 问题诊断
INFO	信息详细程度仅次于DEBUG，通常只记录关键节点信息，用于确认一切都是按照我们预期的那样进行工作
WARNING	当某些不期望的事情发生时记录的信息（如，磁盘可用空间较低），但是此时应用程序还是正常运行的
ERROR	由于一个更严重的问题导致某些功能不能正常运行时记录的信息
CRITICAL	当发生严重错误，导致应用程序不能继续运行时记录的信息

2.logging模块的使用方式

logging模块提供了两种记录日志的方式：

• 第一种方式是使用logging提供的模块级别的函数
• 第二种方式是使用Logging日志系统的四大组件

其实，logging所提供的模块级别的日志记录函数也是对logging日志系统相关类的封装而已。

logging模块定义的模块级别的常用函数

函数	说明
logging.debug(msg, *args, **kwargs)	创建一条严重级别为DEBUG的日志记录
logging.info(msg, *args, **kwargs)	创建一条严重级别为INFO的日志记录
logging.warning(msg, *args, **kwargs)	创建一条严重级别为WARNING的日志记录
logging.error(msg, *args, **kwargs)	创建一条严重级别为ERROR的日志记录
logging.critical(msg, *args, **kwargs)	创建一条严重级别为CRITICAL的日志记录
logging.log(level, *args, **kwargs)	创建一条严重级别为level的日志记录
logging.basicConfig(**kwargs)	对root logger进行一次性配置

其中logging.basicConfig(**kwargs)函数用于指定“要记录的日志级别”、“日志格式”、“日志输出位置”、“日志文件的打开模式”等信息，其他几个都是用于记录各个级别日志的函数。

logging模块的四大组件

组件	说明
loggers	提供应用程序代码直接使用的接口
handlers	用于将日志记录发送到指定的目的位置
filters	提供更细粒度的日志过滤功能，用于决定哪些日志记录将会被输出（其它的日志记录将会被忽略）
formatters	用于控制日志信息的最终输出格式

说明： logging模块提供的模块级别的那些函数实际上也是通过这几个组件的相关实现类来记录日志的，只是在创建这些类的实例时设置了一些默认值。

重复打印问题

排查：是否在某处重复设置了logging.getLogger()

6.warning

忽略警告

import warnings
warnings.filterwarnings('ignore')

7.多进程&多线程&协程

multiprocessing.dummy模块与multiprocessing模块的区别：

dummy模块是多线程，而multiprocessing是多进程， api 都是通用的。 方便代码在多线程和多进程之间切换。

一般CPU密集型的选择用多进程，IO密集型的可以选择多线程

多进程有父子进程。

多线程没有父子线程。

There is no parent/child relationship between threads. If thread A creates thread B and then thread A terminates, then thread B will continue to execute.

The exception to this is when the main thread (that is, the thread that runs the main() function) terminates. When this happens, the process terminates and all other threads stop.

进程有 terminate， 进程关掉terminate，进程里创建的子线程也就关掉了。 start,join后 还是在继续跑（因为进程还在后台，没有关掉。或者得用kill)

线程没有 terminate，线程join后，子线程还继续运行（只要main还在运行）

多进程(multiprocessing)

Pool.map：将提供的可迭代对象转换为列表，并一次性为每个项发出任务；问题是如果数量很大，会占很大内存。

Pool.imap：一种替代方案，是map的惰性版本，用于以惰性方式将目标函数应用到可迭代对象中的每个项。

def mp_fun(args):arg0 = args[0]arg1 = args[1]process(arg0, arg1)returnfrom multiprocessing import Pool
pool = Pool(16)
pool.map(persist_table, zip(arg0_list, arg1_list))
pool.close()
pool.join()

对DataFrame多进程

from multiprocessing import Pooldef stats_wrap(df):"""wrap func for multi process"""    df['periods'], df['detected'] = zip(*df["tdf"].apply(stats_method))return dfdef mltproc_stats_wrap(df, maxproc = 16):"""wrap func for multiprocessing"""    pool = Pool(processes = maxproc)    df_lst = np.array_split(df, maxproc)    # 对DataFrame切分res = pool.map(stats_wrap, df_lst)resdf = pd.concat(res)return resdf

多线程

def mp_fun(args):arg0 = args[0]arg1 = args[1]process(arg0, arg1)returnfrom multiprocessing.dummy import Pool
pool = Pool(16)
pool.map(mp_fun, zip(arg0_list, arg1_list))
pool.close()
pool.join()

协程

协程的由来：

历史上，Python 并不支持专门的异步编程语法，因为不需要。

有了多线程（threading）和多进程（multiprocessing），就没必要一定支持异步了。如果一个线程（或进程）阻塞，新建其他线程（或进程）就可以了，程序不会卡死。

但是，多线程有"线程竞争"的问题，处理起来很复杂，还涉及加锁。

asyncio 模块最大特点就是，只存在一个线程。由于只有一个线程，就不可能多个任务同时运行。asyncio 是"多任务合作"模式（cooperative multitasking），允许异步任务交出执行权给其他任务，等到其他任务完成，再收回执行权继续往下执行。

表面上，这是一个不合理的设计，明明有多线程多进程的能力，为什么放着多余的 CPU 核心不用，而只用一个线程呢？但是就像前面说的，单线程简化了很多问题，使得代码逻辑变得简单，写法符合直觉。

import asyncioasync def hello():print("Hello")await asyncio.sleep(5)  # 模拟IO等待5秒print("World")async def name():print("What")await asyncio.sleep(2)  # 模拟IO等待2秒print("your name")async def main():# 使用asyncio.gather并发执行hello和nameawait asyncio.gather(hello(), name())asyncio.run(main())

多线程（Multithreading）

适用场景：

1. I/O密集型任务：例如文件读写、网络请求、数据库操作等。这类任务在等待I/O完成时，线程可以切换到其他任务，提高资源利用率。
2. 现有库依赖：一些第三方库或现有代码可能已经基于多线程实现，直接使用多线程可能更方便。
3. 需要利用多核CPU（有限）：虽然Python的全局解释器锁（GIL）限制了多线程在CPU密集型任务上的表现，但在某些情况下，通过释放GIL（如I/O操作）仍能部分利用多核。

优点：

• 简单易用，标准库threading模块支持良好。
• 适合与传统同步代码集成。

缺点：

• 受GIL限制，多线程在CPU密集型任务上的性能提升有限。
• 线程间切换和同步可能导致复杂性，如死锁、竞态条件等。
• 资源消耗较高，每个线程都有自己的栈空间，创建和销毁线程的开销较大。

协程（Coroutines）

适用场景：

1. 高度I/O密集型任务：如异步网络编程、大量并发的网络请求等。协程能够在单线程内高效地切换任务，避免线程切换的开销。
2. 需要高并发：例如Web服务器、爬虫等，协程能够处理数以万计的并发任务，而不需要创建大量线程。
3. 事件驱动程序：如GUI应用、游戏引擎等，协程适合处理事件和回调。

优点：

• 轻量高效，单个线程内可以运行数以万计的协程，内存和上下文切换开销低。
• 避免了多线程的竞争和锁机制，编程模型更简单。
• 更适合编写异步、非阻塞代码，asyncio等框架支持良好。

缺点：

• 需要使用async/await语法，代码结构可能相对复杂。
• 不适合CPU密集型任务，依然受限于GIL。
• 需要重构现有同步代码，集成异步代码可能有一定难度。

multitasking 装饰器

通过对threading和multiprocessing进行封装，简单方便的通过装饰器方法进行函数加速。

# pip install multitaskingimport multitasking
import time
import random
import signal# kill all tasks on ctrl-c
signal.signal(signal.SIGINT, multitasking.killall)# or, wait for task to finish on ctrl-c:
# signal.signal(signal.SIGINT, multitasking.wait_for_tasks)@multitasking.task # 通过装饰器方法进行函数并行加速
def hello(count):sleep = random.randint(1,10)/2print("Hello %s (sleeping for %ss)" % (count, sleep))time.sleep(sleep)print("Goodbye %s (after for %ss)" % (count, sleep))if __name__ == "__main__":for i in range(0, 10):hello(i+1)

 

8.traceback

打印报错信息：

import traceback  
try:  fun() 
except Exception,e:  traceback.print_exc()  

print_exc()与format_exc()

format_exc()返回字符串。

print_exc()则直接给打印出来。

即traceback.print_exc()与print(traceback.format_exc())效果是一样的。如果要把错误信息输出或赋值，用traceback.format_exc()

9.eval

ast.literal_eval() (推荐)

eval() 存在安全隐患，

ast.literal_eval()函数：则会判断需要计算的内容计算后是不是合法的python类型，如果是则进行运算，否则就不进行运算。

10.assert

如果为False就报错。

可以代替代码中很多的if条件判断错误。

语法格式如下：

assert expression

等价于：

if not expression:     raise AssertionError

assert 后面也可以紧跟参数:

assert expression [, arguments]

等价于：

if not expression:     raise AssertionError(arguments)

例子：

>>> assert 1==2, '1 不等于 2'
Traceback (most recent call last):File "<stdin>", line 1, in <module>
AssertionError: 1 不等于 2

11. UUID(GUID)

uuid是128位的全局唯一标识符（univeral unique identifier），通常用32位的一个字符串的形式来表现。有时也称guid(global unique identifier)。

python中的uuid模块基于信息如MAC地址、时间戳、命名空间、随机数、伪随机数来uuid。具体方法有如下几个：

• uuid.uuid1() 基于MAC地址，时间戳，随机数来生成唯一的uuid，可以保证全球范围内的唯一性。
• uuid.uuid2() 算法与uuid1相同，不同的是把时间戳的前4位置换为POSIX的UID。不过需要注意的是python中没有基于DCE的算法，所以python的uuid模块中没有uuid2这个方法。
• uuid.uuid3(namespace,name) 通过计算一个命名空间和名字的md5散列值来给出一个uuid，所以可以保证命名空间中的不同名字具有不同的uuid，但是相同的名字就是相同的uuid了。【感谢评论区大佬指出】namespace并不是一个自己手动指定的字符串或其他量，而是在uuid模块中本身给出的一些值。比如uuid.NAMESPACE_DNS，uuid.NAMESPACE_OID，uuid.NAMESPACE_OID这些值。这些值本身也是UUID对象，根据一定的规则计算得出。
• uuid.uuid4() 通过伪随机数得到uuid，是有一定概率重复的
• uuid.uuid5(namespace,name) 和uuid3基本相同，只不过采用的散列算法是sha1

11. Global

模块内全局变量

常见错误：local variable 'xxx' referenced before assignment

原因：

在函数内部可以直接引用全局变量，但是不能在函数内部对全局变量重新assignment（赋值）。如要修改，需要在函数内部加上global。

val = xxx
def test():global valval = xxx

注意：global需要在函数内部声明，若在函数外声明，则函数依然无法操作x！

跨模块的全局变量

在Python中，我们创建一个模块config.py来保存全局变量，并在同一程序中的Python模块之间共享信息。这是我们如何在python模块之间共享全局变量。

示例4：跨Python模块共享全局变量

（直接在config里写，然后import config，就可以用 config.var 来引用）

方法一：

config.py：以存储全局变量

a = 0 
b = "empty"

update.py：更改全局变量

import config 
config.a = 10 
config.b = "alphabet"

main.py：测试值的变化

import config 
import update 
print(config.a)    #10 
print(config.b)    #alphabet

在上文中，我们创建三个文件：config.py，update.py和main.py。

该模块config.py存储a和b的全局变量。在update.py文件中，我们导入config.py模块并修改a和b的值。同样，在main.py文件中，我们同时导入config.py和update.py模块。最后，我们打印并测试全局变量的值，无论它们是否更改。

方法二：（推荐）

1. 全局变量管理模块(globalVar.py)

def _init():""" 初始化 """global _global_dict_global_dict = {}def set_value(key,value):""" 定义一个全局变量 """_global_dict[key] = valuedef get_value(key,defValue=None):""" 获得一个全局变量,不存在则返回默认值 """try:return _global_dict[key]except KeyError:  # 查找字典的key不存在的时候触发return defValue

2. 设置全局变量

import global_var# 初始化全局变量，只在main模块初始化一次即可
global_var._init()# 为全局变量赋值
global_var.set_value('verifyCode', verifyCode)
global_var.set_value('verifyId', verifyId)

3. 使用全局变量

import global_varglobal_var.get_value('verifyId'),
global_var.get_value('verifyCode'),

12.collections

deque 队列

1.指定长度

2.添加队头或队尾

append(item) # 添加一个数据到队列的尾部。与列表的append()方法功能相似。
appendleft(item) # 添加一个数据到队列的头部。与append()的添加方向相反。

13.最大堆/最小堆

heapq

import heapq
data = [1,5,3,2,8,5]
heap = []
for n in data:heapq.heappush(heap, n)
print(heap)
>>>> [1, 2, 3, 5, 8, 5]

99.__future__

Python2 与 Python3 区别

# absolute_import： 在python2.4以前，默认的是相对引入，首先在当前目录下查找目标模块，如果找不到才会去系统的默认目录中查找。
from __future__ import absolute_import# division：在Python 2.x中，如果是整数相除，结果仍是整数，余数会被扔掉。可以通过division实现精确除法。
from __future__ import division# print_function：在Python 3中，print方法仅作为一个函数使用，不加括号会报错。
from __future__ import print_function