编译的时候可能写了个-w,你重新编译看看
[Python] 纯文本查看 复制代码 # coding=utf-8
from deepdanbooru_onnx import DeepDanbooru
from PIL import Image
import json,io,os,sys,time,re
import http.server
import configparser
import threading
import socket
occupancy = 27513
port = 13609
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
def check_port(oport):# bianyi -F ImageTagger.py
try:# bianyi ImageTagger.py
s.bind(('localhost', oport))
return True
except OSError:
return False
if not check_port(occupancy):
print(f'重复运行')
#time.sleep(5)
sys.exit()
print('DeepDanbooru...Init')
config = configparser.ConfigParser()
if os.path.exists('DeepDanbooru.ini'):
config.read('DeepDanbooru.ini')
else:
config['Proxy'] = {'Proxy':'127.0.0.1:1080'}
with open('DeepDanbooru.ini','w') as configfile:
config.write(configfile)
config.read('DeepDanbooru.ini')
Proxy = config.get('Proxy','Proxy')
try:
root = os.getcwd()
if not os.path.exists(root+'/model'):
os.mkdir(root+'/model')
model_path=root+'/model/deepdanbooru.onnx'
tags_path=root+'/model/tags.txt'
if not os.path.exists(model_path):
model_path = None
if not os.path.exists(tags_path):
tags_path = None
danbooru = DeepDanbooru(Proxy=Proxy,model_path=model_path,tags_path=tags_path)
except Exception as e:
print('[Proxy]:'+Proxy)
print("[Exception]:",e)
time.sleep(100)
sys.exit()
print('DeepDanbooru...End')
class ImageServer(http.server.BaseHTTPRequestHandler):
def do_POST(self):
# 读取请求体中的数据
if self.headers['Content']=="End":#关闭
t = threading.Thread(target=shutdown_server)
s.close()
t.start()
sys.exit()
length = int(self.headers['Content-Length'])
data = self.rfile.read(length)
if data:
img = Image.open(io.BytesIO(data))
res = danbooru(img)
self.send_response(200)
self.send_header('Content-Type', 'text/json')
self.end_headers()
res=sortJsonData(res)
otl,tlv = {},{}
for item in res:
tlv[item]=transl[item]
otl['original'] = res
otl['control'] = tlv
result = re.match('^"([\s\S]*?)"$', json.dumps(str(otl).replace("'", '"')))
self.wfile.write(bytes(result.group(1),'UTF-8'))
else:
self.send_response(200)
self.send_header('Content-Type', 'text/plain')
self.end_headers()
self.wfile.write(bytes('Me is Online','UTF-8'))
# 发送响应
# 创建并启动服务器
def shutdown_server():
server.shutdown()
global transl
transl = None
if os.path.exists('./transl.json'):
with open('./transl.json', 'r', encoding='utf-8') as f:
transl = json.load(f)
def sortJsonData(jsondata):
sorted_jsondata = sorted(jsondata.items(), key=lambda item:item[1], reverse=True)
return dict(map(lambda x: (x[0], x[1]), sorted_jsondata))
server = http.server.HTTPServer(('localhost', port), ImageServer)
print('Server:127.0.0.1:'+str(port))
server.serve_forever() |