import time import uuid import threading from datetime import datetime from pathlib import Path from typing import Dict, Any, List, Optional from playwright.sync_api import sync_playwright from automation.browser_manager import BrowserManager from automation.gemini_brain import GeminiBrain from automation.platform_handlers.pinterest_handler import PinterestHandler from automation.platform_handlers.facebook_handler import FacebookHandler from automation.platform_handlers.instagram_handler import InstagramHandler from automation.platform_handlers.twitter_handler import TwitterHandler from automation.platform_handlers.tiktok_handler import TikTokHandler from automation.platform_handlers.generic_ai_handler import GenericAiHandler from tracker_manager import TrackerManager class PosterEngine: """ Main background execution engine for social media posting campaigns. """ def __init__(self, browser_manager: Optional[BrowserManager] = None, brain: Optional[GeminiBrain] = None): self.browser_manager = browser_manager or BrowserManager() self.brain = brain or GeminiBrain() self.tracker_manager = TrackerManager() self.jobs: Dict[str, Dict[str, Any]] = {} def get_job(self, job_id: str) -> Optional[Dict[str, Any]]: return self.jobs.get(job_id) def log(self, job_id: str, message: str, level: str = "INFO"): timestamp = datetime.now().strftime("%H:%M:%S") entry = { "timestamp": timestamp, "level": level, "message": message } if job_id in self.jobs: self.jobs[job_id]["logs"].append(entry) print(f"[{timestamp}] [{level}] {message}") def start_campaign( self, campaign_name: str, flyer_path: str, title: str, caption: str, target_link: Optional[str], selected_domains: List[str] ) -> str: job_id = str(uuid.uuid4())[:8] self.jobs[job_id] = { "id": job_id, "campaign_name": campaign_name, "flyer_path": flyer_path, "title": title, "caption": caption, "target_link": target_link, "selected_domains": selected_domains, "status": "queued", "progress": 0, "current_site": "", "logs": [], "results": [], "should_stop": False, "start_time": datetime.now().isoformat(), "end_time": None } # Start thread thread = threading.Thread(target=self._run_job_thread, args=(job_id,), daemon=True) thread.start() return job_id def stop_campaign(self, job_id: str) -> bool: if job_id in self.jobs and self.jobs[job_id]["status"] == "running": self.jobs[job_id]["should_stop"] = True self.log(job_id, "Permintaan pembatalan job diterima oleh sistem...", "WARN") return True return False def _get_handler(self, domain: str): d = domain.lower() if "pinterest" in d: return PinterestHandler(self.brain) elif "facebook" in d: return FacebookHandler(self.brain) elif "instagram" in d: return InstagramHandler(self.brain) elif "twitter" in d or "x.com" in d: return TwitterHandler(self.brain) elif "tiktok" in d: return TikTokHandler(self.brain) else: return GenericAiHandler(self.brain, domain=d) def _run_job_thread(self, job_id: str): job = self.jobs[job_id] job["status"] = "running" self.log(job_id, f"Memulai kampanye '{job['campaign_name']}' untuk {len(job['selected_domains'])} platform sosmed...") # Find row indices for selected domains all_sites = self.tracker_manager.get_sosmed_sites() sites_map = {s["domain"]: s for s in all_sites} total_targets = len(job["selected_domains"]) completed_count = 0 try: # 1. Ensure Chrome Bot is active cdp_status = self.browser_manager.check_cdp_status() if not cdp_status["active"]: self.log(job_id, "Chrome Bot belum aktif di layar. Meluncurkan jendela Chrome Bot...") self.browser_manager.launch_isolated_chrome() time.sleep(2.5) with sync_playwright() as pw: self.log(job_id, "Menghubungkan ke browser bot terisolasi (Chrome Dev/Persistent)...") browser_or_ctx, context, mode = self.browser_manager.acquire_session(pw) self.log(job_id, f"Browser terhubung dengan sukses [Mode: {mode.upper()}].") for domain in job["selected_domains"]: if job["should_stop"]: self.log(job_id, "Kampanye dihentikan oleh pengguna.", "WARN") break job["current_site"] = domain site_meta = sites_map.get(domain, {"name": domain.capitalize(), "row_idx": None}) platform_name = site_meta["name"] row_idx = site_meta.get("row_idx") self.log(job_id, f"--- Menyiapkan konten untuk {platform_name} ({domain}) ---") # 1. Load / Research Platform Posting Guide via Gemini AI (JSON Knowledge) try: self.log(job_id, f"Memuat / meriset panduan AI untuk upload flyer di {platform_name}...") guide = self.brain.get_or_research_platform_guide(domain, platform_name) upload_url = guide.get("direct_upload_url", f"https://{domain}/") self.log(job_id, f"Panduan {platform_name} aktif (URL Target: {upload_url}).") except Exception as ge: self.log(job_id, f"Gagal memuat panduan AI ({ge}), menggunakan mode default.", "WARN") guide = None # 2. Adapt caption using Gemini AI try: self.log(job_id, f"Gemini AI mengadaptasi copywriting & tagar untuk {platform_name}...") custom_caption = self.brain.generate_social_caption( title=job["title"], description=job["caption"], platform_name=platform_name, target_link=job["target_link"] ) except Exception as e: self.log(job_id, f"Gagal adaptasi AI, menggunakan caption asli: {e}", "WARN") custom_caption = job["caption"] # 3. Execute post in a fresh, dedicated, visible tab handler = self._get_handler(domain) handler.set_logger(lambda msg, level="INFO": self.log(job_id, msg, level)) post_result = {"success": False, "post_url": None, "message": "Unknown error"} target_page = None try: # Ensure healthy context & open a fresh visible tab try: target_page = context.new_page() except Exception as page_err: self.log(job_id, f"Koneksi context terputus ({page_err}), menyambungkan ulang...", "WARN") browser_or_ctx, context, mode = self.browser_manager.acquire_session(pw) target_page = context.new_page() target_page.bring_to_front() target_page.set_default_timeout(35000) self.log(job_id, f"Robot Playwright membuka tab dan mengeksekusi posting di {platform_name}...") post_result = handler.post( page=target_page, flyer_path=job["flyer_path"], title=job["title"], caption=custom_caption, target_link=job["target_link"], guide=guide ) except Exception as err: self.log(job_id, f"Kesalahan saat posting ke {platform_name}: {err}", "ERROR") post_result = {"success": False, "post_url": None, "message": str(err)} finally: # Cleanly close the tab so tabs don't clutter if target_page and not target_page.is_closed(): try: time.sleep(2) target_page.close() except Exception: pass # Record result success = post_result.get("success", False) post_url = post_result.get("post_url") msg = post_result.get("message", "") if success: self.log(job_id, f"[BERHASIL] {platform_name}: {msg}", "SUCCESS") if post_url and row_idx: try: self.tracker_manager.update_campaign_link(row_idx, job["campaign_name"], post_url) self.log(job_id, f"Link {post_url} berhasil dicatat di Excel Baris {row_idx} ({job['campaign_name']}).", "SUCCESS") except Exception as ex: self.log(job_id, f"Gagal update Excel: {ex}", "WARN") else: self.log(job_id, f"[GAGAL] {platform_name}: {msg}", "ERROR") if row_idx: try: if "login" in msg.lower() or "2fa" in msg.lower(): fail_text = f"BELUM LOGIN: {platform_name}" else: clean_reason = msg.replace("FAILED:", "").strip() if "FAILED:" in msg else msg.strip() fail_text = f"FAILED: {clean_reason[:40]}" if clean_reason else "FAILED: Gagal posting" self.tracker_manager.update_campaign_link(row_idx, job["campaign_name"], fail_text) self.log(job_id, f"Status dicatat di Excel Baris {row_idx}: {fail_text}", "INFO") except Exception as ex: self.log(job_id, f"Gagal update status di Excel: {ex}", "WARN") job["results"].append({ "domain": domain, "name": platform_name, "success": success, "post_url": post_url, "message": msg }) completed_count += 1 job["progress"] = int((completed_count / total_targets) * 100) # Pause between platforms if completed_count < total_targets and not job["should_stop"]: self.log(job_id, "Jeda 4 detik sebelum menuju platform berikutnya...") time.sleep(4) # Close context if persistent if mode == "persistent": try: browser_or_ctx.close() except Exception: pass except Exception as global_err: self.log(job_id, f"Terjadi kesalahan fatal pada mesin otomatisasi: {global_err}", "ERROR") job["status"] = "failed" finally: if job["status"] != "failed": job["status"] = "completed" if not job["should_stop"] else "stopped" job["end_time"] = datetime.now().isoformat() job["current_site"] = "" self.log(job_id, f"Kampanye selesai dengan status: {job['status'].upper()}.")