From 683efb6d8daaa5f1b1a1df9bd679c4673ec1bb60 Mon Sep 17 00:00:00 2001
From: ARCshekin <85334866+ARCshekin@users.noreply.github.com>
Date: Sun, 20 Jul 2025 16:48:47 +0300
Subject: [PATCH 1/5] Update main.py
---
python/main.py | 396 ++++++++++++++++++++++++++++++++++++++++---------
1 file changed, 328 insertions(+), 68 deletions(-)
diff --git a/python/main.py b/python/main.py
index a9f4328..b1a2c2c 100644
--- a/python/main.py
+++ b/python/main.py
@@ -1,32 +1,78 @@
import os
import logging
+import sys
+from this import d
from typing import List, Any, Optional
+from datetime import datetime
import requests
from bs4 import BeautifulSoup
from fastapi import FastAPI
from pydantic import BaseModel
from openai import OpenAI
-from selenium import webdriver
+import undetected_chromedriver as uc
from selenium.webdriver.chrome.options import Options
-# --- API client setup ---
-API_KEY = os.getenv("OPENAI_API_KEY", "sk-0M12tbkrnKubF86nHyKPKidkqoqfNzei")
-BASE_URL = "https://api.proxyapi.ru/openai/v1"
+# --- Logging Configuration ---
+def setup_logging():
+ """Configure logging with pretty formatting and colors"""
+ # Create formatter with timestamp, level, and message
+ formatter = logging.Formatter(
+ '%(asctime)s | %(levelname)-8s | %(name)-15s | %(message)s',
+ datefmt='%Y-%m-%d %H:%M:%S'
+ )
+
+ # Console handler with colors
+ console_handler = logging.StreamHandler(sys.stdout)
+ console_handler.setFormatter(formatter)
+ console_handler.setLevel(logging.INFO)
+
+ # Configure root logger
+ root_logger = logging.getLogger()
+ root_logger.setLevel(logging.DEBUG)
+ root_logger.addHandler(console_handler)
+
+ # Create logger for this module
+ logger = logging.getLogger(__name__)
+ logger.info("🚀 Application started - Logging configured")
+ return logger
+
+
+# Initialize logging
+logger = setup_logging()
+
client = OpenAI(
- api_key=API_KEY,
- base_url=BASE_URL
+ api_key="sk-or-vv-332479a389752e4e178aacf60daa99709ebd4dbcaf7ee44b78c3fd2491293534",
+ base_url="https://api.vsegpt.ru/v1",
+
)
def ask_ai(prompt: str) -> str:
- chat_completion = client.chat.completions.create(
- model="gpt-4.1-2025-04-14",
- messages=[{"role": "user", "content": prompt}]
- )
- return chat_completion.choices[0].message.content.strip()
+ """Send prompt to AI and return response"""
+ logger.debug("🤖 Sending prompt to AI")
+ try:
+ messages = [
+ {"role": "system", "content": "Ты - большая языковая модель. Отвечай на вопросы пользователя."},
+ {"role": "user", "content": prompt}
+ ]
+ completion = client.chat.completions.create(
+ model='deepseek/deepseek-r1-alt-0528',
+ messages=messages,
+ temperature=0.1,
+ extra_headers={ "X-Title": "Colab Base Example" },
+ )
+ response = completion.choices[0].message.content
+ if response is None:
+ logger.error("❌ AI returned None response")
+ return ""
+ logger.debug("✅ AI response received")
+ return response.strip()
+ except Exception as e:
+ logger.error(f"❌ AI request failed: {e}")
+ raise
class APIResponse(BaseModel):
@@ -45,93 +91,307 @@ class WebTestRunner:
def __init__(self, start_url: str):
self.current_url = start_url
self.current_html = ""
+ self.logger = logging.getLogger(f"{__name__}.WebTestRunner")
+
def fetch_page(self, url: str) -> str:
- # Use headless Selenium to fetch the page HTML
- chrome_options = Options()
- chrome_options.add_argument("--headless")
- chrome_options.add_argument("--no-sandbox")
- chrome_options.add_argument("--disable-dev-shm-usage")
- driver = webdriver.Chrome(options=chrome_options)
- driver.get(url)
- # Wait for dynamic content if needed (could add explicit waits here)
- html = driver.page_source
- self.current_url = driver.current_url
- self.current_html = html
- driver.quit()
- return html
+ """Fetch page HTML using Playwright"""
+ self.logger.info(f"🌐 Fetching page: {url}")
+ # Use Playwright in headless mode to fetch the page HTML (bypasses Cloudflare)
+ try:
+ from playwright.sync_api import sync_playwright
+ html = ""
+ with sync_playwright() as p:
+ browser = p.chromium.launch(headless=True)
+ context = browser.new_context()
+ page = context.new_page()
+ self.logger.debug("📄 Navigating to page...")
+ page.goto(url, timeout=60000)
+ page.wait_for_load_state('networkidle', timeout=60000)
+ html = page.content()
+ browser.close()
+ self.current_url = url
+ self.current_html = html
+ self.logger.info(f"✅ Page fetched successfully ({len(html)} characters)")
+ return html
+ except Exception as e:
+ self.logger.error(f"❌ Failed to fetch page: {e}")
+ raise
+
def check_page(self, url: str, prompts: List[str]) -> List[bool]:
- # Fetch page via Selenium for JS-rendered content
- raw_html = self.fetch_page(url)
- soup = BeautifulSoup(raw_html, 'lxml')
- results: List[bool] = []
- for criterion in prompts:
- # Extract context for AI if no simple rule
- excerpt = soup
- c_lower = criterion.lower()
- # Choose prompt based on criterion template
- if c_lower.startswith("does") and "exist" in c_lower:
- instruct = (
- "Please answer 'Yes' or 'No'.\n"
- f"Criterion: \"{criterion}\" means check if the specified element exists on the page.\n"
- "Example: Does logo exist? -> Yes if a logo image is present near the company name.\n"
- )
- elif c_lower.startswith("is") and "clickable" in c_lower:
- instruct = (
- "Please answer 'Yes' or 'No'.\n"
- f"Criterion: \"{criterion}\" means check if the specified element is clickable (e.g., links, buttons).\n"
- "Example: Is 'Submit' button clickable? -> Yes if it responds to clicks.\n"
- )
- elif c_lower.startswith("does") and ("attribute" in c_lower or "value" in c_lower):
- instruct = (
- "Please answer 'Yes' or 'No'.\n"
- f"Criterion: \"{criterion}\" means check if the element has the given attribute or value.\n"
- "Example: Does input have attribute 'placeholder'? -> Yes if the input tag includes placeholder attribute.\n"
- )
- else:
- instruct = (
- "Please answer 'Yes' or 'No'.\n"
- f"Criterion: \"{criterion}\". Assess based on page content intelligently.\n"
- )
- question = f"{instruct}Context: {excerpt}"
- answer = ask_ai(question).lower()
- f = open('output.txt', 'w')
- f.write(answer)
- f.close()
- logging.info("AI answer: %s", answer)
- result = answer.startswith("yes")
- results.append(result)
- return results
+ """Check page against given criteria"""
+ self.logger.info(f"🔍 Checking page against {len(prompts)} criteria")
+ try:
+ # Fetch page via Selenium for JS-rendered content
+ raw_html = self.fetch_page(url)
+ soup = BeautifulSoup(raw_html, 'lxml')
+ results: List[bool] = []
+
+ for i, criterion in enumerate(prompts, 1):
+ self.logger.info(f"🔍 Criterion {i}/{len(prompts)}: {criterion}")
+ # Extract context for AI if no simple rule
+ excerpt = soup
+ c_lower = criterion.lower()
+ # Choose prompt based on criterion template
+ if c_lower.startswith("does") and "exist" in c_lower:
+ instruct = (
+ "Please answer 'Yes' or 'No'.\n"
+ f"Criterion: \"{criterion}\" means check if the specified element exists on the page.\n"
+ "Example: Does logo exist? -> Yes if a logo image is present near the company name.\n"
+ )
+ elif c_lower.startswith("is") and "clickable" in c_lower:
+ instruct = (
+ "Please answer 'Yes' or 'No'.\n"
+ f"Criterion: \"{criterion}\" means check if the specified element is clickable (e.g., links, buttons).\n"
+ "Example: Is 'Submit' button clickable? -> Yes if it responds to clicks.\n"
+ )
+ elif c_lower.startswith("does") and ("attribute" in c_lower or "value" in c_lower):
+ instruct = (
+ "Please answer 'Yes' or 'No'.\n"
+ f"Criterion: \"{criterion}\" means check if the element has the given attribute or value.\n"
+ "Example: Does input have attribute 'placeholder'? -> Yes if the input tag includes placeholder attribute.\n"
+ )
+ else:
+ instruct = (
+ "Please answer 'Yes' or 'No'.\n"
+ f"Criterion: \"{criterion}\". Assess based on page content intelligently.\n"
+ )
+ question = f"{instruct}Context: {excerpt}"
+ answer = ask_ai(question).lower()
+ self.logger.info(f"🤖 AI answer for '{criterion}': {answer}")
+ result = answer.startswith("yes")
+ results.append(result)
+ self.logger.info(f"✅ Criterion {i} result: {'PASS' if result else 'FAIL'}")
+
+ self.logger.info(f"📊 Page check completed: {sum(results)}/{len(results)} criteria passed")
+ return results
+ except Exception as e:
+ self.logger.error(f"❌ Error while testing page: {e}")
+ return [False] * len(prompts)
+
+
+ def integration_test(self, test: str, **kwargs) -> tuple[bool, str]:
+ """Run integration test using Playwright"""
+ try:
+ try:
+ from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError
+ except ImportError:
+ self.logger.error("❌ Playwright not installed")
+ return False, "Playwright not installed"
+
+ import time
+ import json
+
+ def parse_llm_json(text: str) -> dict:
+ try:
+ start = text.find("")
+ end = text.find("", start)
+ if start != -1 and end != -1:
+ json_str = text[start + 6:end]
+ return json.loads(json_str)
+ except Exception:
+ pass
+ raise ValueError("Failed to parse JSON from LLM response")
+
+ self.logger.info(f"🧪 Starting integration test: {test}")
+ self.logger.info(f"📋 Test parameters: {kwargs}")
+ history = []
+
+ with sync_playwright() as p:
+ browser = p.chromium.launch(headless=True)
+ context = browser.new_context()
+ page = context.new_page()
+ try:
+ self.logger.info(f"🌐 Navigating to: {self.current_url}")
+ page.goto(self.current_url, timeout=60000)
+ page.wait_for_load_state('networkidle', timeout=60000)
+ self.logger.info("✅ Page loaded successfully")
+
+ while True:
+ html = page.content()
+ # Remove