from __future__ import annotations """ OpenAI-powered screenshot analyzer with automated clicking functionality. Specifically designed for detecting and clicking on supporting document links in invoice screenshots, with configurable wait times and S3 integration. """ import base64 import json import logging import os import time from pathlib import Path try: from decimal import Decimal as _Dec except Exception: _Dec = None # type: ignore from typing import List, Dict, Tuple, Optional from io import BytesIO from .prompt_builder import build_find_value_prompt import requests import pyautogui import boto3 from PIL import Image from PIL import ImageDraw, ImageFont from .utils import column_index_to_label, column_label_to_index try: # Prefer official SDK if available from openai import OpenAI as _OpenAIClient except Exception: _OpenAIClient = None from .constants import ( OPENAI_API_KEY, AWS_REGION, LOCAL_DEV, CHROME_DOWNLOADS_S3_BUCKET, OPENAI_ANALYZER_S3_BUCKET, POST_CLICK_WAIT_TIME, GRID_LINE_COLOR_RGBA, GRID_LABEL_TEXT_COLOR_RGBA, GRID_LABEL_BG_COLOR_RGBA, GRID_COL_LABEL_STYLE, GRID_COL_LABEL_POSITIONS, GRID_LABEL_FONT_SIZE, GRID_USE_HEADER_BANDS, SCREENSHOT_PRE_DELAY_SEC, ) # Allow extending OpenAI request timeout via env (default 180s for vision) _REQUEST_TIMEOUT_SECONDS = int(os.getenv('OPENAI_REQUEST_TIMEOUT', '240')) class OpenAIAnalyzer: """OpenAI-powered screenshot analyzer focused on supporting document links.""" def __init__(self, app=None, s3_bucket: str = None, wait_time: int = None): """ Initialize the OpenAI analyzer. Args: app: Main application instance for UI updates s3_bucket: S3 bucket name for storing screenshots wait_time: Configurable wait time after each click (defaults to constant) """ self.app = app # DPI awareness and scaling self._dpi_scale = (1.0, 1.0) self._ensure_dpi_awareness() self.s3_bucket = s3_bucket or OPENAI_ANALYZER_S3_BUCKET self.wait_time = wait_time or POST_CLICK_WAIT_TIME self.s3_client = None # Vision model (configurable): default to gpt-4-vision-preview, override with env OPENAI_VISION_MODEL try: self.vision_model = os.getenv('OPENAI_VISION_MODEL', 'gpt-5') # self.vision_model = os.getenv('OPENAI_VISION_MODEL', 'gpt-4o') except Exception: self.vision_model = 'gpt-5' try: self.api_mode = os.getenv('OPENAI_API_MODE', 'chat') # force chat unless overridden except Exception: self.api_mode = 'auto' # Lazy-created OpenAI client self._openai_client = None # Setup OpenAI-specific logging self._setup_openai_logging() self._init_s3_client() # Configure pyautogui for safety pyautogui.FAILSAFE = True # Move mouse to corner to abort pyautogui.PAUSE = 0.1 # Small pause between actions pyautogui.MINIMUM_DURATION = 0.1 # Minimum movement duration self.openai_logger.info(f"OpenAI Analyzer initialized with {self.wait_time}s wait time") self.openai_logger.info(f"S3 bucket configured: {self.s3_bucket}") self.openai_logger.info(f"Local development mode: {LOCAL_DEV}") try: self.openai_logger.info(f"OpenAI model: {self.vision_model}; HTTP timeout: {_REQUEST_TIMEOUT_SECONDS}s") except Exception: pass def _temporarily_hide_app(self): """Hide the Tk app window briefly so it doesn't appear in screenshots. Returns a callable to restore the window state.""" def _noop(): pass try: if not self.app: return _noop try: self.app.attributes('-topmost', False) except Exception: pass try: self.app.withdraw() except Exception: pass time.sleep(0.1) def _restore(): try: self.app.deiconify() self.app.attributes('-topmost', True) except Exception: pass return _restore except Exception: return _noop def _get_screen_size(self) -> Tuple[int, int]: """Detect current primary screen size at runtime; fallback to constants.""" try: try: import pyautogui as _pg size = _pg.size() if size and size[0] > 0 and size[1] > 0: return int(size[0]), int(size[1]) except Exception: pass try: from PIL import ImageGrab as _IG img = _IG.grab() w, h = img.size if w and h: return int(w), int(h) except Exception: pass from .constants import SCREEN_WIDTH, SCREEN_HEIGHT return int(SCREEN_WIDTH), int(SCREEN_HEIGHT) except Exception: return 1920, 1080 def _ensure_dpi_awareness(self) -> None: """Make process DPI aware on Windows to avoid scaled click offsets.""" try: import ctypes ctypes.windll.user32.SetProcessDPIAware() except Exception: pass def _resolve_chat_model(self, model_name: Optional[str]) -> str: """Map general model names to recommended Chat API aliases when needed.""" try: m = (model_name or "").lower() if m in ("gpt-5", "gpt-5-main"): return "gpt-5-chat-latest" return model_name or "gpt-5-chat-latest" except Exception: return model_name or "gpt-5-chat-latest" def _apply_display_scale(self, coords: List[int]) -> List[int]: try: sx, sy = getattr(self, '_dpi_scale', (1.0, 1.0)) return [int(coords[0] * sx), int(coords[1] * sy)] except Exception: return coords def _looks_like_amount(self, s: Optional[str]) -> bool: """Heuristic check for currency/amount-like strings. Examples considered true: "440", "1,200", "$165.50", "A$1,650.00". Names, dates, invoice numbers that don't match this numeric pattern return False. """ try: if not s: return False import re txt = str(s).strip() pattern = r"^[A$€£¥]?\s*[0-9]{1,3}(?:,[0-9]{3})*(?:[.,][0-9]{1,2})?$|^[A$€£¥]?\s*[0-9]+(?:[.,][0-9]{1,2})?$" return re.match(pattern, txt) is not None except Exception: return False def _normalize_amount(self, s: Optional[str]) -> Optional[str]: """Normalize currency/amount text to a canonical form like '450.00'.""" try: if not s: return None import re txt = str(s) # Remove currency symbols and spaces txt = re.sub(r"[\sA$€£¥]", "", txt) # Keep digits, dots, and commas only txt = re.sub(r"[^0-9.,]", "", txt) # Remove thousand separators, normalize decimal point if "," in txt and "." in txt: # Assume comma is thousands, dot is decimal txt = txt.replace(",", "") else: # If only commas, treat comma as decimal when there's exactly one if txt.count(",") == 1 and "." not in txt: txt = txt.replace(",", ".") else: txt = txt.replace(",", "") # Ensure two decimals if "." not in txt: txt = f"{txt}.00" else: whole, dec = txt.split(".", 1) dec = (dec + "00")[:2] txt = f"{whole}.{dec}" # Drop leading zeros in whole part except zero whole, dec = txt.split(".") whole = str(int(whole)) if whole.isdigit() else whole return f"{whole}.{dec}" except Exception: return None def _get_debug_dir(self) -> Path: """Return the debug directory path, ensuring it exists.""" try: d = Path('logs') / 'openai_debug' d.mkdir(parents=True, exist_ok=True) return d except Exception: return Path('.') def _save_text(self, path: Path, text: str): try: with open(path, 'w', encoding='utf-8') as f: f.write(text) except Exception: pass def _save_json(self, path: Path, obj): def _json_safe(o): try: if _Dec is not None and isinstance(o, _Dec): try: return float(o) except Exception: return str(o) if isinstance(o, dict): return {k: _json_safe(v) for k, v in o.items()} if isinstance(o, list): return [_json_safe(v) for v in o] return o except Exception: return o try: with open(path, 'w', encoding='utf-8') as f: json.dump(_json_safe(obj), f, ensure_ascii=False, indent=2) except Exception: pass def _save_overlay(self, screenshot_path: str, marks: List[Tuple[int, int]], out_path: Path, label: str = '', box: Optional[List[int]] = None, raw_mark: Optional[Tuple[int, int]] = None): """Save an image copy with red crosshair marks at provided coordinates. - marks: transformed (x,y) - raw_mark: optional original (x,y) marked in blue - box: optional [x1,y1,x2,y2] rectangle in green """ try: img = Image.open(screenshot_path).convert('RGB') draw = ImageDraw.Draw(img) for (x, y) in marks: x = int(x); y = int(y) r = 12 draw.line([(x - r, y), (x + r, y)], fill=(255, 0, 0), width=3) draw.line([(x, y - r), (x, y + r)], fill=(255, 0, 0), width=3) draw.ellipse([(x - 4, y - 4), (x + 4, y + 4)], outline=(255, 0, 0), width=3) if raw_mark is not None: try: rx, ry = int(raw_mark[0]), int(raw_mark[1]) r = 12 draw.line([(rx - r, ry), (rx + r, ry)], fill=(50, 120, 255), width=2) draw.line([(rx, ry - r), (rx, ry + r)], fill=(50, 120, 255), width=2) draw.ellipse([(rx - 4, ry - 4), (rx + 4, ry + 4)], outline=(50, 120, 255), width=2) except Exception: pass if box and isinstance(box, (list, tuple)) and len(box) == 4: try: x1, y1, x2, y2 = [int(v) for v in box] draw.rectangle([(x1, y1), (x2, y2)], outline=(0, 200, 0), width=3) except Exception: pass if label: try: draw.text((10, 10), label, fill=(255, 0, 0)) except Exception: pass img.save(out_path) except Exception: pass def _save_grid_overlay(self, screenshot_path: str, out_path: Path, cols: int, rows: int) -> None: """Save a copy of the screenshot with a visual grid (cols x rows), matching selected renderer.""" try: from .constants import GRID_RENDERER if GRID_RENDERER == 'alpha_numeric': # Use the new alpha-numeric tiled renderer so debug overlay matches input from .add_grid import add_alpha_numeric_grid_to_image as _alpha_grid from .constants import ( GRID_ALPHA_CELL_SIZE, GRID_TEXT_OPACITY, GRID_TEXT_OFFSET_X, GRID_ALPHA_FONT_SIZE, GRID_LINE_COLOR_RGBA, GRID_LABEL_BG_COLOR_RGBA, GRID_ALPHA_TILE_ALPHA, ) grid_color = (GRID_LINE_COLOR_RGBA[0], GRID_LINE_COLOR_RGBA[1], GRID_LINE_COLOR_RGBA[2]) _alpha_grid( base_image_path=screenshot_path, output_path=str(out_path), cell_size=int(GRID_ALPHA_CELL_SIZE), grid_color=grid_color, text_opacity=int(GRID_TEXT_OPACITY), text_offset_x=int(GRID_TEXT_OFFSET_X), label_background_color=GRID_LABEL_BG_COLOR_RGBA, font_size=int(GRID_ALPHA_FONT_SIZE), tile_alpha=int(GRID_ALPHA_TILE_ALPHA), ) else: # Delegate to legacy/internal renderer for consistency from .legacy_grid import save_grid_overlay as _legacy _legacy(screenshot_path, out_path, cols, rows) except Exception: # Silently ignore to keep analyzer robust pass def _setup_openai_logging(self): """Setup dedicated logging for OpenAI analyzer activities.""" # Create logs directory if it doesn't exist logs_dir = Path("logs") logs_dir.mkdir(exist_ok=True) # Create OpenAI logger self.openai_logger = logging.getLogger('OpenAIAnalyzer') self.openai_logger.setLevel(logging.INFO) # Prevent duplicate handlers if not self.openai_logger.handlers: # File handler for openai.log log_file = logs_dir / "openai.log" file_handler = logging.FileHandler(log_file, mode='a', encoding='utf-8') file_handler.setLevel(logging.INFO) # Console handler for immediate feedback console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) # Create formatter formatter = logging.Formatter( '[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) # Add handlers self.openai_logger.addHandler(file_handler) self.openai_logger.addHandler(console_handler) # Log initial setup self.openai_logger.info("OpenAI Analyzer logging system initialized") self.openai_logger.info(f"Log file: {log_file.absolute()}") def _init_s3_client(self): """Initialize S3 client if running in production.""" if LOCAL_DEV: self.openai_logger.info("Running in local development mode - S3 operations disabled") self.s3_client = None return try: self.s3_client = boto3.client("s3", region_name=AWS_REGION) # Test AWS credentials self.s3_client.list_buckets() self.openai_logger.info("S3 client initialized successfully") except Exception as e: self.openai_logger.error(f"Failed to initialize S3 client: {e}") self.s3_client = None def analyze_and_click_links(self, screenshot_path: str, playlist_name: str = None) -> Dict: """ Main function: Analyze screenshot for supporting document links and click them automatically. Args: screenshot_path: Path to the screenshot to analyze playlist_name: Name of the playlist for organization Returns: Dict containing analysis results and click actions performed """ try: self.openai_logger.info(f"Starting analysis of screenshot: {screenshot_path}") self.openai_logger.info(f"Playlist: {playlist_name or 'Not specified'}") if self.app: self.app.show_loader('Analyzing screenshot for supporting document links...') # Step 1: Analyze screenshot with OpenAI for links only # Send heartbeat before analysis begins try: if self.app: from os import getenv as _getenv hb_interval = float(_getenv("HB_INTERVAL_SEC", "60")) self.app.send_heartbeat_throttled(min_interval_sec=hb_interval) except Exception: pass clickable_links = self._analyze_screenshot_for_links(screenshot_path) if not clickable_links: self.openai_logger.warning("No supporting document links found in screenshot") if self.app: self.app.hide_loader() return {"success": False, "message": "No supporting document links found"} self.openai_logger.info(f"Found {len(clickable_links)} supporting document links to process") if self.app: self.app.hide_loader() self.app.show_loader(f'Found {len(clickable_links)} supporting document links. Starting automated clicking...') # Step 2: Perform automated clicking on links # Send heartbeat before long clicking sequence try: if self.app: from os import getenv as _getenv hb_interval = float(_getenv("HB_INTERVAL_SEC", "60")) self.app.send_heartbeat_throttled(min_interval_sec=hb_interval) except Exception: pass results = self._perform_automated_link_clicking( screenshot_path, clickable_links, playlist_name ) if self.app: self.app.hide_loader() # Log final results successful_clicks = sum(1 for r in results if r.get('success', False)) self.openai_logger.info(f"Analysis complete. {successful_clicks}/{len(clickable_links)} links processed successfully") return { "success": True, "clickable_links": clickable_links, "click_results": results, "total_links": len(clickable_links), "wait_time_used": self.wait_time } except Exception as e: self.openai_logger.error(f"Error in analyze_and_click_links: {e}") if self.app: self.app.hide_loader() return {"success": False, "error": str(e)} # --- OpenAI helper ----------------------------------------------------- def _vision_api_mode(self) -> str: """Decide API mode to use based on env and model name.""" mode = (self.api_mode or 'auto').lower() if mode in ('chat', 'responses'): return mode # Auto: prefer responses for gpt-5 / gpt-4.1 families name = (self.vision_model or '').lower() if name.startswith('gpt-5') or '4.1' in name: return 'responses' return 'chat' def _supports_vision(self, model_name: str) -> bool: """Return True if model supports image inputs via chat/responses content blocks.""" try: m = (model_name or '').lower() return ( m.startswith('gpt-5') or '4.1' in m or 'gpt-4o' in m or 'gpt-4-vision' in m or m == 'gpt-4' or 'omni' in m ) except Exception: return False def _post_vision(self, prompt_text: str, image_url: str, max_tokens: int = 500, temperature: float = 0.0, target_value: Optional[str] = None) -> str: """Call OpenAI using the official SDK (preferred). Falls back to HTTP if SDK missing.""" # Choose effective model that supports vision; no implicit fallback (fail fast) configured_model = self.vision_model model = configured_model if not self._supports_vision(model): msg = f"Configured model '{configured_model}' does not support vision/image inputs. Set OPENAI_VISION_MODEL to a vision model (e.g., 'gpt-4o' or 'gpt-5')." try: self.openai_logger.error(msg) except Exception: pass raise ValueError(msg) mode = self._vision_api_mode() # Use SDK if available if _OpenAIClient is not None: try: if self._openai_client is None: self._openai_client = _OpenAIClient() if mode == 'responses': # Build content blocks: prompt, optional target value, then image _content_blocks = [ {"type": "input_text", "text": prompt_text}, ] if target_value: _content_blocks.append({"type": "input_text", "text": f"TARGET_VALUE={target_value}"}) _content_blocks.append({"type": "input_image", "image_url": image_url}) _responses_kwargs = { "model": model, "input": [ { "role": "user", "content": _content_blocks, } ], # Prefer concise text output; allow enough reasoning for accuracy "reasoning": {"effort": "high"}, "text": {"verbosity": "low"}, } # Only include temperature for non-GPT-5 models on Responses API if not ((model or "").lower().startswith('gpt-5')): _responses_kwargs["temperature"] = temperature result = self._openai_client.responses.create(**_responses_kwargs) # Log the raw SDK response and a concise summary for debugging try: try: self.openai_logger.info(f"Responses SDK raw result: {result!r}") except Exception: pass try: out_list = getattr(result, "output", None) or [] output_types = [] for item in out_list[:10]: t = getattr(item, "type", None) output_types.append(str(t) if t is not None else type(item).__name__) preview_text = str(getattr(result, "output_text", "") or "") self.openai_logger.info( "Responses SDK summary: model=%s status=%s has_output_text=%s output_items=%s output_types=%s output_text_preview=%s" % ( str(getattr(result, "model", "")), str(getattr(result, "status", "")), str(bool(preview_text)), str(len(out_list)), str(output_types), preview_text[:300], ) ) except Exception: pass except Exception: pass # Extract text from Responses result try: out_text = getattr(result, "output_text", None) if out_text: return str(out_text) collected: list[str] = [] out = getattr(result, "output", None) or [] for item in out: # Prefer message.content list when available content = getattr(item, "content", None) if isinstance(content, list) and content: for seg in content: # Support SDK objects or dicts seg_text = getattr(seg, "text", None) if not seg_text and isinstance(seg, dict): seg_text = seg.get("text") if seg_text: collected.append(str(seg_text)) if collected: return "".join(collected) except Exception: pass # If no text content was found, try Chat Completions as fallback return str(result) else: # Chat Completions via SDK: use image_url content format chat_content = [ {"type": "text", "text": prompt_text}, ] if target_value: chat_content.append({"type": "text", "text": f"TARGET_VALUE={target_value}"}) chat_content.append({"type": "image_url", "image_url": {"url": image_url}}) # Newer models use 'max_completion_tokens' instead of 'max_tokens' chat_kwargs = { "model": self._resolve_chat_model(model), "messages": [ { "role": "user", "content": chat_content, } ], "max_completion_tokens": max_tokens, } result = self._openai_client.chat.completions.create(**chat_kwargs) try: # SDK objects: choices[0].message.content choices = getattr(result, "choices", None) if choices and len(choices) > 0: first = choices[0] message = getattr(first, "message", None) if message is not None: content = getattr(message, "content", None) if isinstance(content, str): return content # Some SDKs return list segments if isinstance(content, list): parts = [] for seg in content: txt = getattr(seg, "text", None) if hasattr(seg, "text") else (seg.get("text") if isinstance(seg, dict) else None) if txt: parts.append(str(txt)) if parts: return "".join(parts) except Exception: pass # Fallback: best-effort stringification return str(result) except Exception as e: try: self.openai_logger.error(f"OpenAI SDK call failed ({mode}): {e}") except Exception: pass # Fall through to HTTP as a last resort # Fallback: HTTP paths (previous implementation) headers = { 'Authorization': f'Bearer {OPENAI_API_KEY}', 'Content-Type': 'application/json' } try: if mode == 'responses': payload = { "model": model, "input": [ { "role": "user", "content": [ {"type": "input_text", "text": prompt_text}, # Responses API expects a string URL for image_url {"type": "input_image", "image_url": image_url}, ], } ], # Prefer concise text output; allow enough reasoning for accuracy "reasoning": {"effort": "medium"}, "text": {"verbosity": "low"}, } # Only include temperature for non-GPT-5 models on Responses API if not ((model or "").lower().startswith('gpt-5')): payload["temperature"] = temperature resp = requests.post("https://api.openai.com/v1/responses", headers=headers, json=payload, timeout=_REQUEST_TIMEOUT_SECONDS) resp.raise_for_status() data = resp.json() try: self.openai_logger.info(f"Responses HTTP JSON: {json.dumps(data)[:2000]}") except Exception: pass # Try extracting a consolidated text from HTTP JSON try: if isinstance(data, dict): if data.get('output_text'): return str(data['output_text']) out = data.get('output') or [] collected = [] if isinstance(out, list): for item in out: content = None if isinstance(item, dict): content = item.get('content') if isinstance(content, list): for seg in content: if isinstance(seg, dict) and seg.get('text'): collected.append(str(seg['text'])) if collected: return "".join(collected) except Exception: pass return json.dumps(data) else: payload = { "model": self._resolve_chat_model(model), "messages": [ { "role": "user", "content": [ {"type": "text", "text": prompt_text}, # Chat-compatible image content expects object form {"type": "image_url", "image_url": {"url": image_url}}, ], } ], "max_completion_tokens": max_tokens, } resp = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload, timeout=_REQUEST_TIMEOUT_SECONDS) resp.raise_for_status() data = resp.json() return (data.get('choices') or [{}])[0].get('message', {}).get('content', '') except Exception as e: try: if isinstance(e, requests.HTTPError) and e.response is not None: self.openai_logger.error(f"OpenAI call failed ({mode}) {e.response.status_code}: {e.response.text}") else: self.openai_logger.error(f"OpenAI call failed ({mode}): {e}") except Exception: pass raise def _analyze_screenshot_for_links(self, screenshot_path: str) -> List[Dict]: """ Analyze screenshot using OpenAI Vision API to find supporting document links. Args: screenshot_path: Path to the screenshot Returns: List of clickable links with coordinates """ if not OPENAI_API_KEY: error_msg = "OpenAI API key not set. Please set OPENAI_API_KEY environment variable." self.openai_logger.error(error_msg) raise Exception(error_msg) # Validate API key format if not OPENAI_API_KEY.startswith(('sk-', 'sk-proj-')): error_msg = "Invalid OpenAI API key format. API key should start with 'sk-' or 'sk-proj-'" self.openai_logger.error(error_msg) raise Exception(error_msg) try: self.openai_logger.info("Preparing screenshot for OpenAI analysis") # Prepare the image for OpenAI with open(screenshot_path, 'rb') as f: img_bytes = f.read() img_b64 = base64.b64encode(img_bytes).decode('utf-8') image_url = f"data:image/png;base64,{img_b64}" # For color sampling later try: from PIL import Image as _Img _img_for_color = _Img.open(screenshot_path).convert('RGB') except Exception: _img_for_color = None # Build links analysis prompt via shared prompt_builder sw, sh = self._get_screen_size() from .prompt_builder import get_links_prompt prompt = get_links_prompt(sw, sh) self.openai_logger.info("Sending screenshot to OpenAI GPT-4 Vision for analysis") try: # Also log the prompt to openai.log for troubleshooting self.openai_logger.info("Prompt (links analysis):\n" + prompt) except Exception: pass # Debug: save request preview (no base64) try: dbg_dir = self._get_debug_dir() / f"links_{int(time.time()*1000)}" dbg_dir.mkdir(parents=True, exist_ok=True) self._save_text(dbg_dir / 'request_preview.txt', f"model={self.vision_model}\nimage_path={screenshot_path}\nprompt=\n{prompt}") except Exception: pass # Route vision call through the unified helper used by the working search flow try: from .constants import OPENAI_VISION_TEMPERATURE except Exception: OPENAI_VISION_TEMPERATURE = 0.0 try: content = self._post_vision( prompt, image_url, max_tokens=2000, temperature=float(OPENAI_VISION_TEMPERATURE), ) except Exception as _e: self.openai_logger.error(f"OpenAI API error: {_e}") return [] self.openai_logger.info("Received response from OpenAI vision endpoint") # Debug: save raw response text try: self._save_text(dbg_dir / 'response_raw.txt', str(content)) except Exception: pass # Extract JSON from response try: # Find JSON array in the response start_idx = content.find('[') end_idx = content.rfind(']') + 1 if start_idx != -1 and end_idx != -1: json_str = content[start_idx:end_idx] clickable_links = json.loads(json_str) self.openai_logger.info(f"OpenAI identified {len(clickable_links)} potential links") # Validate coordinates and ensure link_words and button fields exist validated_links = [] for link in clickable_links: try: if isinstance(link, dict) and 'link_words' not in link: desc = str(link.get('description') or '').strip() link['link_words'] = desc except Exception: try: link['link_words'] = '' except Exception: pass # Ensure button field exists, default to false if not provided if 'button' not in link: link['button'] = False # Server-side filtering: exclude buttons, icon-only, and known controls try: lw = str(link.get('link_words') or '').strip() except Exception: lw = '' desc_lower = str(link.get('description') or '').lower() lw_lower = lw.lower() is_button_like = bool(link.get('button')) looks_like_control = ( 'attach' in lw_lower or 'attach' in desc_lower or 'upload' in lw_lower or 'upload' in desc_lower or 'print' in lw_lower or 'print' in desc_lower or lw_lower.startswith('files (') or lw_lower == 'files' or 'files (' in lw_lower ) # Explicit disallow list if 'view details' in lw_lower or 'view details' in desc_lower: continue # Explicitly exclude broad history pages if 'activity history' in lw_lower or 'activity history' in desc_lower: continue # Heuristic: ignore links described as coming from a "From" section (company/contact names) if ('from' in desc_lower and ('section' in desc_lower or 'column' in desc_lower)): continue # Whitelist: keep only document/file related phrases or filenames import re as _re # Navigation-doc-like: invoice/payment/receipt/bill pages (NOT downloads) has_doc_words = ( 'invoice' in lw_lower or 'receipt' in lw_lower or 'statement' in lw_lower or 'bill' in lw_lower or 'credit note' in lw_lower or 'view invoice' in lw_lower or 'view receipt' in lw_lower or 'statement' in lw_lower or 'credit note' in lw_lower ) has_payment = ('payment' in lw_lower) looks_like_invoice_id = bool(_re.search(r"\b(inv[-\s#:]?\s*\d{3,}|invoice\s*#?\s*\d{3,})\b", lw_lower)) looks_like_date = bool(_re.search(r"\b(\d{4}-\d{2}-\d{2}|\d{1,2}\s+[A-Za-z]{3,9}\s+\d{4})\b", lw_lower)) looks_like_filename = bool(_re.search(r"\.(pdf|png|jpe?g|tiff?|docx?|xlsx?)\b", lw_lower)) # Final allow: doc words OR invoice id OR (payment AND date). Explicitly disallow filenames doc_like = (has_doc_words or looks_like_invoice_id or (has_payment and looks_like_date)) and not looks_like_filename # Heuristic: drop plain uppercase invoice IDs with no link context words # e.g., keep only if has_doc_words or (payment & date) or invoice id alongside a link cue like 'view' if looks_like_invoice_id and not (('view' in lw_lower) or has_doc_words or (has_payment and looks_like_date)): continue # Reject icon-only (no text) and any button-like/control items if is_button_like or looks_like_control or lw == '' or not doc_like: continue # Validate coordinate presence if self._validate_coordinates(link.get('coordinates', [])): # Color heuristic: reject near-black text (likely non-link body text) try: if _img_for_color is not None: cx, cy = link.get('coordinates', [None, None]) if isinstance(cx, (int, float)) and isinstance(cy, (int, float)): cx = int(cx); cy = int(cy) w, h = _img_for_color.size if 0 <= cx < w and 0 <= cy < h: # Sample a small 3x3 around the point left = max(0, cx-1); top = max(0, cy-1); right = min(w, cx+2); bottom = min(h, cy+2) region = _img_for_color.crop((left, top, right, bottom)) pixels = list(region.getdata()) # Compute average brightness avg = tuple(sum(c[i] for c in pixels)//len(pixels) for i in range(3)) # Near-black if all channels < 50 (tunable) if avg[0] < 50 and avg[1] < 50 and avg[2] < 50: # Skip this candidate continue except Exception: pass validated_links.append(link) else: self.openai_logger.warning(f"Invalid coordinates for link: {link.get('description', 'Unknown')}") self.openai_logger.info(f"Validated {len(validated_links)} links with valid coordinates") # Debug: save parsed JSON and overlay try: self._save_json(dbg_dir / 'parsed_links.json', validated_links) marks = [] for l in validated_links: try: coords = l.get('coordinates', []) if isinstance(coords, list) and len(coords) == 2: marks.append((int(coords[0]), int(coords[1]))) except Exception: pass if marks: self._save_overlay(screenshot_path, marks, dbg_dir / 'overlay.png', label='supporting document links') except Exception: pass return validated_links else: self.openai_logger.error("No valid JSON array found in OpenAI response") return [] except json.JSONDecodeError as e: self.openai_logger.error(f"Failed to parse OpenAI response as JSON: {e}") self.openai_logger.error(f"Raw response: {content}") return [] except Exception as e: self.openai_logger.error(f"OpenAI API error: {e}") return [] def _analyze_screenshot_for_sd_icons(self, screenshot_path: str) -> List[Dict]: """ Analyze screenshot using OpenAI Vision API to find SD (Supporting Document) icons. Args: screenshot_path: Path to the screenshot Returns: List of clickable SD icons with coordinates """ if not OPENAI_API_KEY: error_msg = "OpenAI API key not set. Please set OPENAI_API_KEY environment variable." self.openai_logger.error(error_msg) raise Exception(error_msg) # Validate API key format if not OPENAI_API_KEY.startswith(('sk-', 'sk-proj-')): error_msg = "Invalid OpenAI API key format. API key should start with 'sk-' or 'sk-proj-'" self.openai_logger.error(error_msg) raise Exception(error_msg) try: self.openai_logger.info("Preparing screenshot for OpenAI SD icon analysis") # Prepare the image for OpenAI with open(screenshot_path, 'rb') as f: img_bytes = f.read() img_b64 = base64.b64encode(img_bytes).decode('utf-8') image_url = f"data:image/png;base64,{img_b64}" # Build SD icons analysis prompt via shared prompt_builder sw, sh = self._get_screen_size() from .prompt_builder import get_sd_icons_prompt prompt = get_sd_icons_prompt(sw, sh) self.openai_logger.info("Sending screenshot to OpenAI GPT-4 Vision for SD icon analysis") try: # Also log the prompt to openai.log for troubleshooting self.openai_logger.info("Prompt (SD icons analysis):\n" + prompt) except Exception: pass # Debug: save request preview (no base64) try: dbg_dir = self._get_debug_dir() / f"sd_icons_{int(time.time()*1000)}" dbg_dir.mkdir(parents=True, exist_ok=True) self._save_text(dbg_dir / 'request_preview.txt', f"model={self.vision_model}\nimage_path={screenshot_path}\nprompt=\n{prompt}") except Exception: pass # Route vision call through the unified helper used by the working search flow try: from .constants import OPENAI_VISION_TEMPERATURE except Exception: OPENAI_VISION_TEMPERATURE = 0.0 try: content = self._post_vision( prompt, image_url, max_tokens=2000, temperature=float(OPENAI_VISION_TEMPERATURE), ) except Exception as _e: self.openai_logger.error(f"OpenAI API error: {_e}") return [] self.openai_logger.info("Received response from OpenAI vision endpoint") # Debug: save raw response text try: self._save_text(dbg_dir / 'response_raw.txt', str(content)) except Exception: pass # Extract JSON from response try: # Find JSON array in the response start_idx = content.find('[') end_idx = content.rfind(']') + 1 if start_idx != -1 and end_idx != -1: json_str = content[start_idx:end_idx] clickable_icons = json.loads(json_str) self.openai_logger.info(f"OpenAI identified {len(clickable_icons)} potential SD icons") # Validate coordinates and ensure required fields exist validated_icons = [] for icon in clickable_icons: try: if isinstance(icon, dict) and 'link_words' not in icon: desc = str(icon.get('description') or '').strip() icon['link_words'] = desc except Exception: try: icon['link_words'] = '' except Exception: pass # Ensure button field exists, default to true for SD icons if 'button' not in icon: icon['button'] = True # Ensure type field exists if 'type' not in icon: icon['type'] = 'sd_icon' if self._validate_coordinates(icon.get('coordinates', [])): validated_icons.append(icon) else: self.openai_logger.warning(f"Invalid coordinates for SD icon: {icon.get('description', 'Unknown')}") self.openai_logger.info(f"Validated {len(validated_icons)} SD icons with valid coordinates") # Debug: save parsed JSON and overlay try: self._save_json(dbg_dir / 'parsed_sd_icons.json', validated_icons) marks = [] for l in validated_icons: try: coords = l.get('coordinates', []) if isinstance(coords, list) and len(coords) == 2: marks.append((int(coords[0]), int(coords[1]))) except Exception: pass if marks: self._save_overlay(screenshot_path, marks, dbg_dir / 'overlay.png', label='SD icons') except Exception: pass return validated_icons else: self.openai_logger.error("No valid JSON array found in OpenAI response") return [] except json.JSONDecodeError as e: self.openai_logger.error(f"Failed to parse OpenAI response as JSON: {e}") self.openai_logger.error(f"Raw response: {content}") return [] except Exception as e: self.openai_logger.error(f"OpenAI API error: {e}") return [] def _transform_coordinates(self, coordinates: List[int]) -> List[int]: """ Transform coordinates to fix any coordinate system mismatches. Currently fixes inverted X coordinates. Args: coordinates: [x, y] coordinates from OpenAI Returns: Transformed [x, y] coordinates for PyAutoGUI """ if not coordinates or len(coordinates) != 2: return coordinates x, y = coordinates # Fix inverted X coordinates: flip X around the center # If X is inverted, transform: x = SCREEN_WIDTH - x from .constants import FIX_INVERTED_X_COORDINATES if FIX_INVERTED_X_COORDINATES: sw, _ = self._get_screen_size() transformed_x = sw - x self.openai_logger.info(f"Coordinate transformation: ({x}, {y}) -> ({transformed_x}, {y})") return [transformed_x, y] else: self.openai_logger.info(f"Coordinate transformation disabled, using original coordinates: ({x}, {y})") return [x, y] def _validate_coordinates(self, coordinates: List[int]) -> bool: """Validate that coordinates are within reasonable screen bounds.""" if not coordinates or len(coordinates) != 2: return False x, y = coordinates # Validate coordinates are within screen bounds sw, sh = self._get_screen_size() if 0 <= x <= sw and 0 <= y <= sh: return True else: self.openai_logger.warning(f"Coordinates ({x}, {y}) out of screen bounds ({sw}x{sh})") return False def _generate_amount_variants(self, amount_text: str) -> set: """Generate robust amount display variants including decimals, thousands separators and currency symbols. Examples for 1650 -> {"1650", "1,650", "1650.00", "1,650.00", "$1650", "$1,650", "$1650.00", "$1,650.00", "A$1650", "A$1,650", ...} """ variants: set = set() try: t = str(amount_text).strip() # Strip currency for numeric normalization digits = ''.join(ch for ch in t if ch.isdigit()) if not digits: variants.add(t) return variants # Base integer and with .00 int_form = str(int(digits)) with_dec = f"{int_form}.00" # With thousands separators def add_commas(s: str) -> str: try: return f"{int(s):,}" except Exception: return s int_commas = add_commas(int_form) dec_commas = add_commas(int_form) + ".00" # Currency symbol variants (no space and narrow space) cur_syms = ["$", "A$"] space_opts = ["", "\u2009", " "] base_set = {int_form, with_dec, int_commas, dec_commas} variants.update(base_set) for sym in cur_syms: for sp in space_opts: for b in base_set: variants.add(f"{sym}{sp}{b}") except Exception: variants.add(str(amount_text)) return variants def find_text_coordinates(self, screenshot_path: str, query_text: str, row_hint: Optional[str] = None) -> Optional[List[int]]: """Find coordinates for a UI element matching query_text using OpenAI Vision. Returns [x, y] in screen pixels or None. """ try: if not OPENAI_API_KEY: self.openai_logger.error("OPENAI_API_KEY not set") return None sw, sh = self._get_screen_size() # Prepare input image path; if grid-clicking is enabled, render grid and send THAT image to the model from .constants import USE_GRID_CLICKING, GRID_COLS, GRID_ROWS # Create a debug dir up front so we can store the grid image as the real input we send try: _prefix_for_dir = 'search_ctx' if (row_hint or '') else 'search' dbg_dir_early = self._get_debug_dir() / f"{_prefix_for_dir}_{int(time.time()*1000)}" dbg_dir_early.mkdir(parents=True, exist_ok=True) except Exception: dbg_dir_early = None input_image_path = screenshot_path try: if USE_GRID_CLICKING: if dbg_dir_early is not None: grid_in_path = dbg_dir_early / 'grid_input.png' else: grid_in_path = Path(screenshot_path).with_suffix('.grid.png') # Choose renderer from .constants import GRID_RENDERER if GRID_RENDERER == 'alpha_numeric': try: # Import user-provided renderer from .add_grid import add_alpha_numeric_grid_to_image as _alpha_grid from .constants import ( GRID_ALPHA_CELL_SIZE, GRID_TEXT_OPACITY, GRID_TEXT_OFFSET_X, GRID_ALPHA_FONT_SIZE, GRID_LINE_COLOR_RGBA, GRID_LABEL_BG_COLOR_RGBA, GRID_ALPHA_TILE_ALPHA, ) # Map colors to PIL-friendly forms grid_color = (GRID_LINE_COLOR_RGBA[0], GRID_LINE_COLOR_RGBA[1], GRID_LINE_COLOR_RGBA[2]) label_bg = GRID_LABEL_BG_COLOR_RGBA _alpha_grid( base_image_path=screenshot_path, output_path=str(grid_in_path), cell_size=int(GRID_ALPHA_CELL_SIZE), grid_color=grid_color, text_opacity=int(GRID_TEXT_OPACITY), text_offset_x=int(GRID_TEXT_OFFSET_X), label_background_color=label_bg, font_size=int(GRID_ALPHA_FONT_SIZE), tile_alpha=int(GRID_ALPHA_TILE_ALPHA), ) except Exception: # Fallback to internal overlay if alpha_numeric fails self._save_grid_overlay(screenshot_path, grid_in_path, GRID_COLS, GRID_ROWS) else: self._save_grid_overlay(screenshot_path, grid_in_path, GRID_COLS, GRID_ROWS) input_image_path = str(grid_in_path) except Exception: input_image_path = screenshot_path # Encode the actual input image we will send (grid or original) with open(input_image_path, 'rb') as f: img_b64 = base64.b64encode(f.read()).decode('utf-8') image_url = f"data:image/png;base64,{img_b64}" # Capture actual captured image size to infer DPI scaling try: import pyautogui as _pg img = _pg.screenshot(region=(0, 0, sw, sh)) iw, ih = img.size self._dpi_scale = (iw / sw, ih / sh) except Exception: self._dpi_scale = (1.0, 1.0) # Unified prompt via shared builder qt = str(query_text).strip() hint = (row_hint or '').strip() or None prompt, aux_texts = build_find_value_prompt(qt, sw, sh, row_hint=hint) payload = { "model": self.vision_model, "messages": [ { "role": "user", "content": [ {"type": "text", "text": prompt}, # Include auxiliary fields (TARGET_VALUE) *[{"type": "text", "text": t} for t in aux_texts], # Provide viewport/DPR metadata for better normalization {"type": "text", "text": f"VIEWPORT_CSS_WIDTH={sw}"}, {"type": "text", "text": f"VIEWPORT_CSS_HEIGHT={sh}"}, {"type": "text", "text": f"DEVICE_PIXEL_RATIO={getattr(self, '_dpi_scale', (1.0,1.0))[0]:.3f}"}, {"type": "text", "text": "SCROLL_X=0"}, {"type": "text", "text": "SCROLL_Y=0"}, {"type": "image_url", "image_url": {"url": image_url}}, ], } ], "max_tokens": 400, } headers = { "Authorization": f"Bearer {OPENAI_API_KEY}", "Content-Type": "application/json", } try: self.openai_logger.info(f"Search query: {query_text}") self.openai_logger.info("Prompt (search coordinates):\n" + prompt) except Exception: pass # Debug: save request preview (no base64) try: # Reuse the early directory if we created one; otherwise create now if dbg_dir_early is not None: dbg_dir = dbg_dir_early else: _prefix = 'search_ctx' if hint else 'search' dbg_dir = self._get_debug_dir() / f"{_prefix}_{int(time.time()*1000)}" dbg_dir.mkdir(parents=True, exist_ok=True) extra = f"hint={hint}\n" if hint else '' self._save_text(dbg_dir / 'request_preview.txt', f"model={self.vision_model}\nimage_path={screenshot_path}\ninput_image={input_image_path}\nquery={query_text}\n{extra}prompt=\n{prompt}") # If grid-clicking is enabled, also save a grid overlay for visual debugging try: if USE_GRID_CLICKING: self._save_grid_overlay(screenshot_path, dbg_dir / 'grid_overlay.png', GRID_COLS, GRID_ROWS) except Exception: pass except Exception: dbg_dir = None # Call OpenAI via unified helper try: from .constants import OPENAI_VISION_TEMPERATURE content = self._post_vision(prompt, image_url, max_tokens=500, temperature=float(OPENAI_VISION_TEMPERATURE), target_value=qt) except Exception as _e: self.openai_logger.error(f"Vision call failed: {_e}") return None try: if dbg_dir: self._save_text(dbg_dir / 'response_raw.txt', str(content)) except Exception: pass # Parse JSON object start = content.find('{') end = content.rfind('}') + 1 if start == -1 or end <= start: self.openai_logger.warning("No JSON object found in OpenAI response for coordinates") try: if dbg_dir: self._save_text(dbg_dir / 'parse_error.txt', 'No JSON object in response') except Exception: pass return None try: obj = json.loads(content[start:end]) except Exception: self.openai_logger.warning("Failed to parse coordinates JSON from response") try: if dbg_dir: self._save_text(dbg_dir / 'parse_error.txt', 'JSON parse failed for coordinates object') except Exception: pass return None # Prefer grid cell when grid-clicking enabled from .constants import USE_GRID_CLICKING, GRID_COLS, GRID_ROWS grid_id = obj.get('grid_id') if isinstance(obj, dict) else None # Numeric grid support: if model returned grid_row/grid_column, use them deterministically numeric_grid_used = False try: if USE_GRID_CLICKING and isinstance(obj, dict): grid_row = obj.get('grid_row') grid_col = obj.get('grid_column') if isinstance(grid_row, int) and isinstance(grid_col, int) and 1 <= grid_row <= GRID_ROWS and 1 <= grid_col <= GRID_COLS: sw_map, sh_map = self._get_screen_size() # Align mapping with overlay math (header bands/gutters) from .constants import GRID_LABEL_FONT_SIZE, GRID_COL_LABEL_POSITIONS, GRID_USE_HEADER_BANDS header_top = int(GRID_LABEL_FONT_SIZE * 1.6) if GRID_USE_HEADER_BANDS else 0 footer_bottom = int(GRID_LABEL_FONT_SIZE * 1.6) if (GRID_USE_HEADER_BANDS and GRID_COL_LABEL_POSITIONS in ('bottom','both')) else 0 left_gutter = int(GRID_LABEL_FONT_SIZE * 1.8) if GRID_USE_HEADER_BANDS else 0 grid_w = max(1, sw_map - left_gutter) grid_h = max(1, sh_map - header_top - footer_bottom) cell_w = grid_w / GRID_COLS cell_h = grid_h / GRID_ROWS cx = int(left_gutter + (int(grid_col) - 0.5) * cell_w) cy = int(header_top + (int(grid_row) - 0.5) * cell_h) coords = [cx, cy] # Back-fill grid_id if missing for logging/debugging if not grid_id: try: col_label = column_index_to_label(int(grid_col)) if col_label: grid_id = f"{col_label}{int(grid_row)}" except Exception: pass numeric_grid_used = True # Verify against substring box even when numeric grid is present try: from .constants import GRID_VERIFY_FROM_BOX, GRID_ROW_SOURCE # Preserve numeric grid_id in alpha-numeric mode if the model provided a numeric tile id try: from .constants import GRID_RENDERER _orig_gid = obj.get('grid_id') _preserve_numeric_gid = bool(_orig_gid) and str(_orig_gid).isdigit() and GRID_RENDERER == 'alpha_numeric' except Exception: _preserve_numeric_gid = False if GRID_VERIFY_FROM_BOX and isinstance(obj.get('box'), (list, tuple)) and len(obj.get('box')) == 4: x1, y1, x2, _ = [int(v) for v in obj['box']] cx_box = (x1 + x2) / 2.0 recomputed_col = max(1, min(GRID_COLS, int((cx_box - left_gutter) // cell_w) + 1)) if int(grid_col) != recomputed_col: grid_col = recomputed_col cx = int(left_gutter + (recomputed_col - 0.5) * cell_w) coords = [cx, cy] if GRID_ROW_SOURCE == 'box_top': recomputed_row = max(1, min(GRID_ROWS, int((y1 - header_top) // cell_h) + 1)) if int(grid_row) != recomputed_row: grid_row = recomputed_row cy = int(header_top + (recomputed_row - 0.5) * cell_h) coords = [cx, cy] # Update grid_id accordingly if not _preserve_numeric_gid: try: col_label = column_index_to_label(int(grid_col)) if col_label: grid_id = f"{col_label}{int(grid_row)}" except Exception: pass except Exception: pass except Exception: pass # Prefer grid-driven click; derive column from substring box center when available from .constants import USE_GRID_CLICKING, GRID_COLS, GRID_ROWS coords = obj.get('coordinates') if not numeric_grid_used else coords # When using alpha-numeric tiles and the model returns a numeric grid_id (e.g., "169"), # do NOT run the legacy header/row/col mapping below. We'll map the tile id later # via _grid_to_center, preserving the numeric id in logs. try: from .constants import GRID_RENDERER s_grid = str(grid_id) if grid_id is not None else "" alpha_tile_id_numeric = (GRID_RENDERER == 'alpha_numeric' and s_grid.isdigit()) except Exception: alpha_tile_id_numeric = False if not numeric_grid_used and not alpha_tile_id_numeric and USE_GRID_CLICKING and (grid_id or obj.get('box') or obj.get('coordinates')): try: sw_map, sh_map = self._get_screen_size() from .constants import GRID_LABEL_FONT_SIZE, GRID_COL_LABEL_POSITIONS, GRID_USE_HEADER_BANDS header_top = int(GRID_LABEL_FONT_SIZE * 1.6) if GRID_USE_HEADER_BANDS else 0 footer_bottom = int(GRID_LABEL_FONT_SIZE * 1.6) if (GRID_USE_HEADER_BANDS and GRID_COL_LABEL_POSITIONS in ('bottom','both')) else 0 left_gutter = int(GRID_LABEL_FONT_SIZE * 1.8) if GRID_USE_HEADER_BANDS else 0 grid_w = max(1, sw_map - left_gutter) grid_h = max(1, sh_map - header_top - footer_bottom) cell_w = grid_w / GRID_COLS cell_h = grid_h / GRID_ROWS # If only normalized coords exist, convert to pixels early try: norm = obj.get('normalized') if isinstance(obj, dict) else None if isinstance(norm, dict) and 'x' in norm and 'y' in norm and coords is None: nx = float(norm['x']); ny = float(norm['y']) # Map normalized coords to inner grid area if header bands are used px = int(left_gutter + nx * grid_w) py = int(header_top + ny * grid_h) coords = [px, py] except Exception: pass # Choose row according to configuration from .constants import GRID_ROW_SOURCE, GRID_COL_SOURCE row_idx: Optional[int] = None if GRID_ROW_SOURCE == 'box_top' and isinstance(obj.get('box'), (list, tuple)) and len(obj.get('box')) == 4: try: y1 = int(obj['box'][1]) row_idx = max(1, min(GRID_ROWS, int((y1 - header_top) // cell_h) + 1)) except Exception: row_idx = None if row_idx is None and grid_id: try: s = str(grid_id).strip().upper() i = 0 while i < len(s) and s[i].isalpha(): i += 1 row_idx = int(s[i:]) except Exception: row_idx = None # Fallback to raw coords row if row_idx is None and isinstance(coords, list) and len(coords) == 2: try: row_idx = max(1, min(GRID_ROWS, int((int(coords[1]) - header_top) // cell_h) + 1)) except Exception: row_idx = None # Determine column according to strategy from .constants import GRID_CLICK_COLUMN, GRID_CLICK_COLUMN_LABEL, GRID_VERIFY_FROM_BOX col_idx: Optional[int] = None # First, derive from box center if configured if GRID_COL_SOURCE == 'box_center' and isinstance(obj.get('box'), (list, tuple)) and len(obj.get('box')) == 4: try: x1, _, x2, _ = [int(v) for v in obj['box']] cx = (x1 + x2) / 2.0 col_idx = max(1, min(GRID_COLS, int((cx - left_gutter) // cell_w) + 1)) except Exception: col_idx = None # If model supplied grid_column and verify flag is set, recompute from box and prefer box if mismatch try: if GRID_VERIFY_FROM_BOX and isinstance(obj.get('grid_column'), int) and isinstance(obj.get('box'), (list, tuple)) and len(obj.get('box')) == 4: x1, _, x2, _ = [int(v) for v in obj['box']] cx = (x1 + x2) / 2.0 recomputed = max(1, min(GRID_COLS, int((cx - left_gutter) // cell_w) + 1)) provided = int(obj['grid_column']) if recomputed != provided: col_idx = recomputed # Also fix grid_id if needed try: col_label = column_index_to_label(recomputed) if col_label and row_idx: grid_id = f"{col_label}{row_idx}" except Exception: pass except Exception: pass # Optional override strategies if col_idx is None and GRID_CLICK_COLUMN == 'center': col_idx = max(1, min(GRID_COLS, (GRID_COLS + 1) // 2)) elif col_idx is None and GRID_CLICK_COLUMN == 'label': forced = column_label_to_index(GRID_CLICK_COLUMN_LABEL) if 1 <= forced <= GRID_COLS: col_idx = forced # else use model-provided grid column if present if col_idx is None and grid_id: try: s = str(grid_id).strip().upper() i = 0 while i < len(s) and s[i].isalpha(): i += 1 col_label = s[:i] col_idx = column_label_to_index(col_label) except Exception: col_idx = None # Final fallback to coords-based column if col_idx is None and isinstance(coords, list) and len(coords) == 2: try: col_idx = max(1, min(GRID_COLS, int(int(coords[0]) // cell_w) + 1)) except Exception: col_idx = None if row_idx and col_idx: cx = int(left_gutter + (col_idx - 0.5) * cell_w) cy = int(header_top + (row_idx - 0.5) * cell_h) coords = [cx, cy] # Ensure grid_id reflects final row/column try: col_label = column_index_to_label(int(col_idx)) if col_label: grid_id = f"{col_label}{int(row_idx)}" except Exception: pass # else fall through to legacy handling below except Exception: pass if not isinstance(coords, list) or len(coords) != 2: try: if dbg_dir: self._save_text(dbg_dir / 'parse_error.txt', f"Unexpected coords payload (no coords and no mappable grid_id): {obj}") except Exception: pass return None # Extract optional extras matched_box = obj.get('box') if isinstance(obj, dict) else None matched_text = obj.get('matched_text') if isinstance(obj, dict) else None reason = obj.get('reason') if isinstance(obj, dict) else None # Optional: compute normalized forms for logging only for amount-like values try: if self._looks_like_amount(query_text) or self._looks_like_amount(matched_text): norm_query = self._normalize_amount(query_text) norm_matched = self._normalize_amount(matched_text) if dbg_dir and (norm_query and norm_matched) and norm_query != norm_matched: self._save_text(dbg_dir / 'parse_error.txt', 'Note: normalized amount values differ; proceeding with model coordinates') else: norm_query = None norm_matched = None except Exception: norm_query = None norm_matched = None # Transform and validate; use grid mapping if provided raw_coords = coords[:] if isinstance(coords, list) else None # If a grid id is provided, map to the center of that cell try: if USE_GRID_CLICKING and grid_id: sw, sh = self._get_screen_size() def _grid_to_center(cell_id: str, cell_position: Optional[str] = None) -> Optional[list[int]]: try: if not cell_id: return None s = str(cell_id).strip().upper() # Support two formats: # 1) Legacy AA27 style (lettered column + numeric row) # 2) Alpha-numeric tile id as pure digits like '168' (when GRID_RENDERER == 'alpha_numeric') from .constants import GRID_RENDERER if s.isdigit() and GRID_RENDERER == 'alpha_numeric': try: from .constants import GRID_ALPHA_CELL_SIZE, GRID_CELL_OFFSET_FRACTION, GRID_USE_CELL_POSITION cell_px = int(GRID_ALPHA_CELL_SIZE) if cell_px <= 0: return None tiles_per_row = max(1, int(sw // cell_px)) idx = int(s) x_index = idx % tiles_per_row y_index = idx // tiles_per_row # Base center cx = x_index * cell_px + cell_px / 2.0 cy = y_index * cell_px + cell_px / 2.0 # Optional offset within cell try: if GRID_USE_CELL_POSITION: pos = (cell_position or '').strip().lower() frac = max(0.0, min(0.49, float(GRID_CELL_OFFSET_FRACTION))) dx = dy = 0.0 if pos == 'left': dx = - (cell_px * frac) elif pos == 'right': dx = (cell_px * frac) elif pos == 'top': dy = - (cell_px * frac) elif pos == 'bottom': dy = (cell_px * frac) cx += dx; cy += dy except Exception: pass return [int(cx), int(cy)] except Exception: return None # Legacy AA27 path i = 0 while i < len(s) and s[i].isalpha(): i += 1 col_label = s[:i] row_str = s[i:] if not col_label or not row_str: return None row_idx = int(row_str) col_idx = column_label_to_index(col_label) if not (1 <= col_idx <= GRID_COLS): return None if not (1 <= row_idx <= GRID_ROWS): return None from .constants import GRID_DISABLE_COORDS, GRID_CLICK_COLUMN, GRID_CLICK_COLUMN_LABEL, GRID_CELL_OFFSET_FRACTION, GRID_USE_CELL_POSITION # Allow column override strategy if GRID_CLICK_COLUMN == 'center': col_idx = max(1, min(GRID_COLS, (GRID_COLS + 1) // 2)) elif GRID_CLICK_COLUMN == 'label': forced = column_label_to_index(GRID_CLICK_COLUMN_LABEL) if 1 <= forced <= GRID_COLS: col_idx = forced # else 'grid': use model-provided column cell_w = sw / GRID_COLS cell_h = sh / GRID_ROWS cx = (col_idx - 0.5) * cell_w cy = (row_idx - 0.5) * cell_h # Optional offset within cell try: if GRID_USE_CELL_POSITION: pos = (cell_position or '').strip().lower() frac = max(0.0, min(0.49, float(GRID_CELL_OFFSET_FRACTION))) dx = dy = 0.0 if pos == 'left': dx = - (cell_w * frac) elif pos == 'right': dx = (cell_w * frac) elif pos == 'top': dy = - (cell_h * frac) elif pos == 'bottom': dy = (cell_h * frac) cx += dx; cy += dy except Exception: pass return [int(cx), int(cy)] except Exception: return None # Extract optional cell_position from response try: cell_position = obj.get('cell_position') if isinstance(obj, dict) else None except Exception: cell_position = None mapped = _grid_to_center(grid_id, cell_position) if mapped: coords = mapped else: coords = self._transform_coordinates(coords) elif matched_box and isinstance(matched_box, (list, tuple)) and len(matched_box) == 4: # Click the bottom-left corner of the matched box x1, y1, x2, y2 = [int(v) for v in matched_box] coords = [int(x1), int(y2)] else: coords = self._transform_coordinates(coords) except Exception: coords = self._transform_coordinates(coords) if not self._validate_coordinates(coords): try: if dbg_dir: self._save_text(dbg_dir / 'parse_error.txt', f"Coords out of bounds after transform: {coords}") except Exception: pass return None # Debug: save parsed coords and overlay try: if dbg_dir: if self._looks_like_amount(query_text) or self._looks_like_amount(matched_text): norm_query = self._normalize_amount(query_text) norm_matched = self._normalize_amount(matched_text) else: norm_query = None norm_matched = None # Determine click mode and optional tile id try: from .constants import GRID_RENDERER, GRID_USE_CELL_POSITION s_id = str(grid_id) if grid_id is not None else None tile_id = int(s_id) if (s_id and s_id.isdigit() and GRID_RENDERER == 'alpha_numeric') else None grid_click_mode = 'alpha_tile_center' if tile_id is not None else 'grid_mapping_or_box' # Preserve numeric grid_id when alpha-numeric: keep it as-is in logs if tile_id is not None: grid_id = s_id except Exception: tile_id = None grid_click_mode = 'grid_mapping_or_box' dbg_obj = { 'query': query_text, 'query_normalized': norm_query, 'coords_transformed': coords, # Optionally hide raw coords when grid-clicking is the source of truth 'coords_raw': raw_coords if not (USE_GRID_CLICKING and grid_id) else None, 'box': matched_box, 'matched_text': matched_text, 'matched_normalized': norm_matched, 'reason': reason, 'normalized_match_equal': (norm_query == norm_matched) if (norm_query and norm_matched) else None, 'dpi_scale': getattr(self, '_dpi_scale', (1.0, 1.0)), 'grid_id': grid_id, 'tile_id': tile_id, 'grid_click_mode': grid_click_mode, 'cell_position': obj.get('cell_position') if isinstance(obj, dict) else None, 'cell_position_applied': (obj.get('cell_position') is not None and GRID_USE_CELL_POSITION) if isinstance(obj, dict) else False, 'grid_row': None if tile_id is not None else (obj.get('grid_row') if isinstance(obj, dict) else None), 'grid_column': None if tile_id is not None else (obj.get('grid_column') if isinstance(obj, dict) else None), 'grid_cols': GRID_COLS, 'grid_rows': GRID_ROWS, } try: # Only compute/attach legacy header-based diagnostics when not using alpha tile ids if tile_id is None: # If we overrode the column due to verification, include a note if isinstance(obj, dict) and isinstance(obj.get('box'), (list, tuple)) and obj.get('grid_column'): sw_map, sh_map = self._get_screen_size() cell_w = sw_map / GRID_COLS x1, _, x2, _ = [int(v) for v in (matched_box or obj.get('box'))] recomputed_col = max(1, min(GRID_COLS, int(((x1 + x2) / 2.0) // cell_w) + 1)) dbg_obj['grid_column_recomputed_from_box'] = recomputed_col dbg_obj['grid_column_corrected'] = (dbg_obj.get('grid_column') != recomputed_col) # Include final indices used for click try: # Derive from final coords final_cx, final_cy = int(coords[0]), int(coords[1]) # Recompute indices with header/gutter from .constants import GRID_LABEL_FONT_SIZE, GRID_COL_LABEL_POSITIONS, GRID_USE_HEADER_BANDS header_top = int(GRID_LABEL_FONT_SIZE * 1.6) if GRID_USE_HEADER_BANDS else 0 left_gutter = int(GRID_LABEL_FONT_SIZE * 1.8) if GRID_USE_HEADER_BANDS else 0 grid_w = max(1, sw_map - left_gutter) grid_h = max(1, sh_map - header_top - (int(GRID_LABEL_FONT_SIZE * 1.6) if (GRID_USE_HEADER_BANDS and GRID_COL_LABEL_POSITIONS in ('bottom','both')) else 0)) cell_w2 = grid_w / GRID_COLS cell_h2 = grid_h / GRID_ROWS col_final = max(1, min(GRID_COLS, int((final_cx - left_gutter) // cell_w2) + 1)) row_final = max(1, min(GRID_ROWS, int((final_cy - header_top) // cell_h2) + 1)) dbg_obj['grid_column_final'] = col_final dbg_obj['grid_row_final'] = row_final except Exception: pass except Exception: pass self._save_json(dbg_dir / 'parsed_coords.json', dbg_obj) self._save_overlay( screenshot_path, [(int(coords[0]), int(coords[1]))], dbg_dir / 'overlay.png', label=f"query: {query_text}", box=matched_box, raw_mark=(raw_coords[0], raw_coords[1]) if isinstance(raw_coords, list) and len(raw_coords)==2 else None, ) except Exception: pass # Apply OS display scaling and optional Y offset before returning/clicking coords = self._apply_display_scale(coords) try: from .constants import Y_CLICK_OFFSET coords = [int(coords[0]), int(coords[1] + int(Y_CLICK_OFFSET))] except Exception: pass return coords except Exception as e: self.openai_logger.error(f"find_text_coordinates error: {e}") return None def find_text_coordinates_rekognition(self, screenshot_path: str, query_text: str) -> Optional[List[int]]: """Find coordinates for text using AWS Rekognition OCR. Returns [x, y] in screen pixels or None. """ try: try: self.openai_logger.info("Rekognition: analyzing screenshot for coordinates") except Exception: pass from .constants import AWS_REGION # Load screenshot and compress to JPEG to stay under Rekognition size limits img = Image.open(screenshot_path).convert('RGB') from io import BytesIO as _BytesIO buf = _BytesIO() # Format selection via env try: from .constants import REKOGNITION_IMAGE_FORMAT, REKOGNITION_JPEG_QUALITY except Exception: REKOGNITION_IMAGE_FORMAT = 'PNG' REKOGNITION_JPEG_QUALITY = 92 img_format = REKOGNITION_IMAGE_FORMAT if REKOGNITION_IMAGE_FORMAT in ('PNG','JPEG') else 'PNG' try: if img_format == 'PNG': img.save(buf, format='PNG') else: img.save(buf, format='JPEG', quality=int(REKOGNITION_JPEG_QUALITY)) except Exception: # Fallback to PNG if save fails buf = _BytesIO() img.save(buf, format='PNG') img_format = 'PNG' image_bytes = buf.getvalue() # Prepare debug directory and save prepared input image dbg_dir = None try: dbg_dir = self._get_debug_dir() / f"rekognition_{int(time.time()*1000)}" dbg_dir.mkdir(parents=True, exist_ok=True) try: # Save the prepared image bytes (full screenshot pre-tiling) ext = 'jpg' if img_format.upper() == 'JPEG' else 'png' with open(dbg_dir / f"prepared_image.{ext}", 'wb') as _f: _f.write(image_bytes) except Exception: pass try: self._save_text( dbg_dir / 'request.txt', f"query_text={query_text}\nimage_size={img.size[0]}x{img.size[1]}\nbytes={len(image_bytes)}\nformat={img_format}\nscreenshot_path={screenshot_path}\n" ) except Exception: pass try: self.openai_logger.info( f"Rekognition request: text='{str(query_text)[:120]}' size={img.size[0]}x{img.size[1]} bytes={len(image_bytes)} format={img_format}" ) except Exception: pass except Exception: pass # Initialize Rekognition client lazily try: _rk_client = boto3.client('rekognition', region_name=AWS_REGION) except Exception as e: try: self.openai_logger.error(f"Rekognition client init failed: {e}") except Exception: pass return None # Primary tiling path (if enabled): search tiles first and return immediately on success try: from .constants import ( REKOGNITION_USE_TILING as __TILING_ON, REKOGNITION_TILE_GRID as __T_GRID, REKOGNITION_TILE_OVERLAP as __T_OVLP, REKOGNITION_UPSCALE_FACTOR as __T_SCALE, REKOGNITION_IMAGE_FORMAT as __T_FMT, REKOGNITION_JPEG_QUALITY as __T_Q, ) except Exception: __TILING_ON = True; __T_GRID = "1x3"; __T_OVLP = 0.08; __T_SCALE = 1.0; __T_FMT = 'PNG'; __T_Q = 92 if __TILING_ON and isinstance(__T_GRID, str) and 'x' in __T_GRID: # Log tiling configuration for visibility and debugging try: if not dbg_dir: dbg_dir = self._get_debug_dir() / f"rekognition_{int(time.time()*1000)}" dbg_dir.mkdir(parents=True, exist_ok=True) self._save_text( dbg_dir / 'tiling_config.txt', f"enabled={__TILING_ON}\ngrid={__T_GRID}\noverlap={__T_OVLP}\nscale={__T_SCALE}\nformat={__T_FMT}\nquality={__T_Q}\nquery={query_text}\n" ) try: self.openai_logger.info(f"Rekognition tiling: enabled={__TILING_ON} grid={__T_GRID} overlap={__T_OVLP} scale={__T_SCALE}") except Exception: pass except Exception: pass # Build phrase/amount/date normalization for scoring import re as _re def _t_norm_phrase(s: str) -> str: try: txt = (s or "").lower().replace("\u2009", " ") txt = _re.sub(r"[\(\)\[\]\{\}:,]", " ", txt) txt = _re.sub(r"\s+", " ", txt).strip() months = {"january":"jan","february":"feb","march":"mar","april":"apr","june":"jun","july":"jul","august":"aug","september":"sep","october":"oct","november":"nov","december":"dec"} for full, abbr in months.items(): txt = txt.replace(full, abbr) return txt except Exception: return str(s or "").lower().strip() target_low = (query_text or '').strip().lower() # Amount handling try: is_amount = self._looks_like_amount(query_text) or ''.join(ch for ch in str(query_text) if ch.isdigit()) == str(query_text).strip() except Exception: is_amount = False amt_variants: set[str] = set(); norm_amt_target: str | None = None if is_amount: try: amt_variants = set(v.lower() for v in (self._generate_amount_variants(query_text) or set())) extra = set() for v in list(amt_variants): if ',' in v: extra.add(v.replace(',', ' ')); extra.add(v.replace(',', '\u2009')) amt_variants.update(e.lower() for e in extra) except Exception: amt_variants = set() try: norm_amt_target = (self._normalize_amount(query_text) or '').lower() except Exception: norm_amt_target = None # Date handling def _looks_like_date_t(s: str) -> bool: try: txt = str(s) if _re.search(r"\b\d{1,2}[./\-\\]\d{1,2}[./\-\\]\d{2,4}\b", txt): return True return _re.search(r"\b\d{1,2}\s+(jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\s+\d{2,4}\b", txt, _re.I) is not None except Exception: return False def _parse_date_digits_t(s: str): try: m = _re.search(r"(\d{1,2})[./\-\\](\d{1,2})[./\-\\](\d{2,4})", str(s)) if not m: return None d = int(m.group(1)); mth = int(m.group(2)); y = int(m.group(3)) if y < 100: y = 2000 + y return d, mth, y except Exception: return None def _date_norm_targets_t(s: str) -> set[str]: v = set(); pm = _parse_date_digits_t(s) if not pm: try: m = _re.search(r"(\d{1,2})\s+(jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\s+(\d{2,4})", str(s), _re.I) if m: d = int(m.group(1)); y = int(m.group(3)) mon_map = {"jan":1,"feb":2,"mar":3,"apr":4,"may":5,"jun":6,"jul":7,"aug":8,"sep":9,"oct":10,"nov":11,"dec":12} mth = mon_map[m.group(2).lower()] if y < 100: y = 2000 + y v.add(f"{d:02d}{mth:02d}{y:04d}"); v.add(f"{d}{mth}{y}") return v except Exception: pass return v d, mth, y = pm try: v.add(f"{d:02d}{mth:02d}{y:04d}"); v.add(f"{d}{mth}{y}") except Exception: pass return v is_date = _looks_like_date_t(query_text) date_norm_targets = _date_norm_targets_t(query_text) if is_date else set() norm_target_phrase = _t_norm_phrase(query_text) def _tile_score_match(det): try: txt = str(det.get('DetectedText', '') or ''); low = txt.lower(); conf = float(det.get('Confidence', 0) or 0) t = (det.get('Type') or '').upper() if is_amount: try: norm_txt = (self._normalize_amount(txt) or '').lower() except Exception: norm_txt = '' is_exact = ((norm_amt_target and norm_txt and norm_txt == norm_amt_target) or (low in amt_variants)) is_partial = (not is_exact and any(v in low for v in amt_variants)) elif is_date: try: norm_digits = ''.join(ch for ch in txt if ch.isdigit()) except Exception: norm_digits = '' is_exact = norm_digits in date_norm_targets is_partial = (not is_exact and any(nd in norm_digits for nd in date_norm_targets)) else: norm_txt = _t_norm_phrase(low) is_exact = (norm_txt == norm_target_phrase) is_partial = (not is_exact and norm_target_phrase in norm_txt) type_bonus = (1.0 if (is_amount and t == 'WORD') else (1.0 if (t == 'LINE' and len(norm_target_phrase.split()) >= 2) else (0.5 if t == 'WORD' else 0.0))) base = (2.0 if is_exact else (1.0 if is_partial else 0.0)) return base + type_bonus + conf / 100.0 except Exception: return 0.0 # Robust grid parsing: accept 'CxR' with separators x, X, ×, 'by', or whitespace try: import re as ___re m = ___re.search(r"(\d+)\s*[xX×\*\-:, ]\s*(\d+)", str(__T_GRID)) if m: cols, rows = int(m.group(1)), int(m.group(2)) else: cols, rows = 1, 1 except Exception: cols, rows = 1, 1 if cols < 1: cols = 1 if rows < 1: rows = 1 if cols * rows > 1: # Prefer returning the FIRST match found in reading order when enabled try: _STOP_AT_FIRST = str(os.getenv('REKOGNITION_FIRST_MATCH', '1')).lower() in ('1','true','yes') except Exception: _STOP_AT_FIRST = True # Collect per-tile raw responses so we can write a single combined log _tile_raw_dump = [] try: w, h = img.size tile_w = max(1, w // cols); tile_h = max(1, h // rows) ovx = max(0, int(tile_w * max(0.0, min(__T_OVLP, 0.5)))) ovy = max(0, int(tile_h * max(0.0, min(__T_OVLP, 0.5)))) # Track best candidate across all tiles: 3=exact WORD, 2=exact LINE, 1=partial best_cat = 0 best_score_global = 0.0 best_coords_global = None for r in range(rows): for c in range(cols): left_nom = c * tile_w; top_nom = r * tile_h left = max(0, left_nom - (ovx if c > 0 else 0)); top = max(0, top_nom - (ovy if r > 0 else 0)) right = min(w, (w if c == cols - 1 else (left_nom + tile_w + (ovx if c < cols - 1 else 0)))) bottom = min(h, (h if r == rows - 1 else (top_nom + tile_h + (ovy if r < rows - 1 else 0)))) if right <= left or bottom <= top: continue box = (left, top, right, bottom) tile_img = img.crop(box) scale = float(__T_SCALE or 1.0) if scale and scale > 1.0: try: tile_img_up = tile_img.resize((int(tile_img.width * scale), int(tile_img.height * scale)), Image.LANCZOS) except Exception: tile_img_up = tile_img; scale = 1.0 else: tile_img_up = tile_img; scale = 1.0 from io import BytesIO as ___B tb = ___B() try: if __T_FMT == 'PNG': tile_img_up.save(tb, format='PNG') else: tile_img_up.save(tb, format='JPEG', quality=int(__T_Q)) except Exception: tb = ___B(); tile_img_up.save(tb, format='PNG') tile_bytes = tb.getvalue() # Save tile bytes for debugging try: if dbg_dir: tile_ext = 'jpg' if __T_FMT.upper() == 'JPEG' else 'png' with open(dbg_dir / 'tiles_index.txt', 'a', encoding='utf-8') as __idx: __idx.write(f"{c+1},{r+1} box={box} size={tile_img_up.size} scale={scale}\n") with open(dbg_dir / f"tile_{c+1}_{r+1}.{tile_ext}", 'wb') as __tf: __tf.write(tile_bytes) except Exception: pass try: tile_resp = _rk_client.detect_text(Image={'Bytes': tile_bytes}) except Exception: continue # Persist raw response for this tile try: if dbg_dir: base = dbg_dir / f"raw_response_tile_{c+1}_{r+1}" self._save_json(base.with_suffix('.json'), tile_resp) try: with open(base.with_suffix('.txt'), 'w', encoding='utf-8') as _ft: json.dump(tile_resp, _ft, ensure_ascii=False, indent=2) except Exception: pass except Exception: pass try: _tile_raw_dump.append({ 'tile': [c+1, r+1], 'box': [int(left), int(top), int(right), int(bottom)], 'scale': float(scale), 'response': tile_resp, }) except Exception: pass tile_dets = tile_resp.get('TextDetections', []) or [] t_best = None; t_best_score = 0.0; t_best_center_override = None best_txt = None; best_type = None for det in tile_dets: s = _tile_score_match(det) if s > t_best_score: t_best_score = s; t_best = det try: best_txt = str(det.get('DetectedText') or '') best_type = str((det.get('Type') or '')).upper() except Exception: best_txt = None; best_type = None # Reconstruct LINE from WORDs to refine center try: from collections import defaultdict as __dd t_lines = [d for d in tile_dets if str((d.get('Type') or '')).upper() == 'LINE'] t_words = [d for d in tile_dets if str((d.get('Type') or '')).upper() == 'WORD'] lid2w = __dd(list) for wdet in t_words: pid = wdet.get('ParentId') if isinstance(pid, int): lid2w[pid].append(wdet) for ln in t_lines: try: lid = int(ln.get('Id')) if ln.get('Id') is not None else None except Exception: lid = None ws = lid2w.get(lid, []) try: ws = sorted(ws, key=lambda d: float((((d.get('Geometry') or {}).get('BoundingBox') or {}).get('Left') or 0.0))) except Exception: pass joined = " ".join(str(wd.get('DetectedText') or '') for wd in ws).strip() if not joined: continue joined_norm = _t_norm_phrase(joined) is_exact = (joined_norm == norm_target_phrase) is_partial = (not is_exact and norm_target_phrase in joined_norm) if not (is_exact or is_partial): continue try: tiw, tih = tile_img_up.size xs = []; ys = []; xe = []; ye = [] for wd in ws: bb = (wd.get('Geometry') or {}).get('BoundingBox') or {} lft = float(bb.get('Left', 0)); tp = float(bb.get('Top', 0)) wid = float(bb.get('Width', 0)); hei = float(bb.get('Height', 0)) xs.append(lft); ys.append(tp); xe.append(lft + wid); ye.append(tp + hei) if xs and ys and xe and ye: l = min(xs); t = min(ys); r2 = max(xe); b2 = max(ye) cx = int((l + r2) / 2.0 * tiw); cy = int((t + b2) / 2.0 * tih) score = (2.0 if is_exact else 1.0) + 1.0 if score > t_best_score: t_best_score = score; t_best = ln; t_best_center_override = [cx, cy] best_txt = joined; best_type = 'LINE' except Exception: pass except Exception: pass if t_best: try: tiw, tih = tile_img_up.size if t_best_center_override is not None: cx_local = int(t_best_center_override[0] / (scale if scale else 1.0)) cy_local = int(t_best_center_override[1] / (scale if scale else 1.0)) else: bb = (t_best.get('Geometry') or {}).get('BoundingBox') or {} lft = int(float(bb.get('Left', 0)) * tiw) tp = int(float(bb.get('Top', 0)) * tih) wid = int(float(bb.get('Width', 0)) * tiw) hei = int(float(bb.get('Height', 0)) * tih) cx_local = int((lft + max(1, wid) // 2) / (scale if scale else 1.0)) cy_local = int((tp + max(1, hei) // 2) / (scale if scale else 1.0)) gx = int(left + cx_local); gy = int(top + cy_local) coords = [gx, gy] except Exception: coords = None else: coords = None # Optional early return on the first match (top-to-bottom, left-to-right tiles) if coords is not None and _STOP_AT_FIRST: try: if not dbg_dir: dbg_dir = self._get_debug_dir() / f"rekognition_{int(time.time()*1000)}"; dbg_dir.mkdir(parents=True, exist_ok=True) # Save all-tiles raw collected so far and a brief summary try: self._save_json(dbg_dir / 'raw_detect_text_all_tiles.json', _tile_raw_dump) with open(dbg_dir / 'raw_detect_text_all_tiles.txt', 'w', encoding='utf-8') as _aft: json.dump(_tile_raw_dump, _aft, ensure_ascii=False, indent=2) except Exception: pass # Save overlay and summary raw_coords = coords[:] self._save_overlay(screenshot_path, [(int(coords[0]), int(coords[1]))], dbg_dir / 'overlay.png', label=f"rekog: {query_text}", raw_mark=(int(raw_coords[0]), int(raw_coords[1]))) try: self._save_text(dbg_dir / 'summary.txt', f"selected=({coords[0]},{coords[1]}) reason=first_match\n") except Exception: pass except Exception: pass return coords # Categorize and keep the best across all tiles (no early break) try: cat = 0 if coords is not None and isinstance(best_txt, str) and best_txt.strip(): low_txt = best_txt.lower() if is_amount: try: norm_txt = (self._normalize_amount(best_txt) or '').lower() except Exception: norm_txt = '' is_exact_amt = ((norm_amt_target and norm_txt and norm_txt == norm_amt_target) or (low_txt in amt_variants)) is_partial_amt = (not is_exact_amt and any(v in low_txt for v in amt_variants)) if is_exact_amt: cat = 3 if best_type == 'WORD' else 2 elif is_partial_amt: cat = 1 elif is_date: try: norm_digits = ''.join(ch for ch in best_txt if ch.isdigit()) except Exception: norm_digits = '' if norm_digits in date_norm_targets: cat = 3 if best_type == 'WORD' else 2 elif any(nd in norm_digits for nd in date_norm_targets): cat = 1 else: norm_txt2 = _t_norm_phrase(low_txt) if norm_txt2 == norm_target_phrase: cat = 3 if best_type == 'WORD' else 2 elif norm_target_phrase in norm_txt2: cat = 1 if cat > 0: if (cat > best_cat) or (cat == best_cat and t_best_score > best_score_global): best_cat = cat best_score_global = t_best_score best_coords_global = coords except Exception: pass # After scanning all tiles, prefer exact WORD > exact LINE > partial if best_coords_global is not None: coords = best_coords_global # Apply optional transforms and overlay then return try: from .constants import ( REKOGNITION_APPLY_TRANSFORM, REKOGNITION_APPLY_DISPLAY_SCALE, REKOGNITION_APPLY_Y_OFFSET, Y_CLICK_OFFSET, ) except Exception: REKOGNITION_APPLY_TRANSFORM = False; REKOGNITION_APPLY_DISPLAY_SCALE = False; REKOGNITION_APPLY_Y_OFFSET = False; Y_CLICK_OFFSET = 0 raw_coords = coords[:] if REKOGNITION_APPLY_TRANSFORM: try: coords = self._transform_coordinates(coords) except Exception: pass if not self._validate_coordinates(coords): return None if REKOGNITION_APPLY_DISPLAY_SCALE: try: coords = self._apply_display_scale(coords) except Exception: pass if REKOGNITION_APPLY_Y_OFFSET: try: coords = [int(coords[0]), int(coords[1] + int(Y_CLICK_OFFSET))] except Exception: pass try: if not dbg_dir: dbg_dir = self._get_debug_dir() / f"rekognition_{int(time.time()*1000)}"; dbg_dir.mkdir(parents=True, exist_ok=True) self._save_overlay( screenshot_path, [(int(coords[0]), int(coords[1]))], dbg_dir / 'overlay.png', label=f"rekog: {query_text}", raw_mark=(int(raw_coords[0]), int(raw_coords[1])) if isinstance(raw_coords, list) and len(raw_coords)==2 else None, ) except Exception: pass # Before returning, write a combined all-tiles file and a brief summary try: if dbg_dir: self._save_json(dbg_dir / 'raw_detect_text_all_tiles.json', _tile_raw_dump) try: with open(dbg_dir / 'raw_detect_text_all_tiles.txt', 'w', encoding='utf-8') as _aft: json.dump(_tile_raw_dump, _aft, ensure_ascii=False, indent=2) except Exception: pass # Minimal summary (selected coords) try: self._save_text(dbg_dir / 'summary.txt', f"selected=({coords[0]},{coords[1]})\n") except Exception: pass except Exception: pass return coords except Exception: pass # If tiling is the main path and no coords were found, do not fall back to full image return None # Detect text try: resp = _rk_client.detect_text(Image={'Bytes': image_bytes}) except Exception as e: try: self.openai_logger.error(f"Rekognition detect_text failed: {e}") except Exception: pass return None detections = resp.get('TextDetections', []) or [] # Debug: save raw response and a concise summary try: if not dbg_dir: dbg_dir = self._get_debug_dir() / f"rekognition_{int(time.time()*1000)}" dbg_dir.mkdir(parents=True, exist_ok=True) try: self._save_json(dbg_dir / 'raw_response.json', resp) except Exception: pass try: # One-line summary: TYPE\tCONF\tTEXT (first 120 chars) lines = [] for det in detections: try: _t = str((det.get('Type') or '')).upper() _c = float(det.get('Confidence', 0) or 0) _tx = str(det.get('DetectedText', '') or '').replace('\n', ' ')[:120] lines.append(f"{_t}\t{_c:.1f}\t{_tx}") except Exception: pass if lines: self._save_text(dbg_dir / 'summary.txt', "\n".join(lines)) except Exception: pass except Exception: dbg_dir = None # Log counts by type try: num_words = sum(1 for d in detections if str(d.get('Type') or '').upper() == 'WORD') num_lines = sum(1 for d in detections if str(d.get('Type') or '').upper() == 'LINE') self.openai_logger.info(f"Rekognition detections: total={len(detections)} WORD={num_words} LINE={num_lines}") except Exception: pass target = (query_text or '').strip().lower() if not target: return None # Prefer matches on WORD over LINE so the center is near the exact token # Build amount-aware variants and normalization # Phrase normalization for general text (handles parentheses and punctuation) import re as _re def _normalize_phrase_for_ocr(s: str) -> str: try: txt = (s or "").lower() # unify whitespace variants txt = txt.replace("\u2009", " ") # normalize common punctuation that OCR may drop txt = _re.sub(r"[\(\)\[\]\{\}:,]", " ", txt) # collapse multiple spaces txt = _re.sub(r"\s+", " ", txt).strip() # normalize month names to 3-letter form months = { "january":"jan","february":"feb","march":"mar","april":"apr","june":"jun","july":"jul", "august":"aug","september":"sep","october":"oct","november":"nov","december":"dec" } for full, abbr in months.items(): txt = txt.replace(full, abbr) return txt except Exception: return str(s or "").lower().strip() try: is_amount = self._looks_like_amount(query_text) or ''.join(ch for ch in str(query_text) if ch.isdigit()) == str(query_text).strip() except Exception: is_amount = False amount_variants: set[str] = set() norm_target: str | None = None if is_amount: try: amount_variants = set(v.lower() for v in (self._generate_amount_variants(query_text) or set())) # Add space/narrow-space thousands forms from comma forms extra = set() for v in list(amount_variants): if ',' in v: extra.add(v.replace(',', ' ')) extra.add(v.replace(',', '\u2009')) amount_variants.update(e.lower() for e in extra) except Exception: amount_variants = set() try: norm_target = (self._normalize_amount(query_text) or '').lower() except Exception: norm_target = None # Date-like handling: treat different separators or zero-padded forms as equivalent def _looks_like_date(s: str) -> bool: try: txt = str(s) # numeric dates like 25/01/2023 or 25-01-23 or 25.01.2023 if _re.search(r"\b\d{1,2}[./\-\\]\d{1,2}[./\-\\]\d{2,4}\b", txt): return True # month-name dates like 25 Jan 2023 return _re.search(r"\b\d{1,2}\s+(jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\s+\d{2,4}\b", txt, _re.I) is not None except Exception: return False def _parse_date_digits(s: str): try: m = _re.search(r"(\d{1,2})[./\-\\](\d{1,2})[./\-\\](\d{2,4})", str(s)) if not m: return None d = int(m.group(1)); mth = int(m.group(2)); y = int(m.group(3)) if y < 100: y = 2000 + y return d, mth, y except Exception: return None def _date_norm_variants_from_query(s: str) -> set[str]: v = set() pm = _parse_date_digits(s) if not pm: # try month-name format: 25 Jan 2023 -> 25012023 try: m = _re.search(r"(\d{1,2})\s+(jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\s+(\d{2,4})", str(s), _re.I) if m: d = int(m.group(1)); y = int(m.group(3)) mon_map = {"jan":1,"feb":2,"mar":3,"apr":4,"may":5,"jun":6,"jul":7,"aug":8,"sep":9,"oct":10,"nov":11,"dec":12} mth = mon_map[m.group(2).lower()] if y < 100: y = 2000 + y v.add(f"{d:02d}{mth:02d}{y:04d}") v.add(f"{d}{mth}{y}") return v except Exception: pass return v d, mth, y = pm try: v.add(f"{d:02d}{mth:02d}{y:04d}") v.add(f"{d}{mth}{y}") except Exception: pass return v is_date = _looks_like_date(query_text) date_norm_targets = _date_norm_variants_from_query(query_text) if is_date else set() norm_target_phrase = _normalize_phrase_for_ocr(query_text) def _score_match(det): try: txt = str(det.get('DetectedText', '') or '') low = txt.lower() conf = float(det.get('Confidence', 0) or 0) t = (det.get('Type') or '').upper() # Exact/partial rules with amount normalization if is_amount: try: norm_txt = (self._normalize_amount(txt) or '').lower() except Exception: norm_txt = '' is_exact = ( (norm_target and norm_txt and norm_txt == norm_target) or (low in amount_variants) ) is_partial = (not is_exact and any(v in low for v in amount_variants)) elif is_date: try: norm_digits = ''.join(ch for ch in txt if ch.isdigit()) except Exception: norm_digits = '' is_exact = norm_digits in date_norm_targets is_partial = (not is_exact and any(nd in norm_digits for nd in date_norm_targets)) else: # Phrase-aware normalization to tolerate punctuation/parentheses differences norm_txt = _normalize_phrase_for_ocr(low) is_exact = (norm_txt == norm_target_phrase) is_partial = (not is_exact and norm_target_phrase in norm_txt) # Prefer WORD for numbers; prefer LINE for multi-word phrases if is_amount: type_bonus = 1.0 if t == 'WORD' else 0.0 else: # phrase length > 1 word → LINE gets a bonus phrase_len = len(norm_target_phrase.split()) type_bonus = (1.0 if (t == 'LINE' and phrase_len >= 2) else (0.5 if t == 'WORD' else 0.0)) # Composite score: exact=2, partial=1 plus type bonus and confidence/100 base = (2.0 if is_exact else (1.0 if is_partial else 0.0)) return base + type_bonus + conf / 100.0 except Exception: return 0.0 best = None best_score = 0.0 best_center_override = None for det in detections: s = _score_match(det) if s > best_score: best_score = s best = det # Fallback/pass 2: reconstruct LINE phrases from WORD children and match phrase try: from collections import defaultdict as _dd lines_only = [d for d in detections if str(d.get('Type') or '').upper() == 'LINE'] words_only = [d for d in detections if str(d.get('Type') or '').upper() == 'WORD'] lid_to_words = _dd(list) for w in words_only: pid = w.get('ParentId') if isinstance(pid, int): lid_to_words[pid].append(w) for ln in lines_only: try: lid = int(ln.get('Id')) if ln.get('Id') is not None else None except Exception: lid = None ws = lid_to_words.get(lid, []) # Sort words left-to-right try: ws = sorted( ws, key=lambda d: float(((d.get('Geometry') or {}).get('BoundingBox') or {}).get('Left', 0.0)), ) except Exception: pass joined = " ".join(str(w.get('DetectedText') or '') for w in ws).strip() if not joined: continue joined_norm = _normalize_phrase_for_ocr(joined) is_exact = (joined_norm == norm_target_phrase) is_partial = (not is_exact and norm_target_phrase in joined_norm) if not (is_exact or is_partial): continue # Compute union bbox of words for precise click center try: iw, ih = img.size xs = []; ys = []; xe = []; ye = [] for w in ws: bb = (w.get('Geometry') or {}).get('BoundingBox') or {} left = float(bb.get('Left', 0)); top = float(bb.get('Top', 0)) width = float(bb.get('Width', 0)); height = float(bb.get('Height', 0)) xs.append(left); ys.append(top); xe.append(left + width); ye.append(top + height) if xs and ys and xe and ye: l = min(xs); t = min(ys); r = max(xe); b = max(ye) cx = int((l + r) / 2.0 * iw) cy = int((t + b) / 2.0 * ih) score = (2.0 if is_exact else 1.0) + 1.0 # phrase bonus if score > best_score: best_score = score best = ln best_center_override = [cx, cy] except Exception: pass except Exception: pass if not best: # Optional tiling fallback: split the screenshot into overlapping tiles and retry Rekognition per tile try: from .constants import ( REKOGNITION_USE_TILING, REKOGNITION_TILE_GRID, REKOGNITION_TILE_OVERLAP, REKOGNITION_UPSCALE_FACTOR, REKOGNITION_IMAGE_FORMAT as _RK_FMT, REKOGNITION_JPEG_QUALITY as _RK_Q, ) except Exception: REKOGNITION_USE_TILING = False REKOGNITION_TILE_GRID = "" REKOGNITION_TILE_OVERLAP = 0.08 REKOGNITION_UPSCALE_FACTOR = 1.0 _RK_FMT = 'PNG' _RK_Q = 92 coords = None if REKOGNITION_USE_TILING and isinstance(REKOGNITION_TILE_GRID, str) and 'x' in REKOGNITION_TILE_GRID: try: parts = REKOGNITION_TILE_GRID.lower().split('x') cols, rows = int(parts[0]), int(parts[1]) except Exception: cols, rows = 1, 1 if cols < 1: cols = 1 if rows < 1: rows = 1 if cols * rows > 1: try: w, h = img.size tile_w = max(1, w // cols) tile_h = max(1, h // rows) ovx = max(0, int(tile_w * max(0.0, min(REKOGNITION_TILE_OVERLAP, 0.5)))) ovy = max(0, int(tile_h * max(0.0, min(REKOGNITION_TILE_OVERLAP, 0.5)))) for r in range(rows): if coords is not None: break for c in range(cols): # Compute tile bounds with overlap left_nom = c * tile_w top_nom = r * tile_h left = max(0, left_nom - (ovx if c > 0 else 0)) top = max(0, top_nom - (ovy if r > 0 else 0)) right = min(w, (w if c == cols - 1 else (left_nom + tile_w + (ovx if c < cols - 1 else 0)))) bottom = min(h, (h if r == rows - 1 else (top_nom + tile_h + (ovy if r < rows - 1 else 0)))) if right <= left or bottom <= top: continue box = (left, top, right, bottom) tile_img = img.crop(box) # Optional upscale for better OCR on small tiles scale = float(REKOGNITION_UPSCALE_FACTOR or 1.0) if scale and scale > 1.0: try: tile_img_up = tile_img.resize((int(tile_img.width * scale), int(tile_img.height * scale)), Image.LANCZOS) except Exception: tile_img_up = tile_img scale = 1.0 else: tile_img_up = tile_img scale = 1.0 # Encode per configured format from io import BytesIO as __BytesIO _buf = __BytesIO() try: if _RK_FMT == 'PNG': tile_img_up.save(_buf, format='PNG') else: tile_img_up.save(_buf, format='JPEG', quality=int(_RK_Q)) except Exception: _buf = __BytesIO() tile_img_up.save(_buf, format='PNG') tile_bytes = _buf.getvalue() # Rekognition on tile try: tile_resp = _rk_client.detect_text(Image={'Bytes': tile_bytes}) except Exception: continue tile_dets = tile_resp.get('TextDetections', []) or [] # Score detections within tile t_best = None t_best_score = 0.0 t_best_center_override = None for det in tile_dets: s = _score_match(det) if s > t_best_score: t_best_score = s t_best = det # Fallback: reconstruct LINE from WORDs per tile try: from collections import defaultdict as __dd t_lines = [d for d in tile_dets if str((d.get('Type') or '')).upper() == 'LINE'] t_words = [d for d in tile_dets if str((d.get('Type') or '')).upper() == 'WORD'] lid2w = __dd(list) for wdet in t_words: pid = wdet.get('ParentId') if isinstance(pid, int): lid2w[pid].append(wdet) for ln in t_lines: try: lid = int(ln.get('Id')) if ln.get('Id') is not None else None except Exception: lid = None ws = lid2w.get(lid, []) try: ws = sorted(ws, key=lambda d: float((((d.get('Geometry') or {}).get('BoundingBox') or {}).get('Left') or 0.0))) except Exception: pass joined = " ".join(str(wd.get('DetectedText') or '') for wd in ws).strip() if not joined: continue joined_norm = _normalize_phrase_for_ocr(joined) is_exact = (joined_norm == norm_target_phrase) is_partial = (not is_exact and norm_target_phrase in joined_norm) if not (is_exact or is_partial): continue try: tiw, tih = tile_img_up.size xs = []; ys = []; xe = []; ye = [] for wd in ws: bb = (wd.get('Geometry') or {}).get('BoundingBox') or {} lft = float(bb.get('Left', 0)); tp = float(bb.get('Top', 0)) wid = float(bb.get('Width', 0)); hei = float(bb.get('Height', 0)) xs.append(lft); ys.append(tp); xe.append(lft + wid); ye.append(tp + hei) if xs and ys and xe and ye: l = min(xs); t = min(ys); r2 = max(xe); b2 = max(ye) cx = int((l + r2) / 2.0 * tiw) cy = int((t + b2) / 2.0 * tih) score = (2.0 if is_exact else 1.0) + 1.0 if score > t_best_score: t_best_score = score t_best = ln t_best_center_override = [cx, cy] except Exception: pass except Exception: pass if t_best: # Compute coords within tile (accounting for optional upscale) try: tiw, tih = tile_img_up.size if t_best_center_override is not None: cx_local = int(t_best_center_override[0] / (scale if scale else 1.0)) cy_local = int(t_best_center_override[1] / (scale if scale else 1.0)) else: bb = (t_best.get('Geometry') or {}).get('BoundingBox') or {} lft = int(float(bb.get('Left', 0)) * tiw) tp = int(float(bb.get('Top', 0)) * tih) wid = int(float(bb.get('Width', 0)) * tiw) hei = int(float(bb.get('Height', 0)) * tih) cx_local = int((lft + max(1, wid) // 2) / (scale if scale else 1.0)) cy_local = int((tp + max(1, hei) // 2) / (scale if scale else 1.0)) # Translate to global within original screenshot gx = int(left + cx_local) gy = int(top + cy_local) coords = [gx, gy] # Save minimal debug info try: if dbg_dir: self._save_text(dbg_dir / 'tile_match.txt', f"matched in tile c={c+1}, r={r+1}, box={box}, coords={coords}\n") except Exception: pass except Exception: coords = None if coords is not None: break except Exception: coords = None if coords is None: try: self.openai_logger.info(f"Rekognition: no match for '{query_text}'") except Exception: pass return None # Found via tiling path; continue with coord transforms below using `coords` best = None # Log best match details try: _t_best = str((best.get('Type') or '')).upper() _c_best = float(best.get('Confidence', 0) or 0) _txt_best = str(best.get('DetectedText', '') or '').replace('\n', ' ') self.openai_logger.info(f"Rekognition best match: type={_t_best} conf={_c_best:.1f} text='{_txt_best}'") if dbg_dir: try: self._save_json(dbg_dir / 'best_match.json', best) except Exception: pass except Exception: pass bbox = (best.get('Geometry') or {}).get('BoundingBox') or {} try: iw, ih = img.size if best is None and 'coords' in locals() and isinstance(coords, list) and len(coords) == 2: # Tiled path already computed coords coords = [int(coords[0]), int(coords[1])] elif best_center_override is not None: coords = [int(best_center_override[0]), int(best_center_override[1])] else: left = int(float(bbox.get('Left', 0)) * iw) top = int(float(bbox.get('Top', 0)) * ih) width = int(float(bbox.get('Width', 0)) * iw) height = int(float(bbox.get('Height', 0)) * ih) cx = left + max(1, width) // 2 cy = top + max(1, height) // 2 coords = [int(cx), int(cy)] except Exception: return None # Optional transforms: default disabled to preserve Rekognition center exactly try: from .constants import ( REKOGNITION_APPLY_TRANSFORM, REKOGNITION_APPLY_DISPLAY_SCALE, REKOGNITION_APPLY_Y_OFFSET, Y_CLICK_OFFSET, ) except Exception: REKOGNITION_APPLY_TRANSFORM = False REKOGNITION_APPLY_DISPLAY_SCALE = False REKOGNITION_APPLY_Y_OFFSET = False Y_CLICK_OFFSET = 0 raw_coords = coords[:] # Compute display scale if requested (derive DPR similar to OpenAI path) if REKOGNITION_APPLY_DISPLAY_SCALE: try: sw, sh = self._get_screen_size() try: import pyautogui as _pg img_full = _pg.screenshot(region=(0, 0, sw, sh)) iw2, ih2 = img_full.size self._dpi_scale = (iw2 / sw, ih2 / sh) try: self.openai_logger.info(f"Rekognition: computed DPI scale {self._dpi_scale}") except Exception: pass except Exception: self._dpi_scale = (1.0, 1.0) except Exception: self._dpi_scale = (1.0, 1.0) if REKOGNITION_APPLY_TRANSFORM: try: coords = self._transform_coordinates(coords) except Exception: pass if not self._validate_coordinates(coords): return None if REKOGNITION_APPLY_DISPLAY_SCALE: try: coords = self._apply_display_scale(coords) except Exception: pass if REKOGNITION_APPLY_Y_OFFSET: try: coords = [int(coords[0]), int(coords[1] + int(Y_CLICK_OFFSET))] except Exception: pass # Optional debug overlay try: if not dbg_dir: dbg_dir = self._get_debug_dir() / f"rekognition_{int(time.time()*1000)}" dbg_dir.mkdir(parents=True, exist_ok=True) self._save_overlay( screenshot_path, [(int(coords[0]), int(coords[1]))], dbg_dir / 'overlay.png', label=f"rekog: {query_text}", raw_mark=(int(raw_coords[0]), int(raw_coords[1])) if isinstance(raw_coords, list) and len(raw_coords)==2 else None, ) except Exception: pass return coords except Exception as e: try: self.openai_logger.error(f"find_text_coordinates_rekognition error: {e}") except Exception: pass return None def _perform_automated_link_clicking( self, original_screenshot_path: str, clickable_links: List[Dict], playlist_name: str = None ) -> List[Dict]: """ Perform automated clicking on all identified supporting document links. Args: original_screenshot_path: Path to original screenshot clickable_links: List of links to click playlist_name: Name of playlist for organization Returns: List of results for each click action """ results = [] # Maintain a list of previously clicked screen coordinates during this run clicked_points: List[Tuple[int, int]] = [] self.openai_logger.info(f"Starting automated clicking process for {len(clickable_links)} links") for i, link in enumerate(clickable_links): try: # Send a throttled heartbeat during long clicking loops try: if self.app: from os import getenv as _getenv hb_interval = float(_getenv("HB_INTERVAL_SEC", "60")) self.app.send_heartbeat_throttled(min_interval_sec=hb_interval) except Exception: pass self.openai_logger.info(f"Processing link {i+1}/{len(clickable_links)}: {link.get('description', 'Unknown')}") # Runtime exclusion: skip payment-related controls that are not supporting docs try: lw_norm = str(link.get('link_words') or '').strip().lower() except Exception: lw_norm = '' disallowed_fragments = ( 'receive payment', 'reveive payment', # common OCR typo 'record payment', 'add payment option', 'add payment method', 'payment options', 'payment methods', ) if lw_norm and any(frag in lw_norm for frag in disallowed_fragments): try: self.openai_logger.info(f"Skipping disallowed payment-related link text: '{lw_norm}'") except Exception: pass results.append({ "link": link, "success": False, "skipped": True, "reason": "disallowed_payment_control", }) continue if self.app: self.app.show_loader(f'Clicking supporting document link {i+1}/{len(clickable_links)}: {link.get("description", "Unknown")}') # Step 1: Prefer Rekognition click using link_words to avoid OpenAI misclicks link_words = '' try: link_words = str(link.get('link_words') or '').strip() except Exception: link_words = '' # Determine target coordinates (prefer Rekognition exact match) target_xy: Optional[Tuple[int, int]] = None if link_words: try: try: self.openai_logger.info(f"Rekognition attempt for link_words='{link_words}'") except Exception: pass try: from .rekognition_tiler import find_text_coordinates_tiled as _tiler except Exception: _tiler = None try: from .constants import AWS_REGION, REKOGNITION_UPSCALE_FACTOR, REKOGNITION_TILE_OVERLAP except Exception: AWS_REGION = 'us-east-1' REKOGNITION_UPSCALE_FACTOR = 2.0 REKOGNITION_TILE_OVERLAP = 0.10 if _tiler is not None: import boto3 as _b3 from PIL import Image as _PIL rk = _b3.client('rekognition', region_name=AWS_REGION) img = _PIL.open(original_screenshot_path).convert('RGB') # Always enable Rekognition debug output directory (no env flag required) _rk_debug_dir = None try: import time as _t from pathlib import Path as _Path _base = _Path('logs') / 'openai_debug' / f"rekognition_{int(_t.time()*1000)}" _base.mkdir(parents=True, exist_ok=True) _link_dir = _base / f"link_{i+1}" _link_dir.mkdir(parents=True, exist_ok=True) _rk_debug_dir = str(_link_dir) _rk_base_dir = str(_base) try: self.openai_logger.info(f"Rekognition debug dir for link {i+1}: {_rk_debug_dir}") except Exception: pass except Exception: _rk_debug_dir = None _rk_base_dir = None try: self.openai_logger.info(f"Rekognition exclude_points count: {len(clicked_points)}") except Exception: pass # First attempt with moderate de-dup radius coords = _tiler( rk, img, query=link_words, is_regex=False, upscale=float(REKOGNITION_UPSCALE_FACTOR or 2.0), overlap_frac=float(REKOGNITION_TILE_OVERLAP or 0.10), debug_dir=_rk_debug_dir, require_exact=True, require_include=True, # Prefer the first include in reading order (top-to-bottom) when duplicates exist stop_at_first_include=True, # Avoid re-clicking the same occurrence when multiple links share text exclude_points=clicked_points, exclude_radius=16, ) # If nothing found, retry with smaller exclusion radius (links may be close together) if not (isinstance(coords, (list, tuple)) and len(coords) == 2): try: self.openai_logger.info("Rekognition retry with smaller exclude radius (10)") except Exception: pass coords = _tiler( rk, img, query=link_words, is_regex=False, upscale=float(REKOGNITION_UPSCALE_FACTOR or 2.0), overlap_frac=float(REKOGNITION_TILE_OVERLAP or 0.10), debug_dir=_rk_debug_dir, require_exact=True, require_include=True, stop_at_first_include=True, exclude_points=clicked_points, exclude_radius=10, ) # Mirror key debug artifacts into the parent rekognition folder so they are easy to find try: if _rk_debug_dir and _rk_base_dir: import shutil as _sh from pathlib import Path as _Path p = _Path(_rk_debug_dir) b = _Path(_rk_base_dir) prefix = f"link_{i+1}__" for name in ( 'summary.json','summary.txt', 'raw_detect_text_all_tiles.json','raw_detect_text_all_tiles.txt', 'raw_detect_text_full.json','raw_detect_text_full.txt', ): src = p / name if src.exists(): try: _sh.copyfile(str(src), str(b / (prefix + name))) except Exception: pass except Exception: pass # Last-resort: pull all candidates and pick the first not near excluded points if not (isinstance(coords, (list, tuple)) and len(coords) == 2): try: cand_list = self._rekognition_list_candidates( original_screenshot_path, link_words, require_exact=True, exclude_points=clicked_points, exclude_radius=10, max_candidates=20, debug_dir=_rk_debug_dir, ) or [] except Exception: cand_list = [] if cand_list: try: ax, ay = cand_list[0] coords = (int(ax), int(ay)) except Exception: coords = None if isinstance(coords, (list, tuple)) and len(coords) == 2: target_xy = (int(coords[0]), int(coords[1])) link['_coords_are_screen'] = True link['coordinates'] = [target_xy[0], target_xy[1]] try: self.openai_logger.info(f"Rekognition selected coords for '{link_words}': {target_xy}") except Exception: pass except Exception: target_xy = None if target_xy is None: # Special-case fallback: handle split line '(No reference' + 'number)' try: if 'no reference' in (link_words or '').lower(): try: self.openai_logger.info("Attempting Rekognition '(No reference ... number)' two-line fallback") coords = self._rekognition_find_no_reference_pair( original_screenshot_path, exclude_points=clicked_points, exclude_radius=6, debug_dir=_rk_debug_dir, ) except Exception: coords = None if isinstance(coords, (list, tuple)) and len(coords) == 2: target_xy = (int(coords[0]), int(coords[1])) link['_coords_are_screen'] = True link['coordinates'] = [target_xy[0], target_xy[1]] try: self.openai_logger.info("Rekognition 'No reference' pair fallback selected coords: %s" % (str(target_xy))) except Exception: pass except Exception: target_xy = None if target_xy is None: # Honor policy: Rekognition-only, no OpenAI fallback try: self.openai_logger.info("Rekognition did not find coordinates; skipping OpenAI fallback by policy") except Exception: pass results.append({"link": link, "success": False, "error": "rekognition_not_found"}) continue if target_xy is None: results.append({"link": link, "success": False, "error": "no_click_coordinates"}) continue # Compute compare region and capture pre-click region try: reg_l, reg_t, reg_w, reg_h = self._compute_compare_region(target_xy[0], target_xy[1]) compare_region2 = (reg_l, reg_t, reg_w, reg_h) except Exception: compare_region2 = None pre_region_path = self._capture_pre_click_screenshot(i, link.get('description', 'link'), region=compare_region2) # Perform the click at target_xy restore_click = self._temporarily_hide_app() try: pyautogui.moveTo(target_xy[0], target_xy[1], duration=0.5) pyautogui.click() finally: try: restore_click() except Exception: pass click_result = {"success": True, "coordinates": [target_xy[0], target_xy[1]], "transformed_coordinates": [target_xy[0], target_xy[1]]} if not click_result['success']: self.openai_logger.error(f"Failed to click link {i+1}: {click_result.get('error', 'Unknown error')}") results.append(click_result) # Continue to next link even if this one fails continue else: try: # Record the successful click location for de-duplication cx, cy = None, None if isinstance(click_result.get('coordinates'), (list, tuple)) and len(click_result.get('coordinates')) == 2: cx, cy = int(click_result['coordinates'][0]), int(click_result['coordinates'][1]) elif isinstance(link.get('coordinates'), (list, tuple)) and len(link.get('coordinates')) == 2: cx, cy = int(link['coordinates'][0]), int(link['coordinates'][1]) if cx is not None and cy is not None: clicked_points.append((cx, cy)) except Exception: pass # Modal handling temporarily disabled per request # (Previously: waited 5s and attempted to close save/cancel dialogs) # Step 2: Wait configurable time self.openai_logger.info(f"Waiting {self.wait_time} seconds after click {i+1}") if self.app: self.app.show_loader(f'Waiting {self.wait_time} seconds after click {i+1}/{len(clickable_links)}...') time.sleep(self.wait_time) # Step 3: Region post-click screenshot and comparison post_region_path = self._capture_post_click_region_screenshot(i, link.get('description', 'link'), region=compare_region2) changed2 = True try: if pre_region_path and post_region_path: changed2 = self._images_look_different(pre_region_path, post_region_path) except Exception: changed2 = True # If unchanged, iterate through additional Rekognition candidates for the same link text if (not changed2) and link_words: try: exclude_pts = [(target_xy[0], target_xy[1])] + clicked_points except Exception: exclude_pts = [(target_xy[0], target_xy[1])] candidates2 = self._rekognition_list_candidates( original_screenshot_path, link_words, require_exact=True, exclude_points=exclude_pts, exclude_radius=28, max_candidates=20, ) or [] for idx2, (ax, ay) in enumerate(candidates2): try: # Fresh pre region for this candidate try: l2, t2, w2, h2 = self._compute_compare_region(ax, ay) cand_region = (l2, t2, w2, h2) except Exception: cand_region = None pre_cand = self._capture_pre_click_screenshot(i, f"{link.get('description','link')}_cand{idx2+1}", region=cand_region) # Click candidate restore_c = self._temporarily_hide_app() try: pyautogui.moveTo(int(ax), int(ay), duration=0.5) pyautogui.click() finally: try: restore_c() except Exception: pass time.sleep(2) post_cand = self._capture_post_click_region_screenshot(i, f"{link.get('description','link')}_cand{idx2+1}", region=cand_region) cand_changed = True try: if pre_cand and post_cand: cand_changed = self._images_look_different(pre_cand, post_cand) except Exception: cand_changed = True if cand_changed: changed2 = True target_xy = (int(ax), int(ay)) break except Exception: continue # If still unchanged, skip post-click flow for this link if not changed2: results.append({ "link": link, "click": click_result, "screenshot": {"success": False, "skipped": True, "reason": "no_change_after_click"}, "s3_upload": {"success": False, "skipped": True}, "navigation": {"success": True, "action": "skipped_post_click_steps"}, "success": False, }) time.sleep(1) continue # Step 3b: Now that we detected a change, take full-screen screenshot for upload screenshot_result = self._take_post_click_screenshot( link, i, playlist_name ) # Guard: Only proceed with upload and Get Supporting Docs if we actually # captured a post-click screenshot successfully (i.e., a link was clicked # and resulted in a document view to capture). If not, skip to next link. if not (isinstance(screenshot_result, dict) and screenshot_result.get('success')): try: self.openai_logger.info("Skip upload/support-docs: no successful post-click screenshot") except Exception: pass results.append({ "link": link, "click": click_result, "screenshot": screenshot_result or {"success": False, "error": "no_screenshot"}, "s3_upload": {"success": False, "skipped": True}, "navigation": {"success": True, "action": "skipped_post_click_steps"}, "success": False, }) # Small delay between actions time.sleep(1) continue # Step 4: Upload analyze screenshot like normal screenshots (no supporting_documents folder) try: s3_result = {"success": False, "skipped": True} mgr = getattr(self.app, 'screenshot_manager', None) if mgr and hasattr(mgr, 'upload_generic_screenshot_to_openai_s3'): step_details2 = getattr(self.app, 'current_di_step_details', None) or {} s3_result = mgr.upload_generic_screenshot_to_openai_s3( screenshot_result.get('path'), step_details=step_details2, ) # Record into manage file processing (same semantics as normal screenshots) if isinstance(s3_result, dict) and s3_result.get('success'): try: from .manage_file_processing import add_screenshot_record as _add_sr user_id = getattr(self.app, 'current_user_id', None) if user_id: try: step_num2 = int(step_details2.get('stepNumber') or 1) except Exception: step_num2 = 1 _add_sr( user_id=user_id, step_number=step_num2, bucket=str(s3_result.get('s3_bucket') or 'big-pond-openai'), key=str(s3_result.get('s3_key') or ''), description=step_details2.get('description') or 'Analyze doc screenshot', di_step_details=step_details2, flow_type_fallback='analyze', ) except Exception: pass # After successful upload, remove local file unless debugging try: import os as _os try: from .constants import KEEP_SCREENSHOTS_FOR_DEBUG as _KEEP except Exception: _KEEP = False if (not _KEEP) and isinstance(screenshot_result, dict): pth = screenshot_result.get('path') if pth and _os.path.isfile(pth): try: _os.remove(pth) except Exception: pass except Exception: pass except Exception: s3_result = {"success": False, "error": "upload_failed"} # Step 5: While the document view is still open, run the dedicated Get Supporting Docs flow # IMPORTANT: For Xero (app_id==6) run the same subplaylist flow as the 'support_docs' action type try: try: self.app.show_loader(f'Getting supporting documents for link {i+1}/{len(clickable_links)}...') except Exception: pass # Use shared router: prefer Sage attachments finder in analyzer context from .support_docs_router import route_supporting_docs def _sage_handler(): from .sage_attachments_finder import run_and_click_attachments as _sage_attach _sage_attach(self.app, analyzer=self) def _fallback_generic(): from .playback import PlaybackManager # PlaybackManager(self.app)._execute_support_docs_action() # In analyzer context: allow Xero to fallback to generic if the specific flow fails route_supporting_docs( self.app, sage_handler=_sage_handler, fallback_generic=_fallback_generic, sleep_before=0.0, allow_xero_fallback=True, ) finally: try: self.app.hide_loader() except Exception: pass # Step 6: Navigate back first to ensure we return to the correct page, # then handle any extra tabs/windows as cleanup. try: from .browser_tabs import ( get_chromium_tab_count, close_current_tab_via_hotkey, get_chrome_pids, kill_new_renderer_processes, ) except Exception: get_chromium_tab_count = None # type: ignore close_current_tab_via_hotkey = None # type: ignore get_chrome_pids = None # type: ignore kill_new_renderer_processes = None # type: ignore # Always try to navigate back before changing tabs back_result = self._go_back_to_previous_screen() # In LOCAL_DEV, skip all browser/tab closing logic to avoid disrupting manual testing try: from .constants import LOCAL_DEV as _LD except Exception: _LD = False if _LD: link_result = { "link": link, "click": click_result, "screenshot": screenshot_result, "s3_upload": s3_result, "navigation": back_result, "success": all([ click_result['success'], screenshot_result['success'], back_result['success'] ]) } results.append(link_result) if link_result['success']: self.openai_logger.info(f"Successfully processed link {i+1} (LOCAL_DEV, no tab cleanup): {link.get('description', 'Unknown')}") else: self.openai_logger.warning(f"Link {i+1} had some failures (LOCAL_DEV) but continuing") time.sleep(1) continue # Process-based cleanup first (safer for Chrome new-tab behavior) did_kill = False try: baseline = getattr(self, '_chrome_pid_baseline', None) if callable(get_chrome_pids) and callable(kill_new_renderer_processes): if baseline is None: setattr(self, '_chrome_pid_baseline', get_chrome_pids() or set()) else: killed = kill_new_renderer_processes(baseline) did_kill = killed > 0 if did_kill: self.openai_logger.info(f"Closed {killed} new Chrome renderer process(es)") except Exception: pass # Stronger fallback: kill all non-baseline chrome processes (including browser subprocesses) if not did_kill: try: from .browser_tabs import kill_extra_chrome_processes baseline2 = getattr(self, '_chrome_pid_baseline', None) or set() killed2 = kill_extra_chrome_processes(baseline2) if killed2 > 0: did_kill = True self.openai_logger.info(f"Force-closed {killed2} extra Chrome process(es)") except Exception: pass # Window-level fallback: close any new top-level Chrome windows not in baseline if not did_kill: try: from .browser_tabs import get_chrome_window_handles, close_new_chrome_windows base_hwnds = getattr(self, '_chrome_hwnd_baseline', None) if base_hwnds is None: setattr(self, '_chrome_hwnd_baseline', get_chrome_window_handles() or set()) else: closed_w = close_new_chrome_windows(base_hwnds) if closed_w > 0: did_kill = True self.openai_logger.info(f"Closed {closed_w} new Chrome window(s) via WM_CLOSE") except Exception: pass tab_count = None try: if callable(get_chromium_tab_count): tab_count = get_chromium_tab_count() except Exception: tab_count = None # Unconditional UIA close attempt (even if tab count unknown) uia_closed = False try: from .browser_tabs import close_tabs_via_uia self.openai_logger.info("Attempting UIA Ctrl+W loop (unconditional)") uia_closed = close_tabs_via_uia() if uia_closed: self.openai_logger.info("Closed tab via UIA Ctrl+W") except Exception: uia_closed = False if not did_kill and not uia_closed and isinstance(tab_count, int) and tab_count > 1: try: self.openai_logger.info(f"Detected {tab_count} tabs; attempting multi-close via Ctrl+W loop") except Exception: pass closed = False try: from .browser_tabs import close_extra_tabs closed = close_extra_tabs() except Exception: pass # Merge navigation result with tab cleanup outcome, keeping navigation success primary back_result = back_result if back_result.get("success") else {"success": bool(closed or uia_closed), "action": "close_tab" if (closed or uia_closed) else "close_tab_failed"} else: # Ensure we are on the first tab after going back switched = False try: from .browser_tabs import switch_to_first_tab switched = switch_to_first_tab() if switched: self.openai_logger.info("Sent Ctrl+1 to switch to first tab") except Exception: switched = False if switched and not back_result.get("success"): back_result = {"success": True, "action": "switch_to_tab_1"} # Enforce switch to first tab regardless of path taken above try: from .browser_tabs import switch_to_first_tab as _switch_enforce enforced = _switch_enforce() if enforced: try: self.openai_logger.info("Enforced switch to first tab via Ctrl+1 (post-navigation)") except Exception: pass back_result = {"success": True, "action": "switch_to_tab_1_enforced"} except Exception: pass # Combine results link_result = { "link": link, "click": click_result, "screenshot": screenshot_result, "s3_upload": s3_result, "navigation": back_result, "success": all([ click_result['success'], screenshot_result['success'], back_result['success'] ]) } results.append(link_result) if link_result['success']: self.openai_logger.info(f"Successfully processed link {i+1}: {link.get('description', 'Unknown')}") else: self.openai_logger.warning(f"Link {i+1} had some failures but continuing to next link") # Small delay between actions time.sleep(1) except Exception as e: self.openai_logger.error(f"Error processing link {i+1}: {e}") results.append({ "link": link, "success": False, "error": str(e) }) # Continue to next link even if this one fails self.openai_logger.info("Automated clicking process completed") return results def _click_link(self, link: Dict) -> Dict: """ Click on a specific supporting document link. Args: link: Link information with coordinates Returns: Dict with click result """ try: coordinates = link.get('coordinates', []) if len(coordinates) != 2: return {"success": False, "error": "Invalid coordinates"} # If coordinates originated from Rekognition tiler, treat them as screen-space from_flag = bool(link.get('_coords_are_screen')) if from_flag: x, y = int(coordinates[0]), int(coordinates[1]) transformed_coords = [x, y] try: self.openai_logger.info(f"Using Rekognition screen coords (no transform): ({x}, {y})") except Exception: pass else: transformed_coords = self._transform_coordinates(coordinates) x, y = transformed_coords # Bounds check against current screen size; skip click if invalid try: sw, sh = self._get_screen_size() if not (isinstance(x, (int, float)) and isinstance(y, (int, float)) and 0 <= x <= sw and 0 <= y <= sh): self.openai_logger.warning(f"Skip click: coordinates out of bounds {transformed_coords} for screen {sw}x{sh}") return {"success": False, "error": "Coordinates out of bounds"} except Exception: # If we cannot get screen size, be safe and do not click return {"success": False, "error": "Screen size unavailable"} description = link.get('description', 'Unknown supporting document link') self.openai_logger.info(f"Clicking supporting document link: {description} at transformed coordinates ({x}, {y})") # Move to coordinates and click with app temporarily hidden restore = self._temporarily_hide_app() try: pyautogui.moveTo(x, y, duration=0.5) pyautogui.click() finally: try: restore() except Exception: pass self.openai_logger.info(f"Successfully clicked at coordinates ({x}, {y})") return { "success": True, "coordinates": [x, y], "original_coordinates": coordinates, "transformed_coordinates": transformed_coords, "description": description } except Exception as e: self.openai_logger.error(f"Click error: {e}") return {"success": False, "error": str(e)} def capture_full_resolution_screenshot(self) -> str: """ Capture a screenshot at full screen resolution for OpenAI analysis. Returns: Path to the captured screenshot file """ try: import os from pathlib import Path # Create screenshots directory if it doesn't exist screenshots_dir = Path("screenshots/openai_analysis") screenshots_dir.mkdir(parents=True, exist_ok=True) # wait for the UI to settle wait_time = float(os.getenv("SCREENSHOT_PRE_DELAY_SEC", "5.0")) time.sleep(5.0) # Generate filename with timestamp and resolution timestamp = int(time.time() * 1000) sw, sh = self._get_screen_size() filename = f"openai_analysis_{timestamp}_{sw}x{sh}.png" screenshot_path = screenshots_dir / filename self.openai_logger.info(f"Capturing full resolution screenshot: {filename} ({sw}x{sh})") # Capture screenshot at full screen resolution with app hidden restore = self._temporarily_hide_app() try: # Pre-capture delay to allow UI overlays to close wait_time = float(os.getenv("SCREENSHOT_PRE_DELAY_SEC", "5.0")) time.sleep(wait_time) screenshot = pyautogui.screenshot(region=(0, 0, sw, sh)) screenshot.save(screenshot_path) finally: try: restore() except Exception: pass self.openai_logger.info(f"Full resolution screenshot saved: {screenshot_path}") return str(screenshot_path) except Exception as e: self.openai_logger.error(f"Error capturing full resolution screenshot: {e}") return None def _take_post_click_screenshot( self, link: Dict, link_index: int, playlist_name: str = None ) -> Dict: """ Take a screenshot after clicking a supporting document link. Args: link: Link that was clicked link_index: Index of the link in the sequence playlist_name: Name of playlist for organization Returns: Dict with screenshot result """ try: # Generate filename timestamp = int(time.time() * 1000) # Prefer the exact visible link words for traceability try: raw_link_words = link.get('link_words') or link.get('text') or '' except Exception: raw_link_words = '' try: raw_desc = link.get('description', 'supporting_doc') except Exception: raw_desc = 'supporting_doc' # Sanitize components for safe filenames import re as _re safe_desc = _re.sub(r'[^A-Za-z0-9 _.-]', '_', str(raw_desc)).strip().replace(' ', '_')[:32] or 'supporting_doc' safe_link_name = _re.sub(r'[^A-Za-z0-9 _.-]', '_', str(raw_link_words)).strip().replace(' ', '_')[:48] # Compose filename: prefer exact link words; fallback to description # Do NOT include timestamp here; the uploader prefixes one already if safe_link_name: filename = f"{safe_desc}__{safe_link_name}_{timestamp}.png" else: filename = f"{safe_desc}_{timestamp}.png" # Create screenshots directory to match normal screenshots (no supporting_documents/invoice folder) screenshots_base = Path("screenshots") try: # If playlist provided, mirror normal screenshot folder structure if playlist_name and str(playlist_name).strip() and str(playlist_name).strip().lower() != 'select playlist': safe_playlist = _re.sub(r'[^A-Za-z0-9 _.-]', '_', str(playlist_name)).strip() screenshots_dir = screenshots_base / (safe_playlist or 'default') else: screenshots_dir = screenshots_base except Exception: screenshots_dir = screenshots_base screenshots_dir.mkdir(parents=True, exist_ok=True) screenshot_path = screenshots_dir / filename self.openai_logger.info(f"Taking post-click screenshot: {filename}") # Take full-page scroll+stitch screenshot of the active window with app hidden restore = self._temporarily_hide_app() try: # Small pre-capture delay to let content stabilize try: wait_time = float(os.getenv("SCREENSHOT_PRE_DELAY_SEC", "0.5")) except Exception: wait_time = 0.5 time.sleep(wait_time) # Use the v2 full-page capture (no cropping by default) from .fullpage_screenshot_v2 import save_fullpage_screenshot_v2 save_fullpage_screenshot_v2( str(screenshot_path), window_title_contains=None, # use active window activate=False, # window should already be focused maximize=False, # avoid resizing after click ) except Exception: # Fallback to single full-screen capture if full-page capture fails try: sw, sh = self._get_screen_size() screenshot = pyautogui.screenshot(region=(0, 0, sw, sh)) screenshot.save(screenshot_path) except Exception: raise finally: try: restore() except Exception: pass self.openai_logger.info(f"Post-click screenshot saved: {screenshot_path}") return { "success": True, "path": str(screenshot_path), "filename": filename, "timestamp": timestamp } except Exception as e: self.openai_logger.error(f"Screenshot error: {e}") return {"success": False, "error": str(e)} def _upload_screenshot_to_s3(self, screenshot_path: str) -> Dict: """ Upload screenshot to S3 bucket. Args: screenshot_path: Path to screenshot file Returns: Dict with upload result """ if not self.s3_client: self.openai_logger.warning("S3 client not available, skipping upload") return {"success": False, "error": "S3 client not available"} try: # Generate S3 key to match normal screenshot convention filename = Path(screenshot_path).name timestamp = int(time.time() * 1000) # Try to include user context similar to normal screenshots try: user_uuid = getattr(self.app, 'current_user_id', None) or '' except Exception: user_uuid = '' s3_key = f"openai/{AWS_REGION}:{user_uuid}/{timestamp}_{filename}" self.openai_logger.info(f"Uploading screenshot to S3: {s3_key}") # Upload to S3 self.s3_client.upload_file( screenshot_path, self.s3_bucket, s3_key ) self.openai_logger.info(f"Screenshot uploaded to S3: s3://{self.s3_bucket}/{s3_key}") return { "success": True, "s3_bucket": self.s3_bucket, "s3_key": s3_key, "s3_url": f"s3://{self.s3_bucket}/{s3_key}" } except Exception as e: self.openai_logger.error(f"S3 upload error: {e}") return {"success": False, "error": str(e)} def _go_back_to_previous_screen(self) -> Dict: """ Navigate back to the previous screen using Alt+Left with a Backspace fallback if needed. Returns: Dict with navigation result """ try: self.openai_logger.info("Navigating back to previous screen (Alt+Left, fallback Backspace)") # Prefer browser-native back: Alt+Left try: pyautogui.hotkey('alt', 'left') time.sleep(1.2) primary_ok = True except Exception: primary_ok = False # Fallback to Backspace if Alt+Left failed if not primary_ok: try: pyautogui.press('backspace') time.sleep(1.2) primary_ok = True except Exception: primary_ok = False if primary_ok: self.openai_logger.info("Successfully navigated back to previous screen") else: self.openai_logger.warning("Back navigation hotkeys may not have executed successfully") return {"success": bool(primary_ok), "action": "alt_left_or_backspace"} except Exception as e: self.openai_logger.error(f"Navigation error: {e}") return {"success": False, "error": str(e)} def set_wait_time(self, wait_time: int): """Set configurable wait time after each click.""" old_wait_time = self.wait_time self.wait_time = wait_time self.openai_logger.info(f"Wait time updated from {old_wait_time}s to {wait_time}s") def get_status(self) -> Dict: """Get current status of the analyzer.""" status = { "s3_available": bool(self.s3_client), "s3_bucket": self.s3_bucket, "openai_configured": bool(OPENAI_API_KEY), "local_dev": LOCAL_DEV, "current_wait_time": self.wait_time } self.openai_logger.info(f"Status requested: {status}") return status def get_log_file_path(self) -> str: """Get the path to the OpenAI log file.""" log_file = Path("logs/openai.log") return str(log_file.absolute()) def perform_automated_clicking(self, clickable_elements: List[Dict], screenshot_path: str): """ Perform automated clicking on all identified supporting document links. """ try: self.openai_logger.info(f"Starting automated clicking on {len(clickable_elements)} elements") for i, element in enumerate(clickable_elements): try: # Runtime exclusion: skip payment-related controls try: lw_norm = str(element.get('link_words') or '').strip().lower() except Exception: lw_norm = '' disallowed_fragments = ( 'receive payment', 'reveive payment', 'record payment', 'add payment option', 'add payment method', 'payment options', 'payment methods', ) if lw_norm and any(frag in lw_norm for frag in disallowed_fragments): try: self.openai_logger.info(f"Skipping disallowed payment-related element: '{lw_norm}'") except Exception: pass continue # Extract coordinates coords = element.get('coordinates', []) if len(coords) != 2: self.openai_logger.warning(f"Invalid coordinates for element {i}: {coords}") continue # Transform and bounds-check coordinates try: x, y = self._transform_coordinates(coords) except Exception: self.openai_logger.warning(f"Skipping element {i}: transform failed for coords {coords}") continue try: sw, sh = self._get_screen_size() if not (0 <= x <= sw and 0 <= y <= sh): self.openai_logger.warning(f"Skipping element {i}: out-of-bounds click ({x},{y}) for screen {sw}x{sh}") continue except Exception: self.openai_logger.warning(f"Skipping element {i}: screen size unavailable") continue description = element.get('description', 'Unknown element') self.openai_logger.info(f"Clicking element {i+1}/{len(clickable_elements)}: {description} at ({x}, {y})") # Compute a region around the click point for change detection try: region_left, region_top, region_w, region_h = self._compute_compare_region(x, y) compare_region = (region_left, region_top, region_w, region_h) except Exception: compare_region = None # Capture pre-click screenshot for change detection pre_click_screenshot = self._capture_pre_click_screenshot(i, description, region=compare_region) # Also capture a full-screen baseline for Rekognition alternate attempts pre_fullscreen_screenshot = None try: sw, sh = self._get_screen_size() restore_fs = self._temporarily_hide_app() try: _pre_full = pyautogui.screenshot(region=(0, 0, sw, sh)) ts = int(time.time()) pre_fullscreen_screenshot = f"screenshots/pre_click_debug/pre_full_{i}_{ts}_{description.replace(' ', '_')}.png" os.makedirs(os.path.dirname(pre_fullscreen_screenshot), exist_ok=True) _pre_full.save(pre_fullscreen_screenshot) finally: try: restore_fs() except Exception: pass except Exception: pre_fullscreen_screenshot = None # Click on the element with app temporarily hidden restore = self._temporarily_hide_app() try: pyautogui.click(x, y) finally: try: restore() except Exception: pass # Wait for the specified time self.openai_logger.info(f"Waiting {self.wait_time} seconds after click...") time.sleep(self.wait_time) # Take screenshot after click (same region as pre-click) post_click_screenshot = self._capture_post_click_region_screenshot(i, description, region=compare_region) # Determine if the screen changed changed = True try: if pre_click_screenshot and post_click_screenshot: changed = self._images_look_different(pre_click_screenshot, post_click_screenshot) except Exception: changed = True # If no change, attempt alternate Rekognition matches for the same link text if not changed: try: self.openai_logger.info("No visual change detected; attempting alternate Rekognition match") except Exception: pass # Prepare for alternate attempts try: link_words_exact = str(element.get('link_words') or '').strip() except Exception: link_words_exact = '' exclude_points = [(int(x), int(y))] # Build a candidate list from Rekognition response for the same link word candidates = [] try: if pre_fullscreen_screenshot and link_words_exact: candidates = self._rekognition_list_candidates( pre_fullscreen_screenshot, link_words_exact, require_exact=True, exclude_points=exclude_points, exclude_radius=28, max_candidates=20, ) or [] except Exception: candidates = [] attempt_idx = 0 while attempt_idx < len(candidates): try: new_coords = candidates[attempt_idx] # Click alternate coordinates ax, ay = int(new_coords[0]), int(new_coords[1]) exclude_points.append((ax, ay)) try: self.openai_logger.info(f"Alternate click attempt {attempt_idx+1} at ({ax}, {ay})") except Exception: pass # Recompute compare region centered on the new coords try: alt_left, alt_top, alt_w, alt_h = self._compute_compare_region(ax, ay) alt_region = (alt_left, alt_top, alt_w, alt_h) except Exception: alt_region = None # Fresh pre-click region for this alternate attempt pre_click_screenshot_alt_pre = self._capture_pre_click_screenshot(i, f"{description}_alt{attempt_idx+1}", region=alt_region) restore2 = self._temporarily_hide_app() try: pyautogui.click(ax, ay) finally: try: restore2() except Exception: pass # Wait and capture again time.sleep(self.wait_time) post_click_screenshot_alt = self._capture_post_click_region_screenshot(i, f"{description}_alt{attempt_idx+1}", region=alt_region) try: if pre_click_screenshot_alt_pre and post_click_screenshot_alt: changed = self._images_look_different(pre_click_screenshot_alt_pre, post_click_screenshot_alt) except Exception: changed = True if changed: post_click_screenshot = post_click_screenshot_alt or post_click_screenshot break attempt_idx += 1 except Exception: break # If changed, continue with upload and navigation; else skip if changed: # Capture a full-screen screenshot for upload to S3 upload_fullscreen_path = None try: # Prefer a fresh fullscreen capture after change sw, sh = self._get_screen_size() restore3 = self._temporarily_hide_app() try: fs_img = pyautogui.screenshot(region=(0, 0, sw, sh)) ts2 = int(time.time()) upload_fullscreen_path = f"screenshots/post_click_debug/full_upload_{i}_{ts2}_{description.replace(' ', '_')}.png" os.makedirs(os.path.dirname(upload_fullscreen_path), exist_ok=True) fs_img.save(upload_fullscreen_path) finally: try: restore3() except Exception: pass except Exception: upload_fullscreen_path = post_click_screenshot # Upload full-screen screenshot to S3 if upload_fullscreen_path: s3_url = self._upload_to_s3(upload_fullscreen_path, f"post_click_full_{i}_{description.replace(' ', '_')}") self.openai_logger.info(f"Uploaded full-screen post-click screenshot to S3: {s3_url}") # Go back to previous screen self.openai_logger.info("Navigating back to previous screen...") pyautogui.press('backspace') # Wait a moment for navigation time.sleep(2) else: try: self.openai_logger.info("No screen change after all attempts; skipping upload/back and moving to next element") except Exception: pass except Exception as e: self.openai_logger.error(f"Error clicking element {i}: {e}") continue self.openai_logger.info("Automated clicking sequence completed") except Exception as e: self.openai_logger.error(f"Error in perform_automated_clicking: {e}") raise def _capture_post_click_screenshot(self, index: int, description: str) -> str: """Capture a full-screen screenshot after clicking for upload/logging.""" try: timestamp = int(time.time()) filename = f"post_click_{index}_{timestamp}_{description.replace(' ', '_')}.png" filepath = f"screenshots/post_click_debug/{filename}" # Ensure directory exists os.makedirs(os.path.dirname(filepath), exist_ok=True) # Take screenshot at full screen resolution with app hidden restore = self._temporarily_hide_app() try: sw, sh = self._get_screen_size() screenshot = pyautogui.screenshot(region=(0, 0, sw, sh)) screenshot.save(filepath) finally: try: restore() except Exception: pass self.openai_logger.info(f"Captured post-click screenshot: {filepath}") return filepath except Exception as e: self.openai_logger.error(f"Error capturing post-click screenshot: {e}") return None def _capture_post_click_region_screenshot(self, index: int, description: str, region=None) -> str: """Capture a region screenshot after clicking (used only for change detection).""" try: timestamp = int(time.time()) filename = f"post_click_region_{index}_{timestamp}_{description.replace(' ', '_')}.png" filepath = f"screenshots/post_click_debug/{filename}" os.makedirs(os.path.dirname(filepath), exist_ok=True) restore = self._temporarily_hide_app() try: if isinstance(region, (list, tuple)) and len(region) == 4: l, t, w, h = [int(v) for v in region] screenshot = pyautogui.screenshot(region=(l, t, w, h)) else: # Fallback to small top-left region if region missing screenshot = pyautogui.screenshot() screenshot.save(filepath) finally: try: restore() except Exception: pass self.openai_logger.info(f"Captured post-click region screenshot: {filepath}") return filepath except Exception as e: self.openai_logger.error(f"Error capturing post-click region screenshot: {e}") return None def _rekognition_list_candidates( self, screenshot_path: str, link_words: str, require_exact: bool = True, exclude_points: Optional[List[Tuple[int, int]]] = None, exclude_radius: int = 28, max_candidates: int = 20, debug_dir: Optional[str] = None, ) -> List[Tuple[int, int]]: """Return a list of candidate (x,y) centers for the same link text from a single Rekognition pass. This uses the tiler once and extracts all include-line matches, ordered by confidence, excluding points near previously tried coordinates. """ try: from .constants import ( REKOGNITION_UPSCALE_FACTOR, REKOGNITION_TILE_OVERLAP, AWS_REGION, ) except Exception: REKOGNITION_UPSCALE_FACTOR = 2.0 REKOGNITION_TILE_OVERLAP = 0.10 AWS_REGION = 'us-east-1' try: from PIL import Image as _PILImage import boto3 as _b3 from .rekognition_tiler import find_text_coordinates_tiled as _tiler except Exception: return [] try: full = _PILImage.open(screenshot_path).convert('RGB') except Exception: return [] # Single Rekognition call to get raw response via tiler path try: rk = _b3.client('rekognition', region_name=AWS_REGION) from io import BytesIO as _BytesIO buf = _BytesIO(); full.save(buf, format='PNG') resp = rk.detect_text(Image={'Bytes': buf.getvalue()}) # Optional debug save of raw response if debug_dir: try: with open(f"{str(debug_dir).rstrip('/').rstrip('\\')}/raw_list_candidates.json", 'w', encoding='utf-8') as f: import json as _json _json.dump(resp, f, ensure_ascii=False, indent=2) except Exception: pass except Exception: return [] iw, ih = full.size q_low = (link_words or '').lower() q_soft = None try: import re as _re def _soft_norm(s: str) -> str: s = (s or '').lower() s = _re.sub(r"[^a-z0-9]+", " ", s) s = _re.sub(r"\s+", " ", s).strip() return s q_soft = _soft_norm(link_words) except Exception: pass def _is_near(px: int, py: int) -> bool: try: pts = exclude_points or [] r2 = max(1, int(exclude_radius)) ** 2 for ex, ey in pts: dx = int(px) - int(ex) dy = int(py) - int(ey) if dx * dx + dy * dy <= r2: return True return False except Exception: return False # Collect all LINE matches for the same text (exact-only) lines = [d for d in (resp.get('TextDetections') or []) if d.get('Type') == 'LINE'] matches = [] for d in lines: txt = str(d.get('DetectedText') or '') low = txt.lower() soft = _soft_norm(txt) if q_soft is not None else None is_match = False if require_exact: if (q_low and q_low == low) or (q_soft and soft == q_soft): is_match = True else: if (q_low and q_low in low) or (q_soft and (soft and q_soft in soft)): is_match = True if not is_match: continue try: bb = d['Geometry']['BoundingBox'] cx = int((bb['Left'] + bb['Width'] / 2) * iw) cy = int((bb['Top'] + bb['Height'] / 2) * ih) if not _is_near(cx, cy): conf = float(d.get('Confidence', 0) or 0) matches.append(((cx, cy), conf)) except Exception: continue # Sort by confidence desc and limit matches.sort(key=lambda x: x[1], reverse=True) coords = [m[0] for m in matches[:max_candidates]] return coords def _rekognition_find_no_reference_pair( self, screenshot_path: str, exclude_points: Optional[List[Tuple[int, int]]] = None, exclude_radius: int = 0, debug_dir: Optional[str] = None, ) -> Optional[Tuple[int, int]]: """Detect the specific two-line pattern '(No reference' + 'number)' and return union center. Uses Rekognition LINE detections on the full image. Returns None if pattern not found. """ try: from .constants import AWS_REGION except Exception: AWS_REGION = 'us-east-1' try: from PIL import Image as _PIL import boto3 as _b3 except Exception: return None # Helper: near-point exclusion def _is_near(px: int, py: int) -> bool: try: pts = exclude_points or [] r2 = max(1, int(exclude_radius)) ** 2 if not pts or r2 <= 1: return False for ex, ey in pts: dx = int(px) - int(ex) dy = int(py) - int(ey) if dx * dx + dy * dy <= r2: return True return False except Exception: return False # Normalize helpers import re as _re def _norm(s: str) -> str: s = (s or '').lower() s = _re.sub(r"[^a-z0-9]+", " ", s) return _re.sub(r"\s+", " ", s).strip() try: full = _PIL.open(screenshot_path).convert('RGB') except Exception: return None try: rk = _b3.client('rekognition', region_name=AWS_REGION) from io import BytesIO as _BytesIO buf = _BytesIO(); full.save(buf, format='PNG') resp = rk.detect_text(Image={'Bytes': buf.getvalue()}) except Exception: return None iw, ih = full.size lines = [d for d in (resp.get('TextDetections') or []) if d.get('Type') == 'LINE'] # Sort by reading order def _key(d): bb = d.get('Geometry', {}).get('BoundingBox', {}) return (float(bb.get('Top', 0) or 0), float(bb.get('Left', 0) or 0)) lines.sort(key=_key) # Search anchor then adjacent 'number' line (choose highest combined confidence) best = None best_score = -1.0 best_anchor = None best_anchor_conf = -1.0 for idx in range(len(lines)): a = lines[idx] a_txt = str(a.get('DetectedText') or '') a_norm = _norm(a_txt) if 'no reference' not in a_norm: continue try: abb = a['Geometry']['BoundingBox'] a_left, a_top, a_w, a_h = float(abb['Left']), float(abb['Top']), float(abb['Width']), float(abb['Height']) except Exception: continue # Track anchor as fallback if pair not found try: cx_a = int(((a_left + a_w / 2.0) * iw)) cy_a = int(((a_top + a_h / 2.0) * ih)) if not _is_near(cx_a, cy_a): conf_a = float(a.get('Confidence', 0) or 0) if conf_a > best_anchor_conf: best_anchor_conf = conf_a best_anchor = (cx_a, cy_a, a) except Exception: pass # Look ahead a small window for 'number' line for j in range(idx + 1, min(idx + 4, len(lines))): b = lines[j] b_txt = str(b.get('DetectedText') or '') b_norm = _norm(b_txt) if 'number' not in b_norm: continue try: bbb = b['Geometry']['BoundingBox'] b_left, b_top, b_w, b_h = float(bbb['Left']), float(bbb['Top']), float(bbb['Width']), float(bbb['Height']) except Exception: continue # Spatial checks: horizontal overlap and vertical closeness r1, r2 = a_left + a_w, b_left + b_w overlap = max(0.0, min(r1, r2) - max(a_left, b_left)) minw = max(1e-6, min(a_w, b_w)) horiz_ok = (overlap / minw) >= 0.30 vgap = abs(b_top - a_top) avg_h = max(1e-6, (a_h + b_h) / 2.0) vert_ok = vgap <= (2.0 * avg_h) if not (horiz_ok and vert_ok): continue # Union bbox -> center in image pixels left_px = min(a_left * iw, b_left * iw) top_px = min(a_top * ih, b_top * ih) right_px = max((a_left + a_w) * iw, (b_left + b_w) * iw) bottom_px = max((a_top + a_h) * ih, (b_top + b_h) * ih) cx = int((left_px + right_px) / 2) cy = int((top_px + bottom_px) / 2) if _is_near(cx, cy): continue score = float(a.get('Confidence', 0) or 0) + float(b.get('Confidence', 0) or 0) if score > best_score: best = (cx, cy, a, b) best_score = score if best is not None: cx, cy, a, b = best if debug_dir: try: import json as _json with open(f"{str(debug_dir).rstrip('/').rstrip('\\')}/no_reference_pair.json", 'w', encoding='utf-8') as f: _json.dump({ 'anchor': a, 'follower': b, 'center': {'x': cx, 'y': cy}, 'score': best_score, }, f, ensure_ascii=False, indent=2) except Exception: pass return (cx, cy) # Fallback to clicking the anchor '(No reference' line center if pair not found if best_anchor is not None: cx, cy, a = best_anchor if debug_dir: try: import json as _json with open(f"{str(debug_dir).rstrip('/').rstrip('\\')}/no_reference_anchor.json", 'w', encoding='utf-8') as f: _json.dump({ 'anchor': a, 'center': {'x': cx, 'y': cy}, 'score': best_anchor_conf, }, f, ensure_ascii=False, indent=2) except Exception: pass return (cx, cy) return None # On any failure, return empty return [] def _compute_compare_region(self, center_x: int, center_y: int, width: int = 600, height: int = 400) -> tuple: """Compute a clamped rectangular region centered on (x,y) for visual comparisons. Returns (left, top, width, height). Width/height defaults may be tuned. """ try: sw, sh = self._get_screen_size() w = max(100, int(width)) h = max(100, int(height)) left = int(center_x - w // 2) top = int(center_y - h // 2) left = max(0, min(left, max(0, sw - w))) top = max(0, min(top, max(0, sh - h))) return left, top, w, h except Exception: # Fallback to top-left small region to avoid failures return 0, 0, 600, 400 def _capture_pre_click_screenshot(self, index: int, description: str, region=None) -> str: """Capture screenshot before clicking an element for change detection. If region is provided as (left, top, width, height), captures only that area. """ try: timestamp = int(time.time()) filename = f"pre_click_{index}_{timestamp}_{description.replace(' ', '_')}.png" filepath = f"screenshots/pre_click_debug/{filename}" # Ensure directory exists os.makedirs(os.path.dirname(filepath), exist_ok=True) # Take screenshot at full screen resolution with app hidden restore = self._temporarily_hide_app() try: if isinstance(region, (list, tuple)) and len(region) == 4: l, t, w, h = [int(v) for v in region] screenshot = pyautogui.screenshot(region=(l, t, w, h)) else: sw, sh = self._get_screen_size() screenshot = pyautogui.screenshot(region=(0, 0, sw, sh)) screenshot.save(filepath) finally: try: restore() except Exception: pass self.openai_logger.info(f"Captured pre-click screenshot: {filepath}") return filepath except Exception as e: self.openai_logger.error(f"Error capturing pre-click screenshot: {e}") return None def _images_look_different(self, img_path_a: str, img_path_b: str, pixel_delta_threshold: int = 10, change_ratio_threshold: float = 0.002) -> bool: """Return True if images look significantly different. - pixel_delta_threshold: per-channel absolute difference to count a pixel as changed - change_ratio_threshold: fraction of changed pixels to consider as a real change """ try: from PIL import Image, ImageChops import numpy as _np if not (img_path_a and img_path_b and os.path.isfile(img_path_a) and os.path.isfile(img_path_b)): return True a = Image.open(img_path_a).convert('RGB') b = Image.open(img_path_b).convert('RGB') if a.size != b.size: # Resize B to A for fair comparison b = b.resize(a.size, Image.BILINEAR) diff = ImageChops.difference(a, b) # Quick identical check if not diff.getbbox(): return False arr = _np.asarray(diff, dtype=_np.int16) # Count pixels where any channel exceeds threshold changed_mask = (arr > pixel_delta_threshold).any(axis=2) changed_ratio = float(changed_mask.sum()) / float(changed_mask.size) return changed_ratio >= float(change_ratio_threshold) except Exception: # On errors, assume changed to avoid infinite loops return True def _upload_to_s3(self, file_path: str, key_prefix: str) -> str: """Upload file to S3 bucket.""" try: if LOCAL_DEV: self.openai_logger.info(f"Local dev mode - skipping S3 upload for {file_path}") return f"local://{file_path}" s3_client = boto3.client('s3', region_name=AWS_REGION) # Generate S3 key to match normal screenshot convention timestamp = int(time.time()) try: user_uuid = getattr(self.app, 'current_user_id', None) or '' except Exception: user_uuid = '' # key_prefix is kept in filename for traceability s3_key = f"openai/{AWS_REGION}:{user_uuid}/{key_prefix}_{timestamp}.png" # Upload file s3_client.upload_file(file_path, self.s3_bucket, s3_key) # Generate URL s3_url = f"https://{self.s3_bucket}.s3.{AWS_REGION}.amazonaws.com/{s3_key}" self.openai_logger.info(f"Successfully uploaded to S3: {s3_url}") return s3_url except Exception as e: self.openai_logger.error(f"Error uploading to S3: {e}") return None