Skip to content

P
python_open_ai
Commit 7394c5da by lei
Options

1

parent cf0fa24e
Showing with 531 additions and 56 deletions
ai_pyppeteer.py 0 → 100644
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.common.exceptions import TimeoutException
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.chrome.service import Service as ChromeService
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support import expected_conditions as EC
import pyperclip
import time
from loguru import logger
from bit_api import *
import asyncio
from Helper import RedisCon
import json
from multiprocessing import Process
import random
import pyppeteer
from multiprocessing import Process, Lock
# 获取当前时间
def get_current_time():
# 返回秒级时间戳
return int(time.mktime(time.localtime(time.time())))
class ToWork:
def __init__(self, browser_item, lock):
self.init_prompt = '/imagine'
self.browser_item = browser_item
self.browser_id = browser_item['id']
self.lock = lock
self.key = 'laravel_database_midjourney_prompt'
# 获取浏览器窗口
async def browser(self, browser_id):
"""
:return:
"""
self.browser_id = browser_id
ws = await self.bit_get_ws()
return await pyppeteer.connect({
"browserWSEndpoint": ws
})
# 随机等待
def random_sleep(self, min, max):
time.sleep(random.uniform(min, max))
# 粘贴板插入数据
def driver_copy(self, driver, value):
pyperclip.copy(value)
# 粘贴
def driver_paste(self, dricer, value):
ActionChains(dricer).send_keys(Keys.CONTROL, value).perform()
# 元素出现的循环次数
async def CheckElement(self, page, path, num=10, check_type='only'):
for i in range(num):
if check_type == 'only':
element = await self.sleepElementShow(page, path)
if element:
# 随机等待
self.random_sleep(1, 2)
return element
else:
# 列表
element = self.sleepElementListShow(page, path)
if element:
# 随机等待
self.random_sleep(1, 2)
return element
time.sleep(1)
return False
# 等待元素出现--自定义
async def sleepElementShow(self, page, path):
try:
element = (await page.xpath(xpath))[0]
if element:
return element
return False
except Exception as e:
logger.info(e)
return False
# 找元素，只找一次，不等待
def FindElement(self, custom_el, find_type, value):
try:
el = custom_el.find_element(find_type, value)
return el
except Exception as e:
return False
# 获取元素列表
def sleepElementListShow(self, dricer, type, value):
try:
element = dricer.find_elements(type, value)
if element:
return element
return False
except Exception as e:
return False
# 元素点击
def ElementClick(self, dricer, type, value, name=''):
try:
task = dricer.find_element(type, value)
if task:
# 随机等待
self.random_sleep(1, 2)
# js 点击
dricer.execute_script("arguments[0].click()", task)
# task.click()
logger.info(f'{name}-元素存在，开始点击')
return True
logger.info(f'{name}-元素不存在')
return False
except Exception as e:
logger.info(f'{name}-元素不存在')
logger.error(e)
return False
# 循环输入指令
def enter_prompt(self, dricer, tasks):
if True:
# 获取开始时间
start_time = get_current_time()
while True:
# 再次获取当前时间
current_time = get_current_time()
# 最多等待一分钟
if current_time - start_time >= 60:
# end
break
# 开始判断
element = '//ol[@data-list-id="chat-messages"]/li'
# 消息列表
message_list = self.CheckElement(dricer, 'xpath', element, 20, 'list')
if message_list:
logger.info('消息列表存在')
# 找到所有子元素
for e in message_list:
logger.info(e.get_attribute('id'))
# 找到文本并匹配
path = '//div[contains(@data-list-item-id, "chat-messages___chat-messages")]'
label_element = self.FindElement(e, By.XPATH, path)
if label_element:
logger.info(label_element.get_attribute('data-list-item-id'))
else:
logger.info('未找到文本')
logger.info('等待300秒')
time.sleep(300)
return
# task是一个列表
for task in tasks:
# 输入初始指令
logger.info('输入初始指令')
self.driver_paste(dricer, self.init_prompt)
# 选择指令
prompt_el = '//*[@id="autocomplete-0"]'
# 等待选择指令元素出现
element = self.CheckElement(dricer, 'xpath', prompt_el, 5)
if element:
status = self.ElementClick(dricer, 'xpath', prompt_el, '选择指令')
if status:
change_prompt = task['prompt']
# 图片链接
prompt_img = task['prompt_img']
if task['type'] == 1 and prompt_img and len(json.loads(prompt_img)):
prompt_img = json.loads(prompt_img)
st = ''
for item in prompt_img:
# 加入图片-链接
st += ' ' + item
change_prompt = st + ' ' + change_prompt
self.driver_paste(dricer, change_prompt)
# 回车
ActionChains(dricer).send_keys(Keys.ENTER).perform()
time.sleep(2)
# 输入完成,循环获取生成成功后的链接
# 获取开始时间
start_time = get_current_time()
while True:
# 再次获取当前时间
current_time = get_current_time()
# 最多等待一分钟
if current_time - start_time >= 60:
# end
break
# 开始判断
message_list = '//*[@data-list-id="chat-messages"]'
# 获取任务列表
def get_current_tasks(self):
# 测试
# 当前浏览器需要执行的任务
cur_tasks = []
for j in range(2):
# task = RedisCon.get_task(key)
task = {
'prompt': "a very cute boys, 16 years old, fly a kite,in beautifulstreet,full body，3d art, c4d, octane render, ray tracting,clay material,popmart blind box，Pixar trend, animationlighting, depth of field, ultra detailed --q 5",
'user_id': 1,
'task_id': 1,
'prompt_img': [],
'result_img': "",
'type': 2,
'callback': 'http://test.phpgpt.com/api/users/prompt/callback',
}
if task:
# 将任务添加到列表
cur_tasks.append(task)
if len(cur_tasks):
return cur_tasks
return []
async def open_browser(self, browser_item):
try:
logger.info(f'浏览器id-{browser_item["num"]},开始执行')
res = openBrowser(browser_item)
driverPath = res['data']['driver']
debuggerAddress = res['data']['http']
ws = res['data']['ws']
chrome_options = Options()
chrome_options.add_experimental_option("debuggerAddress", debuggerAddress)
browser = await pyppeteer.connect({
"browserWSEndpoint": ws
})
page = await browser.newPage()
logger.info('打开窗口')
await page.goto('https://discord.com/channels/@me')
return page
except Exception as e:
logger.error(e)
return False
# 初始化任务
async def on_start(self):
while True:
# 先获取任务
tasks = self.get_current_tasks()
if len(tasks):
# 打开窗口
page = await self.open_browser(self.browser_item)
if not page:
# 浏览器打开失败
logger.error(f'{self.browser_item["num"]},窗口打开失败,检查代理ip')
else:
# 开始执行任务
logger.info('开始执行任务')
status = await self.run_task(self.browser_item, tasks, page)
if status == False:
logger.info('请检查自动化')
else:
time.sleep(3)
logger.info('不存在任务,等待3秒')
async def run_task(self, browser_item, tasks, page):
try:
logger.info('等待登录成功')
# 判断页面是否加载完成
page_load = '//*[@data-list-item-id="guildsnav___home"]'
element = await self.CheckElement(page, page_load)
if not element:
logger.error('登录失败')
# 去登陆
return False
# 登录成功
logger.info('登录成功')
# 等待服务器元素出现
server_el = '//div[contains(@aria-label, "测试服务器")]'
element = self.CheckElement(page, 'xpath', server_el)
if element:
# 点击指定服务器
status = self.ElementClick(page, 'xpath', server_el, '测试服务器')
if status:
# 等待聊天窗口出现
input_el = '//*[@role="textbox"]'
element = self.CheckElement(page, 'xpath', input_el)
if element:
logger.info('聊天窗口存在')
# 输入指令
self.enter_prompt(page, tasks)
logger.info('元素未找到,任务暂停')
return False
except Exception as e:
logger.error(e)
return False
def process_start(browser_item, lock):
loop = asyncio.get_event_loop()
keyword = ToWork(browser_item, lock)
loop.run_until_complete(keyword.on_start())
if __name__ == '__main__':
# 测试-两个任务
browser_list = [
{
# 7
'num': 7,
'id': '1115ce586624431a8b7bcba8b1e82035'
},
# {
# # 9
# 'num': 9,
# 'id': 'db474b6936a24a7a949377a6d90b4dcb'
# }
]
lock = Lock()
l = []
for item in browser_list:
p = Process(target=process_start, args=(item, lock))
p.start()
l.append(p) # 将进程加入列表中，这样 5 个子进程就会全部执行
for p in l:
p.join() # 当5 个子进程全部执行完，才会执行主进程
bit_selenium.py
......@@ -17,6 +17,12 @@ import json
from multiprocessing import Process
import random
# 获取当前时间
def get_current_time():
# 返回秒级时间戳
return int(time.mktime(time.localtime(time.time())))
class ToWork:
def __init__(self):
self.init_prompt = '/imagine'
......@@ -34,15 +40,26 @@ class ToWork:
ActionChains(dricer).send_keys(Keys.CONTROL, value).perform()
# 元素出现的循环次数
def CheckElement(self, dricer, type, value, num=10):
def CheckElement(self, dricer, type, value, num=10, check_type='only'):
for i in range(num):
element = self.sleepElementShow(dricer, type, value)
if element:
# 随机等待
self.random_sleep(1, 2)
return element
if check_type == 'only':
element = self.sleepElementShow(dricer, type, value)
if element:
# 随机等待
self.random_sleep(1, 2)
return element
else:
# 列表
element = self.sleepElementListShow(dricer, type, value)
if element:
# 随机等待
self.random_sleep(1, 2)
return element
time.sleep(1)
return False
# 等待元素出现--自定义
def sleepElementShow(self, dricer, type, value):
try:
......@@ -53,6 +70,24 @@ class ToWork:
except Exception as e:
return False
# 找元素，只找一次，不等待
def FindElement(self, custom_el, find_type, value):
try:
el = custom_el.find_element(find_type, value)
return el
except Exception as e:
return False
# 获取元素列表
def sleepElementListShow(self, dricer, type, value):
try:
element = dricer.find_elements(type, value)
if element:
return element
return False
except Exception as e:
return False
# 元素点击
def ElementClick(self, dricer, type, value, name=''):
try:
......@@ -74,6 +109,35 @@ class ToWork:
# 循环输入指令
def enter_prompt(self, dricer, tasks):
if True:
# 获取开始时间
start_time = get_current_time()
while True:
# 再次获取当前时间
current_time = get_current_time()
# 最多等待一分钟
if current_time - start_time >= 60:
# end
break
# 开始判断
element = '//ol[@data-list-id="chat-messages"]/li'
# 消息列表
message_list = self.CheckElement(dricer, 'xpath', element, 20, 'list')
if message_list:
logger.info('消息列表存在')
# 找到所有子元素
for e in message_list:
logger.info(e.get_attribute('id'))
# 找到文本并匹配
path = '//div[contains(@data-list-item-id, "chat-messages___chat-messages")]'
label_element = self.FindElement(e, By.XPATH, path)
if label_element:
logger.info(label_element.get_attribute('data-list-item-id'))
else:
logger.info('未找到文本')
logger.info('等待300秒')
time.sleep(300)
return
# task是一个列表
for task in tasks:
# 输入初始指令
......@@ -86,13 +150,35 @@ class ToWork:
if element:
status = self.ElementClick(dricer, 'xpath', prompt_el, '选择指令')
if status:
self.driver_paste(dricer, task['prompt'])
change_prompt = task['prompt']
# 图片链接
prompt_img = task['prompt_img']
if task['type'] == 1 and prompt_img and len(json.loads(prompt_img)):
prompt_img = json.loads(prompt_img)
st = ''
for item in prompt_img:
# 加入图片-链接
st += ' ' + item
change_prompt = st + ' ' + change_prompt
self.driver_paste(dricer, change_prompt)
# 回车
ActionChains(dricer).send_keys(Keys.ENTER).perform()
time.sleep(2)
# 输入完成,循环获取生成成功后的链接
# 获取当前事件
# 获取开始时间
start_time = get_current_time()
while True:
# 再次获取当前时间
current_time = get_current_time()
# 最多等待一分钟
if current_time - start_time >= 60:
# end
break
# 开始判断
message_list = '//*[@data-list-id="chat-messages"]'
......@@ -134,7 +220,7 @@ class ToWork:
logger.info('打开窗口')
dricer.get('https://discord.com/channels/@me')
# dricer.get('https://discord.com/login')
return True
return dricer
except Exception as e:
logger.error(e)
return False
......@@ -146,21 +232,21 @@ class ToWork:
tasks = self.get_current_tasks()
if len(tasks):
# 打开窗口
browser_status = self.open_browser(browser_item)
if not browser_status:
dricer = self.open_browser(browser_item)
if not dricer:
# 浏览器打开失败
logger.error(f'{browser_item["num"]},窗口打开失败,检查代理ip')
else:
# 开始执行任务
statsu = self.run_task(browser_item, tasks)
if statsu == False:
status = self.run_task(browser_item, tasks, dricer)
if status == False:
logger.info('请检查自动化')
else:
time.sleep(3)
logger.info('不存在任务,等待3秒')
def run_task(self, browser_item, tasks):
def run_task(self, browser_item, tasks, dricer):
try:
logger.info('等待登录成功')
# 判断页面是否加载完成
......@@ -229,17 +315,17 @@ if __name__ == '__main__':
'num': 7,
'id': '1115ce586624431a8b7bcba8b1e82035'
},
{
# 9
'num': 9,
'id': 'db474b6936a24a7a949377a6d90b4dcb'
}
# {
# # 9
# 'num': 9,
# 'id': 'db474b6936a24a7a949377a6d90b4dcb'
# }
]
#
l = []
for item in browser_list:
p = Process(target=bro.OpenBros, args=(item,))
p = Process(target=bro.on_start, args=(item,))
p.start()
l.append(p) # 将进程加入列表中，这样 5 个子进程就会全部执行
......
browser_api.py 0 → 100644
# 获取浏览器窗口
async def browser(self, browser_id):
"""
:return:
"""
self.browser_id = browser_id
ws = await self.bit_get_ws()
return await pyppeteer.connect({
"browserWSEndpoint": ws
})
gpt.py
import asyncio
import json
import time
import requests
import re
import openai
from loguru import logger
from Helper import RedisCon, Helper
# key
api_key = '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'
api_key = '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'
# 组织标识
organization = 'org-ZMM4WGwi3vmM7o5y51IljTcW'
organization = 'org-cflsyp3YsvDovzOu3g8G9OFS'
# 回调接口
def gpt_callback(res_prompt, task):
url = task['callback']
print(1)
try:
url = task['callback']
if len(res_prompt):
params = {
'prompt': json.dumps(res_prompt),
'user_id': task['user_id'],
'task_id': task['task_id'],
'prompt_img': task['prompt_img'],
'type': task['type'],
}
res = requests.post(url, data=params)
else:
logger.error(f'指令任务不正确，{res_prompt}')
except Exception as e:
logger.error(e)
# 拆分指令
def split_prompt(response, task):
try:
text = response.choices[0].text
if text:
# 使用正则定义分隔符
pattern = r'\n[1-9]{1,2}.*.'
# 拆分字符串
chemical_list = re.findall(pattern, text)
for i in range(len(chemical_list)):
# 匹配标题并替换
chemical_list[i] = re.sub(r'\n[1-9]{1,3}.', '', chemical_list[i], count=0, flags=0)
return chemical_list
except Exception as e:
logger.error(e)
logger.error('gpt指令拆分失败，请查看原因')
return []
# 获取模型列表
def get_module_list():
header = {
'Authorization': 'Bearer ' + api_key,
}
x = requests.get('https://api.openai.com/v1/models', headers=header)
print(x.text)
def split_prompt(response):
logger.info(response)
def get_gpt_message(task):
openai.organization = organization
......@@ -28,29 +72,28 @@ def get_gpt_message(task):
logger.info('开始访问open_ai')
try:
response = openai.Completion.create(
model="text-davinci-003",
prompt="请根据我给你的指令生成"+str(prompt_num)+"条相似的指令，指令如下:"+prompt,
temperature=1,
max_tokens=500,
top_p=1,
frequency_penalty=0.0,
presence_penalty=0.0,
stop=None
model="text-davinci-003",
prompt="请根据我给你的指令生成" + str(prompt_num) + "条相似的指令，指令如下:" + prompt,
temperature=1,
max_tokens=1000,
top_p=1,
frequency_penalty=0.0,
presence_penalty=0.0,
stop=None
)
logger.info(response)
# 拆分指令
res_prompt = split_prompt(response, task)
# 回调
gpt_callback(res_prompt, task)
logger.info('open_ai_结束')
except Exception as e:
logger.info(f"错误 {e}")
# 拆分指令
res_prompt = split_prompt(response, task)
# 回调
gpt_callback(res_prompt, task)
logger.info('open_ai_结束')
if __name__ == '__main__':
key = 'laravel_database_gpt_text_prompt'
logger.info('开始执行')
while True:
while False:
tasks = list()
# 获取任务数量
task_num = RedisCon.get_task_num(key)
......@@ -72,19 +115,11 @@ if __name__ == '__main__':
tasks.append(get_gpt_message(task))
loop.run_until_complete((asyncio.wait(tasks, timeout=30)))
else:
time.sleep(3)
logger.info('等待100秒')
time.sleep(100)
continue
# test_task = {
# 'type': 2,
# 'prompt': 'a very cute boys, 16 years old, fly a kite,in beautifulstreet,full body，3d art, c4d, octane render, ray tracting,clay material,popmart blind box，Pixar trend, animationlighting, depth of field, ultra detailed --q 5',
# 'prompt_img': '',
# 'user_id': 1,
# 'prompt_num': 5,
# 'task_id': 10,
# 'callback': 'http://test.phpgpt.com/api/users/gpt/callback'
# }
# get_gpt_message(test_task)
test_task = {'type': 2,
'prompt': 'a very cute boys, 16 years old, fly a kite,in beautifulstreet,full body，3d art, c4d, octane render, ray tracting,clay material,popmart blind box，Pixar trend, animationlighting, depth of field, ultra detailed',
'prompt_img': None, 'user_id': 1, 'prompt_num': 5,
'callback': 'http://test.phpgpt.com/api/users/gpt/callback', 'task_id': 1}
get_gpt_message(test_task)
test.py
import time
import re
# 获取当前时间
# print(int(time.mktime(time.localtime(time.time()))))
text = '\u6761\u76f8\u4f3c\u7684\u6307\u4ee4\u53ef\u80fd\u662f\uff1a\n\n1.A attractive girls, 12 years old, dancing with a rainbow umbrella, in a wide street, full body, V-Ray render, Octane render, X-Particles, cartoon material, Popmart keychains, CG trend, night lighting, long-focus, ultra high definition. \n2.A very handsome teenagers, 10 years old, riding a skateboard, in a colorful playground, 3/4 body, Arnold render, Corona render, hard surface modeling, photorealistic material, Popmart toys, Pixar tech, interior lighting, soft focus, ultra realistic. \n3.A very lovely kids, 5 years old, playing a guitar, in an old alley, full body, Blender render, Redshift render, polygon modeling, 3d printed material, Popmart collectibles, anime trend, day lighting, depth of field, ultra stunning. \n4.A cute baby, 2 years old, flying a kite, in a green park, upper body, Maxwell render, Arnold render, sculpting, traditional material, Popmart accessories, motion graphics trend, art direction lighting, bokeh, ultra hyper real. \n5.A really cute teens, 14 years old, drawing a picture, in a quiet garden, full body, Octane render, Art Renderer, low polygon modeling, abstract material, Popmart dolls, Hollywood trend, environment lighting, vignette, ultra vibrant.'
# 拆分
# 使用正则定义分隔符
pattern = r'\n[1-9]{1,2}.*.'
# 拆分字符串
chemical_list = re.findall(pattern, text)
for item in chemical_list:
# 匹配标题并替换
item = re.sub(r'\n[1-9]{1,3}.', '', item, count=0, flags=0)
print(item)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment