main #22

Merged
ccyj merged 68 commits from sangge/tpre-python:main into main 2023-10-26 14:45:47 +08:00
Showing only changes of commit 55add91467 - Show all commits

View File

@ -6,33 +6,40 @@ import sqlite3
import asyncio import asyncio
import time import time
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI): async def lifespan(app: FastAPI):
init() init()
yield yield
clean_env() clean_env()
app = FastAPI(lifespan = lifespan)
app = FastAPI(lifespan=lifespan)
# 连接到数据库(如果数据库不存在,则会自动创建) # 连接到数据库(如果数据库不存在,则会自动创建)
conn = sqlite3.connect('server.db') conn = sqlite3.connect("server.db")
# 创建游标对象用于执行SQL语句 # 创建游标对象用于执行SQL语句
cursor = conn.cursor() cursor = conn.cursor()
# 创建表: id: int; ip: TEXT # 创建表: id: int; ip: TEXT
cursor.execute('''CREATE TABLE IF NOT EXISTS nodes ( cursor.execute(
"""CREATE TABLE IF NOT EXISTS nodes (
id INTEGER PRIMARY KEY AUTOINCREMENT, id INTEGER PRIMARY KEY AUTOINCREMENT,
ip TEXT NOT NULL, ip TEXT NOT NULL,
last_heartbeat INTEGER last_heartbeat INTEGER
)''') )"""
)
def init(): def init():
asyncio.create_task(receive_heartbeat_internal()) asyncio.create_task(receive_heartbeat_internal())
def clean_env(): def clean_env():
# 关闭游标和连接 # 关闭游标和连接
cursor.close() cursor.close()
conn.close() conn.close()
@app.get("/server/show_nodes") @app.get("/server/show_nodes")
async def show_nodes() -> list: async def show_nodes() -> list:
nodes_list = [] nodes_list = []
@ -43,15 +50,16 @@ async def show_nodes() -> list:
nodes_list.append(row) nodes_list.append(row)
return nodes_list return nodes_list
@app.get("/server/get_node") @app.get("/server/get_node")
async def get_node(ip: str) -> int: async def get_node(ip: str) -> int:
''' """
中心服务器与节点交互, 节点发送ip, 中心服务器接收ip存入数据库并将ip转换为int作为节点id返回给节点 中心服务器与节点交互, 节点发送ip, 中心服务器接收ip存入数据库并将ip转换为int作为节点id返回给节点
params: params:
ip: node ip ip: node ip
return: return:
id: ip按点分割成四部分, 每部分转二进制后拼接再转十进制作为节点id id: ip按点分割成四部分, 每部分转二进制后拼接再转十进制作为节点id
''' """
ip_parts = ip.split(".") ip_parts = ip.split(".")
ip_int = 0 ip_int = 0
for i in range(4): for i in range(4):
@ -61,19 +69,23 @@ async def get_node(ip: str) -> int:
current_time = int(time.time()) current_time = int(time.time())
# 插入数据 # 插入数据
cursor.execute("INSERT INTO nodes (id, ip, last_heartbeat) VALUES (?, ?, ?)", (ip_int, ip, current_time)) cursor.execute(
"INSERT INTO nodes (id, ip, last_heartbeat) VALUES (?, ?, ?)",
(ip_int, ip, current_time),
)
conn.commit() conn.commit()
return ip_int return ip_int
@app.get("/server/delete_node") @app.get("/server/delete_node")
async def delete_node(ip: str) -> None: async def delete_node(ip: str) -> None:
''' """
param: param:
ip: 待删除节点的ip地址 ip: 待删除节点的ip地址
return: return:
None None
''' """
# 查询要删除的节点 # 查询要删除的节点
cursor.execute("SELECT * FROM nodes WHERE ip=?", (ip,)) cursor.execute("SELECT * FROM nodes WHERE ip=?", (ip,))
row = cursor.fetchone() row = cursor.fetchone()
@ -85,12 +97,16 @@ async def delete_node(ip: str) -> None:
else: else:
print(f"Node with IP {ip} not found.") print(f"Node with IP {ip} not found.")
# 接收节点心跳包 # 接收节点心跳包
@app.post("/server/heartbeat") @app.post("/server/heartbeat")
async def receive_heartbeat(ip: str): async def receive_heartbeat(ip: str):
cursor.execute("UPDATE nodes SET last_heartbeat = ? WHERE ip = ?", (time.time(), ip)) cursor.execute(
"UPDATE nodes SET last_heartbeat = ? WHERE ip = ?", (time.time(), ip)
)
return {"status": "received"} return {"status": "received"}
async def receive_heartbeat_internal(): async def receive_heartbeat_internal():
while 1: while 1:
timeout = 70 timeout = 70
@ -99,32 +115,34 @@ async def receive_heartbeat_internal():
# conn.commit() # conn.commit()
await asyncio.sleep(timeout) await asyncio.sleep(timeout)
@app.get("/server/send_nodes_list") @app.get("/server/send_nodes_list")
async def send_nodes_list(count: int) -> list: async def send_nodes_list(count: int) -> list:
''' """
中心服务器与客户端交互, 客户端发送所需节点个数, 中心服务器从数据库中顺序取出节点封装成list格式返回给客户端 中心服务器与客户端交互, 客户端发送所需节点个数, 中心服务器从数据库中顺序取出节点封装成list格式返回给客户端
params: params:
count: 所需节点个数 count: 所需节点个数
return: return:
nodes_list: list nodes_list: list
''' """
nodes_list = [] nodes_list = []
# 查询数据库中的节点数据 # 查询数据库中的节点数据
cursor.execute("SELECT * FROM nodes LIMIT ?", (count,)) cursor.execute("SELECT * FROM nodes LIMIT ?", (count,))
rows = cursor.fetchall() rows = cursor.fetchall()
for row in rows: for row in rows:
id, ip, last_heartbeat = row id, ip, last_heartbeat = row
nodes_list.append(ip) nodes_list.append(ip)
return nodes_list return nodes_list
@app.get("/server/clear_database") @app.get("/server/clear_database")
async def clear_database() -> None: async def clear_database() -> None:
cursor.execute("DELETE FROM nodes") cursor.execute("DELETE FROM nodes")
conn.commit() conn.commit()
if __name__ == "__main__": if __name__ == "__main__":
import uvicorn # pylint: disable=e0401 import uvicorn # pylint: disable=e0401