import cv2 import os 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"]) if request.json['state'] == str(479169837): sendmsg(request.json["imageUrl"],479169837) return 'ok' sendmsg(request.json["imageUrl"],5464741) 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 and request.json["group_id"] != 479169837: 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 } if data["group_id"] == 479169837: params['state'] = str(479169837) 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}\{name}" if not os.path.exists(local_filename): # 如果目录不存在，则创建它 os.makedirs(local_filename) print(f'目录{local_filename}已创建') local_filename += "\\"+datetime.now().strftime('%Y%m%d_%H%M%S') 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,id): # 设置 go-cqhttp 服务的 URL 和端口 api_url = 'http://localhost:5700/send_msg' # 设置要发送的消息参数 data = { 'group_id': id, # 接收消息的用户的 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)