
15 个实用的网络编程脚本覆盖 HTTP、TCP、UDP、FTP、异步、邮件、DNS 等常见场景,代码改改路径和参数基本能直接跑。
往期阅读>>>
Python 20 个文本分析的库:效率提升 10 倍的秘密武器
Python 自动化管理Jenkins的15个实用脚本,提升效率
App2Docker:如何无需编写Dockerfile也可以创建容器镜像
Python 自动化识别Nginx配置并导出为excel文件,提升Nginx管理效率
用 requests 发 GET 请求,拿到网页 HTML。最简单的网络操作,检查网站可访问性也可以用这个。
importrequestsurl = 'https://www.example.com'response = requests.get(url)print(response.text) # 输出网页HTML源码
向接口提交 JSON 数据,模拟用户登录、API 调用等场景。
importrequestsdata = {'username': 'test', 'password': '123456'}response = requests.post('https://httpbin.org/post', json=data)print(response.json()) # 解析并打印服务器返回的JSON响应
用 socket 模块建立 TCP 连接,适合和非 HTTP 服务通信,比如自定义协议的数据库或游戏服务器。
importsocketclient = socket.socket(socket.AF_INET, socket.SOCK_STREAM)client.connect(('www.google.com', 80))client.sendall(b'GET / HTTP/1.1\r\nHost: google.com\r\n\r\n')response = client.recv(4096)print(response.decode('utf-8'))client.close()
尝试连接指定端口,检测是否开放。只在自己有权限的机器上用,扫别人的服务器请先授权。
importsocketdefport_scanner(host, port):sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)sock.settimeout(1)try:sock.connect((host, port))print(f'端口 {port} 开放')returnTrueexcept (socket.timeout, ConnectionRefusedError):returnFalsefinally:sock.close()# 扫描本地80和443端口port_scanner('localhost', 80)port_scanner('localhost', 443)
在局域网内广播消息,所有监听该端口的设备都能收到。适合局域网设备发现、服务通告。
importsocketsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)message = b'Device Discovery Hello!'sock.sendto(message, ('<broadcast>', 37020))sock.close()
用 ftplib 连接 FTP 服务器下载文件,适合自动化备份或从公共 FTP 拉取资源。
fromftplibimportFTPdefdownload_file(ftp, remote_path, local_path):withopen(local_path, 'wb') asf:ftp.retrbinary('RETR '+remote_path, f.write)ftp = FTP('ftp.example.com')ftp.login(user='your_username', passwd='your_password')download_file(ftp, '/pub/example.txt', './local_example.txt')ftp.quit()
用标准库 http.server 起一个本地服务,本地开发测试、快速共享文件用得上。
fromhttp.serverimportHTTPServer, BaseHTTPRequestHandlerclassSimpleHandler(BaseHTTPRequestHandler):defdo_GET(self):self.send_response(200)self.send_header('Content-type', 'text/html')self.end_headers()self.wfile.write(b'<h1>Hello from Python Web Server!</h1>')server = HTTPServer(('localhost', 8080), SimpleHandler)print('服务器运行在 http://localhost:8080')server.serve_forever()
用 ThreadPoolExecutor 同时发多个 HTTP 请求,批量抓取或并行调 API 时能明显加速。
importrequestsfromconcurrent.futuresimportThreadPoolExecutor, as_completedurls = ['https://httpbin.org/get','https://api.github.com','https://www.example.com']deffetch_url(url):resp = requests.get(url)returnurl, resp.status_codewithThreadPoolExecutor(max_workers=3) asexecutor:future_to_url = {executor.submit(fetch_url, url): urlforurlinurls}forfutureinas_completed(future_to_url):url, status = future.result()print(f'{url} -> 状态码: {status}')
每来一个连接就开一个线程处理,适合学习多线程网络编程,或者搭个局域网内的简单即时通讯原型。
importsocketimportthreadingdefhandle_client(client_socket, address):print(f'新连接来自: {address}')whileTrue:try:message = client_socket.recv(1024).decode('utf-8')ifnotmessage:breakprint(f'收到来自 {address} 的消息: {message}')# 这里可以添加广播给其他客户端的逻辑client_socket.send(f'服务器已收到: {message}'.encode('utf-8'))exceptConnectionResetError:breakclient_socket.close()print(f'连接关闭: {address}')server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)server.bind(('0.0.0.0', 9999))server.listen(5)print('聊天室服务器启动,监听端口 9999...')whileTrue:client_sock, addr = server.accept()client_thread = threading.Thread(target=handle_client, args=(client_sock, addr))client_thread.start()
用 requests 抓页面,BeautifulSoup 解析 HTML,提取所有超链接。用之前记得看目标站的 robots.txt。
importrequestsfrombs4importBeautifulSoupurl = 'https://www.example.com'try:response = requests.get(url, timeout=5)soup = BeautifulSoup(response.text, 'html.parser')links = [a.get('href') forainsoup.find_all('a', href=True)]forlinkinlinks[:10]: # 打印前10个链接print(link)exceptrequests.RequestExceptionase:print(f'请求失败: {e}')
加上 User-Agent、Cookie 等请求头,绕过简单的反爬检测,模拟浏览器访问。
importrequestsheaders = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36','Accept-Language': 'zh-CN,zh;q=0.9',}cookies = {'session_id': 'your_session_id_here'}response = requests.get('https://httpbin.org/headers', headers=headers, cookies=cookies)print(response.json())
aiohttp + asyncio 做异步请求,I/O 密集的场景比多线程更省资源,适合高并发爬虫或微服务间通信。
importaiohttpimportasyncioasyncdeffetch(session, url):asyncwithsession.get(url) asresponse:returnawaitresponse.text()asyncdefmain():asyncwithaiohttp.ClientSession() assession:html = awaitfetch(session, 'https://www.example.com')print(html[:500]) # 打印前500个字符asyncio.run(main())
用标准库 xmlrpc.client 调用 XML-RPC 接口,WordPress、MediaWiki 等系统都支持这个协议。
fromxmlrpc.clientimportServerProxy# 连接一个公共的测试服务server = ServerProxy('https://httpbin.org/xml')# 调用远程方法(示例,实际方法名需查看API文档)try:# 假设存在一个 `example.method` 的方法# result = server.example.method('arg1')# print(result)print('XML-RPC客户端已就绪,请根据实际API文档调用方法。')exceptExceptionase:print(f'调用失败: {e}')
用 smtplib 连 SMTP 服务器发邮件,常见于系统告警、注册验证码、自动报表。注意用应用专用密码而不是账户密码。
importsmtplibfromemail.mime.textimportMIMETextfromemail.headerimportHeadersender = 'your_email@gmail.com'receiver = 'receiver_email@example.com'password = 'your_app_password'# 注意:使用应用专用密码message = MIMEText('这是一封来自Python的测试邮件。', 'plain', 'utf-8')message['From'] = Header(sender, 'utf-8')message['To'] = Header(receiver, 'utf-8')message['Subject'] = Header('Python SMTP 测试邮件', 'utf-8')try:smtp_obj = smtplib.SMTP_SSL('smtp.gmail.com', 465)smtp_obj.login(sender, password)smtp_obj.sendmail(sender, [receiver], message.as_string())print('邮件发送成功')smtp_obj.quit()exceptsmtplib.SMTPExceptionase:print(f'邮件发送失败: {e}')
用 socket.getaddrinfo 把域名解析成 IP,写网络诊断工具或需要获取服务 IP 时用得上。
importsocketdefresolve_dns(hostname):try:# 获取地址信息,返回 (family, type, proto, canonname, sockaddr)addr_info = socket.getaddrinfo(hostname, None)ips = {info[4][0] forinfoinaddr_info} # 提取IP地址,去重returnlist(ips)exceptsocket.gaierrorase:returnf'DNS解析失败: {e}'hostname = 'www.google.com'ip_addresses = resolve_dns(hostname)print(f'{hostname} 解析到的IP地址: {ip_addresses}')
