Spaces:

webmazta
/

contentai

Running

webmazta
Move dashboard sign-in off reserved login path
f393d93 3 months ago
71.2 kB
#!/usr/bin/env python3
"""
Content Publishing Agent — Flask Backend
=========================================
Web dashboard + background agent for automated content publishing.
Integrates Instagram-to-WP and Article Rewriter pipelines.
Run:  python app.py
Visit: http://localhost:5000
"""

import json
import hmac
import os
import secrets
import smtplib
import sqlite3
import textwrap
import threading
import time
import traceback
import concurrent.futures
from datetime import datetime, timezone
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from pathlib import Path
from urllib.parse import quote_plus
import sys
import collections
import uuid
from werkzeug.utils import secure_filename
import signal

import feedparser
import requests
from flask import Flask, Response, jsonify, render_template, request, session, redirect, url_for, flash
import functools

from dotenv import load_dotenv
from gemma_ai import gemini_generate
from social_publisher import (
    PartialFacebookPublishError,
    SocialPublishError,
    post_facebook_comment,
    publish_facebook_page_photo_story,
    publish_facebook_page_photo_with_first_comment,
    render_template_text,
)

load_dotenv()

# ──────────────────────────────────────────────
# App Setup
# ──────────────────────────────────────────────

app = Flask(__name__)
app.secret_key = os.getenv("FLASK_SECRET_KEY") or secrets.token_hex(16)

# Use persistent storage directory if available, otherwise local file
# On Hugging Face Spaces, /data is the persistent volume (prevents restart loops)
DATA_DIR = Path("/data")
if DATA_DIR.is_dir() or os.getenv("HUGGING_FACE_SPACES"):
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    DB_PATH = DATA_DIR / "agent.db"
else:
    # Resolve to absolute path relative to this script's directory
    DB_PATH = Path(__file__).parent / "agent.db"

# Upload directory for media files (use persistent storage on HF Spaces)
if DB_PATH.parent == DATA_DIR:
    UPLOAD_DIR = DATA_DIR / "uploads"
else:
    UPLOAD_DIR = Path(__file__).parent / "uploads"
UPLOAD_DIR.mkdir(exist_ok=True)

LOCK = threading.Lock()
STOP_PROCESSING = threading.Event()  # Set this to halt queue processing
LAST_LLM_JOB_TIME = 0  # Timestamp of last LLM-heavy job (for proactive rate-limit spacing)
JOB_TIMEOUT_SECONDS = 600  # 10 minutes max per job
WATCHDOG_INTERVAL = 30  # Check for stuck jobs every 30 seconds

# ── Worker Thread Infrastructure ──
_WORK_AVAILABLE = threading.Event()  # Signal the worker thread to process jobs
_SOCIAL_WORK_AVAILABLE = threading.Event()  # Signal the social worker thread to process queued posts
_current_job_future = None  # concurrent.futures.Future for the running job
_current_job_id = None  # ID of the currently running job
_current_job_lock = threading.Lock()  # Protects _current_job_future and _current_job_id

# ──────────────────────────────────────────────
# Logging Setup
# ──────────────────────────────────────────────
LOG_BUFFER = collections.deque(maxlen=1000)

class LogCapturer:
    def __init__(self, original_stdout):
        self.original_stdout = original_stdout

    def write(self, message):
        self.original_stdout.write(message)
        if message and message.strip() != "":
            timestamp = datetime.now().strftime("%H:%M:%S")
            LOG_BUFFER.append(f"[{timestamp}] {message.rstrip()}")

    def flush(self):
        self.original_stdout.flush()

sys.stdout = LogCapturer(sys.stdout)

# ──────────────────────────────────────────────
# Database
# ──────────────────────────────────────────────

def get_db():
    conn = sqlite3.connect(str(DB_PATH), timeout=30, isolation_level="DEFERRED")
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA busy_timeout=30000")
    conn.execute("PRAGMA foreign_keys=ON")
    conn.execute("PRAGMA synchronous=NORMAL")
    return conn

def init_db():
    conn = get_db()
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS jobs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            url TEXT NOT NULL,
            title TEXT,
            mode TEXT NOT NULL DEFAULT 'article',
            status TEXT NOT NULL DEFAULT 'pending',
            error TEXT,
            result_json TEXT,
            created_at TEXT NOT NULL DEFAULT (datetime('now')),
            started_at TEXT,
            finished_at TEXT
        );
        CREATE TABLE IF NOT EXISTS posts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            wp_url TEXT,
            wp_post_id INTEGER,
            source_url TEXT,
            mode TEXT,
            featured_image_url TEXT,
            flyer_url TEXT,
            flyer_media_id INTEGER,
            story_flyer_url TEXT,
            story_flyer_media_id INTEGER,
            youtube_count INTEGER DEFAULT 0,
            image_count INTEGER DEFAULT 0,
            social_status TEXT NOT NULL DEFAULT 'not_queued',
            created_at TEXT NOT NULL DEFAULT (datetime('now'))
        );
        CREATE TABLE IF NOT EXISTS social_targets (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            platform TEXT NOT NULL DEFAULT 'facebook',
            target_type TEXT NOT NULL,
            name TEXT NOT NULL,
            external_id TEXT NOT NULL,
            access_token TEXT,
            enabled INTEGER NOT NULL DEFAULT 1,
            auto_post INTEGER NOT NULL DEFAULT 1,
            post_to_story INTEGER NOT NULL DEFAULT 0,
            caption_template TEXT,
            created_at TEXT NOT NULL DEFAULT (datetime('now')),
            updated_at TEXT,
            UNIQUE(platform, target_type, external_id)
        );
        CREATE TABLE IF NOT EXISTS social_posts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            post_id INTEGER NOT NULL,
            job_id INTEGER,
            target_id INTEGER NOT NULL,
            platform TEXT NOT NULL DEFAULT 'facebook',
            target_type TEXT NOT NULL,
            status TEXT NOT NULL DEFAULT 'pending',
            caption TEXT,
            comment_text TEXT,
            media_url TEXT,
            wp_url TEXT,
            external_post_id TEXT,
            external_comment_id TEXT,
            external_post_url TEXT,
            story_status TEXT NOT NULL DEFAULT 'disabled',
            story_media_url TEXT,
            external_story_id TEXT,
            external_story_photo_id TEXT,
            story_error TEXT,
            story_attempt_count INTEGER DEFAULT 0,
            story_posted_at TEXT,
            story_completed_at TEXT,
            error TEXT,
            attempt_count INTEGER DEFAULT 0,
            created_at TEXT NOT NULL DEFAULT (datetime('now')),
            updated_at TEXT,
            posted_at TEXT,
            completed_at TEXT,
            FOREIGN KEY(post_id) REFERENCES posts(id) ON DELETE CASCADE,
            FOREIGN KEY(job_id) REFERENCES jobs(id) ON DELETE SET NULL,
            FOREIGN KEY(target_id) REFERENCES social_targets(id) ON DELETE CASCADE,
            UNIQUE(post_id, target_id)
        );
        CREATE TABLE IF NOT EXISTS feeds (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            url TEXT NOT NULL UNIQUE,
            active INTEGER NOT NULL DEFAULT 1,
            last_checked TEXT,
            articles_found INTEGER DEFAULT 0,
            created_at TEXT NOT NULL DEFAULT (datetime('now'))
        );
        CREATE TABLE IF NOT EXISTS settings (
            key TEXT PRIMARY KEY,
            value TEXT NOT NULL
        );
        CREATE TABLE IF NOT EXISTS processed_urls (
            url TEXT PRIMARY KEY,
            status TEXT,
            created_at TEXT NOT NULL DEFAULT (datetime('now'))
        );
    """)

    # Migrations
    try:
        cursor = conn.execute("PRAGMA table_info(jobs)")
        columns = [row[1] for row in cursor.fetchall()]
        if "title" not in columns:
            conn.execute("ALTER TABLE jobs ADD COLUMN title TEXT")
            conn.commit()
    except: pass
    try:
        cursor = conn.execute("PRAGMA table_info(posts)")
        columns = [row[1] for row in cursor.fetchall()]
        if "flyer_url" not in columns:
            conn.execute("ALTER TABLE posts ADD COLUMN flyer_url TEXT")
        if "flyer_media_id" not in columns:
            conn.execute("ALTER TABLE posts ADD COLUMN flyer_media_id INTEGER")
        if "story_flyer_url" not in columns:
            conn.execute("ALTER TABLE posts ADD COLUMN story_flyer_url TEXT")
        if "story_flyer_media_id" not in columns:
            conn.execute("ALTER TABLE posts ADD COLUMN story_flyer_media_id INTEGER")
        if "social_status" not in columns:
            conn.execute("ALTER TABLE posts ADD COLUMN social_status TEXT NOT NULL DEFAULT 'not_queued'")
        conn.commit()
    except Exception as e:
        print(f"[DB] Post migration warning: {e}")
    try:
        cursor = conn.execute("PRAGMA table_info(social_targets)")
        columns = [row[1] for row in cursor.fetchall()]
        if "post_to_story" not in columns:
            conn.execute("ALTER TABLE social_targets ADD COLUMN post_to_story INTEGER NOT NULL DEFAULT 0")
        conn.commit()
    except Exception as e:
        print(f"[DB] Social target migration warning: {e}")
    try:
        cursor = conn.execute("PRAGMA table_info(social_posts)")
        columns = [row[1] for row in cursor.fetchall()]
        story_columns = {
            "story_status": "TEXT NOT NULL DEFAULT 'disabled'",
            "story_media_url": "TEXT",
            "external_story_id": "TEXT",
            "external_story_photo_id": "TEXT",
            "story_error": "TEXT",
            "story_attempt_count": "INTEGER DEFAULT 0",
            "story_posted_at": "TEXT",
            "story_completed_at": "TEXT",
        }
        for column, ddl in story_columns.items():
            if column not in columns:
                conn.execute(f"ALTER TABLE social_posts ADD COLUMN {column} {ddl}")
        conn.commit()
    except Exception as e:
        print(f"[DB] Social post migration warning: {e}")

    # Default settings
    defaults = {
        "agent_enabled": "false",
        "check_interval_minutes": "30",
        "auto_publish": "true",
        "screening_enabled": "true",
        "screening_criteria": "Ghana news, African entertainment, sports highlights, trending stories.",
        "review_feeds_enabled": "true",
        "wp_url": os.getenv("WP_URL", "").rstrip("/"),
        "wp_username": os.getenv("WP_USERNAME", ""),
        "wp_app_password": os.getenv("WP_APP_PASSWORD", ""),
        "wp_category_id": os.getenv("WP_CATEGORY_ID", "1"),
        "wp_category_news": os.getenv("WP_REWRITE_CATEGORY_ID", "1"),
        "wp_category_slay": os.getenv("WP_SLAY_CATEGORY_ID", "408"),
        "wp_post_status": os.getenv("WP_POST_STATUS", "publish"),
        "google_api_key": os.getenv("GOOGLE_API_KEY", ""),
        "openrouter_api_key": os.getenv("OPENROUTER_API_KEY", ""),
        "ig_username": os.getenv("IG_USERNAME", ""),
        "ig_password": os.getenv("IG_PASSWORD", ""),
        "llm_model": os.getenv("LLM_MODEL", "gemma-4-31b-it"),
        "social_auto_post": "true",
        "social_graph_version": os.getenv("SOCIAL_GRAPH_VERSION", "v25.0"),
        "social_request_timeout_seconds": os.getenv("SOCIAL_REQUEST_TIMEOUT_SECONDS", "120"),
        "social_connect_timeout_seconds": os.getenv("SOCIAL_CONNECT_TIMEOUT_SECONDS", "60"),
        "social_post_gap_seconds": os.getenv("SOCIAL_POST_GAP_SECONDS", "20"),
        "social_caption_template": "{title}",
        "social_first_comment_template": "Read more: {wp_url}",
    }
    for key, value in defaults.items():
        conn.execute("INSERT OR IGNORE INTO settings (key, value) VALUES (?, ?)", (key, value))
    conn.commit()
    conn.close()

def get_setting(key, default=""):
    conn = get_db()
    row = conn.execute("SELECT value FROM settings WHERE key = ?", (key,)).fetchone()
    conn.close()
    return row["value"] if row else default

def get_all_settings():
    conn = get_db()
    rows = conn.execute("SELECT key, value FROM settings").fetchall()
    conn.close()
    return {r["key"]: r["value"] for r in rows}

def set_setting(key, value):
    conn = get_db()
    conn.execute("INSERT INTO settings (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value", (key, value))
    conn.commit()
    conn.close()

# ──────────────────────────────────────────────
# Social Publishing Queue
# ──────────────────────────────────────────────

def _truthy(value):
    return str(value).strip().lower() in ("1", "true", "yes", "on")


def _mask_secret(value):
    if not value:
        return ""
    value = str(value)
    return "••••••" + value[-4:] if len(value) > 4 else "••••••"


def _render_social_template(template, context, fallback=""):
    try:
        rendered = render_template_text(template or "", context)
        return rendered or fallback
    except Exception as e:
        print(f"[SOCIAL] Template render warning: {e}")
        return fallback


def _social_request_timeout_seconds():
    try:
        timeout = int(get_setting("social_request_timeout_seconds", "120"))
    except (TypeError, ValueError):
        timeout = 120
    return max(45, min(timeout, 300))


def _social_connect_timeout_seconds():
    try:
        timeout = int(get_setting("social_connect_timeout_seconds", "60"))
    except (TypeError, ValueError):
        timeout = 60
    return max(10, min(timeout, 120))


def _social_post_gap_seconds():
    try:
        gap = int(get_setting("social_post_gap_seconds", "20"))
    except (TypeError, ValueError):
        gap = 20
    return max(0, min(gap, 120))


def _social_context(title, wp_url, source_url, mode, target):
    return {
        "title": title or "Untitled",
        "wp_url": wp_url or "",
        "source_url": source_url or "",
        "mode": mode or "",
        "target_name": target["name"] if target else "",
        "target_type": target["target_type"] if target else "",
    }


def _refresh_post_social_status(post_id):
    conn = get_db()
    rows = conn.execute(
        "SELECT status, story_status FROM social_posts WHERE post_id = ?",
        (post_id,)
    ).fetchall()
    if not rows:
        status = "no_targets"
    else:
        statuses = {row["status"] for row in rows}
        story_statuses = {row["story_status"] for row in rows if row["story_status"] and row["story_status"] != "disabled"}
        active_statuses = statuses.union(story_statuses)
        if active_statuses.intersection({"pending", "running", "comment_pending"}):
            status = "queued"
        elif active_statuses.intersection({"failed", "comment_failed"}):
            status = "needs_attention"
        elif "review" in active_statuses:
            status = "review"
        elif "manual" in active_statuses:
            status = "manual"
        elif active_statuses == {"posted"}:
            status = "posted"
        elif "posted" in active_statuses:
            status = "partly_posted"
        else:
            status = sorted(active_statuses)[0] if active_statuses else "not_queued"

    conn.execute("UPDATE posts SET social_status = ? WHERE id = ?", (status, post_id))
    conn.commit()
    conn.close()
    return status


def _enqueue_social_for_post(local_post_id, job_id, result, mode):
    """Create social queue rows for every enabled target after a WP post is published."""
    title = result.get("title", "Untitled")
    wp_url = result.get("wp_url", "") or result.get("post_url", "")
    source_url = result.get("source_url", "")
    flyer_url = result.get("flyer_url", "")
    story_flyer_url = result.get("story_flyer_url", "") or flyer_url

    if not wp_url:
        conn = get_db()
        conn.execute("UPDATE posts SET social_status = 'no_wp_url' WHERE id = ?", (local_post_id,))
        conn.commit()
        conn.close()
        return {"created": 0, "status": "no_wp_url"}

    if not flyer_url:
        conn = get_db()
        conn.execute("UPDATE posts SET social_status = 'no_flyer' WHERE id = ?", (local_post_id,))
        conn.commit()
        conn.close()
        return {"created": 0, "status": "no_flyer"}

    conn = get_db()
    targets = conn.execute(
        "SELECT * FROM social_targets WHERE enabled = 1 ORDER BY target_type, name"
    ).fetchall()
    if not targets:
        conn.execute("UPDATE posts SET social_status = 'no_targets' WHERE id = ?", (local_post_id,))
        conn.commit()
        conn.close()
        return {"created": 0, "status": "no_targets"}

    global_auto_post = _truthy(get_setting("social_auto_post", "true"))
    default_caption = get_setting("social_caption_template", "{title}")
    first_comment_template = get_setting("social_first_comment_template", "Read more: {wp_url}")
    status_counts = collections.Counter()
    created = 0

    for target in targets:
        context = _social_context(title, wp_url, source_url, mode, target)
        caption_template = target["caption_template"] or default_caption
        caption = _render_social_template(caption_template, context, title)
        comment_text = _render_social_template(first_comment_template, context, wp_url)

        if target["target_type"] == "facebook_page":
            status = "pending" if global_auto_post and target["auto_post"] else "review"
            story_status = "pending" if target["post_to_story"] and story_flyer_url else "disabled"
        elif target["target_type"] == "facebook_group":
            status = "manual"
            story_status = "disabled"
        else:
            status = "review"
            story_status = "disabled"

        cursor = conn.execute("""
            INSERT OR IGNORE INTO social_posts (
                post_id, job_id, target_id, platform, target_type, status,
                caption, comment_text, media_url, wp_url, story_status, story_media_url
            )
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            local_post_id,
            job_id,
            target["id"],
            target["platform"],
            target["target_type"],
            status,
            caption,
            comment_text,
            flyer_url,
            wp_url,
            story_status,
            story_flyer_url,
        ))
        if cursor.rowcount:
            created += 1
            status_counts[status] += 1
            status_counts[f"story_{story_status}"] += 1
        elif story_status == "pending":
            updated = conn.execute("""
                UPDATE social_posts
                SET story_status = 'pending',
                    story_media_url = COALESCE(NULLIF(story_media_url, ''), ?),
                    updated_at = datetime('now')
                WHERE post_id = ? AND target_id = ? AND story_status = 'disabled'
            """, (story_flyer_url, local_post_id, target["id"])).rowcount
            if updated:
                status_counts["story_pending"] += 1

    conn.commit()
    conn.close()
    final_status = _refresh_post_social_status(local_post_id)
    if status_counts["pending"] or status_counts["story_pending"]:
        _SOCIAL_WORK_AVAILABLE.set()
    print(f"[SOCIAL] Queued {created} social target(s) for post #{local_post_id} ({final_status})")
    return {"created": created, "status": final_status, "status_counts": dict(status_counts)}


def _process_pending_social_story():
    conn = get_db()
    row = conn.execute("""
        SELECT sp.*, st.name AS target_name, st.external_id, st.access_token
        FROM social_posts sp
        JOIN social_targets st ON st.id = sp.target_id
        WHERE sp.story_status = 'pending'
          AND sp.target_type = 'facebook_page'
          AND st.enabled = 1
          AND st.post_to_story = 1
        ORDER BY sp.created_at ASC
        LIMIT 1
    """).fetchone()
    if not row:
        conn.close()
        return False

    social_post_id = row["id"]
    post_id = row["post_id"]
    conn.execute("""
        UPDATE social_posts
        SET story_status = 'running',
            story_attempt_count = story_attempt_count + 1,
            story_error = NULL,
            updated_at = datetime('now')
        WHERE id = ? AND story_status = 'pending'
    """, (social_post_id,))
    conn.commit()
    conn.close()

    try:
        graph_version = get_setting("social_graph_version", "v25.0")
        print(f"[SOCIAL] Publishing Story for post #{post_id} to Facebook Page: {row['target_name']}")
        result = publish_facebook_page_photo_story(
            graph_version=graph_version,
            page_id=row["external_id"],
            page_access_token=row["access_token"] or "",
            story_image_url=row["story_media_url"] or row["media_url"] or "",
            timeout=_social_request_timeout_seconds(),
            connect_timeout=_social_connect_timeout_seconds(),
        )
        story_id = result.get("facebook_story_id", "")
        story_url = f"https://www.facebook.com/{story_id}" if story_id else ""

        conn = get_db()
        conn.execute("""
            UPDATE social_posts
            SET story_status = 'posted',
                external_story_id = ?,
                external_story_photo_id = ?,
                story_posted_at = datetime('now'),
                story_completed_at = datetime('now'),
                external_post_url = COALESCE(NULLIF(external_post_url, ''), ?),
                story_error = NULL,
                updated_at = datetime('now')
            WHERE id = ?
        """, (
            story_id,
            result.get("facebook_story_photo_id", ""),
            story_url,
            social_post_id,
        ))
        conn.commit()
        conn.close()
        _refresh_post_social_status(post_id)
        print(f"[SOCIAL] Published Facebook Page Story: {story_id}")
        return True

    except (SocialPublishError, Exception) as e:
        error = str(e)[:2000]
        conn = get_db()
        conn.execute("""
            UPDATE social_posts
            SET story_status = 'failed',
                story_error = ?,
                updated_at = datetime('now'),
                story_completed_at = datetime('now')
            WHERE id = ?
        """, (error, social_post_id))
        conn.commit()
        conn.close()
        _refresh_post_social_status(post_id)
        print(f"[SOCIAL] Facebook Story publish failed for queue #{social_post_id}: {error[:200]}")
        return True


def _process_pending_social_posts():
    conn = get_db()
    row = conn.execute("""
        SELECT sp.*, st.name AS target_name, st.external_id, st.access_token
        FROM social_posts sp
        JOIN social_targets st ON st.id = sp.target_id
        WHERE sp.status IN ('pending', 'comment_pending')
          AND sp.target_type = 'facebook_page'
          AND st.enabled = 1
        ORDER BY sp.created_at ASC
        LIMIT 1
    """).fetchone()
    if not row:
        conn.close()
        return _process_pending_social_story()

    social_post_id = row["id"]
    post_id = row["post_id"]
    conn.execute("""
        UPDATE social_posts
        SET status = 'running',
            attempt_count = attempt_count + 1,
            error = NULL,
            updated_at = datetime('now')
        WHERE id = ? AND status IN ('pending', 'comment_pending')
    """, (social_post_id,))
    conn.commit()
    conn.close()

    try:
        graph_version = get_setting("social_graph_version", "v25.0")
        timeout = _social_request_timeout_seconds()
        connect_timeout = _social_connect_timeout_seconds()
        if row["external_post_id"]:
            print(f"[SOCIAL] Adding first comment to Facebook post {row['external_post_id']}")
            comment_data = post_facebook_comment(
                graph_version=graph_version,
                facebook_post_id=row["external_post_id"],
                page_access_token=row["access_token"] or "",
                message=row["comment_text"] or "",
                timeout=timeout,
                connect_timeout=connect_timeout,
            )
            result = {
                "facebook_post_id": row["external_post_id"],
                "facebook_comment_id": comment_data.get("id", ""),
            }
        else:
            print(f"[SOCIAL] Publishing post #{post_id} to Facebook Page: {row['target_name']}")
            result = publish_facebook_page_photo_with_first_comment(
                graph_version=graph_version,
                page_id=row["external_id"],
                page_access_token=row["access_token"] or "",
                flyer_url=row["media_url"] or "",
                caption=row["caption"] or "",
                first_comment=row["comment_text"] or "",
                timeout=timeout,
                connect_timeout=connect_timeout,
            )
        external_post_id = result.get("facebook_post_id", "")
        external_url = f"https://www.facebook.com/{external_post_id}" if external_post_id else ""

        conn = get_db()
        conn.execute("""
            UPDATE social_posts
            SET status = 'posted',
                external_post_id = ?,
                external_comment_id = ?,
                external_post_url = ?,
                posted_at = datetime('now'),
                completed_at = datetime('now'),
                updated_at = datetime('now'),
                error = NULL
            WHERE id = ?
        """, (
            external_post_id,
            result.get("facebook_comment_id", ""),
            external_url,
            social_post_id,
        ))
        conn.commit()
        conn.close()
        _refresh_post_social_status(post_id)
        print(f"[SOCIAL] Published Facebook Page post: {external_post_id}")
        return True

    except PartialFacebookPublishError as e:
        error = str(e)[:2000]
        partial = e.partial_result
        external_post_id = partial.get("facebook_post_id", "")
        external_url = f"https://www.facebook.com/{external_post_id}" if external_post_id else ""
        conn = get_db()
        conn.execute("""
            UPDATE social_posts
            SET status = 'comment_failed',
                external_post_id = ?,
                external_comment_id = '',
                external_post_url = ?,
                posted_at = COALESCE(posted_at, datetime('now')),
                error = ?,
                updated_at = datetime('now'),
                completed_at = datetime('now')
            WHERE id = ?
        """, (external_post_id, external_url, error, social_post_id))
        conn.commit()
        conn.close()
        _refresh_post_social_status(post_id)
        print(f"[SOCIAL] Photo posted but first comment failed for queue #{social_post_id}: {error[:200]}")
        return True

    except (SocialPublishError, Exception) as e:
        error = str(e)[:2000]
        conn = get_db()
        conn.execute("""
            UPDATE social_posts
            SET status = 'failed',
                error = ?,
                updated_at = datetime('now'),
                completed_at = datetime('now')
            WHERE id = ?
        """, (error, social_post_id))
        conn.commit()
        conn.close()
        _refresh_post_social_status(post_id)
        print(f"[SOCIAL] Facebook publish failed for queue #{social_post_id}: {error[:200]}")
        return True


def _process_all_pending_social():
    while True:
        if not _process_pending_social_posts():
            break
        gap = _social_post_gap_seconds()
        if gap:
            print(f"[SOCIAL] Waiting {gap}s before the next Facebook Page action...")
            time.sleep(gap)

# ──────────────────────────────────────────────
# Pipeline Wrappers
# ──────────────────────────────────────────────

def _apply_settings_to_env():
    """
    Sync DB settings to environment — but ONLY for vars that aren't already set.
    This protects deployment secrets (Hugging Face Space secrets, Render env vars)
    from being overwritten by stale or empty DB values.
    """
    settings = get_all_settings()
    for k in ["wp_url", "wp_username", "wp_app_password", "wp_category_id", "wp_category_news", "wp_category_slay", "wp_post_status", "ig_username", "ig_password", "llm_model", "google_api_key", "openrouter_api_key"]:
        env_key = k.upper()
        if not os.environ.get(env_key):  # Only set if not already set from deployment
            val = settings.get(k, "")
            if val:
                os.environ[env_key] = val

def run_upload_pipeline(video_path):
    _apply_settings_to_env()
    import main as m
    return m.process_uploaded_video(video_path)

def run_image_upload_pipeline(image_path):
    _apply_settings_to_env()
    import main as m
    return m.process_uploaded_image(image_path)

# ──────────────────────────────────────────────
# Job Processor (with timeout + cancellation)
# ──────────────────────────────────────────────

def _fail_job(job_id, error_msg, url="", mode=""):
    """Centralized helper to mark a job as failed."""
    conn = get_db()
    conn.execute(
        "UPDATE jobs SET status = 'failed', finished_at = datetime('now'), error = ? WHERE id = ?",
        (str(error_msg)[:2000], job_id)
    )
    conn.commit()
    conn.close()
    print(f"[AGENT] ❌ Job #{job_id} failed: {str(error_msg)[:200]}")
    if url:
        send_notification("failed", url, str(error_msg)[:200], mode)


def _run_job_pipeline(url_data, mode):
    """Execute the actual job pipeline. Runs inside a timeout-controlled thread."""
    if mode == "article":
        from article_rewriter import process_article
        return process_article(url_data)
    elif mode in ["instagram", "social"]:
        from main import process_video
        _apply_settings_to_env()
        return process_video(url_data)
    elif mode in ["upload", "upload_video"]:
        return run_upload_pipeline(url_data)
    elif mode == "upload_image":
        return run_image_upload_pipeline(url_data)
    else:
        raise ValueError(f"Unknown mode: {mode}")


def process_pending_jobs():
    """Pick up the next pending job and execute it with a timeout."""
    global _current_job_future, _current_job_id, LAST_LLM_JOB_TIME

    if STOP_PROCESSING.is_set():
        return False

    conn = get_db()
    job = conn.execute("SELECT * FROM jobs WHERE status = 'pending' ORDER BY created_at ASC LIMIT 1").fetchone()
    conn.close()
    if not job:
        return False

    job_id = job["id"]
    url = job["url"]
    mode = job["mode"]

    # Mark as running
    conn = get_db()
    conn.execute("UPDATE jobs SET status = 'running', started_at = datetime('now') WHERE id = ?", (job_id,))
    conn.commit()
    conn.close()

    # Parse URL data
    try:
        url_data = json.loads(url)
        if not isinstance(url_data, (list, dict)):
            url_data = url
    except (json.JSONDecodeError, TypeError):
        url_data = url

    # Clean URL for logging and UI displaying
    clean_url_str = url
    if isinstance(url_data, dict):
        clean_url_str = url_data.get("url") or (url_data.get("urls")[0] if url_data.get("urls") else str(url))
    elif isinstance(url_data, list) and url_data:
        clean_url_str = str(url_data[0])

    print(f"[AGENT] Processing job #{job_id}: {mode} (timeout: {JOB_TIMEOUT_SECONDS}s)")

    # Execute pipeline with hard timeout using ThreadPoolExecutor
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=1, thread_name_prefix=f"job-{job_id}")
    future = executor.submit(_run_job_pipeline, url_data, mode)

    # Track the running job so it can be cancelled externally
    with _current_job_lock:
        _current_job_future = future
        _current_job_id = job_id

    try:
        result = future.result(timeout=JOB_TIMEOUT_SECONDS)

        # Success — save result
        conn = get_db()
        conn.execute(
            "UPDATE jobs SET status = 'done', finished_at = datetime('now'), result_json = ? WHERE id = ?",
            (json.dumps(result), job_id)
        )
        wp_url = result.get("wp_url", "") or result.get("post_url", "")
        final_source_url = result.get("source_url") or clean_url_str
        flyer_url = result.get("flyer_url", "")
        cursor = conn.execute(
            """
            INSERT INTO posts (
                title, wp_url, wp_post_id, source_url, mode, featured_image_url,
                flyer_url, flyer_media_id, story_flyer_url, story_flyer_media_id,
                youtube_count, image_count, social_status
            )
            VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)
            """,
            (result.get("title", "Untitled"), wp_url, result.get("post_id", 0), final_source_url, mode,
             result.get("featured_image_url", ""), flyer_url, result.get("flyer_media_id", 0),
             result.get("story_flyer_url", ""), result.get("story_flyer_media_id", 0),
             result.get("youtube_videos", 0), result.get("image_count", 0), "not_queued")
        )
        local_post_id = cursor.lastrowid
        conn.commit()
        conn.close()
        try:
            _enqueue_social_for_post(local_post_id, job_id, result, mode)
        except Exception as e:
            social_error = str(e)[:500]
            conn = get_db()
            conn.execute("UPDATE posts SET social_status = 'queue_failed' WHERE id = ?", (local_post_id,))
            conn.commit()
            conn.close()
            print(f"[SOCIAL] Queue setup failed for post #{local_post_id}: {social_error}")
        print(f"[AGENT] ✅ Job #{job_id} completed: {result.get('title', 'N/A')}")
        send_notification("published", result.get("title", "Untitled"), wp_url, mode)

    except concurrent.futures.TimeoutError:
        _fail_job(job_id, f"Job timed out after {JOB_TIMEOUT_SECONDS}s (10 min). The pipeline did not finish in time.", clean_url_str, mode)
        future.cancel()

    except concurrent.futures.CancelledError:
        _fail_job(job_id, "Job was cancelled by user.", clean_url_str, mode)

    except Exception as e:
        error_msg = f"{e}\n{traceback.format_exc()}"
        _fail_job(job_id, error_msg, clean_url_str, mode)

    finally:
        # Clear the running job tracker
        with _current_job_lock:
            _current_job_future = None
            _current_job_id = None
        executor.shutdown(wait=False)

    # Track completion time for proactive rate-limit spacing
    LAST_LLM_JOB_TIME = time.time()
    return True


def _process_all_pending():
    """Process all pending jobs sequentially with rate-limit spacing."""
    global LAST_LLM_JOB_TIME
    while True:
        if STOP_PROCESSING.is_set():
            print("[AGENT] ⛔ Stop signal received, halting job processing.")
            break

        # Proactive rate-limit spacing (Gemini 4: 15 RPM = 4s per call)
        MIN_JOB_GAP = 10  # 10s between jobs (gemma_ai.py handles 4s intra-call gap)
        elapsed = time.time() - LAST_LLM_JOB_TIME
        if LAST_LLM_JOB_TIME > 0 and elapsed < MIN_JOB_GAP:
            wait = MIN_JOB_GAP - elapsed
            print(f"[AGENT] ⏳ Proactive cooldown: waiting {wait:.0f}s before next job...")
            time.sleep(wait)

        if not process_pending_jobs():
            break

        time.sleep(1)

# ──────────────────────────────────────────────
# Notifications
# ──────────────────────────────────────────────

def send_notification(event, title, post_url, mode):
    try:
        if get_setting(f"notify_on_{event}", "true") != "true": return
        emoji = "✅" if event == "published" else "❌"
        subject = f"{'Published' if event == 'published' else 'Failed'}: {title[:50]}"
        body = f"{emoji} {mode} job {event}!\nTitle: {title}\nURL: {post_url}"
        if get_setting("notify_email_enabled") == "true": send_email_alert(subject, body)
        if get_setting("notify_telegram_enabled") == "true": send_telegram_alert(f"{emoji} {subject}\n\n{body}")
    except Exception as e: print(f"[AGENT] Notification error: {e}")

def send_email_alert(subject, body):
    try:
        to_addr = get_setting("notify_email_to")
        user = get_setting("notify_email_smtp_user")
        pwd = get_setting("notify_email_smtp_password")
        if not all([to_addr, user, pwd]): return
        msg = MIMEMultipart(); msg["From"] = user; msg["To"] = to_addr; msg["Subject"] = f"🤖 {subject}"
        msg.attach(MIMEText(body, "plain"))
        with smtplib.SMTP(get_setting("notify_email_smtp_host", "smtp.gmail.com"), int(get_setting("notify_email_smtp_port", "587"))) as s:
            s.starttls(); s.login(user, pwd); s.send_message(msg)
    except Exception as e: print(f"[AGENT] Email failed: {e}")

def send_telegram_alert(message):
    try:
        token = get_setting("notify_telegram_bot_token")
        chat_id = get_setting("notify_telegram_chat_id")
        if token and chat_id:
            requests.post(f"https://api.telegram.org/bot{token}/sendMessage", json={"chat_id": chat_id, "text": message, "parse_mode": "HTML"}, timeout=10)
    except Exception as e: print(f"[AGENT] Telegram failed: {e}")

# ──────────────────────────────────────────────
# RSS Feeds
# ──────────────────────────────────────────────

def screen_article(title, summary, criteria):
    try:
        if not os.getenv("GOOGLE_API_KEY"):
            return True
        prompt = (
            f"Decide if this article matches the following criteria: {criteria}\n\n"
            f"Title: {title}\nSummary: {summary}\n\n"
            f"Respond ONLY with JSON: {{\"publish\": true/false, \"reason\": \"brief explanation\"}}"
        )
        resp, retries = gemini_generate(prompt, temperature=0.3, json_mode=True)
        if retries > 0:
            print(f"[AGENT] [RETRY] Screening succeeded after {retries} retry(ies)")
        result = json.loads(resp.text.strip())
        publish = result.get("publish", True)
        reason = result.get("reason", "")
        print(f"[AGENT] Screening: {publish} - {reason}")
        return publish
    except Exception as e:
        print(f"[AGENT] Screening error: {e} — defaulting to True")
        return True

def check_rss_feeds():
    conn = get_db()
    feeds = conn.execute("SELECT * FROM feeds WHERE active = 1").fetchall()
    conn.close()
    screening = get_setting("screening_enabled", "true") == "true"
    criteria = get_setting("screening_criteria")
    new_articles = 0
    for feed in feeds:
        try:
            resp = requests.get(feed["url"], headers={"User-Agent": "Mozilla/5.0"}, timeout=20)
            parsed = feedparser.parse(resp.content)
            for entry in parsed.entries[:10]:
                url = entry.get("link")
                if not url: continue
                conn = get_db()
                if conn.execute("SELECT status FROM processed_urls WHERE url = ?", (url,)).fetchone():
                    conn.close(); continue
                if screening and not screen_article(entry.get("title", ""), entry.get("summary", ""), criteria):
                    conn.execute("INSERT INTO processed_urls (url, status) VALUES (?, ?)", (url, 'rejected'))
                else:
                    review = get_setting("review_feeds_enabled", "true") == "true"
                    status = "review" if review else "pending"
                    conn.execute("INSERT INTO jobs (url, title, mode, status) VALUES (?, ?, 'article', ?)", (url, entry.get("title", "Untitled"), status))
                    conn.execute("INSERT INTO processed_urls (url, status) VALUES (?, ?)", (url, 'queued'))
                    new_articles += 1
                conn.commit(); conn.close()
            conn = get_db(); conn.execute("UPDATE feeds SET last_checked = datetime('now'), articles_found = articles_found + ? WHERE id = ?", (new_articles, feed["id"])); conn.commit(); conn.close()
        except Exception as e: print(f"[AGENT] RSS failed for {feed['url']}: {e}")
    return new_articles

def agent_loop():
    global LAST_LLM_JOB_TIME
    print("[AGENT] 🤖 Agent thread started")
    last_rss = 0
    while True:
        try:
            if get_setting("agent_enabled") == "true":
                if time.time() - last_rss > int(get_setting("check_interval_minutes", "30")) * 60:
                    check_rss_feeds(); last_rss = time.time()
                # Proactive rate-limit spacing in agent loop
                MIN_JOB_GAP = 120
                elapsed = time.time() - LAST_LLM_JOB_TIME
                if LAST_LLM_JOB_TIME > 0 and elapsed < MIN_JOB_GAP:
                    time.sleep(MIN_JOB_GAP - elapsed)
                process_pending_jobs()
            time.sleep(10)
        except Exception as e: print(f"[AGENT] Loop error: {e}"); time.sleep(30)


def _watchdog_loop():
    """Background watchdog: auto-fails jobs stuck in 'running' beyond the timeout."""
    print(f"[WATCHDOG] 🐕 Started (timeout={JOB_TIMEOUT_SECONDS}s, interval={WATCHDOG_INTERVAL}s)")
    while True:
        try:
            conn = get_db()
            stuck = conn.execute("""
                UPDATE jobs SET status = 'failed',
                    error = 'Auto-killed by watchdog: exceeded max runtime of ' || ? || ' seconds',
                    finished_at = datetime('now')
                WHERE status = 'running'
                  AND started_at IS NOT NULL
                  AND (julianday('now') - julianday(started_at)) * 86400 > ?
            """, (JOB_TIMEOUT_SECONDS, JOB_TIMEOUT_SECONDS)).rowcount
            conn.commit()
            conn.close()
            if stuck:
                print(f"[WATCHDOG] ⏰ Auto-failed {stuck} stuck job(s) that exceeded {JOB_TIMEOUT_SECONDS}s")
                # Also cancel the current future if it's still running
                with _current_job_lock:
                    if _current_job_future and not _current_job_future.done():
                        _current_job_future.cancel()
                        print(f"[WATCHDOG] Cancelled running future for job #{_current_job_id}")
        except Exception as e:
            print(f"[WATCHDOG] Error: {e}")
        time.sleep(WATCHDOG_INTERVAL)


def _worker_loop():
    """Single persistent worker thread that processes jobs when signaled."""
    print("[WORKER] 🔧 Worker thread started (single persistent thread)")
    while True:
        # Wait until work is available or check periodically
        _WORK_AVAILABLE.wait(timeout=5)
        _WORK_AVAILABLE.clear()
        if not STOP_PROCESSING.is_set():
            _process_all_pending()


def _social_worker_loop():
    """Persistent worker for Graph API social posting."""
    print("[SOCIAL] Social worker thread started")
    while True:
        _SOCIAL_WORK_AVAILABLE.wait(timeout=10)
        _SOCIAL_WORK_AVAILABLE.clear()
        _process_all_pending_social()

# ──────────────────────────────────────────────
# Routes
# ──────────────────────────────────────────────

@app.before_request
def require_login():
    public_endpoints = {
        'login', 'static', 'health_check', 'api_wp_relay_next',
        'api_wp_relay_file', 'api_wp_relay_result',
    }
    if request.endpoint not in public_endpoints and not session.get('logged_in'):
        return redirect(url_for('login'))


def _wp_relay_authorized() -> bool:
    expected = os.getenv("WP_PULL_RELAY_TOKEN", "")
    supplied = request.headers.get("Authorization", "")
    return bool(expected) and hmac.compare_digest(supplied, f"Bearer {expected}")


@app.route("/api/wp-upload-relay/next", methods=["POST"])
def api_wp_relay_next():
    if not _wp_relay_authorized():
        return jsonify({"error": "Unauthorized"}), 401
    from upload_pull_queue import claim_next

    item = claim_next()
    if item is None:
        response = Response(status=204)
        response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate"
        return response
    response = jsonify(item)
    response.headers["Cache-Control"] = "no-store"
    return response


@app.route("/api/wp-upload-relay/file/<upload_id>", methods=["GET"])
def api_wp_relay_file(upload_id):
    if not _wp_relay_authorized():
        return jsonify({"error": "Unauthorized"}), 401
    from upload_pull_queue import get_file

    item = get_file(upload_id)
    if item is None:
        return jsonify({"error": "Upload not found"}), 404
    file_data, mime_type, filename = item
    response = Response(file_data, mimetype=mime_type)
    response.headers["Content-Disposition"] = f'attachment; filename="{filename}"'
    response.headers["Cache-Control"] = "no-store"
    return response


@app.route("/api/wp-upload-relay/result/<upload_id>", methods=["POST"])
def api_wp_relay_result(upload_id):
    if not _wp_relay_authorized():
        return jsonify({"error": "Unauthorized"}), 401
    from upload_pull_queue import complete

    data = request.get_json(silent=True) or {}
    completed = complete(
        upload_id,
        int(data.get("status", 502)),
        str(data.get("content_type", "application/json")),
        str(data.get("body", ""))[:1_000_000],
    )
    return jsonify({"completed": completed}), 200 if completed else 404

@app.route("/signin", methods=["GET", "POST"])
def login():
    if request.method == "POST":
        expected_username = os.getenv("DASHBOARD_USERNAME", "")
        expected_password = os.getenv("DASHBOARD_PASSWORD", "")
        supplied_username = request.form.get("username", "")
        supplied_password = request.form.get("password", "")
        credentials_configured = bool(expected_username and expected_password)
        credentials_match = (
            credentials_configured
            and hmac.compare_digest(supplied_username, expected_username)
            and hmac.compare_digest(supplied_password, expected_password)
        )
        if credentials_match:
            session.clear()
            session["logged_in"] = True
            return redirect(url_for("dashboard"))
        flash("Invalid credentials", "error")
    return render_template("login.html")

@app.route("/logout")
def logout(): session.clear(); return redirect(url_for("login"))

@app.route("/")
def dashboard(): return render_template("dashboard.html")

@app.route("/api/stats")
def api_stats():
    conn = get_db()
    stats = {
        "total_posts": conn.execute("SELECT COUNT(*) FROM posts").fetchone()[0],
        "today_posts": conn.execute("SELECT COUNT(*) FROM posts WHERE date(created_at) = date('now')").fetchone()[0],
        "queue_size": conn.execute("SELECT COUNT(*) FROM jobs WHERE status IN ('pending', 'running')").fetchone()[0],
        "total_jobs": conn.execute("SELECT COUNT(*) FROM jobs").fetchone()[0],
        "failed_jobs": conn.execute("SELECT COUNT(*) FROM jobs WHERE status = 'failed'").fetchone()[0],
        "active_feeds": conn.execute("SELECT COUNT(*) FROM feeds WHERE active = 1").fetchone()[0],
        "agent_enabled": get_setting("agent_enabled") == "true"
    }
    stats["success_rate"] = round((stats["total_jobs"] - stats["failed_jobs"]) / stats["total_jobs"] * 100) if stats["total_jobs"] > 0 else 100
    conn.close(); return jsonify(stats)

@app.route("/api/jobs", methods=["GET"])
def api_jobs_list():
    status = request.args.get("status", "")
    conn = get_db()
    if status: rows = conn.execute("SELECT * FROM jobs WHERE status = ? ORDER BY created_at DESC LIMIT 50", (status,)).fetchall()
    else: rows = conn.execute("SELECT * FROM jobs ORDER BY created_at DESC LIMIT 50").fetchall()
    conn.close(); return jsonify([dict(r) for r in rows])

@app.route("/api/jobs", methods=["POST"])
def api_jobs_create():
    data = request.get_json()
    urls = data.get("urls", [])
    mode = data.get("mode", "article")
    group_media = data.get("group_media", False)
    is_redo = data.get("is_redo", False)
    if isinstance(urls, str): urls = [u.strip() for u in urls.split("\n") if u.strip() and not u.strip().startswith("#")]
    if not urls: return jsonify({"error": "No URLs provided"}), 400
    conn = get_db()
    if group_media and len(urls) > 1:
        payload = {"urls": urls, "is_redo": is_redo} if is_redo else urls
        conn.execute("INSERT INTO jobs (url, mode) VALUES (?, ?)", (json.dumps(payload), mode))
    else:
        for u in urls:
            payload = {"url": u, "is_redo": is_redo} if is_redo else u
            url_val = json.dumps(payload) if isinstance(payload, dict) else payload
            conn.execute("INSERT INTO jobs (url, mode) VALUES (?, ?)", (url_val, mode))
    conn.commit(); conn.close()
    _WORK_AVAILABLE.set()  # Signal the persistent worker thread
    count = len(urls) if not group_media or len(urls) <= 1 else 1
    return jsonify({"success": True, "created": count}), 201

@app.route("/api/upload", methods=["POST"])
def api_upload():
    """Handle media file uploads. Supports multiple files and group_media flag."""
    files = request.files.getlist("video")
    if not files:
        return jsonify({"error": "No files uploaded"}), 400

    group_media = request.form.get("group_media", "false").lower() in ("true", "1", "yes")

    saved_paths = []
    for f in files:
        if f.filename:
            # Generate unique filename to avoid collisions
            ext = Path(f.filename).suffix.lower()
            unique_name = f"{uuid.uuid4().hex[:12]}_{secure_filename(f.filename)}"
            save_path = UPLOAD_DIR / unique_name
            f.save(str(save_path))
            saved_paths.append(str(save_path))
            print(f"[UPLOAD] Saved: {save_path} ({save_path.stat().st_size / (1024*1024):.1f} MB)")

    if not saved_paths:
        return jsonify({"error": "No valid files uploaded"}), 400

    conn = get_db()
    if group_media and len(saved_paths) > 1:
        # Group all files as ONE job
        conn.execute("INSERT INTO jobs (url, mode) VALUES (?, ?)", (json.dumps(saved_paths), "upload"))
        job_count = 1
        print(f"[UPLOAD] Created 1 grouped job with {len(saved_paths)} files")
    else:
        # Each file is a SEPARATE job
        for p in saved_paths:
            ext = Path(p).suffix.lower()
            mode = "upload_image" if ext in [".jpg", ".jpeg", ".png", ".webp"] else "upload_video"
            conn.execute("INSERT INTO jobs (url, mode) VALUES (?, ?)", (p, mode))
        job_count = len(saved_paths)
        print(f"[UPLOAD] Created {job_count} separate job(s)")
    conn.commit()
    conn.close()

    _WORK_AVAILABLE.set()  # Signal the persistent worker thread
    return jsonify({"success": True, "message": f"{job_count} job(s) created from {len(saved_paths)} file(s)", "created": job_count}), 201


@app.route("/api/agent/toggle", methods=["POST"])
def api_agent_toggle():
    current = get_setting("agent_enabled", "false")
    new_val = "false" if current == "true" else "true"
    set_setting("agent_enabled", new_val)
    return jsonify({"agent_enabled": new_val == "true"})

@app.route("/api/jobs/<int:job_id>", methods=["DELETE"])
def api_job_delete(job_id):
    conn = get_db()
    deleted = conn.execute(
        "DELETE FROM jobs WHERE id = ? AND status IN ('failed', 'review')", (job_id,)
    ).rowcount
    conn.commit(); conn.close()
    return jsonify({"deleted": deleted > 0}), 200 if deleted else 409


@app.route("/api/jobs/<int:job_id>/cancel", methods=["POST"])
def api_job_cancel_pending(job_id):
    conn = get_db()
    cancelled = conn.execute(
        """
        UPDATE jobs
        SET status = 'failed', error = 'Cancelled by user', finished_at = datetime('now')
        WHERE id = ? AND status = 'pending'
        """,
        (job_id,),
    ).rowcount
    conn.commit(); conn.close()
    if not cancelled:
        return jsonify({"error": "Only pending jobs can be cancelled"}), 409
    print(f"[CANCEL] Pending job #{job_id} moved to failed")
    return jsonify({"cancelled": True, "status": "failed"})

@app.route("/api/jobs/<int:job_id>/retry", methods=["POST"])
def api_job_retry(job_id):
    conn = get_db()
    conn.execute("UPDATE jobs SET status = 'pending', error = NULL, started_at = NULL, finished_at = NULL WHERE id = ?", (job_id,))
    conn.commit(); conn.close()
    _WORK_AVAILABLE.set()  # Signal the persistent worker thread
    return jsonify({"retried": True})

@app.route("/api/jobs/<int:job_id>/approve", methods=["POST"])
def api_job_approve(job_id):
    conn = get_db()
    conn.execute("UPDATE jobs SET status = 'pending' WHERE id = ? AND status = 'review'", (job_id,))
    conn.commit(); conn.close()
    _WORK_AVAILABLE.set()  # Signal the persistent worker thread
    return jsonify({"approved": True})

@app.route("/api/jobs/stop", methods=["POST"])
def api_jobs_stop():
    """
    Stop queue processing AND force-fail any running jobs.
    Marks running jobs as 'failed' (not 'pending') to prevent the stuck
    thread from immediately re-grabbing the same job.
    """
    STOP_PROCESSING.set()

    # Cancel the currently running future
    cancelled = 0
    with _current_job_lock:
        if _current_job_future and not _current_job_future.done():
            _current_job_future.cancel()
            cancelled = 1
            print(f"[STOP] Cancelled running future for job #{_current_job_id}")

    # Mark running jobs as FAILED (not pending) to prevent re-grab
    conn = get_db()
    stuck = conn.execute(
        "UPDATE jobs SET status = 'failed', error = 'Manually stopped by user', finished_at = datetime('now') WHERE status = 'running'"
    ).rowcount
    conn.commit(); conn.close()

    # Clear the stop flag after a short delay so future jobs can run
    threading.Timer(3.0, STOP_PROCESSING.clear).start()
    print(f"[STOP] ⛔ Queue stopped. {stuck} running job(s) force-failed.")
    return jsonify({"stopped": True, "cancelled_running": stuck})


@app.route("/api/jobs/<int:job_id>/force-fail", methods=["POST"])
def api_job_force_fail(job_id):
    """
    Force-fail a specific running job. Marks it as failed and cancels
    the running future if it matches the currently executing job.
    """
    conn = get_db()
    updated = conn.execute(
        "UPDATE jobs SET status = 'failed', error = 'Force-killed by user', finished_at = datetime('now') WHERE id = ? AND status = 'running'",
        (job_id,)
    ).rowcount
    conn.commit(); conn.close()

    # If this was the currently running job, cancel its future
    with _current_job_lock:
        if _current_job_id == job_id and _current_job_future and not _current_job_future.done():
            _current_job_future.cancel()
            print(f"[FORCE-KILL] Cancelled running future for job #{job_id}")

    if updated:
        print(f"[FORCE-KILL] ⛔ Job #{job_id} force-failed by user")

    return jsonify({"force_failed": updated > 0})

@app.route("/api/jobs/cancel-all", methods=["POST"])
def api_jobs_cancel_all():
    conn = get_db()
    cancelled = conn.execute(
        """
        UPDATE jobs
        SET status = 'failed', error = 'Cancelled by user', finished_at = datetime('now')
        WHERE status = 'pending'
        """
    ).rowcount
    conn.commit(); conn.close()
    if cancelled:
        print(f"[CANCEL] {cancelled} pending job(s) moved to failed")
    return jsonify({"cancelled": cancelled})

@app.route("/api/posts")
def api_posts():
    conn = get_db()
    rows = conn.execute("SELECT * FROM posts ORDER BY created_at DESC LIMIT 50").fetchall()
    conn.close()
    return jsonify([dict(r) for r in rows])

@app.route("/api/posts/<int:post_id>/social/queue", methods=["POST"])
def api_post_social_queue(post_id):
    conn = get_db()
    post = conn.execute("SELECT * FROM posts WHERE id = ?", (post_id,)).fetchone()
    conn.close()
    if not post:
        return jsonify({"error": "Post not found"}), 404

    result = {
        "title": post["title"],
        "wp_url": post["wp_url"],
        "post_url": post["wp_url"],
        "post_id": post["wp_post_id"],
        "source_url": post["source_url"],
        "featured_image_url": post["featured_image_url"],
        "flyer_url": post["flyer_url"],
        "flyer_media_id": post["flyer_media_id"],
        "story_flyer_url": post["story_flyer_url"],
        "story_flyer_media_id": post["story_flyer_media_id"],
    }
    try:
        queued = _enqueue_social_for_post(post_id, None, result, post["mode"])
    except Exception as e:
        return jsonify({"error": str(e)[:500]}), 500
    return jsonify({"queued": True, **queued})

@app.route("/api/social/targets", methods=["GET"])
def api_social_targets_list():
    conn = get_db()
    rows = conn.execute("SELECT * FROM social_targets ORDER BY target_type, name").fetchall()
    conn.close()
    targets = []
    for row in rows:
        item = dict(row)
        item["has_access_token"] = bool(item.get("access_token"))
        item["access_token_masked"] = _mask_secret(item.get("access_token"))
        item.pop("access_token", None)
        targets.append(item)
    return jsonify(targets)

@app.route("/api/social/targets", methods=["POST"])
def api_social_targets_create():
    data = request.get_json(silent=True) or {}
    target_type = str(data.get("target_type", "facebook_page")).strip()
    name = str(data.get("name", "")).strip()
    external_id = str(data.get("external_id", "")).strip()
    access_token = str(data.get("access_token", "")).strip()

    if target_type not in ("facebook_page", "facebook_group"):
        return jsonify({"error": "Target type must be facebook_page or facebook_group"}), 400
    if not name or not external_id:
        return jsonify({"error": "Name and Page ID/group URL are required"}), 400
    if target_type == "facebook_page" and not access_token:
        return jsonify({"error": "Facebook Page targets require a Page access token"}), 400
    if target_type == "facebook_group" and not external_id.lower().startswith(("http://", "https://")):
        return jsonify({"error": "Facebook Group targets need a group URL"}), 400

    conn = get_db()
    try:
        conn.execute("""
            INSERT INTO social_targets (
                platform, target_type, name, external_id, access_token,
                enabled, auto_post, post_to_story, caption_template
            )
            VALUES ('facebook', ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            target_type,
            name,
            external_id,
            access_token,
            1 if _truthy(data.get("enabled", "true")) else 0,
            1 if _truthy(data.get("auto_post", "true")) else 0,
            1 if target_type == "facebook_page" and _truthy(data.get("post_to_story", "false")) else 0,
            str(data.get("caption_template", "")).strip(),
        ))
        conn.commit()
    except sqlite3.IntegrityError:
        conn.close()
        return jsonify({"error": "That social target already exists"}), 409
    conn.close()
    return jsonify({"created": True}), 201

@app.route("/api/social/targets/<int:target_id>", methods=["PATCH"])
def api_social_targets_update(target_id):
    data = request.get_json(silent=True) or {}
    allowed = {"name", "external_id", "caption_template"}
    updates = []
    params = []

    for key in allowed:
        if key in data:
            updates.append(f"{key} = ?")
            params.append(str(data.get(key, "")).strip())

    if "enabled" in data:
        updates.append("enabled = ?")
        params.append(1 if _truthy(data.get("enabled")) else 0)
    if "auto_post" in data:
        updates.append("auto_post = ?")
        params.append(1 if _truthy(data.get("auto_post")) else 0)
    if "post_to_story" in data:
        updates.append("post_to_story = ?")
        params.append(1 if _truthy(data.get("post_to_story")) else 0)
    if "access_token" in data and "••••••" not in str(data.get("access_token", "")):
        updates.append("access_token = ?")
        params.append(str(data.get("access_token", "")).strip())

    if not updates:
        return jsonify({"updated": False})

    updates.append("updated_at = datetime('now')")
    params.append(target_id)
    conn = get_db()
    conn.execute(f"UPDATE social_targets SET {', '.join(updates)} WHERE id = ?", params)
    conn.commit()
    conn.close()
    return jsonify({"updated": True})

@app.route("/api/social/targets/<int:target_id>", methods=["DELETE"])
def api_social_targets_delete(target_id):
    conn = get_db()
    conn.execute("DELETE FROM social_targets WHERE id = ?", (target_id,))
    conn.commit()
    conn.close()
    return jsonify({"deleted": True})

@app.route("/api/social/posts", methods=["GET"])
def api_social_posts_list():
    conn = get_db()
    rows = conn.execute("""
        SELECT
            sp.*,
            p.title,
            p.source_url,
            st.name AS target_name,
            st.external_id AS target_external_id,
            st.enabled AS target_enabled
        FROM social_posts sp
        JOIN posts p ON p.id = sp.post_id
        JOIN social_targets st ON st.id = sp.target_id
        ORDER BY sp.created_at DESC
        LIMIT 200
    """).fetchall()
    conn.close()
    return jsonify([dict(row) for row in rows])

@app.route("/api/social/posts/<int:social_post_id>/retry", methods=["POST"])
def api_social_post_retry(social_post_id):
    conn = get_db()
    row = conn.execute("SELECT * FROM social_posts WHERE id = ?", (social_post_id,)).fetchone()
    if not row:
        conn.close()
        return jsonify({"error": "Social queue item not found"}), 404
    if row["target_type"] != "facebook_page":
        conn.close()
        return jsonify({"error": "Only Facebook Page posts can be retried automatically"}), 400

    next_status = "comment_pending" if row["external_post_id"] else "pending"
    retry_feed = row["status"] in ("failed", "review", "comment_failed")
    retry_story = row["story_status"] == "failed"

    if retry_feed:
        conn.execute("""
        UPDATE social_posts
        SET status = ?,
            error = NULL,
            completed_at = NULL,
            updated_at = datetime('now')
        WHERE id = ?
        """, (next_status, social_post_id))
    if retry_story:
        conn.execute("""
            UPDATE social_posts
            SET story_status = 'pending',
                story_error = NULL,
                story_completed_at = NULL,
                updated_at = datetime('now')
            WHERE id = ?
        """, (social_post_id,))
    if not retry_feed and not retry_story:
        conn.close()
        return jsonify({"error": "Nothing retryable on this social queue item"}), 400
    conn.commit()
    conn.close()
    _refresh_post_social_status(row["post_id"])
    _SOCIAL_WORK_AVAILABLE.set()
    return jsonify({"retried": True})

@app.route("/api/social/posts/<int:social_post_id>/mark-posted", methods=["POST"])
def api_social_post_mark_posted(social_post_id):
    data = request.get_json(silent=True) or {}
    conn = get_db()
    row = conn.execute("SELECT post_id FROM social_posts WHERE id = ?", (social_post_id,)).fetchone()
    if not row:
        conn.close()
        return jsonify({"error": "Social queue item not found"}), 404
    conn.execute("""
        UPDATE social_posts
        SET status = 'posted',
            external_post_url = COALESCE(NULLIF(?, ''), external_post_url),
            error = NULL,
            posted_at = datetime('now'),
            completed_at = datetime('now'),
            updated_at = datetime('now')
        WHERE id = ?
    """, (data.get("external_post_url", "").strip(), social_post_id))
    conn.commit()
    conn.close()
    _refresh_post_social_status(row["post_id"])
    return jsonify({"posted": True})

@app.route("/api/feeds", methods=["GET"])
def api_feeds_list():
    conn = get_db()
    rows = conn.execute("SELECT * FROM feeds ORDER BY created_at DESC").fetchall()
    conn.close()
    return jsonify([dict(r) for r in rows])

@app.route("/api/feeds", methods=["POST"])
def api_feeds_add():
    data = request.get_json()
    name = data.get("name", "").strip()
    url = data.get("url", "").strip()
    if not name or not url:
        return jsonify({"error": "Name and URL required"}), 400
    conn = get_db()
    try:
        conn.execute("INSERT INTO feeds (name, url) VALUES (?, ?)", (name, url))
        conn.commit()
    except sqlite3.IntegrityError:
        conn.close()
        return jsonify({"error": "Feed URL already exists"}), 409
    conn.close()
    return jsonify({"added": True}), 201

@app.route("/api/feeds/<int:feed_id>", methods=["DELETE"])
def api_feed_delete(feed_id):
    conn = get_db()
    conn.execute("DELETE FROM feeds WHERE id = ?", (feed_id,))
    conn.commit(); conn.close()
    return jsonify({"deleted": True})

@app.route("/api/feeds/<int:feed_id>/toggle", methods=["POST"])
def api_feed_toggle(feed_id):
    conn = get_db()
    conn.execute("UPDATE feeds SET active = CASE WHEN active = 1 THEN 0 ELSE 1 END WHERE id = ?", (feed_id,))
    conn.commit(); conn.close()
    return jsonify({"toggled": True})

@app.route("/api/settings", methods=["GET"])
def api_settings_get():
    s = get_all_settings()
    for k in ["wp_app_password", "ig_password", "google_api_key", "openrouter_api_key"]:
        if s.get(k): s[k] = "••••••" + s[k][-4:]
    return jsonify(s)

@app.route("/api/settings", methods=["POST"])
def api_settings_update():
    data = request.get_json()
    # Batch all settings into a single transaction to avoid 30+ separate
    # DB connections — critical on HF Spaces where /data is a network volume
    conn = get_db()
    try:
        for k, v in data.items():
            if "••••••" not in str(v):
                conn.execute(
                    "INSERT INTO settings (key, value) VALUES (?, ?) "
                    "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
                    (k, str(v))
                )
        conn.commit()
    finally:
        conn.close()
    return jsonify({"updated": True})

@app.route("/api/logs", methods=["GET"])
def api_logs_get(): return jsonify({"logs": list(LOG_BUFFER)})

@app.route("/api/logs/clear", methods=["POST"])
def api_logs_clear():
    LOG_BUFFER.clear()
    return jsonify({"cleared": True})

@app.route("/api/test-notification", methods=["POST"])
def api_test_notification():
    results = {}
    if get_setting("notify_email_enabled") == "true":
        try:
            send_email_alert("🔔 Test Notification", "This is a test from Content Agent!")
            results["email"] = {"ok": True, "message": "Email sent successfully!"}
        except Exception as e:
            results["email"] = {"ok": False, "error": str(e)}
    else:
        results["email"] = {"ok": False, "error": "Email notifications disabled"}
    if get_setting("notify_telegram_enabled") == "true":
        try:
            send_telegram_alert("🔔 Test notification from Content Agent!")
            results["telegram"] = {"ok": True, "message": "Telegram sent successfully!"}
        except Exception as e:
            results["telegram"] = {"ok": False, "error": str(e)}
    else:
        results["telegram"] = {"ok": False, "error": "Telegram notifications disabled"}
    return jsonify(results)

@app.route("/health")
@app.route("/api/health")
def health_check(): return jsonify({"status": "ok"}), 200

# ──────────────────────────────────────────────
# Main
# ──────────────────────────────────────────────

_started = False
def _start_agent():
    global _started
    if not _started: 
        _started = True
        threading.Thread(target=agent_loop, daemon=True, name="agent-loop").start()
        threading.Thread(target=_watchdog_loop, daemon=True, name="watchdog").start()
        threading.Thread(target=_worker_loop, daemon=True, name="job-worker").start()
        threading.Thread(target=_social_worker_loop, daemon=True, name="social-worker").start()
        _SOCIAL_WORK_AVAILABLE.set()

print("[STARTUP] Initializing database...")
init_db()

# Ensure working directories exist (downloads, output, uploads)
from pathlib import Path
_DATA_DIR = Path("/data") if Path("/data").is_dir() or os.getenv("HUGGING_FACE_SPACES") else Path(__file__).parent
if _DATA_DIR == Path("/data"):
    _DATA_DIR.mkdir(parents=True, exist_ok=True)
(_DATA_DIR / "downloads").mkdir(exist_ok=True)
(_DATA_DIR / "output").mkdir(exist_ok=True)
(_DATA_DIR / "uploads").mkdir(exist_ok=True)
print(f"[STARTUP] Working directories ensured: {_DATA_DIR}")

# Recover any jobs stuck in 'running' from a previous crash/deploy
_conn = get_db()
_recovered = _conn.execute("UPDATE jobs SET status = 'pending', started_at = NULL WHERE status = 'running'").rowcount
_conn.commit(); _conn.close()
if _recovered:
    print(f"[STARTUP] Recovered {_recovered} stuck 'running' job(s) → reset to 'pending'")

# AI Engine status
if not os.getenv("GOOGLE_API_KEY") and not os.getenv("OPENROUTER_API_KEY"):
    print("[WARN] Neither GOOGLE_API_KEY nor OPENROUTER_API_KEY is set. All AI operations will fail.")
    print("[WARN] Set at least one API key in .env or environment to enable AI features.")
elif not os.getenv("GOOGLE_API_KEY"):
    print("[INFO] GOOGLE_API_KEY not set. Using OpenRouter as primary AI engine.")
elif not os.getenv("OPENROUTER_API_KEY"):
    print("[INFO] OPENROUTER_API_KEY not set. Backup fallback (Nemotron) will be unavailable.")
print(f"[STARTUP] Active AI model: {os.getenv('LLM_MODEL', 'gemma-4-31b-it')}")

@app.before_request
def _ensure_agent(): _start_agent()

if __name__ == "__main__":
    port = int(os.environ.get("PORT", 5000))
    _start_agent(); app.run(host="0.0.0.0", port=port, debug=False)