246 lines
7.5 KiB
Python
246 lines
7.5 KiB
Python
import cv2
|
||
import json
|
||
import time
|
||
import logging
|
||
import threading
|
||
import requests
|
||
import supervision as sv
|
||
|
||
from datetime import datetime
|
||
from ultralytics import YOLO
|
||
from flask import Flask,request,Response
|
||
from multiprocessing import Process, Event
|
||
|
||
app = Flask(__name__)
|
||
|
||
@app.route('/show',methods=['POST'])
|
||
def show():
|
||
# print(request.json)
|
||
data = request.json["content"]
|
||
balls = data["balls"]
|
||
table = data["table"]
|
||
text = f"table:{table:>3} "
|
||
for i in range(16):
|
||
ball = "#" if "ball"+str(i) in balls else " "
|
||
text += f"| {i} : {ball} "
|
||
print(datetime.now().strftime('%Y-%m-%d %H:%M:%S'),text)
|
||
return 'ok'
|
||
import re
|
||
@app.route('/mj',methods=['POST'])
|
||
def mj():
|
||
print(request.json)
|
||
if request.json["progress"] == "100%":
|
||
print(request.json["imageUrl"])
|
||
sendmsg(request.json["imageUrl"])
|
||
download_image(request.json["imageUrl"], "MJ", request.json["state"])
|
||
return 'ok'
|
||
|
||
@app.route('/cq',methods=['POST'])
|
||
def cq():
|
||
print(request.json)
|
||
if request.json["post_type"] != "message":
|
||
return 'ok'
|
||
if request.json["message_type"] != "group":
|
||
return 'ok'
|
||
if request.json["group_id"] != 116277572:
|
||
return 'ok'
|
||
# if request.json["user_id"] != 2132960:
|
||
# return 'ok'
|
||
|
||
thread = threading.Thread(target=handle, args=(request.json,))
|
||
thread.start()
|
||
return "ok"
|
||
|
||
def handle(data):
|
||
print(data['raw_message'])
|
||
|
||
# 正则表达式来匹配URL
|
||
url_pattern = re.compile(r'url=(.*?)\]')
|
||
|
||
# 使用findall来找到所有的URL
|
||
urls = re.findall(url_pattern, data['raw_message'])
|
||
|
||
result = re.sub(r'\[.*?\]', '', data['raw_message'])
|
||
result = result.replace('\r\n', '')
|
||
|
||
name = data["sender"]["card"] or data["sender"]["nickname"]
|
||
|
||
base64Array = []
|
||
# 输出找到的URL
|
||
for url in urls:
|
||
print(url)
|
||
|
||
download_image(url, "QQ", name)
|
||
# 从URL下载图片
|
||
response = requests.get(url)
|
||
response.raise_for_status() # 如果请求失败，此处将引发异常
|
||
|
||
# 将图片数据转换为Base64编码的字节数组
|
||
image_base64_bytes = base64.b64encode(response.content)
|
||
|
||
# 如果您需要将其转换为字符串，您可以使用.decode('utf-8')将字节数组转换为字符串
|
||
image_base64_str = image_base64_bytes.decode('utf-8')
|
||
|
||
# 打印Base64编码的字节数组
|
||
# print(image_base64_bytes)
|
||
base64Array.append('data:image/jpeg;base64,'+image_base64_str)
|
||
|
||
if len(base64Array) == 0:
|
||
return 'ok'
|
||
prompt = 'game assets,game ui,animal,flower,plant,forest,cute,baby,magical,spirit,fairy,elf,glowing light,anime style,seeming very happy,from above,ultra detailed,soft,on white background'
|
||
if result != '':
|
||
prompt = result
|
||
# 示例调用
|
||
params = {
|
||
'base64Array':base64Array,
|
||
'notifyHook': 'http://hk.luanhailiang.cn:5000/mj',
|
||
'prompt': prompt,
|
||
'state': name
|
||
}
|
||
|
||
response = imagine(params)
|
||
print(response.json())
|
||
time.sleep(3)
|
||
params['prompt'] = prompt + " --iw 2 --niji 5"
|
||
params['state'] = name+"_niji"
|
||
response = imagine(params)
|
||
print(response.json())
|
||
time.sleep(3)
|
||
params['prompt'] = prompt + " --iw 2 --style raw"
|
||
params['state'] = name+"_raw"
|
||
response = imagine(params)
|
||
print(response.json())
|
||
|
||
return 'ok'
|
||
|
||
def download_image(url, t, name):
|
||
"""
|
||
从指定URL下载图片并保存到本地文件。
|
||
|
||
:param url: 图片的URL。
|
||
:param local_filename: 要保存图片的本地文件名。
|
||
"""
|
||
|
||
local_filename = f"y:\美术\参考收集\{t}\{datetime.now().strftime('%Y%m%d%H%M%S')}_{name}"
|
||
response = requests.get(url)
|
||
if response.status_code == 200: # HTTP状态码200表示请求成功
|
||
# 获取Content-Type响应头
|
||
content_type = response.headers['Content-Type']
|
||
# 判断图片格式
|
||
ext = ""
|
||
if 'jpeg' in content_type or 'jpg' in content_type:
|
||
ext = '.jpg'
|
||
print('The image is in JPEG format.')
|
||
elif 'png' in content_type:
|
||
ext = '.png'
|
||
print('The image is in PNG format.')
|
||
elif 'gif' in content_type:
|
||
ext = '.gif'
|
||
print('The image is in GIF format.')
|
||
elif 'webp' in content_type:
|
||
ext = '.webp'
|
||
print('The image is in WEBP format.')
|
||
else:
|
||
ext = '.png'
|
||
print(f'Unknown image format: {content_type}')
|
||
local_filename += ext
|
||
with open(local_filename, 'wb') as f:
|
||
f.write(response.content)
|
||
print(f"Image successfully downloaded to {local_filename}")
|
||
else:
|
||
print(f"Error downloading image, HTTP status code: {response.status_code}")
|
||
|
||
|
||
import base64
|
||
import requests
|
||
|
||
|
||
def imagine(params):
|
||
"""
|
||
提交Imagine任务
|
||
|
||
:param params: imagineDTO，一个字典，包含以下键:
|
||
- base64Array: 垫图base64数组
|
||
- notifyHook: 回调地址, 为空时使用全局notifyHook
|
||
- prompt: 提示词
|
||
- state: 自定义参数
|
||
:return: 返回服务器的响应
|
||
"""
|
||
# API端点
|
||
url = 'http://127.0.0.1:8080/mj/submit/imagine'
|
||
|
||
# 发送POST请求
|
||
response = requests.post(url, json=params)
|
||
|
||
# 如果需要，处理响应（例如，检查响应状态，解析响应主体等）
|
||
|
||
return response
|
||
|
||
|
||
def sendmsg(url):
|
||
# 设置 go-cqhttp 服务的 URL 和端口
|
||
api_url = 'http://localhost:5700/send_msg'
|
||
|
||
# 设置要发送的消息参数
|
||
data = {
|
||
'group_id': 5464741, # 接收消息的用户的 QQ 号
|
||
'message_type': 'group', # 消息类型
|
||
'message': f'[CQ:image,file={url}]' # 消息内容，使用 CQ码 格式
|
||
}
|
||
|
||
# 发送 POST 请求到 go-cqhttp 服务
|
||
response = requests.post(api_url, data=data)
|
||
|
||
# 输出响应
|
||
print(response.json())
|
||
|
||
if __name__ == '__main__':
|
||
|
||
# # 这是包含URL的示例文本
|
||
# text = '[CQ:image,file=42c93be3dbe10bfa79ad93ce1baed561.image,subType=1,url=https://gchat.qpic.cn/gchatpic_new/1183464602/691534145-2408237736-42C93BE3DBE10BFA79AD93CE1BAED561/0?term=2&is_origin=0][CQ:image,file=42c93be3dbe10bfa79ad93ce1baed561.image,subType=1,url=https://gchat.qpic.cn/gchatpic_new/1183464602/691534145-2408237736-42C93BE3DBE10BFA79AD93CE1BAED561/0?term=2&is_origin=0]'
|
||
|
||
# # 正则表达式来匹配URL
|
||
# url_pattern = re.compile(r'url=(.*?)\]')
|
||
|
||
|
||
|
||
# # 使用findall来找到所有的URL
|
||
# urls = re.findall(url_pattern, text)
|
||
|
||
# # 输出找到的URL
|
||
# for url in urls:
|
||
# print(url)
|
||
|
||
|
||
|
||
# # 从URL下载图片
|
||
# response = requests.get(url)
|
||
# response.raise_for_status() # 如果请求失败，此处将引发异常
|
||
|
||
# # 将图片数据转换为Base64编码的字节数组
|
||
# image_base64_bytes = base64.b64encode(response.content)
|
||
|
||
# # 如果您需要将其转换为字符串，您可以使用.decode('utf-8')将字节数组转换为字符串
|
||
# image_base64_str = image_base64_bytes.decode('utf-8')
|
||
|
||
# # 打印Base64编码的字节数组
|
||
# # print(image_base64_bytes)
|
||
|
||
# # 示例调用
|
||
# params = {
|
||
# 'base64Array': ['data:image/jpeg;base64,'+image_base64_str],
|
||
# 'notifyHook': 'http://hk.luanhailiang.cn:5000/mj',
|
||
# 'prompt': 'character design, game assets, game ui, cute, soft',
|
||
# 'state': ''
|
||
# }
|
||
|
||
# response = imagine(params)
|
||
# print(response.json())
|
||
|
||
|
||
|
||
|
||
|
||
app.logger.setLevel(logging.ERROR)
|
||
app.run("0.0.0.0",port=8000,threaded=True)
|