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