人生边上
0%

Quick and dirty Tools in Python

发表于 更新于

https://github.com/91Quin/Quick-and-dirty-Tools-in-Python

设置你的Python环境

Python-venv

之前并没有听说过这个包，这么一看真是非常整洁

廖雪峰Python教程 venv

基础的网络编程工具

TCP客户端

廖雪峰 TCP编程

1
2
3
4
5
6
7
8
9
10
11
12
13
import socket

HOST = 'www.google.com'
PORT = 80

client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect((HOST, PORT))

client.send(b'GET / HTTP/1.1\r\nHost: google.com\r\n\r\n')
response = client.recv(4096)
print(response.decode('utf-8'))

client.close()
  • 创建Socket时，AF_INET指定使用IPv4协议，如果要用更先进的IPv6，就指定为AF_INET6。SOCK_STREAM指定使用面向流的TCP协议
  • recvfrom（）在接受数据的同时也会给出地址
    廖雪峰教程中的示例：
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    # 接收数据:
    buffer = []
    while True:
        # 每次最多接收1k字节:
        d = s.recv(1024)
        if d:
            buffer.append(d)
        else:
            break
    data = b''.join(buffer)
    1
    2
    3
    4
    5
    6
    7
    8
    header, html = data.split(b'\r\n\r\n', 1) #分割header和body
    print(header.decode('utf-8'))
    # 把接收的数据写入文件:

    #以二进制打开一个文件只用于写入，如果该文件已存在则打开文件，并从开头开始编辑，即原有内容会被删除。如果该文件不存在，创建新文件。挨为什么我这个#是多行注释？

    with open('sina.html', 'wb') as f: 
        f.write(html)
  • 接收数据时，调用recv(max)方法，一次最多接收指定的字节数，因此，在一个while循环中反复接收，直到recv()返回空数据，表示接收完毕，退出循环。
  • b’’.join(buffer)将字符串标识为bytes类型
  • ```python
    str.split(str=””, num=string.count(str))
    1. str 分隔符
    2. num 分割次数
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      import socket

      s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
      s.connect(("www.qdu.edu.cn", 80))  # 里层括号为了传入'address:port'
      s.send(b"GET / HTTP/1.1\r\nHost: www.qdu.edu.cn\r\nConnection: close\r\n\r\n")
      buffer = []
      while True:
          d = s.recv(2048)
          if d:
              buffer.append(d)
          else:
              break
      data = b"".join(buffer)

      s.close()
      header, body = data.split(b"\r\n\r\n", 1)
      print(header.decode("utf-8"))
      with open("qdu.html", "wb") as f:
          f.write(body)

TCP服务端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import socket
import threading

IP = '0.0.0.0'
PORT = 9998


def main():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.bind((IP, PORT))
    server.listen(5)
    print(f'[*] Listening on {IP}:{PORT}')

    while True:
        client, address = server.accept()
        print(f'[*] Accepted connection from {address[0]}:{address[1]}')
        client_handler = threading.Thread(target=handle_client, args=(client,))
        client_handler.start()


def handle_client(client_socket):
    with client_socket as sock:
        request = sock.recv(1024)
        print(f'[*] Received: {request.decode("utf-8")}')
        sock.send(b'ACKKK')


if __name__ == '__main__':
    main()
  • 注意 在云服务器上进行代码测试时，可能需要修改安全组规则，放通相应端口。我说我怎么老是不行