最新备份简单的fastapi+socketserver_fastapi socket

# -*- coding:utf-8 -*-
import uvicorn
from fastapi import FastAPI, Request,WebSocket
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
import emoji
import time
yt=time.time()
def i():
    num = 0
    while True:
        yield num
        num += 1
y=i() 

hd=('name','age','years')                       
#t= time.strftime("%Y/%m/%d  %r",time.localtime())

s= emoji.emojize('Python is :thumbs_up:') 


app = FastAPI()
app.mount("/static", StaticFiles(directory="/data/data/com.termux/files/home/static"), name="static")#js.css.images等静态文件目录
templates = Jinja2Templates(directory="/data/data/com.termux/files/home/www") # 模板目录(html等的目录)

@app.get("/")#,response_class=HTMLResponse)
async def home(request: Request):
    data_all={"request":request,
            "hd":hd,
            "s":s
            }
    return templates.TemplateResponse("index.html",data_all)
@app.get("/html") 
async def h(request: Request):        
    return templates.TemplateResponse("h.html",{"request": request})

@app.websocket("/ws")
async def websocket_endpoint(websocket:WebSocket):    
    try:
        await websocket.accept()
        while True:
             data = await websocket.receive_text()
             await websocket.send_text(f"Message text was:{data}")
    except:
        print('断开连接')
        print('当前连接次数'+str(next(y)))
        #c=time.time()-yt
        #print(c)
        u=time.time()
        d=time.strftime("此连接断开时间: %Y-%m-%d %H:%M:%S",time.localtime(u))
        print(d)
#websocket.close()
#websocket.close(code=10000)
# await websocket.send_bytes(data) 
#await websocket.send_json(data)
#websocket.receive_json(data, mode="binary")通过二进制数据帧接收 JSON。
#await websocket.send(message)
#data = await websocket.receive
#await websocket.receive()
 
if __name__ == '__main__':
    uvicorn.run(app, host="127.0.0.1", port=5000)


`阻止107等每个断开连接反馈信息``
#self.transport.write_eof()1312行注释掉/data/data/com.termux/files/usr/lib/python3.10/site-packages/websockets/legacy/protocol.py


await self.close() 238行添加try: except:pass
/data/data/com.termux/files/usr/lib/python3.10/site-packages/websockets/legacy/server.py

```java
@app.websocket("/ws")
async def websocket_endpoint(websocket:WebSocket):    
    try:
        #等待socket连接并建立连接
        await websocket.accept()
        while True:
            #获取客服端传回的数据
             data = await websocket.receive_text()
             await websocket.send_text(f"主动发消息给客户端，此消息只要有连接就会发送")
             if data=='1':
                 g='2'
             # 启动处理程序并把处理程序的结果返回给客户端  
             #if判断客服端发来的数据并处理客户端发来的数据data后重新发送给客户端
             #print(data)
             #向客户端发送的的数据
                 await websocket.send_text(f"Message text was:data: {data} g: {2}")
             else:
                 await websocket.send_text(f"不匹配")



project/
├── __init__.py
├── __pycache__/
│   ├── database.cpython-310.pyc
│   ├── hero_model.cpython-310.pyc
│   ├── models.cpython-310.pyc
│   └── team_model.cpython-310.pyc
├── app.py
├── database.db
├── database.py
├── hero_model.py
├── models.py
└── team_model.py

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
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108