import os import io import json import time import random from pathlib import Path from typing import Dict, Any, Optional from PIL import Image, ImageDraw try: import google.generativeai as legacy_genai HAS_LEGACY_GENAI = True except (ImportError, AttributeError): HAS_LEGACY_GENAI = False legacy_genai = None try: from google import genai genai_types = getattr(genai, "types", None) HAS_NEW_GENAI = True except (ImportError, AttributeError): HAS_NEW_GENAI = False genai = None genai_types = None class GeminiBrain: """ The AI Brain of DSE Automation Sosmed. Uses Google Gemini API to analyze web pages, identify dynamic selectors, solve modal dialogues, and generate optimized platform-specific captions. """ def __init__(self, api_key: Optional[str] = None, model_name: str = "gemini-3.5-flash"): self.api_key = api_key or os.getenv("GEMINI_API_KEY", "") self.model_name = model_name or os.getenv("GEMINI_MODEL", "gemini-3.5-flash") self.knowledge_path = Path(__file__).resolve().parent / "site_knowledge.json" self._load_knowledge() # Visual Debugger & Inspector directory self.vision_debug_dir = Path(__file__).resolve().parent.parent / "static" / "vision_debug" self.vision_debug_dir.mkdir(parents=True, exist_ok=True) self.vision_history_file = self.vision_debug_dir / "vision_history.json" def set_api_key(self, api_key: str): self.api_key = api_key.strip() def _load_knowledge(self): if self.knowledge_path.exists(): try: with open(self.knowledge_path, "r", encoding="utf-8") as f: self.knowledge = json.load(f) except Exception: self.knowledge = {} else: self.knowledge = {} def _save_knowledge(self): try: with open(self.knowledge_path, "w", encoding="utf-8") as f: json.dump(self.knowledge, f, indent=2, ensure_ascii=False) except Exception as e: print(f"[WARN] Failed to save site knowledge: {e}") CANDIDATE_MODELS = [ "gemini-3.5-flash", "gemini-flash-lite-latest", "gemini-flash-latest", "gemini-3.6-flash", "gemini-1.5-flash", "gemma-4-26b-a4b-it" ] def test_connection(self, api_key: Optional[str] = None) -> Dict[str, Any]: """Test if the provided or stored Gemini API key is valid with automatic model fallback.""" key = (api_key or self.api_key).strip() if not key: return {"success": False, "error": "API Key belum diisi."} # Try models starting with current model_name then candidates models_to_try = [self.model_name] + [m for m in self.CANDIDATE_MODELS if m != self.model_name] last_error = "" # First try legacy google.generativeai if available if HAS_LEGACY_GENAI: try: legacy_genai.configure(api_key=key) for m_name in models_to_try: try: model = legacy_genai.GenerativeModel(m_name) res = model.generate_content("Hello! Respond with: 'OK'") self.model_name = m_name return { "success": True, "message": f"Koneksi Gemini AI Berhasil! (Model: {m_name})", "model": m_name, "response": res.text.strip() } except Exception as me: last_error = str(me) continue except Exception as e: last_error = str(e) # Second try modern google.genai if available if HAS_NEW_GENAI: try: client = genai.Client(api_key=key) for m_name in models_to_try: try: response = client.models.generate_content( model=m_name, contents="Hello! Respond with: 'OK'" ) self.model_name = m_name return { "success": True, "message": f"Koneksi Gemini AI Berhasil! (Model: {m_name})", "model": m_name, "response": response.text.strip() } except Exception as me: last_error = str(me) continue except Exception as e: if not last_error: last_error = str(e) if not HAS_LEGACY_GENAI and not HAS_NEW_GENAI: return {"success": False, "error": "Library Google Gemini SDK (google-generativeai / google-genai) tidak ditemukan."} return {"success": False, "error": f"Gagal menghubungkan ke Gemini: {last_error}"} def _call_gemini_with_retry(self, prompt: str, max_retries: int = 3) -> str: """Call Gemini API with automatic model candidate fallback and rate limit backoff.""" if not self.api_key: raise ValueError("GEMINI_API_KEY belum dikonfigurasi.") delay = 4.0 # safe interval for Free Tier models_to_try = [self.model_name] + [m for m in self.CANDIDATE_MODELS if m != self.model_name] last_error = None for attempt in range(max_retries): # Try with legacy_genai (installed and primary in this environment) if HAS_LEGACY_GENAI: try: legacy_genai.configure(api_key=self.api_key) for m_name in models_to_try: try: model = legacy_genai.GenerativeModel(m_name) res = model.generate_content(prompt) self.model_name = m_name time.sleep(delay) return res.text except Exception as me: err_text = str(me).lower() last_error = me # If model not available (404), quota (429), or server capacity (503), try next candidate model! if any(k in err_text for k in ["404", "not found", "no longer available", "429", "quota", "resource_exhausted", "limit", "503", "unavailable", "capacity", "overloaded", "internal"]): continue raise me except Exception as e_leg: last_error = e_leg # Try with new google.genai if available if HAS_NEW_GENAI: try: client = genai.Client(api_key=self.api_key) for m_name in models_to_try: try: res = client.models.generate_content( model=m_name, contents=prompt ) self.model_name = m_name time.sleep(delay) return res.text except Exception as me: err_text = str(me).lower() last_error = me if any(k in err_text for k in ["404", "not found", "no longer available", "429", "quota", "resource_exhausted", "limit", "503", "unavailable", "capacity", "overloaded", "internal"]): continue raise me except Exception as e_new: last_error = e_new # If all candidates exhausted this attempt, back off and retry if attempt < max_retries - 1: wait_time = (2 ** (attempt + 1)) * 3 + random.uniform(1.0, 3.0) print(f"[GEMINI RETRY] Backing off for {wait_time:.1f}s (percobaan {attempt+1}/{max_retries})...") time.sleep(wait_time) if last_error: raise last_error raise RuntimeError("Gagal memanggil Gemini setelah beberapa kali percobaan.") def _draw_vision_annotation( self, img: Image.Image, coords: Optional[Dict[str, float]], bbox: Optional[Dict[str, float]], label: str = "" ) -> Image.Image: """ Draws precise spatial annotation overlay (bounding box, red target crosshair, and high-contrast badge) on top of the original screenshot. """ annotated = img.copy().convert("RGBA") overlay = Image.new("RGBA", annotated.size, (0, 0, 0, 0)) draw = ImageDraw.Draw(overlay) w, h = annotated.size # 1. Bounding box overlay (if provided) if bbox and isinstance(bbox, dict): try: ymin = max(0.0, min(100.0, float(bbox.get("ymin", 0)))) xmin = max(0.0, min(100.0, float(bbox.get("xmin", 0)))) ymax = max(0.0, min(100.0, float(bbox.get("ymax", 0)))) xmax = max(0.0, min(100.0, float(bbox.get("xmax", 0)))) box_x1 = int(w * (xmin / 100.0)) box_y1 = int(h * (ymin / 100.0)) box_x2 = int(w * (xmax / 100.0)) box_y2 = int(h * (ymax / 100.0)) if box_x2 > box_x1 and box_y2 > box_y1: draw.rectangle( [box_x1, box_y1, box_x2, box_y2], fill=(239, 68, 68, 35), outline=(239, 68, 68, 230), width=3 ) except Exception as e_box: print(f"[WARN] Bounding box draw error: {e_box}") # 2. Target Crosshair & Concentric Rings (if coordinates provided) if coords and isinstance(coords, dict) and "x" in coords and "y" in coords: try: cx = int(w * (float(coords["x"]) / 100.0)) cy = int(h * (float(coords["y"]) / 100.0)) # Outer soft ring draw.ellipse([cx - 28, cy - 28, cx + 28, cy + 28], outline=(239, 68, 68, 120), width=6) # Outer crisp ring draw.ellipse([cx - 20, cy - 20, cx + 20, cy + 20], outline=(255, 255, 255, 240), width=2) # Primary red ring draw.ellipse([cx - 16, cy - 16, cx + 16, cy + 16], outline=(239, 68, 68, 255), width=3) # Center white bullseye draw.ellipse([cx - 4, cy - 4, cx + 4, cy + 4], fill=(255, 255, 255, 255), outline=(239, 68, 68, 255), width=2) # Crosshair ticks draw.line([cx - 36, cy, cx - 10, cy], fill=(239, 68, 68, 255), width=2) draw.line([cx + 10, cy, cx + 36, cy], fill=(239, 68, 68, 255), width=2) draw.line([cx, cy - 36, cx, cy - 10], fill=(239, 68, 68, 255), width=2) draw.line([cx, cy + 10, cx, cy + 36], fill=(239, 68, 68, 255), width=2) # Target Badge clean_label = str(label or "Target").replace("\n", " ").strip() badge_text = f"🎯 {clean_label} ({coords['x']:.1f}%, {coords['y']:.1f}%)" badge_w = len(badge_text) * 7 + 16 badge_h = 22 badge_x = min(w - badge_w - 10, max(10, cx - (badge_w // 2))) badge_y = cy - 48 if cy > 55 else cy + 32 draw.rectangle( [badge_x, badge_y, badge_x + badge_w, badge_y + badge_h], fill=(220, 38, 38, 230), outline=(255, 255, 255, 210), width=1 ) draw.text((badge_x + 8, badge_y + 4), badge_text, fill=(255, 255, 255, 255)) except Exception as e_draw: print(f"[WARN] Target crosshair draw error: {e_draw}") return Image.alpha_composite(annotated, overlay).convert("RGB") def _record_vision_history(self, record: Dict[str, Any]): """Append record to vision_history.json thread-safely.""" try: history = [] if self.vision_history_file.exists(): try: with open(self.vision_history_file, "r", encoding="utf-8") as f: history = json.load(f) except Exception: history = [] history.insert(0, record) with open(self.vision_history_file, "w", encoding="utf-8") as f: json.dump(history, f, indent=2, ensure_ascii=False) except Exception as e: print(f"[WARN] Failed to record vision history: {e}") def _prune_vision_debug_history(self, max_keep: int = 25): """ Enforce rolling file cleanup to prevent filling drive C disk space. Keeps only the most recent max_keep sessions and deletes older image files. """ try: if not self.vision_history_file.exists(): return with open(self.vision_history_file, "r", encoding="utf-8") as f: history = json.load(f) if len(history) > max_keep: keep_records = history[:max_keep] prune_records = history[max_keep:] for r in prune_records: raw_name = Path(r.get("raw_image_url", "")).name ann_name = Path(r.get("annotated_image_url", "")).name if raw_name: raw_p = self.vision_debug_dir / raw_name if raw_p.exists() and raw_p.is_file(): try: raw_p.unlink() except Exception: pass if ann_name: ann_p = self.vision_debug_dir / ann_name if ann_p.exists() and ann_p.is_file(): try: ann_p.unlink() except Exception: pass with open(self.vision_history_file, "w", encoding="utf-8") as f: json.dump(keep_records, f, indent=2, ensure_ascii=False) except Exception as pe: print(f"[WARN] Failed to prune vision debug files: {pe}") def get_vision_history(self, limit: int = 20) -> list: """Returns recent vision analysis debug records.""" if not self.vision_history_file.exists(): return [] try: with open(self.vision_history_file, "r", encoding="utf-8") as f: history = json.load(f) return history[:limit] except Exception: return [] def clear_vision_history(self) -> bool: """Deletes all debug screenshots and clears history.""" try: if self.vision_debug_dir.exists(): for p in self.vision_debug_dir.glob("*.png"): try: p.unlink() except Exception: pass if self.vision_history_file.exists(): self.vision_history_file.unlink() return True except Exception as ce: print(f"[WARN] Failed to clear vision history: {ce}") return False def record_vision_success(self, domain: str, step_name: str, target_text: str, coordinates: Optional[Dict[str, float]] = None): """ Record verified successful vision detection into site_knowledge.json so future runs can prioritize this proven location and knowledge. """ clean_d = domain.lower().replace("www.", "") if clean_d not in self.knowledge: self.knowledge[clean_d] = {} if "learned_vision_steps" not in self.knowledge[clean_d]: self.knowledge[clean_d]["learned_vision_steps"] = {} self.knowledge[clean_d]["learned_vision_steps"][step_name] = { "target_text": target_text, "coordinates_percent": coordinates, "updated_at": time.time() } self._save_knowledge() def analyze_screenshot_with_vision( self, screenshot_bytes: bytes, question_or_prompt: str, max_retries: int = 2, source_tag: str = "automation" ) -> Dict[str, Any]: """ Multimodal AI Vision with Spatial Target Localization: Analyzes full screenshot image of the active page, saves debug artifacts, determines exact (x, y) coordinates percent and bounding boxes, and returns transparent structured guidance for Playwright and human inspector. """ if not self.api_key: return {"success": False, "error": "GEMINI_API_KEY belum dikonfigurasi"} # Generate unique session ID for debug artifacts session_id = f"{int(time.time())}_{random.randint(1000, 9999)}" raw_filename = f"raw_{session_id}.png" annotated_filename = f"annotated_{session_id}.png" raw_filepath = self.vision_debug_dir / raw_filename annotated_filepath = self.vision_debug_dir / annotated_filename # 1. Save raw unedited screenshot directly to disk try: with open(raw_filepath, "wb") as f: f.write(screenshot_bytes) except Exception as se: print(f"[WARN] Failed to write raw vision screenshot: {se}") # Open image to get dimensions img = Image.open(io.BytesIO(screenshot_bytes)) img_w, img_h = img.size vision_instruction = f""" You are an expert web automation vision assistant analyzing a screenshot of a live social media web interface. Task / Question: {question_or_prompt} Examine the full screenshot with extreme precision and respond STRICTLY in valid JSON with these exact keys: {{ "page_state": "e.g. normal_feed, create_post_modal, pin_builder, loading, login_required, upload_in_progress, done_or_success, or unknown", "is_modal_open": true or false, "is_page_loading": true or false, "roadblock_type": "one of: 'none', 'login_required', 'captcha_detected', 'security_checkpoint', 'overlay_popup', 'media_uploading', 'button_disabled', or 'unknown'", "action_needed": "e.g. click_target, wait_patiently, type_caption, upload_file, close_modal, solve_roadblock, or none", "target_element_text": "exact label or text on the button/element (e.g. 'Post', 'Kirim', 'Publish', 'Simpan', 'Share', 'Next', 'Berikutnya', 'What\\'s on your mind') or null", "target_element_type": "e.g. button, input, textarea, modal_close, file_dropzone, icon, or null", "coordinates_percent": {{ "x": 88.5, "y": 92.0 }}, "bounding_box_percent": {{ "ymin": 89.0, "xmin": 84.0, "ymax": 95.0, "xmax": 93.0 }}, "visual_location": "approximate location on screen (e.g. 'bottom-right of modal popup', 'top-right header', 'center')", "feedback": "2-3 comprehensive sentences in Indonesian explaining: 1) What page/modal is currently open, 2) Where the target element was found and why it should be clicked, 3) Any potential blocker like unselected boards or missing text.", "notes": "practical tip for Playwright automation execution" }} IMPORTANT RULES FOR ROADBLOCKS & COORDINATES: - If the page requires login, 2FA, or shows a captcha, set roadblock_type accordingly and action_needed to 'solve_roadblock'. - If the target button is visible but grayed out / disabled, set roadblock_type to 'button_disabled' and action_needed to 'wait_patiently'. - `coordinates_percent.x` MUST be a number between 0.0 and 100.0 measuring horizontal position from left edge to the center of the target element. - `coordinates_percent.y` MUST be a number between 0.0 and 100.0 measuring vertical position from top edge to the center of the target element. - If no specific interactive element needs clicking, set coordinates_percent and bounding_box_percent to null. - Provide honest, accurate locations. Do not guess if element is not visible on screen. """ models_to_try = [self.model_name] + [m for m in self.CANDIDATE_MODELS if m != self.model_name] parsed = None raw_text = "" for attempt in range(max_retries): # 1. Try legacy google.generativeai with PIL if HAS_LEGACY_GENAI: try: legacy_genai.configure(api_key=self.api_key) for m_name in models_to_try: try: model = legacy_genai.GenerativeModel(m_name) res = model.generate_content([vision_instruction, img]) self.model_name = m_name raw_text = res.text.strip() clean_raw = raw_text if clean_raw.startswith("```json"): clean_raw = clean_raw[7:] if clean_raw.startswith("```"): clean_raw = clean_raw[3:] if clean_raw.endswith("```"): clean_raw = clean_raw[:-3] parsed = json.loads(clean_raw.strip()) break except Exception as me: err = str(me).lower() if any(k in err for k in ["404", "not found", "no longer available", "429", "quota", "resource_exhausted", "limit", "503", "unavailable", "capacity", "overloaded", "internal"]): continue raise me if parsed: break except Exception as ge_leg: print(f"[VISION RETRY {attempt+1}] Catatan visi AI: {ge_leg}") # 2. Try modern google.genai if available if not parsed and HAS_NEW_GENAI: try: client = genai.Client(api_key=self.api_key) image_part = genai_types.Part.from_bytes( data=screenshot_bytes, mime_type="image/png" ) for m_name in models_to_try: try: res = client.models.generate_content( model=m_name, contents=[vision_instruction, image_part] ) self.model_name = m_name raw_text = res.text.strip() clean_raw = raw_text if clean_raw.startswith("```json"): clean_raw = clean_raw[7:] if clean_raw.startswith("```"): clean_raw = clean_raw[3:] if clean_raw.endswith("```"): clean_raw = clean_raw[:-3] parsed = json.loads(clean_raw.strip()) break except Exception as me: err = str(me).lower() if any(k in err for k in ["404", "not found", "no longer available", "429", "quota", "resource_exhausted", "limit", "503", "unavailable", "capacity", "overloaded", "internal"]): continue raise me if parsed: break except Exception as ge_new: print(f"[VISION RETRY {attempt+1}] Catatan visi AI (genai): {ge_new}") time.sleep(2) if not parsed: parsed = { "page_state": "unknown", "is_modal_open": False, "action_needed": "fallback_to_standard", "target_element_text": None, "target_element_type": None, "coordinates_percent": None, "bounding_box_percent": None, "visual_location": None, "feedback": "Vision AI belum dapat menganalisis gambar atau respon tidak valid.", "notes": "Vision processing timed out or failed to parse." } parsed["success"] = True if parsed.get("action_needed") != "fallback_to_standard" else False # Extract coordinates and draw visual annotations coords = parsed.get("coordinates_percent") bbox = parsed.get("bounding_box_percent") target_text = parsed.get("target_element_text") or parsed.get("action_needed") or "Target" try: annotated_img = self._draw_vision_annotation(img, coords, bbox, label=target_text) annotated_img.save(annotated_filepath, format="PNG") except Exception as ae: print(f"[WARN] Failed to draw vision annotation: {ae}") try: img.save(annotated_filepath, format="PNG") except Exception: pass # Calculate exact pixel coordinates on the captured image pixel_x = None pixel_y = None if coords and isinstance(coords, dict) and "x" in coords and "y" in coords: try: pixel_x = int(img_w * (float(coords["x"]) / 100.0)) pixel_y = int(img_h * (float(coords["y"]) / 100.0)) except Exception: pass record = { "id": session_id, "timestamp": time.time(), "timestamp_str": time.strftime("%Y-%m-%d %H:%M:%S"), "source": source_tag, "prompt": question_or_prompt, "model": self.model_name, "image_width": img_w, "image_height": img_h, "raw_image_url": f"/static/vision_debug/{raw_filename}", "annotated_image_url": f"/static/vision_debug/{annotated_filename}", "coordinates_percent": coords, "bounding_box_percent": bbox, "pixel_coordinates": {"x": pixel_x, "y": pixel_y} if pixel_x is not None else None, "parsed": parsed, "raw_response": raw_text } # Save session to persistent JSON and prune old records self._record_vision_history(record) self._prune_vision_debug_history(max_keep=25) # Enrich parsed result with debug metadata parsed["session_id"] = session_id parsed["raw_image_url"] = record["raw_image_url"] parsed["annotated_image_url"] = record["annotated_image_url"] parsed["pixel_coordinates"] = record["pixel_coordinates"] parsed["image_dimensions"] = {"width": img_w, "height": img_h} parsed["raw_response"] = raw_text return parsed def analyze_posting_interface(self, domain: str, condensed_dom: Dict[str, Any], force_refresh: bool = False) -> Dict[str, Any]: """ Analyze the condensed DOM of a social media page to locate key posting selectors. Returns cached result if available. """ # Return cached knowledge if available and not forced to refresh if not force_refresh and domain in self.knowledge: return self.knowledge[domain] prompt = f""" You are an expert browser automation engineer. Analyze the following condensed DOM elements from social media site '{domain}' (Title: {condensed_dom.get('title')}, URL: {condensed_dom.get('url')}). Identify the best CSS selectors or element locators to perform a marketing post (image flyer + caption). Condensed Elements: {json.dumps(condensed_dom.get('elements', []), indent=1)} Respond ONLY in valid JSON with these exact keys: {{ "is_logged_in": true/false (does the user appear to be already logged in, or is there a login wall?), "modal_dismiss_selector": "selector or text to close any initial pop-up modal like notifications/cookies, or null", "create_post_button": "selector or text to open the create post/tweet/pin modal/screen", "file_input_selector": "selector for file upload input (usually input[type=file] or dropzone)", "caption_input_selector": "selector for text/caption/description area (often div[contenteditable=true] or textarea)", "submit_button": "selector or text to submit/publish the post", "share_link_strategy": "hint on how to retrieve the published post URL (e.g. current_url, click_last_post, read_toast)" }} """ try: raw_response = self._call_gemini_with_retry(prompt) # Extract JSON clean_json = raw_response.strip() if clean_json.startswith("```json"): clean_json = clean_json[7:] if clean_json.startswith("```"): clean_json = clean_json[3:] if clean_json.endswith("```"): clean_json = clean_json[:-3] clean_json = clean_json.strip() result = json.loads(clean_json) # Store in knowledge cache self.knowledge[domain] = result self._save_knowledge() return result except Exception as e: print(f"[GEMINI DOM ERROR] Failed to analyze DOM for {domain}: {e}") return { "is_logged_in": True, "modal_dismiss_selector": None, "create_post_button": None, "file_input_selector": "input[type='file']", "caption_input_selector": "textarea, [contenteditable='true']", "submit_button": "button:has-text('Post'), button:has-text('Publish'), button:has-text('Tweet')", "share_link_strategy": "current_url", "error": str(e) } def generate_social_caption(self, title: str, description: str, platform_name: str, target_link: Optional[str] = None) -> str: """ Adapts the flyer marketing title and description into an engaging, platform-optimized social media caption with suitable hashtags and tone. """ if not self.api_key: # Fallback formatted caption if API key not provided yet hashtags = "#marketing #bisnis #promo #viral #digitalmarketing" link_text = f"\n\nInfo lengkap: {target_link}" if target_link else "" return f"🔥 {title} 🔥\n\n{description}{link_text}\n\n{hashtags}" prompt = f""" You are a top-tier Social Media Copywriter and Growth Marketer. Generate a captivating, high-converting social media caption for the following marketing campaign flyer specifically tailored for: **{platform_name}**. Campaign Details: - Title: {title} - Description: {description} - Call-to-action Link: {target_link or 'Link in bio'} Platform Guidelines: - If Instagram: Engaging hook, readable paragraphs with emojis, strong call to action, 8-12 targeted hashtags. - If Twitter/X: Concise, punchy, within 250 characters, 2-3 trending hashtags, compelling link placement. - If Pinterest: Catchy Pin title recommendation followed by search-optimized descriptive text with keywords. - If Facebook: Friendly conversational tone, clear value proposition, emojis, and direct call to action. - If TikTok: Very catchy hook in the first sentence, conversational short text, 4-6 hashtags. - Other: Clean professional marketing copy with emojis and hashtags. Output ONLY the final caption text ready to copy-paste. Do not include introductory notes or meta explanations. """ try: caption = self._call_gemini_with_retry(prompt) return caption.strip() except Exception as e: print(f"[GEMINI CAPTION ERROR] Fallback caption used due to: {e}") hashtags = "#marketing #bisnis #promo #viral #digitalmarketing" link_text = f"\n\nInfo lengkap: {target_link}" if target_link else "" return f"{title}\n\n{description}{link_text}\n\n{hashtags}" # Pre-built baseline guides for major platforms BASE_PLATFORM_GUIDES = { "pinterest.com": { "platform_name": "Pinterest", "domain": "pinterest.com", "direct_upload_url": "https://www.pinterest.com/pin-builder/", "login_indicators": ["input[name='id']", "a[href*='/login']", "button[data-test-id='simple-login-button']"], "dismiss_modal_selectors": ["button:has-text('Not Now')", "button:has-text('Cancel')"], "create_post_button_selectors": ["button[data-test-id='pin-builder-button']", "a[href*='pin-builder']"], "file_input_selectors": ["input[type='file']", "div[data-test-id='media-upload']"], "title_input_selectors": ["input[placeholder*='Add a title']", "input[placeholder*='Tambahkan judul']", "textarea[placeholder*='title']", "input[id*='pin-draft-title']"], "caption_input_selectors": ["div[placeholder*='Tell everyone']", "textarea[placeholder*='Tell everyone']", "div[role='textbox']", "div[contenteditable='true']"], "link_input_selectors": ["input[placeholder*='Add a link']", "input[placeholder*='Tambahkan tautan']", "input[id*='pin-draft-link']"], "submit_button_selectors": ["button[data-test-id='board-dropdown-save-button']", "button:has-text('Publish')", "button:has-text('Terbitkan')", "button:has-text('Save')"], "share_link_guide": { "strategy": "read_toast_or_view_pin", "selectors": ["a:has-text('See your Pin')", "a:has-text('Lihat Pin Anda')", "div[role='alert'] a"], "fallback_profile_url": "https://www.pinterest.com/" } }, "instagram.com": { "platform_name": "Instagram", "domain": "instagram.com", "direct_upload_url": "https://www.instagram.com/", "login_indicators": ["input[name='username']", "a[href*='/accounts/login/']"], "dismiss_modal_selectors": ["button:has-text('Not Now')", "button:has-text('Lain Kali')", "button:has-text('Cancel')"], "create_post_button_selectors": ["svg[aria-label='New post']", "svg[aria-label='Postingan baru']", "span:has-text('Create')", "span:has-text('Buat')"], "file_input_selectors": ["input[type='file']", "button:has-text('Select from computer')"], "title_input_selectors": [], "caption_input_selectors": ["div[aria-label*='Write a caption']", "div[aria-label*='Tulis keterangan']", "div[contenteditable='true']"], "submit_button_selectors": ["div[role='button']:has-text('Share')", "button:has-text('Share')", "div[role='button']:has-text('Bagikan')"], "share_link_guide": { "strategy": "read_profile_latest_or_current", "selectors": ["article a[href*='/p/']", "a[href*='/p/']"], "fallback_profile_url": "https://www.instagram.com/" } }, "facebook.com": { "platform_name": "Facebook", "domain": "facebook.com", "direct_upload_url": "https://www.facebook.com/", "login_indicators": ["input[name='email']", "button[name='login']"], "dismiss_modal_selectors": ["div[aria-label='Close']", "button:has-text('Not Now')"], "create_post_button_selectors": ["div[role='button']:has-text(\"What's on your mind\")", "div[role='button']:has-text('Apa yang Anda pikirkan')", "span:has-text('Photo/video')"], "file_input_selectors": ["input[type='file']", "div[aria-label*='Photo/video']"], "title_input_selectors": [], "caption_input_selectors": ["div[role='textbox']", "div[contenteditable='true']", "div[aria-label*='What\'s on your mind']"], "submit_button_selectors": ["div[aria-label='Post']", "div[aria-label='Kirim']", "div[role='button']:has-text('Post')"], "share_link_guide": { "strategy": "read_last_post_link", "selectors": ["a[href*='/posts/']", "a[href*='story.php']"], "fallback_profile_url": "https://www.facebook.com/me" } }, "twitter.com": { "platform_name": "Twitter / X", "domain": "twitter.com", "direct_upload_url": "https://x.com/compose/post", "login_indicators": ["a[href*='/login']", "span:has-text('Sign in to X')"], "dismiss_modal_selectors": ["div[aria-label='Close']", "button:has-text('Not now')"], "create_post_button_selectors": ["a[data-testid='SideNav_NewTweet_Button']", "div[aria-label='Post']"], "file_input_selectors": ["input[data-testid='fileInput']", "input[type='file']"], "title_input_selectors": [], "caption_input_selectors": ["div[data-testid='tweetTextarea_0']", "div[role='textbox']"], "submit_button_selectors": ["button[data-testid='tweetButton']", "button[data-testid='tweetButtonInline']", "button:has-text('Post')"], "share_link_guide": { "strategy": "read_toast_or_user_status", "selectors": ["a[href*='/status/']", "div[data-testid='toast'] a"], "fallback_profile_url": "https://x.com/" } }, "tiktok.com": { "platform_name": "TikTok", "domain": "tiktok.com", "direct_upload_url": "https://www.tiktok.com/creator-center/upload", "login_indicators": ["button:has-text('Log in')", "a[href*='login']"], "dismiss_modal_selectors": ["div[role='dialog'] button:has-text('Cancel')"], "create_post_button_selectors": ["a[href*='upload']", "button:has-text('Upload')"], "file_input_selectors": ["input[type='file']", "iframe[src*='upload'] input[type='file']"], "title_input_selectors": [], "caption_input_selectors": ["div[contenteditable='true']", "div.DraftEditor-editorContainer", "textarea"], "submit_button_selectors": ["button:has-text('Post')", "button:has-text('Publish')"], "share_link_guide": { "strategy": "wait_manage_posts", "selectors": ["a:has-text('Manage your posts')", "a:has-text('View profile')"], "fallback_profile_url": "https://www.tiktok.com/" } }, "reddit.com": { "platform_name": "Reddit", "domain": "reddit.com", "direct_upload_url": "https://www.reddit.com/submit", "login_indicators": ["a[href*='login']", "button:has-text('Log In')"], "dismiss_modal_selectors": ["button:has-text('Close')"], "create_post_button_selectors": ["a[href*='/submit']", "button:has-text('Create')"], "file_input_selectors": ["input[type='file']", "button:has-text('Images & Video')"], "title_input_selectors": ["textarea[placeholder*='Title']", "input[placeholder*='Title']"], "caption_input_selectors": ["div[contenteditable='true']", "textarea[placeholder*='Text']"], "submit_button_selectors": ["button:has-text('Post')"], "share_link_guide": { "strategy": "current_url_or_post_link", "selectors": ["a[data-click-id='body']", "h1 a"], "fallback_profile_url": "https://www.reddit.com/user/" } }, "threads.net": { "platform_name": "Threads", "domain": "threads.net", "direct_upload_url": "https://www.threads.net/", "login_indicators": ["a[href*='login']", "button:has-text('Log in')"], "dismiss_modal_selectors": ["button:has-text('Not now')"], "create_post_button_selectors": ["svg[aria-label='Create']", "div[role='button']:has-text('Start a thread')"], "file_input_selectors": ["input[type='file']"], "title_input_selectors": [], "caption_input_selectors": ["div[contenteditable='true']", "div[role='textbox']"], "submit_button_selectors": ["div[role='button']:has-text('Post')", "button:has-text('Post')"], "share_link_guide": { "strategy": "read_toast_or_user_post", "selectors": ["a[href*='/post/']", "div[role='alert'] a"], "fallback_profile_url": "https://www.threads.net/" } } } def get_or_research_platform_guide(self, domain: str, platform_name: str, force_refresh: bool = False) -> Dict[str, Any]: """ Retrieves the structured upload & share link guide from JSON knowledge. If not yet researched or force_refresh is True, prompts Gemini AI to research the exact workflow and persists the result to site_knowledge.json. """ d = domain.lower().replace("www.", "") # Check existing cached guide in knowledge if not force_refresh and d in self.knowledge and "direct_upload_url" in self.knowledge[d]: return self.knowledge[d] # Use Gemini AI to research if API key is configured if self.api_key: research_prompt = f""" You are an expert Social Media Automation Engineer and Playwright Specialist. Generate a comprehensive technical web posting & share-link extraction workflow guide for the social media platform: **"{platform_name}"** (domain: {domain}). Analyze the 2026 desktop web interface for this platform and provide the exact instructions. Respond STRICTLY in valid JSON matching this exact structure: {{ "platform_name": "{platform_name}", "domain": "{d}", "direct_upload_url": "Direct creation URL e.g. https://www.pinterest.com/pin-builder/ or https://x.com/compose/post or https://{d}/", "login_indicators": ["CSS selector 1 that shows user is not logged in", "CSS selector 2"], "dismiss_modal_selectors": ["button:has-text('Not Now')", "button[aria-label='Close']"], "create_post_button_selectors": ["button or link selectors to start a new post if not at direct URL"], "file_input_selectors": ["input[type='file']", "button or area to trigger upload"], "title_input_selectors": ["input or textarea selector for post/pin title, or empty list if none"], "caption_input_selectors": ["div[contenteditable='true']", "textarea", "description input selector"], "link_input_selectors": ["input for website link if applicable e.g. Pinterest, or empty list"], "submit_button_selectors": ["button:has-text('Post')", "button:has-text('Publish')", "button:has-text('Share')"], "share_link_guide": {{ "strategy": "read_toast_link, read_profile_first_post, or current_url", "selectors": ["CSS selector to click or read published post link"], "fallback_profile_url": "https://{d}/" }} }} """ try: print(f"[GEMINI RESEARCH] Meriset panduan upload & share untuk {platform_name} ({d})...") raw = self._call_gemini_with_retry(research_prompt) clean = raw.strip() if clean.startswith("```json"): clean = clean[7:] if clean.startswith("```"): clean = clean[3:] if clean.endswith("```"): clean = clean[:-3] guide = json.loads(clean.strip()) guide["researched_at"] = time.strftime("%Y-%m-%d %H:%M:%S") guide["source"] = "gemini_ai" # Save to knowledge JSON self.knowledge[d] = guide self._save_knowledge() print(f"[GEMINI RESEARCH] Berhasil menyimpan panduan untuk {d} ke site_knowledge.json!") return guide except Exception as e: print(f"[GEMINI RESEARCH ERROR] Gagal meriset via AI ({e}). Menggunakan panduan dasar.") # Fallback to curated base guides if d in self.BASE_PLATFORM_GUIDES: guide = dict(self.BASE_PLATFORM_GUIDES[d]) guide["source"] = "built_in" else: # Generic fallback guide = { "platform_name": platform_name, "domain": d, "direct_upload_url": f"https://{d}/", "login_indicators": ["a[href*='login']", "input[type='password']"], "dismiss_modal_selectors": ["button:has-text('Not Now')", "button:has-text('Close')"], "create_post_button_selectors": ["button:has-text('Create')", "button:has-text('Post')", "a[href*='submit']"], "file_input_selectors": ["input[type='file']"], "title_input_selectors": ["input[placeholder*='title' i]"], "caption_input_selectors": ["div[contenteditable='true']", "textarea"], "link_input_selectors": [], "submit_button_selectors": ["button:has-text('Post')", "button:has-text('Publish')", "button:has-text('Share')"], "share_link_guide": { "strategy": "current_url", "selectors": ["article a", "div[role='alert'] a"], "fallback_profile_url": f"https://{d}/" }, "source": "generic_fallback" } self.knowledge[d] = guide self._save_knowledge() return guide