سلام! من متصل به هوشواره گوگل هستم. میتوانید علاوه بر متن، تصویر هم برایم بفرستید تا بررسی کنم.
✕ حذف
"""
@app.get("/", response_class=HTMLResponse)
async def chat_interface():
return HTML_CHAT_INTERFACE
@app.post("/api/chat")
async def chat_endpoint(message: str = Form(...), image: UploadFile = File(None)):
saved_image_path = None
if image:
try:
saved_image_path = os.path.join(UPLOAD_DIR, f"{int(time.time())}_{image.filename}")
with open(saved_image_path, "wb") as buffer:
buffer.write(await image.read())
except Exception as e:
raise HTTPException(status_code=500, detail=f"خطا در ذخیرهسازی تصویر: {str(e)}")
url = "https://www.google.com/search?q=سلام&udm=50"
ai_response = ""
async with async_playwright() as p:
try:
browser = await p.chromium.launch(
headless=True,
args=[
"--no-sandbox",
"--disable-setuid-sandbox",
"--disable-dev-shm-usage",
"--disable-blink-features=AutomationControlled"
]
)
context = await browser.new_context(
user_agent="Mozilla/5.0 (Linux; Android 10; K) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Mobile Safari/537.36",
viewport={"width": 412, "height": 915},
locale="fa-IR",
timezone_id="Asia/Tehran"
)
page = await context.new_page()
await page.goto(url, wait_until="networkidle", timeout=45000)
await page.wait_for_timeout(3000)
# پروسه مالتیمدال و انتقال تصویر با ضریب ایمنی بالا و گامهای مستقل خطاپذیر
if saved_image_path and os.path.exists(saved_image_path):
uploaded = False
# استراتژی ۱: تلاش برای تزریق مستقیم به لایه بارگذاری فایل (بدون نیاز به کلیک و باز کردن منو)
try:
file_input = await page.query_selector("input[type='file']")
if file_input:
await file_input.set_input_files(saved_image_path)
await page.wait_for_timeout(3000)
uploaded = True
print("Direct file input injection succeeded.")
except Exception as e:
print(f"Direct injection bypassed: {e}")
# استراتژی ۲: اگر تزریق مستقیم عمل نکرد، کلیک روی آیکون پلاس با تایماوت کوتاه و باز کردن منوی گالری
if not uploaded:
try:
# فوکوس روی کادر چت با تایماوت ۲ ثانیهای برای باز شدن ابزارهای کناری
chat_inputs = await page.query_selector_all("textarea, input[type='text'], [contenteditable='true']")
for inp in chat_inputs:
try:
if await inp.is_visible():
await inp.click(timeout=2000)
await page.wait_for_timeout(500)
break
except:
continue
# پیدا کردن آیکون مثبت یا دوربین و فعالسازی شبیهساز فایل چوزر
elements = await page.query_selector_all("button, div[role='button'], svg, [aria-label]")
for el in elements:
try:
if await el.is_visible():
aria = (await el.get_attribute("aria-label") or "").lower()
html = (await el.inner_html() or "").lower()
text = (await el.inner_text() or "").lower()
combined = aria + html + text
if any(k in combined for k in ["+", "plus", "camera", "lens", "دوربین", "تصویر", "عکس"]):
async with page.expect_file_chooser(timeout=3000) as fc_info:
await el.click(timeout=2000) # تایماوت امن ۲ ثانیهای به جای ۳۰ ثانیه معطلی
file_chooser = await fc_info.value
await file_chooser.set_files(saved_image_path)
await page.wait_for_timeout(3000)
uploaded = True
print("Uploaded via main UI icon trigger.")
break
except:
continue
except Exception as e:
print(f"UI icon flow failed: {e}")
# استراتژی ۳: کلیک مستقیم روی منوی پاپآپ پایینی (گالری یا فایلها) در صورت باز شدن لایه شیت گوگل
if not uploaded:
try:
menu_items = await page.query_selector_all("div, span, button, p")
for item in menu_items:
try:
item_text = (await item.inner_text() or "").strip()
if item_text in ["گالری", "فایلها", "Gallery", "Files"]:
if await item.is_visible():
async with page.expect_file_chooser(timeout=3000) as fc_info:
await item.click(timeout=2000)
file_chooser = await fc_info.value
await file_chooser.set_files(saved_image_path)
await page.wait_for_timeout(3000)
uploaded = True
print("Uploaded via bottom sheet item click.")
break
except:
continue
except Exception as e:
print(f"Bottom sheet workflow failed: {e}")
# یافتن کادر متنی گفتگو جهت وارد کردن متن پیام کاربر
input_box = None
candidates = await page.query_selector_all("textarea, input, [contenteditable='true']")
for el in candidates:
try:
if await el.is_visible():
placeholder = await el.get_attribute("placeholder") or ""
if any(w in placeholder for w in ["بپرسید", "ask", "پیام", "چطور", "هرچه"]):
input_box = el
break
except:
continue
if not input_box:
input_box = await page.query_selector("textarea") or await page.query_selector("input[type='text']")
if input_box:
await input_box.focus()
await input_box.fill(message)
await page.wait_for_timeout(500)
await input_box.press("Enter")
# کلیک بر روی آیکون ارسال فرم (فلش آبی بالا) با استفاده از زمانبندی محدود ۲ ثانیهای
try:
send_buttons = await page.query_selector_all("button, [role='button']")
for btn in send_buttons:
if await btn.is_visible():
aria = (await btn.get_attribute("aria-label") or "").lower()
if any(w in aria for w in ["send", "ارسال", "arrow", "up"]):
await btn.click(timeout=2000)
break
except:
pass
# انتظار تعمدی به مدت ۹ ثانیه جهت تایپ کامل جواب توسط هوش مصنوعی گوگل
await page.wait_for_timeout(9000)
full_text = await page.evaluate("() => document.body.innerText")
await browser.close()
# فیلترسازی خطوط خروجی برای تمیز کردن تگهای اضافی سرچ گوگل
lines = full_text.split('\n')
cleaned_lines = []
skip_headers = ["تصاویر", "ویدیوها", "اخبار", "نقشهها", "خرید کردن", "کتابها", "مالی", "حالت موضوعمحور", "ورود"]
for line in lines:
line_str = line.strip()
if not line_str:
continue
if any(header in line_str for header in skip_headers):
continue
if "در پاسخهای" in line_str or "بیشتر بدانید" in line_str:
continue
cleaned_lines.append(line_str)
ai_response = "\n".join(cleaned_lines)
if not ai_response:
ai_response = full_text
except Exception as e:
if 'browser' in locals():
await browser.close()
raise HTTPException(status_code=500, detail=f"خطا در رندر سرور هوش مصنوعی: {str(e)}")
finally:
# پاکسازی حتمی فایل تصویر از روی سرور داکر
if saved_image_path and os.path.exists(saved_image_path):
try:
os.remove(saved_image_path)
except:
pass
return JSONResponse(content={"response": ai_response})
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=7860)