Python系統(tǒng)交互庫解析過程
系統(tǒng)交互庫
os 模塊
os 模塊提供了跨平臺的文件和目錄操作方法,是 Python 標準庫中最常用的系統(tǒng)接口之一:
文件操作:
os.rename(src, dst)- 重命名文件或目錄os.remove(path)- 刪除文件(不能刪除目錄)
os.path 子模塊提供:
os.path.exists(path)- 檢查路徑是否存在os.path.isfile(path)/os.path.isdir(path)- 判斷路徑類型os.path.getsize(path)- 獲取文件大小
目錄操作:
os.mkdir(path)- 創(chuàng)建單個目錄os.makedirs(path)- 遞歸創(chuàng)建多級目錄os.listdir(path)- 列出目錄內容os.chdir(path)- 改變當前工作目錄os.walk(top)- 目錄樹生成器,遞歸遍歷
路徑處理:
os.path.join(path1, path2)- 智能拼接路徑(跨平臺兼容)os.path.abspath(path)- 獲取絕對路徑os.path.dirname(path)/os.path.basename(path)- 獲取目錄名/文件名os.path.splitext(path)- 分離文件名和擴展名
系統(tǒng)信息:
os.name- 操作系統(tǒng)名稱('posix', 'nt', 'java')os.environ- 環(huán)境變量字典os.getcwd()- 獲取當前工作目錄os.cpu_count()- 獲取CPU核心數
sys 模塊
sys 模塊實現了與 Python 解釋器的深度交互:
命令行參數:
sys.argv- 腳本參數列表(第一個元素是腳本名稱)- 示例:
python script.py arg1 arg2→sys.argv = ['script.py', 'arg1', 'arg2']
模塊搜索路徑:
sys.path- Python 模塊搜索路徑列表- 可動態(tài)修改:
sys.path.append('/custom/path')
標準輸入輸出:
sys.stdin/sys.stdout/sys.stderr- 標準 I/O 流對象- 重定向示例:
sys.stdout = open('output.log', 'w')
print("這將寫入文件")
解釋器信息:
sys.version- Python 版本信息sys.platform- 操作系統(tǒng)平臺標識符sys.executable- Python 解釋器路徑sys.exit([status])- 退出程序(0表示成功)
其他重要功能:
sys.getsizeof(obj)- 獲取對象占用的內存大小sys.getrecursionlimit()- 獲取遞歸深度限制sys.settrace()- 設置跟蹤函數用于調試
subprocess 模塊
subprocess 提供了更安全的進程管理方式,替代傳統(tǒng)的 os.system():
基本用法:
# 運行簡單命令
result = subprocess.run(["ls", "-l"], capture_output=True, text=True)
print(result.stdout)
# 檢查返回碼
if result.returncode != 0:
print("命令執(zhí)行失敗")
高級控制:
subprocess.Popen()- 創(chuàng)建子進程并返回句柄
proc = subprocess.Popen(["python", "worker.py"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE)
管道連接:
subprocess.PIPE- 創(chuàng)建管道
p1 = subprocess.Popen(["cat", "file.txt"], stdout=subprocess.PIPE) p2 = subprocess.Popen(["grep", "keyword"], stdin=p1.stdout, stdout=subprocess.PIPE)
錯誤處理:
subprocess.CalledProcessError- 命令返回非零狀態(tài)時拋出- 安全執(zhí)行:
try:
subprocess.run(["rm", "important_file"], check=True)
except subprocess.CalledProcessError as e:
print(f"命令失敗: {e}")
其他功能:
timeout參數 - 設置命令超時時間shell=True- 通過系統(tǒng)shell執(zhí)行命令(有安全風險)env參數 - 設置子進程環(huán)境變量
遠程管理工具
paramiko 庫
paramiko 是純 Python 實現的 SSHv2 協議庫:
SSH 連接:
import paramiko
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('hostname', username='user', password='pass')
執(zhí)行命令:
stdin, stdout, stderr = client.exec_command('ls -l')
print(stdout.read().decode())
文件傳輸:
sftp = client.open_sftp()
sftp.put('local.txt', 'remote.txt') # 上傳
sftp.get('remote.txt', 'local.txt') # 下載
sftp.close()
密鑰認證:
key = paramiko.RSAKey.from_private_key_file('/path/to/key.pem')
client.connect('host', username='user', pkey=key)
高級功能:
- SSH 隧道
- 保持連接
- 主機密鑰驗證
- 交互式 shell
Fabric 工具
Fabric2 是新一代的批量任務框架,構建在 Paramiko 之上:
任務定義:
from fabric import task
@task
def deploy(c):
c.run('git pull')
c.run('pip install -r requirements.txt')
主機管理:
@task
def check_disk(c):
# 單個連接
result = c.run('df -h', hide=True)
print(f"磁盤使用:\n{result.stdout}")
# 多主機
for connection in c:
with connection.cd('/var/log'):
connection.run('ls -l')
批量執(zhí)行:
from fabric import Connection, ThreadingGroup
hosts = ['web1', 'web2', 'db1']
group = ThreadingGroup(*hosts, user='admin')
result = group.run('hostname')
for connection, result in result.items():
print(f"{connection.host}: {result.stdout.strip()}")
配置系統(tǒng):
fabric.yml配置文件:
user: deploy connect_kwargs: key_filename: /path/to/key.pem hosts: - web1.example.com - web2.example.com
- 或通過 Python 配置:
c = Connection('host', user='user', connect_kwargs={'password': 'pass'})
其他特性:
- 并行執(zhí)行
- 失敗處理
- 上下文管理器
- 本地和遠程操作混合
網絡通信庫
socket 模塊
socket 提供底層網絡接口:
TCP 通信:
# 服務器端
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('0.0.0.0', 8000))
server_socket.listen(1)
conn, addr = server_socket.accept()
# 客戶端
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.connect(('localhost', 8000))
UDP 通信:
# 服務器端
udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp_socket.bind(('0.0.0.0', 8000))
data, addr = udp_socket.recvfrom(1024)
# 客戶端
udp_socket.sendto(b'Hello', ('localhost', 8000))
常用方法:
socket.settimeout()- 設置超時時間socket.getaddrinfo()- 地址解析socket.gethostname()- 獲取主機名socket.shutdown()- 優(yōu)雅關閉連接
高級功能:
- 非阻塞 I/O
- 套接字選項設置
- 多路復用 (select/poll)
- SSL 包裝
requests 庫
requests 提供了人性化的 HTTP 客戶端:
簡單請求:
# GET 請求
response = requests.get('https://api.example.com/data')
print(response.json())
# POST 請求
payload = {'key1': 'value1', 'key2': 'value2'}
response = requests.post('https://api.example.com/post', data=payload)
參數傳遞:
params- URL 查詢參數
requests.get('https://api.example.com/search', params={'q': 'python'})
json- JSON 請求體
requests.post('https://api.example.com/data', json={'key': 'value'})
會話保持:
session = requests.Session()
session.get('https://example.com/login', auth=('user', 'pass'))
response = session.get('https://example.com/dashboard')
高級功能:
- 代理支持:
proxies = {'http': 'http://proxy.example.com:8080'}
requests.get('http://example.com', proxies=proxies)
- 超時設置:
requests.get('http://example.com', timeout=5)
- SSL 驗證:
requests.get('https://example.com', verify='/path/to/cert.pem')
響應處理:
response.status_code- HTTP 狀態(tài)碼response.headers- 響應頭response.content- 二進制響應體response.text- 文本響應體response.json()- JSON 解析
數據處理工具
json 模塊
json 模塊用于 API 數據交互:
序列化:
data = {'name': 'Alice', 'age': 30, 'skills': ['Python', 'SQL']}
json_str = json.dumps(data, indent=2) # 美化輸出
反序列化:
data = json.loads('{"name": "Alice", "age": 30}')
文件操作:
# 寫入文件
with open('data.json', 'w') as f:
json.dump(data, f)
# 讀取文件
with open('data.json') as f:
data = json.load(f)
高級功能:
- 自定義編碼器:
class CustomEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
return super().default(obj)
json.dumps(obj, cls=CustomEncoder)
- 解析參數:
json.loads(json_str, object_hook=my_decoder)
csv 模塊
csv 模塊用于報表生成:
讀取 CSV:
with open('data.csv') as f:
reader = csv.reader(f)
for row in reader:
print(row)
寫入 CSV:
with open('output.csv', 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(['Name', 'Age'])
writer.writerow(['Alice', 30])
字典格式:
# 讀取
with open('data.csv') as f:
reader = csv.DictReader(f)
for row in reader:
print(row['Name'], row['Age'])
# 寫入
with open('output.csv', 'w') as f:
fieldnames = ['Name', 'Age']
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerow({'Name': 'Alice', 'Age': 30})
方言支持:
csv.register_dialect('mydialect', delimiter='|', quoting=csv.QUOTE_MINIMAL)
with open('data.psv') as f:
reader = csv.reader(f, dialect='mydialect')
yaml 模塊
yaml 用于配置文件解析:
基本用法:
import yaml
# 解析 YAML
with open('config.yaml') as f:
config = yaml.safe_load(f)
# 生成 YAML
data = {'server': {'host': '127.0.0.1', 'port': 8000}}
yaml_str = yaml.dump(data, default_flow_style=False)
高級特性:
- 錨點和別名:
defaults: &defaults adapter: postgres host: localhost development: <<: *defaults database: dev
- 自定義標簽
- 多文檔流
安全考慮:
- 始終優(yōu)先使用
yaml.safe_load()而非yaml.load() - 避免加載不受信任的 YAML 文件
- 復雜對象需要自定義表示器/構造器
與其他格式對比:
- 比 JSON 更易讀
- 支持注釋
- 比 XML 更簡潔
- 適合配置文件和復雜數據結構
總結
以上為個人經驗,希望能給大家一個參考,也希望大家多多支持腳本之家。
相關文章
python使用Selenium和cookie繞過驗證碼實現登錄示例代碼
文章介紹了如何使用Selenium獲取登錄過程中的cookie信息,并通過抓包工具獲取cookie,通過示例展示了如何在Dsmall和CRM系統(tǒng)中實現免登陸操作,總結了兩種方法:一種是在獲取到cookie后直接添加到當前會話中,另一種是將cookie存儲到本地文件并在后續(xù)請求中使用2025-01-01
Python實戰(zhàn)基礎之Pandas統(tǒng)計某個數據列的空值個數
我們在處理數據的時候,經常需要檢查數據的質量,也需要知道出問題的數據在哪個位置,下面這篇文章主要給大家介紹了關于Python實戰(zhàn)基礎之利用Pandas統(tǒng)計某個數據列空值個數的相關資料,需要的朋友可以參考下2022-08-08

