
15 个实用的网络编程脚本覆盖 HTTP、TCP、UDP、FTP、异步、邮件、DNS 等常见场景,代码改改路径和参数基本能直接跑。
用 requests 发 GET 请求,拿到网页 HTML。最简单的网络操作,检查网站可访问性也可以用这个。
importrequests
url = 'https://www.example.com'
response = requests.get(url)
print(response.text) # 输出网页HTML源码向接口提交 JSON 数据,模拟用户登录、API 调用等场景。
importrequests
data = {'username': 'test', 'password': '123456'}
response = requests.post('https://httpbin.org/post', json=data)
print(response.json()) # 解析并打印服务器返回的JSON响应用 socket 模块建立 TCP 连接,适合和非 HTTP 服务通信,比如自定义协议的数据库或游戏服务器。
importsocket
client = 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()尝试连接指定端口,检测是否开放。只在自己有权限的机器上用,扫别人的服务器请先授权。
importsocket
defport_scanner(host, port):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(1)
try:
sock.connect((host, port))
print(f'端口 {port} 开放')
returnTrue
except (socket.timeout, ConnectionRefusedError):
returnFalse
finally:
sock.close()
# 扫描本地80和443端口
port_scanner('localhost', 80)
port_scanner('localhost', 443)在局域网内广播消息,所有监听该端口的设备都能收到。适合局域网设备发现、服务通告。
importsocket
sock = 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 拉取资源。
fromftplibimportFTP
defdownload_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, BaseHTTPRequestHandler
classSimpleHandler(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 时能明显加速。
importrequests
fromconcurrent.futuresimportThreadPoolExecutor, as_completed
urls = [
'https://httpbin.org/get',
'https://api.github.com',
'https://www.example.com'
]
deffetch_url(url):
resp = requests.get(url)
returnurl, resp.status_code
withThreadPoolExecutor(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}')每来一个连接就开一个线程处理,适合学习多线程网络编程,或者搭个局域网内的简单即时通讯原型。
importsocket
importthreading
defhandle_client(client_socket, address):
print(f'新连接来自: {address}')
whileTrue:
try:
message = client_socket.recv(1024).decode('utf-8')
ifnotmessage:
break
print(f'收到来自 {address} 的消息: {message}')
# 这里可以添加广播给其他客户端的逻辑
client_socket.send(f'服务器已收到: {message}'.encode('utf-8'))
exceptConnectionResetError:
break
client_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。
importrequests
frombs4importBeautifulSoup
url = '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 等请求头,绕过简单的反爬检测,模拟浏览器访问。
importrequests
headers = {
'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 密集的场景比多线程更省资源,适合高并发爬虫或微服务间通信。
importaiohttp
importasyncio
asyncdeffetch(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 服务器发邮件,常见于系统告警、注册验证码、自动报表。注意用应用专用密码而不是账户密码。
importsmtplib
fromemail.mime.textimportMIMEText
fromemail.headerimportHeader
sender = '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 时用得上。
importsocket
defresolve_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}')“无他,惟手熟尔”!有需要的用起来!
如果你觉得这篇文章有用,欢迎点赞、转发、收藏、留言、推荐❤!
本文分享自 Nicholas与Pypi 微信公众号,前往查看
如有侵权,请联系 cloudcommunity@tencent.com 删除。
本文参与 腾讯云自媒体同步曝光计划 ,欢迎热爱写作的你一起参与!