| """ |
| FB Auto Poster |
| ============== |
| AI image generation + scheduled automatic posting to a Facebook Page. |
| |
| Designed to run as a Hugging Face Gradio Space. |
| |
| What it does |
| ------------ |
| 1. Aap bahut saare prompts (ek line mein ek) daalte hain. |
| 2. Bot har prompt se AI image generate karta hai (Hugging Face Inference API). |
| 3. Har post ek queue mein chala jaata hai (image + caption). |
| 4. Ek scheduler roz set kiye gaye time par queue se agli post apne aap |
| Facebook Page par upload kar deta hai. |
| |
| Required Space secrets / env vars |
| --------------------------------- |
| FB_PAGE_ID -> aapke Facebook Page ka numeric ID |
| FB_PAGE_ACCESS_TOKEN -> long-lived Page access token (setup guide dekhein) |
| HF_TOKEN -> Hugging Face token (image generate karne ke liye) |
| |
| Optional env vars |
| ----------------- |
| IMAGE_MODEL (default: black-forest-labs/FLUX.1-schnell) |
| POST_TIMES (default: 09:00,14:00,19:00 -> IST) |
| GRAPH_VERSION (default: v21.0) |
| """ |
|
|
| import os |
| import io |
| import json |
| import time |
| import logging |
| import threading |
| import datetime |
|
|
| import requests |
| import gradio as gr |
| from PIL import Image |
| from huggingface_hub import InferenceClient |
| from apscheduler.schedulers.background import BackgroundScheduler |
| from apscheduler.triggers.cron import CronTrigger |
|
|
| |
| |
| |
| FB_PAGE_ID = os.getenv("FB_PAGE_ID", "").strip() |
| FB_TOKEN = os.getenv("FB_PAGE_ACCESS_TOKEN", "").strip() |
| HF_TOKEN = os.getenv("HF_TOKEN", "").strip() |
|
|
| IMAGE_MODEL = os.getenv("IMAGE_MODEL", "black-forest-labs/FLUX.1-schnell").strip() |
| GRAPH_VER = os.getenv("GRAPH_VERSION", "v21.0").strip() |
| DEFAULT_TIMES = os.getenv("POST_TIMES", "09:00,14:00,19:00").strip() |
| TIMEZONE = os.getenv("TIMEZONE", "Asia/Kolkata").strip() |
|
|
| STATE_FILE = os.getenv("STATE_FILE", "state.json") |
| IMAGES_DIR = os.getenv("IMAGES_DIR", "generated_images") |
| os.makedirs(IMAGES_DIR, exist_ok=True) |
|
|
| |
| |
| |
| LOG_BUFFER = [] |
|
|
|
|
| class BufferHandler(logging.Handler): |
| def emit(self, record): |
| LOG_BUFFER.append(self.format(record)) |
| del LOG_BUFFER[:-400] |
|
|
|
|
| logging.basicConfig( |
| level=logging.INFO, |
| format="%(asctime)s | %(levelname)s | %(message)s", |
| handlers=[logging.StreamHandler(), BufferHandler()], |
| ) |
| log = logging.getLogger("autoposter") |
|
|
| |
| |
| |
| def _default_state(): |
| return { |
| "queue": [], |
| "times": [t.strip() for t in DEFAULT_TIMES.split(",") if t.strip()], |
| "enabled": False, |
| "history": [], |
| } |
|
|
|
|
| def load_state(): |
| if os.path.exists(STATE_FILE): |
| try: |
| with open(STATE_FILE) as f: |
| data = json.load(f) |
| data.setdefault("queue", []) |
| data.setdefault("times", _default_state()["times"]) |
| data.setdefault("enabled", False) |
| data.setdefault("history", []) |
| return data |
| except Exception as e: |
| log.warning(f"state.json padha nahi ja saka, naya bana rahe hain: {e}") |
| return _default_state() |
|
|
|
|
| def save_state(state): |
| with open(STATE_FILE, "w") as f: |
| json.dump(state, f, indent=2, ensure_ascii=False) |
|
|
|
|
| STATE = load_state() |
| LOCK = threading.RLock() |
|
|
|
|
| def new_id(): |
| return int(time.time() * 1000) % 10**12 |
|
|
|
|
| |
| |
| |
| _client = None |
|
|
|
|
| def get_client(): |
| global _client |
| if _client is None: |
| _client = InferenceClient(token=HF_TOKEN or None) |
| return _client |
|
|
|
|
| def generate_image(prompt): |
| """Prompt se ek AI image banata hai, disk par save karta hai.""" |
| prompt = (prompt or "").strip() |
| if not prompt: |
| raise ValueError("Prompt khaali hai") |
| log.info(f"Image generate ho rahi hai: {prompt[:70]}") |
| result = get_client().text_to_image(prompt, model=IMAGE_MODEL) |
| if isinstance(result, (bytes, bytearray)): |
| img = Image.open(io.BytesIO(result)) |
| elif isinstance(result, Image.Image): |
| img = result |
| else: |
| img = Image.open(io.BytesIO(result)) |
| path = os.path.join(IMAGES_DIR, f"img_{new_id()}.png") |
| img.convert("RGB").save(path) |
| log.info(f"Image ban gayi -> {path}") |
| return path, img |
|
|
|
|
| |
| |
| |
| def _fb_check_config(): |
| if not FB_PAGE_ID or not FB_TOKEN: |
| raise RuntimeError( |
| "FB_PAGE_ID ya FB_PAGE_ACCESS_TOKEN set nahi hai. " |
| "Space ke Settings -> Secrets mein daalein." |
| ) |
|
|
|
|
| def _fb_result(resp): |
| try: |
| data = resp.json() |
| except Exception: |
| data = {"raw": resp.text} |
| if resp.status_code >= 400: |
| log.error(f"Facebook error {resp.status_code}: {data}") |
| raise RuntimeError(f"Facebook error {resp.status_code}: {data}") |
| log.info(f"Facebook post publish ho gaya: {data}") |
| return data |
|
|
|
|
| def fb_post_photo(image_path, caption): |
| """Image + caption ko page par post karta hai.""" |
| _fb_check_config() |
| url = f"https://graph.facebook.com/{GRAPH_VER}/{FB_PAGE_ID}/photos" |
| with open(image_path, "rb") as f: |
| resp = requests.post( |
| url, |
| data={"message": caption or "", "access_token": FB_TOKEN}, |
| files={"source": f}, |
| timeout=180, |
| ) |
| return _fb_result(resp) |
|
|
|
|
| def fb_post_text(caption): |
| """Sirf text post (bina image).""" |
| _fb_check_config() |
| url = f"https://graph.facebook.com/{GRAPH_VER}/{FB_PAGE_ID}/feed" |
| resp = requests.post( |
| url, |
| data={"message": caption or "", "access_token": FB_TOKEN}, |
| timeout=60, |
| ) |
| return _fb_result(resp) |
|
|
|
|
| def fb_page_info(): |
| _fb_check_config() |
| url = f"https://graph.facebook.com/{GRAPH_VER}/{FB_PAGE_ID}" |
| resp = requests.get( |
| url, params={"fields": "name,fan_count", "access_token": FB_TOKEN}, timeout=30 |
| ) |
| if resp.status_code >= 400: |
| raise RuntimeError(f"Facebook error {resp.status_code}: {resp.text}") |
| return resp.json() |
|
|
|
|
| |
| |
| |
| scheduler = BackgroundScheduler(timezone=TIMEZONE) |
|
|
|
|
| def publish_next(reason="schedule"): |
| """Queue se agli pending post publish karta hai.""" |
| with LOCK: |
| pending = [q for q in STATE["queue"] if q.get("status") == "pending"] |
| if not pending: |
| log.info("Queue khaali hai - abhi post karne ke liye kuch nahi.") |
| return None |
|
|
| item = pending[0] |
| try: |
| if not item.get("image") or not os.path.exists(item["image"]): |
| path, _ = generate_image(item["prompt"]) |
| item["image"] = path |
| caption = item.get("caption") or item["prompt"] |
| res = fb_post_photo(item["image"], caption) |
| item["status"] = "posted" |
| item["posted_at"] = datetime.datetime.now().isoformat(timespec="seconds") |
| item["fb_post_id"] = res.get("id") or res.get("post_id") |
| STATE["history"].append( |
| { |
| "time": item["posted_at"], |
| "caption": caption, |
| "image": item.get("image"), |
| "reason": reason, |
| } |
| ) |
| log.info(f"POST OK ({reason}): {caption[:60]}") |
| except Exception as e: |
| item["status"] = "error" |
| item["error"] = str(e) |
| log.error(f"Post FAIL: {e}") |
| save_state(STATE) |
| return item |
|
|
|
|
| def rebuild_jobs(times): |
| scheduler.remove_all_jobs() |
| for t in times: |
| try: |
| hh, mm = t.strip().split(":") |
| scheduler.add_job( |
| publish_next, |
| CronTrigger(hour=int(hh), minute=int(mm), timezone=TIMEZONE), |
| kwargs={"reason": "schedule"}, |
| id=f"job_{t}", |
| replace_existing=True, |
| ) |
| log.info(f"Scheduled: roz {t} ({TIMEZONE})") |
| except Exception as e: |
| log.warning(f"Galat time '{t}': {e}") |
| if not scheduler.running and times: |
| scheduler.start() |
|
|
|
|
| |
| |
| |
| def queue_rows(): |
| rows = [] |
| for q in STATE["queue"]: |
| rows.append( |
| [ |
| q.get("id"), |
| q.get("status", "pending"), |
| (q.get("caption") or q.get("prompt") or "")[:60], |
| (q.get("image") or "").split("/")[-1], |
| ] |
| ) |
| return rows |
|
|
|
|
| def ui_generate_and_queue(prompts_text, captions_text, limit): |
| prompts = [p.strip() for p in (prompts_text or "").split("\n") if p.strip()] |
| caps = [c.strip() for c in (captions_text or "").split("\n")] if captions_text else [] |
| if not prompts: |
| return [], queue_rows(), "Koi prompt nahi mila. Ek line mein ek prompt likhein." |
| try: |
| limit = int(limit) |
| if limit > 0: |
| prompts = prompts[:limit] |
| except Exception: |
| pass |
|
|
| previews, added = [], 0 |
| for i, p in enumerate(prompts): |
| try: |
| path, img = generate_image(p) |
| except Exception as e: |
| log.error(f"Generate fail ('{p[:40]}'): {e}") |
| continue |
| with LOCK: |
| STATE["queue"].append( |
| { |
| "id": new_id(), |
| "prompt": p, |
| "caption": caps[i] if i < len(caps) else "", |
| "image": path, |
| "status": "pending", |
| } |
| ) |
| added += 1 |
| previews.append(img) |
| with LOCK: |
| save_state(STATE) |
| msg = f"{added} post queue mein add ho gaye (images ready)." |
| if added < len(prompts): |
| msg += " Kuch prompts fail hue - Logs tab dekhein." |
| return previews, queue_rows(), msg |
|
|
|
|
| def ui_start(times_text): |
| times = [t.strip() for t in (times_text or "").split(",") if t.strip()] |
| if not times: |
| return "Kam se kam ek time daalein, jaise: 09:00,14:00,19:00" |
| with LOCK: |
| STATE["times"] = times |
| STATE["enabled"] = True |
| save_state(STATE) |
| rebuild_jobs(times) |
| return f"Scheduler CHALU. Roz {', '.join(times)} ({TIMEZONE}) par queue se post hoga." |
|
|
|
|
| def ui_stop(): |
| with LOCK: |
| STATE["enabled"] = False |
| save_state(STATE) |
| scheduler.remove_all_jobs() |
| return "Scheduler BAND kar diya. Queue safe hai." |
|
|
|
|
| def ui_post_now(): |
| item = publish_next(reason="manual") |
| if item is None: |
| return "Queue khaali hai.", queue_rows() |
| if item.get("status") == "posted": |
| return f"Post ho gaya! FB id: {item.get('fb_post_id')}", queue_rows() |
| return f"Post fail: {item.get('error')}", queue_rows() |
|
|
|
|
| def ui_clear_posted(): |
| with LOCK: |
| STATE["queue"] = [q for q in STATE["queue"] if q.get("status") != "posted"] |
| save_state(STATE) |
| return queue_rows(), "Posted items queue se hata diye." |
|
|
|
|
| def ui_reset_all(): |
| with LOCK: |
| STATE["queue"] = [] |
| STATE["history"] = [] |
| save_state(STATE) |
| return queue_rows(), "Poori queue clear kar di." |
|
|
|
|
| def ui_logs(): |
| return "\n".join(LOG_BUFFER[-250:]) or "(abhi koi log nahi)" |
|
|
|
|
| def ui_test_fb(): |
| try: |
| info = fb_page_info() |
| return f"Facebook connected ✔ Page: {info.get('name')} | Fans: {info.get('fan_count', 'NA')}" |
| except Exception as e: |
| return f"Facebook connection fail: {e}" |
|
|
|
|
| |
| |
| |
| with gr.Blocks(title="FB Auto Poster") as demo: |
| gr.Markdown( |
| "# 📮 FB Auto Poster\n" |
| "AI images banao aur apne Facebook Page par **roz automatic** post karo.\n\n" |
| "> Pehle Space ke **Secrets** mein `FB_PAGE_ID`, `FB_PAGE_ACCESS_TOKEN` aur " |
| "`HF_TOKEN` daalna zaroori hai (README dekhein)." |
| ) |
|
|
| with gr.Tab("1️⃣ Content banayein"): |
| gr.Markdown( |
| "Ek line mein ek **prompt** likhein. Har prompt se ek image banegi aur " |
| "queue mein add ho jayegi. Captions optional hain - khaali chhodein to " |
| "prompt hi caption ban jayega." |
| ) |
| with gr.Row(): |
| prompts_in = gr.Textbox( |
| label="Prompts (ek line mein ek)", |
| lines=8, |
| placeholder="Diwali sale banner, bright colors, product in center\nMorning coffee flat-lay, warm tones, instagram style", |
| ) |
| captions_in = gr.Textbox( |
| label="Captions (optional, ek line mein ek)", |
| lines=8, |
| placeholder="🪔 Diwali Dhamaka - 50% OFF!\nStart your day right ☕", |
| ) |
| with gr.Row(): |
| limit_in = gr.Number(label="Max images is baar (0 = sab)", value=0) |
| gen_btn = gr.Button("🖼️ Images banao & Queue mein daalo", variant="primary") |
| gen_status = gr.Textbox(label="Status", interactive=False) |
| gallery = gr.Gallery(label="Generated images", columns=4, height=320) |
|
|
| with gr.Tab("2️⃣ Queue & Schedule"): |
| gr.Markdown( |
| "Schedule set karein (IST). Bot roz in time par queue se agli post " |
| "apne aap publish karega." |
| ) |
| with gr.Row(): |
| times_in = gr.Textbox(label="Post times (comma se alag, 24h)", value="09:00,14:00,19:00") |
| with gr.Column(): |
| start_btn = gr.Button("▶️ Start scheduler", variant="primary") |
| stop_btn = gr.Button("⏸️ Stop scheduler") |
| sched_status = gr.Textbox(label="Scheduler status", interactive=False) |
| with gr.Row(): |
| post_now_btn = gr.Button("🚀 Post next now") |
| test_btn = gr.Button("🔌 Test Facebook connection") |
| queue_df = gr.Dataframe( |
| headers=["id", "status", "caption/prompt", "image"], |
| value=queue_rows(), |
| interactive=False, |
| label="Queue", |
| ) |
| with gr.Row(): |
| clear_btn = gr.Button("🧹 Posted items hatao") |
| reset_btn = gr.Button("🗑️ Poori queue clear karo") |
|
|
| with gr.Tab("3️⃣ Logs"): |
| logs_out = gr.Textbox(label="Recent logs", lines=22, interactive=False) |
| logs_btn = gr.Button("🔄 Refresh logs") |
|
|
| |
| gen_btn.click( |
| ui_generate_and_queue, |
| inputs=[prompts_in, captions_in, limit_in], |
| outputs=[gallery, queue_df, gen_status], |
| ) |
| start_btn.click(ui_start, inputs=[times_in], outputs=[sched_status]) |
| stop_btn.click(ui_stop, outputs=[sched_status]) |
| post_now_btn.click(ui_post_now, outputs=[sched_status, queue_df]) |
| test_btn.click(ui_test_fb, outputs=[sched_status]) |
| clear_btn.click(ui_clear_posted, outputs=[queue_df, sched_status]) |
| reset_btn.click(ui_reset_all, outputs=[queue_df, sched_status]) |
| logs_btn.click(ui_logs, outputs=[logs_out]) |
|
|
| demo.load(ui_logs, outputs=[logs_out]) |
|
|
|
|
| |
| |
| |
| if STATE.get("enabled") and STATE.get("times"): |
| try: |
| rebuild_jobs(STATE["times"]) |
| log.info("Scheduler saved state se dobara chalu kar diya.") |
| except Exception as e: |
| log.error(f"Scheduler resume fail: {e}") |
|
|
|
|
| if __name__ == "__main__": |
| demo.queue(default_concurrency_limit=2).launch() |
|
|
