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