Files
python----/端口检测.ipynb
admin aac9f5934d feat: 添加多个功能模块和工具脚本
- 新增websocket客户端和服务端实现
- 添加图片压缩工具和快速压缩脚本
- 实现学生信息处理相关API
- 添加MQTT客户端和消息处理功能
- 更新.gitignore忽略更多文件类型
- 添加数据库操作工具和示例
- 实现多个测试脚本和工具类
2026-04-13 14:47:50 +08:00

34 KiB

In [1]:
import telnetlib
C:\Users\Administrator\AppData\Local\Temp\ipykernel_29352\3305242955.py:1: DeprecationWarning: 'telnetlib' is deprecated and slated for removal in Python 3.13
  import telnetlib
In [10]:
flag = False
try:
    with telnetlib.Telnet("192.168.0.166", "161", timeout=1):
        print("端口5411被占用")
except Exception as e:
    print(e)
    print("端口5411可用")
timed out
端口5411可用
In [ ]:
1619434628
In [2]:
from pysnmp.hlapi import *

def get_connected_macs(interface_index):
    # 1. 获取FDB表中该接口学习到的MAC
    error_indication, error_status, _, var_binds = next(
        getCmd(SnmpEngine(),
               CommunityData('zabbix'),
               UdpTransportTarget(('192.168.0.112', 161)),
               ContextData(),
               ObjectType(ObjectIdentity('BRIDGE-MIB', 'dot1dTpFdbPort', interface_index)))
    )

    # 2. 提取MAC地址列表
    mac_list = []
    if not error_indication and not error_status:
        for var_bind in var_binds:
            mac = str(var_bind[1]).replace(' ', ':')  # 转换MAC格式
            mac_list.append(mac)
    return mac_list

# 测试获取索引1接口(GigabitEthernet1/0/1)连接的设备
print(get_connected_macs(1))
# 可能输出: ['F8:38:8D:CF:7E:1A', '00:11:22:33:44:55'] 
---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
Cell In[2], line 22
     19     return mac_list
     21 # 测试获取索引1接口(GigabitEthernet1/0/1)连接的设备
---> 22 print(get_connected_macs(1))

Cell In[2], line 6, in get_connected_macs(interface_index)
      3 def get_connected_macs(interface_index):
      4     # 1. 获取FDB表中该接口学习到的MAC
      5     error_indication, error_status, _, var_binds = next(
----> 6         getCmd(SnmpEngine(),
      7                CommunityData('zabbix'),
      8                UdpTransportTarget(('192.168.0.112', 161)),
      9                ContextData(),
     10                ObjectType(ObjectIdentity('BRIDGE-MIB', 'dot1dTpFdbPort', interface_index)))
     11     )
     13     # 2. 提取MAC地址列表
     14     mac_list = []

NameError: name 'getCmd' is not defined
In [2]:
'1' in {'1': 'GigabitEthernet1/0/1', '2': 'GigabitEthernet1/0/2', '3': 'GigabitEthernet1/0/3', '4': 'GigabitEthernet1/0/4', '5': 'GigabitEthernet1/0/5', '6': 'GigabitEthernet1/0/6', '7': 'GigabitEthernet1/0/7', '8': 'GigabitEthernet1/0/8', '9': 'GigabitEthernet1/0/9', '10': 'GigabitEthernet1/0/10', '716': 'NULL0', '717': 'InLoopBack0', '719': 'Vlan-interface1'}
Out [2]:
True
In [7]:
from pysnmp.hlapi import *
def snmp_walk(host, oid, format='str', strip_prefix=True, community='zabbix'):
    res = {}
    # 使用正确的 nextCmd 函数
    for (errorIndication,
         errorStatus,
         errorIndex,
         varBinds) in nextCmd(SnmpEngine(),
                             CommunityData(community),
                             UdpTransportTarget((host, 161), timeout=4.0, retries=3),
                             ContextData(),
                             ObjectType(ObjectIdentity(oid)),
                             lookupMib=False,
                             lexicographicMode=False):
        if errorIndication:
            raise ConnectionError(f'SNMP error: "{str(errorIndication)}". Status={str(errorStatus)}')
        elif errorStatus:
            raise ConnectionError(f'SNMP error: "{str(errorStatus)}" at index {str(errorIndex)}')
        else:
            for varBind in varBinds:
                oid_str = str(varBind[0])
                value = varBind[1]
                
                # 如果指定了去除前缀
                if strip_prefix:
                    oid_str = oid_str.replace(oid, '', 1).lstrip('.')
                
                # 根据格式处理值
                if format == 'hex':
                    res[oid_str] = value.asOctets().hex()
                elif format == 'str':
                    res[oid_str] = str(value)
                else:
                    res[oid_str] = value
    return res
In [1]:
from requests import post,get
# post("https://szihj.idsp.yunxiao.com/Open/Access/GetToken?corpId=869376250979578017&corpSecret=GQSYGTLMKNJHVLHZMHYDMHATIGTTOBYP").text
In [ ]:
post("http://127.0.0.1:8100/door/tongtong/log",json={'lockId': '8686987', 'notifyType': '1', 'records': '[{"lockId":8686987,"electricQuantity":100,"serverDate":1758100706693,"recordTypeFromLock":17,"recordType":7,"success":1,"lockMac":"FD:15:3A:7E:D1:E2","keyboardPwd":"3157220353","lockDate":1758100692000,"username":"夏天乐带0"}]', 'admin': 'xtlft@qq.com', 'lockMac': 'FD:15:3A:7E:D1:E2'}).json()
---------------------------------------------------------------------------
ConnectionRefusedError                    Traceback (most recent call last)
File d:\Anaconda3\Lib\site-packages\urllib3\connection.py:196, in HTTPConnection._new_conn(self)
    195 try:
--> 196     sock = connection.create_connection(
    197         (self._dns_host, self.port),
    198         self.timeout,
    199         source_address=self.source_address,
    200         socket_options=self.socket_options,
    201     )
    202 except socket.gaierror as e:

File d:\Anaconda3\Lib\site-packages\urllib3\util\connection.py:85, in create_connection(address, timeout, source_address, socket_options)
     84 try:
---> 85     raise err
     86 finally:
     87     # Break explicitly a reference cycle

File d:\Anaconda3\Lib\site-packages\urllib3\util\connection.py:73, in create_connection(address, timeout, source_address, socket_options)
     72     sock.bind(source_address)
---> 73 sock.connect(sa)
     74 # Break explicitly a reference cycle

ConnectionRefusedError: [WinError 10061] 由于目标计算机积极拒绝,无法连接。

The above exception was the direct cause of the following exception:

NewConnectionError                        Traceback (most recent call last)
File d:\Anaconda3\Lib\site-packages\urllib3\connectionpool.py:789, in HTTPConnectionPool.urlopen(self, method, url, body, headers, retries, redirect, assert_same_host, timeout, pool_timeout, release_conn, chunked, body_pos, preload_content, decode_content, **response_kw)
    788 # Make the request on the HTTPConnection object
--> 789 response = self._make_request(
    790     conn,
    791     method,
    792     url,
    793     timeout=timeout_obj,
    794     body=body,
    795     headers=headers,
    796     chunked=chunked,
    797     retries=retries,
    798     response_conn=response_conn,
    799     preload_content=preload_content,
    800     decode_content=decode_content,
    801     **response_kw,
    802 )
    804 # Everything went great!

File d:\Anaconda3\Lib\site-packages\urllib3\connectionpool.py:495, in HTTPConnectionPool._make_request(self, conn, method, url, body, headers, retries, timeout, chunked, response_conn, preload_content, decode_content, enforce_content_length)
    494 try:
--> 495     conn.request(
    496         method,
    497         url,
    498         body=body,
    499         headers=headers,
    500         chunked=chunked,
    501         preload_content=preload_content,
    502         decode_content=decode_content,
    503         enforce_content_length=enforce_content_length,
    504     )
    506 # We are swallowing BrokenPipeError (errno.EPIPE) since the server is
    507 # legitimately able to close the connection after sending a valid response.
    508 # With this behaviour, the received response is still readable.

File d:\Anaconda3\Lib\site-packages\urllib3\connection.py:398, in HTTPConnection.request(self, method, url, body, headers, chunked, preload_content, decode_content, enforce_content_length)
    397     self.putheader(header, value)
--> 398 self.endheaders()
    400 # If we're given a body we start sending that in chunks.

File d:\Anaconda3\Lib\http\client.py:1331, in HTTPConnection.endheaders(self, message_body, encode_chunked)
   1330     raise CannotSendHeader()
-> 1331 self._send_output(message_body, encode_chunked=encode_chunked)

File d:\Anaconda3\Lib\http\client.py:1091, in HTTPConnection._send_output(self, message_body, encode_chunked)
   1090 del self._buffer[:]
-> 1091 self.send(msg)
   1093 if message_body is not None:
   1094 
   1095     # create a consistent interface to message_body

File d:\Anaconda3\Lib\http\client.py:1035, in HTTPConnection.send(self, data)
   1034 if self.auto_open:
-> 1035     self.connect()
   1036 else:

File d:\Anaconda3\Lib\site-packages\urllib3\connection.py:236, in HTTPConnection.connect(self)
    235 def connect(self) -> None:
--> 236     self.sock = self._new_conn()
    237     if self._tunnel_host:
    238         # If we're tunneling it means we're connected to our proxy.

File d:\Anaconda3\Lib\site-packages\urllib3\connection.py:211, in HTTPConnection._new_conn(self)
    210 except OSError as e:
--> 211     raise NewConnectionError(
    212         self, f"Failed to establish a new connection: {e}"
    213     ) from e
    215 # Audit hooks are only available in Python 3.8+

NewConnectionError: <urllib3.connection.HTTPConnection object at 0x0000023A69EC9B80>: Failed to establish a new connection: [WinError 10061] 由于目标计算机积极拒绝,无法连接。

The above exception was the direct cause of the following exception:

MaxRetryError                             Traceback (most recent call last)
File d:\Anaconda3\Lib\site-packages\requests\adapters.py:589, in HTTPAdapter.send(self, request, stream, timeout, verify, cert, proxies)
    588 try:
--> 589     resp = conn.urlopen(
    590         method=request.method,
    591         url=url,
    592         body=request.body,
    593         headers=request.headers,
    594         redirect=False,
    595         assert_same_host=False,
    596         preload_content=False,
    597         decode_content=False,
    598         retries=self.max_retries,
    599         timeout=timeout,
    600         chunked=chunked,
    601     )
    603 except (ProtocolError, OSError) as err:

File d:\Anaconda3\Lib\site-packages\urllib3\connectionpool.py:843, in HTTPConnectionPool.urlopen(self, method, url, body, headers, retries, redirect, assert_same_host, timeout, pool_timeout, release_conn, chunked, body_pos, preload_content, decode_content, **response_kw)
    841     new_e = ProtocolError("Connection aborted.", new_e)
--> 843 retries = retries.increment(
    844     method, url, error=new_e, _pool=self, _stacktrace=sys.exc_info()[2]
    845 )
    846 retries.sleep()

File d:\Anaconda3\Lib\site-packages\urllib3\util\retry.py:519, in Retry.increment(self, method, url, response, error, _pool, _stacktrace)
    518     reason = error or ResponseError(cause)
--> 519     raise MaxRetryError(_pool, url, reason) from reason  # type: ignore[arg-type]
    521 log.debug("Incremented Retry for (url='%s'): %r", url, new_retry)

MaxRetryError: HTTPConnectionPool(host='127.0.0.1', port=8100): Max retries exceeded with url: /door/tongtong/log (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x0000023A69EC9B80>: Failed to establish a new connection: [WinError 10061] 由于目标计算机积极拒绝,无法连接。'))

During handling of the above exception, another exception occurred:

ConnectionError                           Traceback (most recent call last)
Cell In[5], line 1
----> 1 post("http://127.0.0.1:8100/door/tongtong/log",json={'lockId': '8686987', 'notifyType': '1', 'records': '[{"lockId":8686987,"electricQuantity":100,"serverDate":1758100706693,"recordTypeFromLock":17,"recordType":7,"success":1,"lockMac":"FD:15:3A:7E:D1:E2","keyboardPwd":"3157220353","lockDate":1758100692000,"username":"夏天乐带0"}]', 'admin': 'xtlft@qq.com', 'lockMac': 'FD:15:3A:7E:D1:E2'}).json()

File d:\Anaconda3\Lib\site-packages\requests\api.py:115, in post(url, data, json, **kwargs)
    103 def post(url, data=None, json=None, **kwargs):
    104     r"""Sends a POST request.
    105 
    106     :param url: URL for the new :class:`Request` object.
   (...)
    112     :rtype: requests.Response
    113     """
--> 115     return request("post", url, data=data, json=json, **kwargs)

File d:\Anaconda3\Lib\site-packages\requests\api.py:59, in request(method, url, **kwargs)
     55 # By using the 'with' statement we are sure the session is closed, thus we
     56 # avoid leaving sockets open which can trigger a ResourceWarning in some
     57 # cases, and look like a memory leak in others.
     58 with sessions.Session() as session:
---> 59     return session.request(method=method, url=url, **kwargs)

File d:\Anaconda3\Lib\site-packages\requests\sessions.py:589, in Session.request(self, method, url, params, data, headers, cookies, files, auth, timeout, allow_redirects, proxies, hooks, stream, verify, cert, json)
    584 send_kwargs = {
    585     "timeout": timeout,
    586     "allow_redirects": allow_redirects,
    587 }
    588 send_kwargs.update(settings)
--> 589 resp = self.send(prep, **send_kwargs)
    591 return resp

File d:\Anaconda3\Lib\site-packages\requests\sessions.py:703, in Session.send(self, request, **kwargs)
    700 start = preferred_clock()
    702 # Send the request
--> 703 r = adapter.send(request, **kwargs)
    705 # Total elapsed time of the request (approximately)
    706 elapsed = preferred_clock() - start

File d:\Anaconda3\Lib\site-packages\requests\adapters.py:622, in HTTPAdapter.send(self, request, stream, timeout, verify, cert, proxies)
    618     if isinstance(e.reason, _SSLError):
    619         # This branch is for urllib3 v1.22 and later.
    620         raise SSLError(e, request=request)
--> 622     raise ConnectionError(e, request=request)
    624 except ClosedPoolError as e:
    625     raise ConnectionError(e, request=request)

ConnectionError: HTTPConnectionPool(host='127.0.0.1', port=8100): Max retries exceeded with url: /door/tongtong/log (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x0000023A69EC9B80>: Failed to establish a new connection: [WinError 10061] 由于目标计算机积极拒绝,无法连接。'))
在当前单元格或上一个单元格中执行代码时 Kernel 崩溃。

请查看单元格中的代码,以确定故障的可能原因。

单击<a href='https://aka.ms/vscodeJupyterKernelCrash'>此处</a>了解详细信息。

有关更多详细信息,请查看 Jupyter <a href='command:jupyter.viewOutput'>log</a>。
In [2]:
get('''https://szihj.idsp.yunxiao.com/open/api/v1/students?pageNo=1&pageSize=20&filter={"studyCode":["280106","280107","280108"]}''',headers={
    "authorization":"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzemloaiIsImF1ZCI6Ijg2OTM3NjI1MDk3OTU3ODAxNyIsImlhdCI6MTc1NzQ0MTMwNSwiZXhwIjoxNzU3NDQ4NTA1LCJzY29wZSI6bnVsbH0.3VH9VAp6uq_pMoACGUtRZqpAq3ewMYmNVNqyXy0YK_E"},verify=False,proxies={
    "http":"http://127.0.0.1:8080",
    "https":"http://127.0.0.1:8080"
    }
)
Out [2]:
d:\Anaconda3\Lib\site-packages\urllib3\connectionpool.py:1099: InsecureRequestWarning: Unverified HTTPS request is being made to host '127.0.0.1'. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/latest/advanced-usage.html#tls-warnings
  warnings.warn(
<Response [200]>
In [3]:
get("https://szihj.idsp.yunxiao.com/open/api/v1/students?pageNo=1&pageSize=20&filter=%7B%22studyCode%22%3A%5B%22280106%22%2C%22280107%22%2C%22280108%22%5D%7D",headers={
    "authorization":"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzemloaiIsImF1ZCI6Ijg2OTM3NjI1MDk3OTU3ODAxNyIsImlhdCI6MTc1NzQ0MTMwNSwiZXhwIjoxNzU3NDQ4NTA1LCJzY29wZSI6bnVsbH0.3VH9VAp6uq_pMoACGUtRZqpAq3ewMYmNVNqyXy0YK_E"},verify=False,proxies={
    "http":"http://127.0.0.1:8080",
    "https":"http://127.0.0.1:8080"
    }

    )
Out [3]:
d:\Anaconda3\Lib\site-packages\urllib3\connectionpool.py:1099: InsecureRequestWarning: Unverified HTTPS request is being made to host '127.0.0.1'. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/latest/advanced-usage.html#tls-warnings
  warnings.warn(
<Response [200]>