Spaces:

mison2264
/

Facebook-Agent

Sleeping

Update app.py
03fff46 verified 3 months ago
40.7 kB
"""
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": "<caption text>", "image_idea": "<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(
            "<h1 class='app-title'>📼 USA Funny Nostalgia — Autonomous 2-Agent Bot</h1>"
            "<p style='color:#333;'>80s / 90s vs Modern Era • Powered by Gemini 2.5 Flash, "
            "Pollinations AI & Facebook Graph API v19.0</p>"
        )

        # ---------------- 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("<div class='section-banner'>📊 Live Status Dashboard</div>")
                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("<div class='section-banner'>🔑 Environment</div>")
                env_status_md = gr.Markdown(build_env_status_markdown())

        # ---------------- Scheduler ----------------
        gr.HTML("<div class='section-banner'>⏰ Dynamic Custom Scheduler (UTC)</div>")
        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("<div class='section-banner'>🖼️ Content & Image Prompt Preview</div>")
        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("<div class='section-banner'>📜 Execution Logs (newest first)</div>")
        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()