Python 实现简单 Web 服务器：使用 PidFile 加锁
这是一个 Python 的程序，用于实现一个简单的 Web 服务器。程序首先导入了一些必要的模块，然后使用 PidFile 模块对代码块加锁，防止多个进程同时运行。接着定义了一个 response 函数，该函数用于响应客户端的请求，并根据请求的 URL 返回相应的内容。程序的 main 函数则用于初始化套接字、绑定端口、设置监听并等待接收请求。最后，如果该程序被直接运行，则调用 main 函数开始运行 Web 服务器。
# 会对with中的代码块加锁
with PidFile():
import socket
import re
import threading
import win32api
import msg
import time
import os # 执行文件
import pyautogui
import 通知内容
import 在线数据
from urllib import parse # 转网址码
import pymysql
msg.se('1001', '[发送消息]WEB服务启动', '1')
#######################隐藏执行后对话框##################
import ctypes # 隐藏执行后对话框
whnd = ctypes.windll.kernel32.GetConsoleWindow() # 隐藏执行后对话框
if whnd != 0: # 隐藏执行后对话框
ctypes.windll.user32.ShowWindow(whnd, 0) # 隐藏执行后对话框
ctypes.windll.kernel32.CloseHandle(whnd) # 隐藏执行后对话框
#######################隐藏执行后对话框##################
def response(new_socket):
data = new_socket.recv(1024).decode('GBK').splitlines()
# print('请求： ')
for item in data:
pass # 空指令
# print(item )
ret = re.search(r'[^/]+(/[^ ]*)', data[0])
# print(ret )
if ret:
file_name = ret.group(1)
# print('*'*50, file_name)
真实路径 =file_name
#print (真实路径,'DDDDDDDDDD')
if file_name != '':
f2 = os.path.exists('D:/开机启动/web/web' + file_name)
if f2 == True:
file_name = file_name
else:
if file_name == '/':
file_name = '/index.html'
else:
file_name = '/index.html'
def rex(r):
html_content=r.encode('GBK')
response = 'HTTP/1.1 200 OK\r\n'
response += '\r\n'
new_socket.send(response.encode('GBK'))
new_socket.send(html_content)
new_socket.close()
# 等于Msg调用模块
if 真实路径.endswith('.msg'):
env={
'file_name':真实路径,
}
内容=通知内容.真实路径(env)
rex(内容)
#等于Line调用模块
if 真实路径.endswith('.line'):
env={
'file_name':真实路径,
}
内容=在线数据.真实路径(env)
rex(内容)
else:
try:
# 网页文件的路径自己设置
f = open('D:/开机启动/web/web' + file_name, 'rb')
except:
response = 'HTTP/1.1 404 NOT FOUND\r\n'
response += '\r\n'
response += ' '
new_socket.send(response.encode('GBK'))
# 新进程中关闭套接字
new_socket.close()
else:
html_content = f.read()
f.close()
# ww='jijiiih数据转为可以'
# html_content=ww.encode('gb2312')
# 2.1 准备发送给浏览器的数据---header
response = 'HTTP/1.1 200 OK\r\n'
response += '\r\n'
# 2.2 准备发送给浏览器的数据---boy
# response += 'hahahhah'
# 将response header发送给浏览器
new_socket.send(response.encode('GBK'))
# 将response body发送给浏览器
new_socket.send(html_content)
# 新进程中关闭套接字
new_socket.close()
######################################################
# 2. 返回http格式的数据，给浏览器
数据 = data
print(数据)
str1 = (str(数据)) # 定义数据类型
str1 = str(str1)
list1 = str1.split(' ')
str2 = (str(list1[1]))
str2 = str(str2)
网址 = parse.unquote(str2)
#print(网址)
if ('/⧭Rex' in 网址):
listRex = 网址.split('⧭')
msg.se(listRex[2],listRex[3], listRex[4])
if ('/logger' in 网址):
list2 = 网址.split('⧭') # 进行分割
# print(list2[1]) #包名
# print(list2[2]) #名称
# print(list2[3]) #内容
sms = (str(list2[3])[0:-19])
if sms[0] == '[':
r1 = sms.index(']') # 检查]在字符串什么位置
r2 = (sms[r1 + 1:])
r3 = r2.index(':') # 检查]在字符串什么位置
发信名称 = (r2[:r3]) # 得出发信名称
list2[3] = (r2[r3 + 2:]) # 发送内容
else:
if (':' in sms):
r5 = sms.index(':')
发信名称 = (sms[:r5]) # 得出发信名称
list2[3] = (sms[r5 + 2:]) # 发送内容
else:
发信名称 =list2[2]
需要转换字符串内容=str(list2[3])
第一部分内容= 需要转换字符串内容.split(' ')
list2[3]=第一部分内容[0]
检查发信人 = os.path.exists('D:/开机启动/web/web/' + list2[1] + '/' + list2[2]+'.py')
if 检查发信人 == True: # 文件夹没有找到
conn = pymysql.connect(host='i.rex-robot.com', user='root', password='Rex888888', database='rex', charset='GBK')
cursor = conn.cursor()
#获取名称记录 = 'SELECT * FROM line WHERE 名称= '999999#' and 使用名='888888#''
获取名称记录 = 'SELECT * FROM line WHERE 名称 = '999999#''
替换字符串 = (获取名称记录.replace('999999#', list2[2]))
替换记录1 = cursor.execute(替换字符串)
for i in range(替换记录1):
替换记录2 = (cursor.fetchone())
if (str(替换记录2[2]) in sms):
发信名称 ='Rex'
conn.commit()
# 关闭光标
cursor.close()
# 关闭连接 conn.cu
conn.close()
else:
conn = pymysql.connect(host='i.rex-robot.com', user='root', password='Rex888888', database='rex', charset='GBK')
cursor = conn.cursor()
#获取名称记录 = 'SELECT * FROM line WHERE 名称= '999999#' and 使用名='888888#''
获取名称记录 = 'SELECT * FROM line WHERE 名称 = '999999#''
替换字符串 = (获取名称记录.replace('999999#', 'Rex'))
替换记录1 = cursor.execute(替换字符串)
for i in range(替换记录1):
替换记录2 = (cursor.fetchone())
if (str(替换记录2[2]) in sms):
发信名称 ='Rex'
conn.commit()
# 关闭光标
cursor.close()
# 关闭连接 conn.cu
conn.close()
if 发信名称 == 'Rex':
#print(' ' + list2[1] + ' ' + list2[2] + ' ' + list2[3])
f1 = os.path.exists('D:/开机启动/web/web/' + list2[1])
#print(f1)
if f1 == True:
#print('找到文件夹')
f2 = os.path.exists('D:/开机启动/web/web/' + list2[1] + '/' + list2[2] + '.py')
if f2 == True: # 文件夹没有找到
win32api.ShellExecute(0, 'open', 'D:/开机启动/web/web/' + list2[1] + '/' + list2[2] + '.py', '', '',
1) # 运行程序
conn = pymysql.connect(host='i.rex-robot.com', user='root', password='Rex888888', database='rex', charset='GBK')
cursor = conn.cursor()
sql = 'INSERT INTO sms ( 类型,消息 ) VALUES ('8888','9999');'
sql1 = (sql.replace('8888', list2[2]))
sql2 = (sql1.replace('9999', list2[3]))
cursor.execute(sql2)
conn.commit()
# 关闭光标
cursor.close()
# 关闭连接 conn.cu
conn.close()
# print ('找到文件')
else:
未找到ID消息 = 'ID: ' + list2[1] + ' 用户 ' + list2[2] + ' 内容: ' + list2[3]
if ('[发送文件]' in 未找到ID消息):
msg.se('1001', '[发送消息]' + 'ID: ' + list2[1] + ' 用户 ' + list2[2] + ' 内容:' + ' 文件消息', '1')
else:
f8 = os.path.exists('D:/开机启动/web/web/' + list2[1] + '/' + '未找到用户.py')
if f8 == True: # 文件夹没有找到
conn = pymysql.connect(host='127.0.0.1', user='root', password='Rex888888', database='rex', charset='GBK')
cursor = conn.cursor()
sql = 'INSERT INTO sms ( 类型,消息 ) VALUES ('8888','9999');'
sql1 = (sql.replace('8888', '未找到用户'))
sql2 = (sql1.replace('9999', ' 用户 ' + list2[2] + ' 内容:' + list2[3]))
cursor.execute(sql2)
conn.commit()
# 关闭光标
cursor.close()
# 关闭连接
conn.close()
win32api.ShellExecute(0, 'open', 'D:/开机启动/web/web/' + list2[1] + '/' + '未找到用户.py', '',
'', 1) # 运行程序
else:
#msg.se('1001', '[发送消息]' + 未找到ID消息, '1')
# print ('文件夹找到了 文件找不到')
pass
else:
未找到程序消息 = '应用: ' + list2[1] + ' 用户 ' + list2[2] + ' 内容: ' + list2[3]
if ('[发送文件]' in 未找到程序消息):
#msg.se('1001', '[发送消息]' + '应用: ' + list2[1] + ' 用户 ' + list2[2] + ' 内容: ' + '文件消息', '1')
pass
else:
#msg.se('1001', '[发送消息]' + 未找到程序消息, '1')
pass
def main():
# 初始化套接字
tcp_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
tcp_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# 1、绑定端口
tcp_server.bind(('', 8888))
# 2、设置监听
tcp_server.listen(128)
while True:
# 3、等待接收
new_server, client_adder = tcp_server.accept()
# 4、响应
threading.Thread(target=response, args=(new_server,)).start()
# 5、关闭新套接字------此处关闭会导致新套接字无法正常使用
# new_server.close()
# 6、关闭服务器套接字
tcp_server.close()
if __name__ == '__main__':
main()
原文地址: https://www.cveoy.top/t/topic/mzYi 著作权归作者所有。请勿转载和采集!