import time import os from typing import Dict, Any, Optional from automation.platform_handlers.base_handler import BasePlatformHandler class TwitterHandler(BasePlatformHandler): PLATFORM_NAME = "Twitter / X" BASE_URL = "https://x.com/compose/post" def phase_upload(self, page, flyer_path: str, guide: Optional[Dict[str, Any]] = None) -> bool: """ Sesi 1: Upload Media Membuka halaman Compose Post Twitter/X, memeriksa login, dan mengunggah flyer. """ upload_url = (guide and guide.get("direct_upload_url")) or self.BASE_URL self.log(f"Membuka halaman Compose Post Twitter ({upload_url})...") page.goto(upload_url, wait_until="domcontentloaded", timeout=45000) time.sleep(3) self.dismiss_common_modals(page) self.setup_clipboard_interceptor(page) # Cek apakah akun membutuhkan login if self.check_login_required(page, guide): self.log("Twitter/X meminta login. Silakan login akun terlebih dahulu di jendela Chrome Bot!", "ERROR") return False # Upload Flyer Media self.log("Mengunggah media flyer ke Twitter/X...") abs_flyer = os.path.abspath(flyer_path) file_uploaded = False file_inputs = page.locator("input[data-testid='fileInput'], input[type='file'][accept*='image'], input[type='file']").all() for finp in file_inputs: try: finp.set_input_files(abs_flyer) file_uploaded = True self.log("Media flyer berhasil dipasang ke input file Twitter/X.") time.sleep(3) break except Exception: continue if not file_uploaded and self.brain: self.log("Input file belum ditemukan via CSS. Memanggil AI Vision...", "WARN") v_res = self.consult_vision(page, "Cari tombol atau ikon galeri/media untuk mengunggah gambar di Twitter/X.") if v_res.get("coordinates_percent"): self.click_visual_target(page, v_res, step_name="twitter_media_btn") time.sleep(1.5) try: page.locator("input[type='file']").first.set_input_files(abs_flyer) file_uploaded = True except Exception: pass # Verifikasi lampiran media is_verified = False for _ in range(4): thumb = page.locator("div[data-testid='attachments'], div[aria-label*='Media'], img[src*='blob:'], div[data-testid='previewImage']").first if thumb.is_visible(timeout=1500): is_verified = True self.log("Verifikasi Sesi 1 Sukses: Media flyer terpasang di kotak tweet!", "SUCCESS") break time.sleep(1.5) return is_verified or file_uploaded def phase_metadata(self, page, title: str, caption: str, target_link: Optional[str] = None, guide: Optional[Dict[str, Any]] = None) -> bool: """ Sesi 2: Content & Metadata Mengetik isi Tweet dengan memperhatikan batas 280 karakter. """ self.log("Mengetik isi Tweet di Twitter/X...") tweet_text = caption if target_link and target_link not in tweet_text: tweet_text = f"{caption}\n\n{target_link}" # Jika melebihi batas 280 karakter, ringkas otomatis if len(tweet_text) > 275: tweet_text = f"{title}\n\n{target_link or ''}\n#promo #bisnis".strip() if len(tweet_text) > 275: tweet_text = tweet_text[:270] textarea = page.locator("div[data-testid='tweetTextarea_0'], div[role='textbox'][contenteditable='true']").first if textarea.is_visible(timeout=5000): textarea.click() time.sleep(0.5) page.keyboard.type(tweet_text, delay=10) time.sleep(1.5) self.log("Sesi 2 Sukses: Teks tweet berhasil dimasukkan.", "SUCCESS") return True else: self.log("Kolom tweet textarea tidak ditemukan.", "ERROR") return False def phase_publish(self, page, guide: Optional[Dict[str, Any]] = None) -> bool: """ Sesi 3: Publish Menekan tombol Post / Tweet dan menunggu postingan terkirim. """ self.log("Menekan tombol Post / Tweet...") post_btn = page.locator("button[data-testid='tweetButton'], button[data-testid='tweetButtonInline'], button:has-text('Post')").first submitted = False if post_btn.is_visible(timeout=5000): post_btn.click() submitted = True elif self.brain: self.log("Tombol Post Twitter tidak ditemukan via CSS. Memanggil AI Vision...", "WARN") v_res = self.consult_vision(page, "Cari tombol biru 'Post' di pojok kanan bawah editor Twitter/X.") if v_res.get("coordinates_percent"): submitted = self.click_visual_target(page, v_res, step_name="twitter_post_btn") if not submitted: return False self.log("Menunggu tweet terkirim di server X...") time.sleep(6) self.dismiss_common_modals(page) return True def phase_extract_link(self, page, title: str, guide: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """ Sesi 4: Ekstraksi Tautan Postingan Mendeteksi URL tweet spesifik (/status/...) dari toast, tombol share, atau clipboard interceptor. """ self.log("Mendeteksi tautan postingan Twitter/X...") post_url = None # 1. Cek clipboard interceptor c = self.get_intercepted_clipboard(page) if c and "/status/" in c: post_url = c self.log(f"Tautan postingan terdeteksi dari clipboard interceptor: {post_url}", "SUCCESS") # 2. Cek View tweet toast link if not post_url: try: view_toast = page.locator("a[href*='/status/']").first if view_toast.is_visible(timeout=3000): href = view_toast.get_attribute("href") if href and "/status/" in href: post_url = f"https://x.com{href}" if not href.startswith("http") else href self.log(f"Tautan tweet ditemukan dari toast: {post_url}", "SUCCESS") except Exception: pass # 3. Klik tombol Share post di tweet terbaru if not post_url: try: share_btn = page.locator("button[aria-label='Share post'], button[aria-label='Bagikan postingan']").first if share_btn.is_visible(timeout=2500): share_btn.click() time.sleep(1.2) copy_opt = page.locator("div[role='menuitem']:has-text('Copy link'), span:has-text('Copy link')").first if copy_opt.is_visible(timeout=2000): copy_opt.click() time.sleep(1.0) cb = self.get_intercepted_clipboard(page) if cb and "/status/" in cb: post_url = cb self.log(f"Tautan tweet berhasil disalin dari menu Share: {post_url}", "SUCCESS") except Exception as se: self.log(f"Tombol share Twitter: {se}", "WARN") # 4. Fallback jika URL browser mengarah ke status if not post_url and "/status/" in page.url: post_url = page.url self.log(f"URL browser mengarah ke tweet: {post_url}", "SUCCESS") if not post_url or "/status/" not in post_url: self.log("Tautan postingan spesifik Twitter (/status/...) tidak ditemukan. Tidak mencatat link palsu.", "ERROR") return { "success": False, "post_url": None, "message": "FAILED: Tweet terkirim tetapi tautan spesifik (/status/...) tidak dapat disalin." } self.log(f"Posting Twitter/X sukses! Tautan: {post_url}", "SUCCESS") return { "success": True, "post_url": post_url, "message": "Postingan berhasil dikirimkan di Twitter / X!" }