""" USA Funny Nostalgia — Autonomous 2-Agent Social Media Bot ========================================================= A fully autonomous Gradio application (Hugging Face ZeroGPU compatible) that generates and publishes "USA Funny Nostalgia (80s/90s vs Modern Era)" content to a Facebook Page using a 2-agent pipeline powered by Google Gemini. Agent 1 : Authentic American Viewer & Content Strategist -> caption + image idea Agent 2 : Hyper-Detailed Retro Prompt Engineer & Publisher -> image prompt + publish Author : Senior Python Developer & AI Automation Specialist """ import os import io import time import json import queue import random import logging import threading import datetime import urllib.parse from typing import List, Tuple, Dict, Optional import requests import gradio as gr # --------------------------------------------------------------------------- # # ZeroGPU support (graceful fallback if not on a Spaces GPU environment) # --------------------------------------------------------------------------- # try: import spaces # Provided automatically on Hugging Face ZeroGPU Spaces _SPACES_AVAILABLE = True except Exception: # pragma: no cover - local dev fallback _SPACES_AVAILABLE = False class _DummySpaces: """Fallback so the app runs locally where `spaces` is unavailable.""" @staticmethod def GPU(*dargs, **dkwargs): def decorator(func): return func # Support both @spaces.GPU and @spaces.GPU(duration=...) if len(dargs) == 1 and callable(dargs[0]) and not dkwargs: return dargs[0] return decorator spaces = _DummySpaces() # type: ignore # --------------------------------------------------------------------------- # # Google GenAI SDK (new library) # --------------------------------------------------------------------------- # try: from google import genai from google.genai import types as genai_types _GENAI_AVAILABLE = True except Exception: # pragma: no cover _GENAI_AVAILABLE = False # =========================================================================== # # LOGGING SETUP # =========================================================================== # class BufferLogHandler(logging.Handler): """A logging handler that keeps the last N log lines for the UI dashboard.""" def __init__(self, capacity: int = 400): super().__init__() self.capacity = capacity self._buffer: "collections_deque" = __import__("collections").deque(maxlen=capacity) self._lock = threading.Lock() def emit(self, record: logging.LogRecord) -> None: try: msg = self.format(record) with self._lock: self._buffer.append(msg) except Exception: self.handleError(record) def get_logs(self) -> str: with self._lock: # Newest first for easy reading in the UI return "\n".join(reversed(self._buffer)) LOG_FORMAT = "%(asctime)s | %(levelname)-7s | %(message)s" DATE_FORMAT = "%Y-%m-%d %H:%M:%S UTC" logging.Formatter.converter = time.gmtime # Force UTC timestamps logger = logging.getLogger("nostalgia_bot") logger.setLevel(logging.INFO) logger.propagate = False # Console handler _console = logging.StreamHandler() _console.setFormatter(logging.Formatter(LOG_FORMAT, DATE_FORMAT)) logger.addHandler(_console) # UI buffer handler ui_log_handler = BufferLogHandler(capacity=400) ui_log_handler.setFormatter(logging.Formatter(LOG_FORMAT, DATE_FORMAT)) logger.addHandler(ui_log_handler) # =========================================================================== # # GLOBAL STATE # =========================================================================== # class AppState: """Thread-safe container for shared application state.""" def __init__(self): self._lock = threading.Lock() # Default USA peak schedule (UTC): 11:00 (5PM BD), 16:00 (10PM BD), 22:00 (4AM BD) self.schedule_times: List[str] = ["11:00", "16:00", "22:00"] self.last_caption: str = "No content generated yet." self.last_image_prompt: str = "No image prompt generated yet." self.last_image_path: Optional[str] = None self.last_run_status: str = "Idle — waiting for first run." self.last_run_time: Optional[str] = None self.is_running: bool = False # prevents overlapping pipeline runs # ---- schedule ---- def get_schedule(self) -> List[str]: with self._lock: return list(self.schedule_times) def set_schedule(self, times: List[str]) -> None: with self._lock: self.schedule_times = list(times) # ---- previews ---- def set_preview(self, caption: str, prompt: str, image_path: Optional[str]) -> None: with self._lock: self.last_caption = caption self.last_image_prompt = prompt self.last_image_path = image_path def get_preview(self) -> Tuple[str, str, Optional[str]]: with self._lock: return self.last_caption, self.last_image_prompt, self.last_image_path # ---- run status ---- def set_status(self, status: str) -> None: with self._lock: self.last_run_status = status self.last_run_time = datetime.datetime.utcnow().strftime(DATE_FORMAT) def get_status(self) -> Tuple[str, Optional[str]]: with self._lock: return self.last_run_status, self.last_run_time # ---- run lock ---- def try_acquire_run(self) -> bool: with self._lock: if self.is_running: return False self.is_running = True return True def release_run(self) -> None: with self._lock: self.is_running = False STATE = AppState() # =========================================================================== # # ENVIRONMENT VARIABLE HELPERS # =========================================================================== # ENV_KEYS = ["GEMINI_API_KEY_1", "GEMINI_API_KEY_2", "FB_PAGE_ID", "FB_PAGE_ACCESS_TOKEN"] def get_env(name: str) -> str: return (os.environ.get(name) or "").strip() def check_env_status() -> Dict[str, bool]: """Return a dict {ENV_NAME: is_loaded_bool}.""" return {key: bool(get_env(key)) for key in ENV_KEYS} def pick_gemini_key() -> Optional[str]: """Return an available Gemini key, preferring key 1, falling back to key 2.""" key1 = get_env("GEMINI_API_KEY_1") key2 = get_env("GEMINI_API_KEY_2") if key1: return key1 if key2: return key2 return None # =========================================================================== # # GEMINI CLIENT WRAPPER # =========================================================================== # GEMINI_MODEL = "gemini-3.6-flash" def _build_genai_client(api_key: str) -> "genai.Client": if not _GENAI_AVAILABLE: raise RuntimeError( "The 'google-genai' library is not installed. " "Add 'google-genai' to requirements.txt." ) return genai.Client(api_key=api_key) def call_gemini(system_persona: str, user_task: str, temperature: float = 1.0) -> str: """ Call Gemini with automatic failover between GEMINI_API_KEY_1 and _2. Returns the model's text response. Raises RuntimeError on total failure. """ keys = [k for k in (get_env("GEMINI_API_KEY_1"), get_env("GEMINI_API_KEY_2")) if k] if not keys: raise RuntimeError("No Gemini API key configured (GEMINI_API_KEY_1 / _2).") last_error: Optional[Exception] = None for idx, key in enumerate(keys, start=1): try: logger.info("Calling Gemini (%s) using API key #%d ...", GEMINI_MODEL, idx) client = _build_genai_client(key) config = genai_types.GenerateContentConfig( system_instruction=system_persona, temperature=temperature, max_output_tokens=2048, ) response = client.models.generate_content( model=GEMINI_MODEL, contents=user_task, config=config, ) text = (getattr(response, "text", None) or "").strip() if not text: raise RuntimeError("Gemini returned an empty response.") logger.info("Gemini call succeeded with API key #%d.", idx) return text except Exception as exc: # noqa: BLE001 - we want to try the next key last_error = exc logger.warning("Gemini API key #%d failed: %s", idx, exc) continue raise RuntimeError(f"All Gemini API keys failed. Last error: {last_error}") # =========================================================================== # # AGENT 1 — AUTHENTIC AMERICAN VIEWER & CONTENT STRATEGIST # =========================================================================== # AGENT1_PERSONA = ( "You are a warm, funny 45-year-old American who grew up in the 1980s and 1990s. " "You are NOT a marketer or a brand. You are a real person reminiscing on Facebook " "with friends. You love sharing specific, relatable nostalgic memories and playful " "'Then vs Now' comparisons. Your tone is casual, warm, humorous, and human." ) AGENT1_TASK = ( "Pick ONE highly specific, funny, and relatable American 80s/90s memory OR a " "'Then vs Now' comparison. Good examples: dial-up internet sounds, rewinding VHS " "tapes, drinking from the garden hose, MapQuest printouts vs GPS, blowing into " "Nintendo cartridges, CRT TVs with rabbit-ear antennas, mixtapes, TV Guide magazine, " "waiting by the phone for a call, Blockbuster Friday nights.\n\n" "Write:\n" "1) A CAPTION: an engaging, warm, funny, casual English Facebook post (2-5 sentences) " "written in first person, ending with an organic question that invites people to " "comment and share their own memories. Add a few tasteful, relevant hashtags at the end.\n" "2) An IMAGE_IDEA: one concise sentence describing the core visual memory to illustrate " "the post.\n\n" "Return STRICT JSON only, with exactly these keys and no markdown fences:\n" '{"caption": "", "image_idea": ""}' ) def _safe_parse_json(raw: str) -> Dict[str, str]: """Robustly parse a JSON object from a model response that may contain fences.""" cleaned = raw.strip() # Strip common markdown code fences if cleaned.startswith("```"): cleaned = cleaned.strip("`") # Remove a leading language hint like 'json\n' if "\n" in cleaned: first_line, rest = cleaned.split("\n", 1) if first_line.strip().lower() in ("json", ""): cleaned = rest # Extract the outermost JSON object if extra text surrounds it start = cleaned.find("{") end = cleaned.rfind("}") if start != -1 and end != -1 and end > start: cleaned = cleaned[start : end + 1] return json.loads(cleaned) def run_agent1() -> Tuple[str, str]: """Run Agent 1. Returns (caption, image_idea).""" logger.info("AGENT 1 (Content Strategist) starting ...") raw = call_gemini(AGENT1_PERSONA, AGENT1_TASK, temperature=1.15) try: data = _safe_parse_json(raw) caption = str(data.get("caption", "")).strip() image_idea = str(data.get("image_idea", "")).strip() except Exception as exc: # noqa: BLE001 logger.warning("Agent 1 JSON parse failed (%s). Falling back to raw text.", exc) caption = raw.strip() image_idea = "A nostalgic 1980s/1990s American home scene." if not caption: raise RuntimeError("Agent 1 produced an empty caption.") if not image_idea: image_idea = "A nostalgic 1980s/1990s American home scene." logger.info("AGENT 1 done. Caption length=%d chars.", len(caption)) return caption, image_idea # =========================================================================== # # AGENT 2 — HYPER-DETAILED RETRO PROMPT ENGINEER & PUBLISHER # =========================================================================== # AGENT2_PERSONA = ( "You are an expert AI photographer and prompt engineer specializing in authentic " "1980s-1990s American aesthetics. You craft extremely detailed, hyper-realistic " "image generation prompts that capture true retro nostalgia." ) AGENT2_TASK_TEMPLATE = ( "Transform the following IMAGE_IDEA into ONE extremely detailed, hyper-realistic " "image generation prompt (a single paragraph, 60-120 words). The prompt MUST " "explicitly include ALL of these elements: 35mm film grain, Kodachrome color " "palette, authentic 1980s/1990s American suburban setting details (shag carpet, " "CRT television, wood paneling, denim clothing, bowl-cut hairstyles), candid flash " "photography look, warm nostalgic lighting, and realistic period-accurate props.\n\n" "IMAGE_IDEA: {image_idea}\n\n" "Return ONLY the final image prompt text. No preamble, no quotes, no markdown." ) def run_agent2_prompt(image_idea: str) -> str: """Run Agent 2's prompt-engineering step. Returns the detailed image prompt.""" logger.info("AGENT 2 (Prompt Engineer) starting ...") task = AGENT2_TASK_TEMPLATE.format(image_idea=image_idea) prompt = call_gemini(AGENT2_PERSONA, task, temperature=0.9).strip() # Reinforce mandatory retro keywords in case the model missed any mandatory = [ "35mm film grain", "Kodachrome", "shag carpet", "CRT television", "wood paneling", "candid flash photography", "warm nostalgic lighting", ] lowered = prompt.lower() missing = [kw for kw in mandatory if kw.lower() not in lowered] if missing: logger.info("Reinforcing missing retro keywords: %s", ", ".join(missing)) prompt = f"{prompt} Additional style: {', '.join(missing)}." if not prompt: raise RuntimeError("Agent 2 produced an empty image prompt.") logger.info("AGENT 2 prompt ready. Length=%d chars.", len(prompt)) return prompt # --------------------------------------------------------------------------- # # IMAGE GENERATION (Pollinations AI, free high-res endpoint) # --------------------------------------------------------------------------- # POLLINATIONS_BASE = "https://image.pollinations.ai/prompt/" def generate_image(prompt: str, width: int = 1024, height: int = 1024) -> str: """ Generate an image via Pollinations AI and save it to disk. Returns the local file path. Raises RuntimeError on failure. """ seed = random.randint(1, 2_000_000_000) encoded_prompt = urllib.parse.quote(prompt, safe="") params = { "width": str(width), "height": str(height), "seed": str(seed), "nologo": "true", "model": "flux", } query = urllib.parse.urlencode(params) url = f"{POLLINATIONS_BASE}{encoded_prompt}?{query}" logger.info("Requesting image from Pollinations AI (seed=%d) ...", seed) last_error: Optional[Exception] = None for attempt in range(1, 4): # up to 3 attempts try: resp = requests.get(url, timeout=120) resp.raise_for_status() content_type = resp.headers.get("Content-Type", "") if "image" not in content_type: raise RuntimeError(f"Unexpected content type: {content_type!r}") os.makedirs("generated", exist_ok=True) timestamp = datetime.datetime.utcnow().strftime("%Y%m%d_%H%M%S") file_path = os.path.join("generated", f"nostalgia_{timestamp}_{seed}.jpg") with open(file_path, "wb") as fh: fh.write(resp.content) if os.path.getsize(file_path) < 1024: raise RuntimeError("Downloaded image is suspiciously small (<1KB).") logger.info("Image saved to %s (%d bytes).", file_path, os.path.getsize(file_path)) return file_path except Exception as exc: # noqa: BLE001 last_error = exc logger.warning("Image generation attempt %d/3 failed: %s", attempt, exc) time.sleep(3 * attempt) raise RuntimeError(f"Image generation failed after 3 attempts. Last error: {last_error}") # --------------------------------------------------------------------------- # # FACEBOOK PUBLISHER (Graph API v19.0) # --------------------------------------------------------------------------- # FB_API_VERSION = "v19.0" def publish_to_facebook(image_path: str, caption: str) -> str: """ Publish an image + caption to a Facebook Page using Graph API v19.0. Returns the created post/photo ID. Raises RuntimeError on failure. """ page_id = get_env("FB_PAGE_ID") access_token = get_env("FB_PAGE_ACCESS_TOKEN") if not page_id: raise RuntimeError("FB_PAGE_ID is not configured.") if not access_token: raise RuntimeError("FB_PAGE_ACCESS_TOKEN is not configured.") if not os.path.exists(image_path): raise RuntimeError(f"Image file not found: {image_path}") endpoint = f"https://graph.facebook.com/{FB_API_VERSION}/{page_id}/photos" logger.info("Publishing to Facebook Page %s via %s ...", page_id, FB_API_VERSION) last_error: Optional[Exception] = None for attempt in range(1, 4): # up to 3 attempts try: with open(image_path, "rb") as image_file: files = {"source": image_file} data = { "caption": caption, "access_token": access_token, "published": "true", } resp = requests.post(endpoint, data=data, files=files, timeout=180) # Parse JSON body regardless of status for detailed error logging try: payload = resp.json() except Exception: payload = {"raw": resp.text} if resp.status_code != 200: fb_error = payload.get("error", {}) if isinstance(payload, dict) else {} message = fb_error.get("message", resp.text) raise RuntimeError(f"Facebook API error ({resp.status_code}): {message}") post_id = payload.get("post_id") or payload.get("id") or "unknown" logger.info("Successfully published to Facebook. Post ID: %s", post_id) return str(post_id) except Exception as exc: # noqa: BLE001 last_error = exc logger.warning("Facebook publish attempt %d/3 failed: %s", attempt, exc) time.sleep(5 * attempt) raise RuntimeError(f"Facebook publishing failed after 3 attempts. Last error: {last_error}") # =========================================================================== # # FULL PIPELINE (ZeroGPU-decorated) # =========================================================================== # @spaces.GPU(duration=10) def run_full_pipeline(trigger: str = "manual") -> Dict[str, str]: """ Execute the complete 2-agent pipeline: Agent 1 -> Agent 2 -> Image -> Facebook publish. Never raises; always returns a status dict so the app cannot crash. """ if not STATE.try_acquire_run(): msg = "A pipeline run is already in progress. Skipping this trigger." logger.warning(msg) return {"status": "skipped", "message": msg} logger.info("=" * 60) logger.info("PIPELINE STARTED (trigger=%s)", trigger) result: Dict[str, str] = {"status": "error", "message": "Unknown error."} try: # ---- Agent 1 ---- caption, image_idea = run_agent1() STATE.set_preview(caption, "Generating detailed image prompt ...", None) # ---- Agent 2: prompt ---- image_prompt = run_agent2_prompt(image_idea) STATE.set_preview(caption, image_prompt, None) # ---- Image generation ---- image_path = None try: image_path = generate_image(image_prompt) STATE.set_preview(caption, image_prompt, image_path) except Exception as img_exc: # noqa: BLE001 logger.error("Image generation failed: %s", img_exc) STATE.set_status(f"Image generation failed: {img_exc}") result = {"status": "error", "message": f"Image generation failed: {img_exc}"} return result # ---- Facebook publish ---- try: post_id = publish_to_facebook(image_path, caption) except Exception as fb_exc: # noqa: BLE001 logger.error("Facebook publishing failed: %s", fb_exc) STATE.set_status(f"Publish failed: {fb_exc}") result = {"status": "error", "message": f"Publish failed: {fb_exc}"} return result success_msg = f"Published successfully. Facebook Post ID: {post_id}" logger.info(success_msg) STATE.set_status(success_msg) result = {"status": "success", "message": success_msg, "post_id": post_id} return result except Exception as exc: # noqa: BLE001 - top-level safety net logger.exception("Pipeline crashed unexpectedly: %s", exc) STATE.set_status(f"Pipeline error: {exc}") result = {"status": "error", "message": f"Pipeline error: {exc}"} return result finally: STATE.release_run() logger.info("PIPELINE FINISHED (trigger=%s)", trigger) logger.info("=" * 60) # =========================================================================== # # BACKGROUND SCHEDULER (threading + schedule) # =========================================================================== # import schedule # noqa: E402 (imported after logging is configured) class SchedulerThread(threading.Thread): """Runs the `schedule` library clock in the background, in UTC.""" def __init__(self): super().__init__(daemon=True) self._stop_event = threading.Event() def rebuild_jobs(self) -> None: """Clear and rebuild scheduled jobs from STATE.schedule_times (UTC).""" schedule.clear() for t in STATE.get_schedule(): try: schedule.every().day.at(t).do(self._job_runner, when=t) except Exception as exc: # noqa: BLE001 logger.error("Failed to schedule time %r: %s", t, exc) logger.info("Scheduler rebuilt with UTC times: %s", ", ".join(STATE.get_schedule())) @staticmethod def _job_runner(when: str) -> None: logger.info("Scheduled trigger fired for UTC %s.", when) run_full_pipeline(trigger=f"scheduled@{when}") def run(self) -> None: # `schedule` uses local time by default; we force UTC comparison below. self.rebuild_jobs() logger.info("Background scheduler thread started.") while not self._stop_event.is_set(): try: self._run_pending_utc() except Exception as exc: # noqa: BLE001 logger.error("Scheduler loop error: %s", exc) time.sleep(20) @staticmethod def _run_pending_utc() -> None: """ Run pending jobs using UTC. We monkey-check each job's scheduled time against current UTC to keep everything timezone-consistent. """ now_utc = datetime.datetime.utcnow() for job in list(schedule.jobs): if job.next_run is None: continue # `schedule` computes next_run in local time; normalize to compare in UTC. # We instead trigger by matching HH:MM to avoid local/UTC drift. # Match current UTC HH:MM against configured times. current_hm = now_utc.strftime("%H:%M") SchedulerThread._maybe_fire(current_hm, now_utc) _last_fired_minute: Dict[str, str] = {} @classmethod def _maybe_fire(cls, current_hm: str, now_utc: datetime.datetime) -> None: """Fire the pipeline once per configured time per day (UTC-based).""" today = now_utc.strftime("%Y-%m-%d") for t in STATE.get_schedule(): if t == current_hm: marker_key = f"{today}@{t}" if cls._last_fired_minute.get(t) != marker_key: cls._last_fired_minute[t] = marker_key logger.info("UTC scheduler match at %s — launching pipeline.", t) threading.Thread( target=run_full_pipeline, kwargs={"trigger": f"scheduled@{t}"}, daemon=True, ).start() def stop(self) -> None: self._stop_event.set() scheduler_thread = SchedulerThread() def compute_next_runs() -> str: """Return a human-readable summary of the next scheduled run times (UTC & BD).""" times = STATE.get_schedule() if not times: return "No schedules configured." now_utc = datetime.datetime.utcnow() lines = [] for t in sorted(times): try: hh, mm = [int(x) for x in t.split(":")] candidate = now_utc.replace(hour=hh, minute=mm, second=0, microsecond=0) if candidate <= now_utc: candidate += datetime.timedelta(days=1) # Bangladesh time = UTC + 6 bd_time = candidate + datetime.timedelta(hours=6) delta = candidate - now_utc hrs, rem = divmod(int(delta.total_seconds()), 3600) mins = rem // 60 lines.append( f"• {t} UTC ({bd_time.strftime('%I:%M %p')} BD) — in {hrs}h {mins}m" ) except Exception: lines.append(f"• {t} (invalid format)") return "\n".join(lines) # =========================================================================== # # UI HELPER FUNCTIONS # =========================================================================== # def validate_time_string(t: str) -> bool: """Validate an HH:MM 24-hour time string.""" try: parts = t.strip().split(":") if len(parts) != 2: return False hh, mm = int(parts[0]), int(parts[1]) return 0 <= hh <= 23 and 0 <= mm <= 59 except Exception: return False def normalize_time_string(t: str) -> str: hh, mm = t.strip().split(":") return f"{int(hh):02d}:{int(mm):02d}" def ui_add_schedule(new_time: str) -> Tuple[str, str, str]: """Add a new schedule time. Returns (schedule_display, next_runs, feedback).""" new_time = (new_time or "").strip() if not validate_time_string(new_time): return ( format_schedule_display(), compute_next_runs(), f"❌ Invalid time '{new_time}'. Use HH:MM (24-hour), e.g. 14:30.", ) normalized = normalize_time_string(new_time) current = STATE.get_schedule() if normalized in current: return ( format_schedule_display(), compute_next_runs(), f"⚠️ {normalized} UTC is already scheduled.", ) current.append(normalized) STATE.set_schedule(sorted(current)) scheduler_thread.rebuild_jobs() logger.info("Added schedule time: %s UTC", normalized) return ( format_schedule_display(), compute_next_runs(), f"✅ Added {normalized} UTC to the schedule.", ) def ui_replace_schedule(times_text: str) -> Tuple[str, str, str]: """Replace the entire schedule with a comma/newline separated list of HH:MM.""" raw = (times_text or "").replace("\n", ",") candidates = [c.strip() for c in raw.split(",") if c.strip()] if not candidates: return ( format_schedule_display(), compute_next_runs(), "❌ Please provide at least one HH:MM time.", ) valid, invalid = [], [] for c in candidates: if validate_time_string(c): valid.append(normalize_time_string(c)) else: invalid.append(c) if invalid: return ( format_schedule_display(), compute_next_runs(), f"❌ Invalid time(s): {', '.join(invalid)}. Use HH:MM (24-hour).", ) unique_sorted = sorted(set(valid)) STATE.set_schedule(unique_sorted) scheduler_thread.rebuild_jobs() logger.info("Replaced schedule with: %s", ", ".join(unique_sorted)) return ( format_schedule_display(), compute_next_runs(), f"✅ Schedule updated: {', '.join(unique_sorted)} UTC.", ) def ui_reset_schedule() -> Tuple[str, str, str]: STATE.set_schedule(["11:00", "16:00", "22:00"]) scheduler_thread.rebuild_jobs() logger.info("Schedule reset to USA peak defaults.") return ( format_schedule_display(), compute_next_runs(), "✅ Reset to default USA peak times (11:00, 16:00, 22:00 UTC).", ) def format_schedule_display() -> str: times = STATE.get_schedule() if not times: return "No schedules configured." rows = [] for t in sorted(times): hh, mm = [int(x) for x in t.split(":")] bd = (datetime.datetime(2000, 1, 1, hh, mm) + datetime.timedelta(hours=6)) rows.append(f"• {t} UTC → {bd.strftime('%I:%M %p')} Bangladesh") return "\n".join(rows) def ui_run_now() -> Tuple[str, str, str, Optional[str], str]: """Manual trigger. Returns dashboard, caption, prompt, image, feedback.""" logger.info("Manual 'Run Agent Now' triggered by user.") result = run_full_pipeline(trigger="manual-button") caption, prompt, image_path = STATE.get_preview() feedback = f"{'✅' if result.get('status') == 'success' else '❌'} {result.get('message')}" return (build_dashboard(), caption, prompt, image_path, feedback) def build_dashboard() -> str: """Build the live status dashboard text.""" now_utc = datetime.datetime.utcnow().strftime(DATE_FORMAT) now_bd = (datetime.datetime.utcnow() + datetime.timedelta(hours=6)).strftime( "%Y-%m-%d %I:%M:%S %p" ) status, status_time = STATE.get_status() next_runs = compute_next_runs() lines = [ "🕒 SERVER TIME", f" UTC : {now_utc}", f" BD : {now_bd} (Bangladesh)", "", "📅 NEXT SCHEDULED RUNS", next_runs, "", "📌 LAST RUN STATUS", f" {status}", f" (updated: {status_time or 'never'})", ] return "\n".join(lines) def build_env_status_markdown() -> str: """Build markdown showing environment variable load status with colored dots.""" status = check_env_status() lines = ["### 🔑 Environment Variables Status", ""] for key, loaded in status.items(): dot = "🟢" if loaded else "🔴" state = "Loaded" if loaded else "MISSING" lines.append(f"- {dot} **{key}** — {state}") return "\n".join(lines) def ui_refresh_all() -> Tuple[str, str, str, str, str, Optional[str]]: """Refresh dashboard, logs, env status, caption, prompt, image.""" caption, prompt, image_path = STATE.get_preview() return ( build_dashboard(), ui_log_handler.get_logs(), build_env_status_markdown(), caption, prompt, image_path, ) # =========================================================================== # # CUSTOM CSS — White bg, Sky Blue accents, Red buttons, Parrot Green status # =========================================================================== # CUSTOM_CSS = """ /* Global white background & English-friendly font */ .gradio-container { background-color: #FFFFFF !important; color: #0B0B0B !important; font-family: 'Segoe UI', Roboto, Helvetica, Arial, sans-serif !important; } /* Headings in Sky Blue */ h1, h2, h3, .app-title { color: #1CA3EC !important; font-weight: 700 !important; } /* Sky Blue accent panels/borders */ .gr-box, .gr-panel, .block, .form { border-color: #1CA3EC !important; } /* Primary action buttons in RED */ button.primary, .gr-button-primary, button.lg.primary { background-color: #E23B3B !important; border-color: #E23B3B !important; color: #FFFFFF !important; font-weight: 700 !important; } button.primary:hover, .gr-button-primary:hover { background-color: #C62828 !important; border-color: #C62828 !important; } /* Secondary buttons in Sky Blue */ button.secondary, .gr-button-secondary { background-color: #1CA3EC !important; border-color: #1CA3EC !important; color: #FFFFFF !important; font-weight: 600 !important; } button.secondary:hover, .gr-button-secondary:hover { background-color: #1483C2 !important; } /* Parrot Green (টিয়া) status/log elements */ .parrot-status textarea, .parrot-status .gr-text-input, .parrot-status input { background-color: #E9FBE5 !important; color: #0A5D00 !important; border: 2px solid #33CC33 !important; font-weight: 600 !important; } /* Dashboard box styling (parrot green tint) */ .dashboard-box textarea { background-color: #F2FFF0 !important; color: #0A5D00 !important; border: 2px solid #33CC33 !important; font-family: 'Consolas', 'Courier New', monospace !important; } /* Log box styling */ .log-box textarea { background-color: #0B1F0B !important; color: #7CFC00 !important; /* bright parrot green text on dark */ font-family: 'Consolas', 'Courier New', monospace !important; font-size: 12px !important; } /* Section header banner */ .section-banner { background: linear-gradient(90deg, #1CA3EC 0%, #33CC33 100%); color: #FFFFFF !important; padding: 8px 14px; border-radius: 8px; font-weight: 700; margin-bottom: 6px; } """ # =========================================================================== # # GRADIO UI # =========================================================================== # def build_ui() -> gr.Blocks: with gr.Blocks(css=CUSTOM_CSS, title="USA Funny Nostalgia Bot", theme=gr.themes.Soft()) as demo: gr.HTML( "

📼 USA Funny Nostalgia — Autonomous 2-Agent Bot

" "

80s / 90s vs Modern Era • Powered by Gemini 2.5 Flash, " "Pollinations AI & Facebook Graph API v19.0

" ) # ---------------- Top control row ---------------- with gr.Row(): run_now_btn = gr.Button("▶️ Run Agent Now", variant="primary", scale=1) refresh_btn = gr.Button("🔄 Refresh Dashboard", variant="secondary", scale=1) feedback_box = gr.Textbox( label="Action Feedback", interactive=False, elem_classes=["parrot-status"], ) # ---------------- Dashboard & Env status ---------------- with gr.Row(): with gr.Column(scale=2): gr.HTML("
📊 Live Status Dashboard
") dashboard_box = gr.Textbox( value=build_dashboard(), label="Dashboard (UTC & Bangladesh Time)", lines=12, interactive=False, elem_classes=["dashboard-box"], ) with gr.Column(scale=1): gr.HTML("
🔑 Environment
") env_status_md = gr.Markdown(build_env_status_markdown()) # ---------------- Scheduler ---------------- gr.HTML("
⏰ Dynamic Custom Scheduler (UTC)
") with gr.Row(): with gr.Column(scale=1): schedule_display = gr.Textbox( value=format_schedule_display(), label="Current Schedule", lines=5, interactive=False, elem_classes=["parrot-status"], ) next_runs_box = gr.Textbox( value=compute_next_runs(), label="Next Scheduled Runs", lines=5, interactive=False, elem_classes=["dashboard-box"], ) with gr.Column(scale=1): add_time_input = gr.Textbox( label="Add a single time (HH:MM, 24-hour UTC)", placeholder="e.g. 14:30", ) add_time_btn = gr.Button("➕ Add Time", variant="secondary") replace_times_input = gr.Textbox( label="Replace ALL times (comma or newline separated)", placeholder="e.g. 11:00, 16:00, 22:00", lines=3, ) replace_times_btn = gr.Button("💾 Replace Schedule", variant="secondary") reset_btn = gr.Button("↩️ Reset to USA Peak Defaults", variant="secondary") schedule_feedback = gr.Textbox( label="Scheduler Feedback", interactive=False, elem_classes=["parrot-status"], ) # ---------------- Content & Image preview ---------------- gr.HTML("
🖼️ Content & Image Prompt Preview
") with gr.Row(): with gr.Column(scale=1): caption_preview = gr.Textbox( value=STATE.get_preview()[0], label="Last Generated Caption (Agent 1)", lines=8, interactive=False, ) prompt_preview = gr.Textbox( value=STATE.get_preview()[1], label="Last Generated Image Prompt (Agent 2)", lines=8, interactive=False, ) with gr.Column(scale=1): image_preview = gr.Image( value=STATE.get_preview()[2], label="Last Generated Image", height=420, ) # ---------------- Logs ---------------- gr.HTML("
📜 Execution Logs (newest first)
") log_box = gr.Textbox( value=ui_log_handler.get_logs(), label="Logs", lines=16, interactive=False, elem_classes=["log-box"], ) # ---------------- Event wiring ---------------- run_now_btn.click( fn=ui_run_now, inputs=None, outputs=[dashboard_box, caption_preview, prompt_preview, image_preview, feedback_box], ).then( fn=ui_refresh_all, inputs=None, outputs=[dashboard_box, log_box, env_status_md, caption_preview, prompt_preview, image_preview], ) refresh_btn.click( fn=ui_refresh_all, inputs=None, outputs=[dashboard_box, log_box, env_status_md, caption_preview, prompt_preview, image_preview], ) add_time_btn.click( fn=ui_add_schedule, inputs=[add_time_input], outputs=[schedule_display, next_runs_box, schedule_feedback], ) replace_times_btn.click( fn=ui_replace_schedule, inputs=[replace_times_input], outputs=[schedule_display, next_runs_box, schedule_feedback], ) reset_btn.click( fn=ui_reset_schedule, inputs=None, outputs=[schedule_display, next_runs_box, schedule_feedback], ) # Auto-refresh dashboard/logs every 15 seconds (Gradio timer) timer = gr.Timer(15.0) timer.tick( fn=ui_refresh_all, inputs=None, outputs=[dashboard_box, log_box, env_status_md, caption_preview, prompt_preview, image_preview], ) return demo # =========================================================================== # # ENTRY POINT # =========================================================================== # def main() -> None: # Startup environment diagnostics logger.info("Starting USA Funny Nostalgia Bot ...") logger.info("ZeroGPU 'spaces' available: %s", _SPACES_AVAILABLE) logger.info("google-genai available: %s", _GENAI_AVAILABLE) for key, loaded in check_env_status().items(): logger.info("ENV %s: %s", key, "LOADED" if loaded else "MISSING") # Start the background scheduler thread scheduler_thread.start() # Build and launch the Gradio app demo = build_ui() demo.queue(max_size=8).launch( server_name="0.0.0.0", server_port=int(os.environ.get("PORT", 7860)), show_error=True, ) if __name__ == "__main__": main()