Phan Văn Năng

---
name: ai-agent
description: Facebook Publisher Skill (Automate Page Posts via Graph API)
version: 1.0.1
env:
  - FB_APP_ID
  - FB_APP_SECRET
  - FB_PAGE_ID
  - FB_PAGE_ACCESS_TOKEN
---
# Facebook Graph API Skill (Advanced)

## Purpose
Production-oriented guide for building Facebook Graph API workflows for Pages:
publishing posts (text + image), managing tokens, and operating Page content
safely using direct HTTPS calls.

## Best fit
- Page posting automation with images (DALL-E generated or external URL)
- Token management (short-lived → long-lived → page token)
- Retry-safe, rate-limit-aware production pipelines

## Not a fit
- Personal profile posting (not supported by Graph API for third-party apps)
- Ads / Marketing API workflows
- Browser-based OAuth flows

## Quick orientation
```
agents/fb_token_helper.py     ← Get & exchange tokens (run this first!)
agents/fb_publisher_agent.py  ← Post text / images to Page
config.py                     ← All env vars
test_fb_connection.py         ← Verify token is working
```

## Token Flow
```
Short-lived User Token (1-2h)
        ↓  GET /oauth/access_token?grant_type=fb_exchange_token
Long-lived User Token (60 days)
        ↓  GET /me/accounts
Page Access Token (never expires*)
```
*Until user changes password or revokes app.

## Required Environment Variables
```env
FB_APP_ID=...           # From Meta for Developers
FB_APP_SECRET=...       # App secret
FB_PAGE_ID=...          # Target Fanpage ID
FB_PAGE_ACCESS_TOKEN=... # From fb_token_helper.py
```

## Key API Endpoints

### Post text
```
POST /v21.0/{page_id}/feed
  message=...
  access_token={page_token}
```

### Upload photo (unpublished)
```
POST /v21.0/{page_id}/photos
  url={image_url}
  published=false
  access_token={page_token}
→ Returns: { "id": "PHOTO_ID" }
```

### Post with photo
```
POST /v21.0/{page_id}/feed
  message=...
  attached_media[0]={"media_fbid":"PHOTO_ID"}
  access_token={page_token}
```

### Scheduled post
```
POST /v21.0/{page_id}/feed
  message=...
  scheduled_publish_time={unix_timestamp}
  published=false
  access_token={page_token}
```

## Required Permissions
| Permission | Purpose |
|-----------|---------|
| `pages_manage_posts` | Create/edit posts |
| `pages_read_engagement` | Read reactions, comments |
| `pages_show_list` | List managed pages |
| `public_profile` | Basic user identity |

## Rate Limits
- 200 calls/hour/user token
- Implement retry with exponential backoff (see fb_publisher_agent.py)
- POST 4-5 times/day max per Page for safety

## Security
- Never log tokens or app secrets
- Store all secrets in .env (ignored by git)
- Validate webhook signatures if using webhooks
- Monitor token validity daily with a cron job

FILE:config.py
"""
config.py — Centralized configuration loaded from .env
"""
import os
from dotenv import load_dotenv

load_dotenv()

# ── OpenAI ─────────────────────────────────────────────────
OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "")
OPENAI_MODEL: str = os.getenv("OPENAI_MODEL", "gpt-4o")
IMAGE_MODEL: str = os.getenv("IMAGE_MODEL", "dall-e-3")

# ── Apify ──────────────────────────────────────────────────
APIFY_API_TOKEN: str = os.getenv("APIFY_API_TOKEN", "")

# ── Facebook App ───────────────────────────────────────────
FB_APP_ID: str = os.getenv("FB_APP_ID", "")
FB_APP_SECRET: str = os.getenv("FB_APP_SECRET", "")
FB_CLIENT_TOKEN: str = os.getenv("FB_CLIENT_TOKEN", "")

# ── Facebook Page ──────────────────────────────────────────
FB_PAGE_ID: str = os.getenv("FB_PAGE_ID", "")
FB_USER_ACCESS_TOKEN: str = os.getenv("FB_USER_ACCESS_TOKEN", "")
FB_PAGE_ACCESS_TOKEN: str = os.getenv("FB_PAGE_ACCESS_TOKEN", "")
FB_API_VERSION: str = os.getenv("API_VERSION", "v21.0")
FB_BASE_URL: str = f"https://graph.facebook.com/{FB_API_VERSION}"

# ── Runtime ────────────────────────────────────────────────
MAX_POST_RETRIES: int = int(os.getenv("MAX_POST_RETRIES", "3"))
POST_DELAY_SECONDS: int = int(os.getenv("POST_DELAY_SECONDS", "2"))


def validate():
    """Kiểm tra các biến môi trường bắt buộc."""
    missing = []
    required = {
        "OPENAI_API_KEY": OPENAI_API_KEY,
        "APIFY_API_TOKEN": APIFY_API_TOKEN,
        "FB_APP_ID": FB_APP_ID,
        "FB_APP_SECRET": FB_APP_SECRET,
        "FB_PAGE_ID": FB_PAGE_ID,
    }
    for name, val in required.items():
        if not val:
            missing.append(name)
    if missing:
        raise EnvironmentError(
            f"❌ Thiếu biến môi trường: {', '.join(missing)}\n"
            "→ Kiểm tra file .env"
        )

FILE:test_fb_connection.py
"""
test_fb_connection.py
──────────────────────
Script test nhanh kết nối Facebook Graph API.
Chạy cái này TRƯỚC khi dùng main.py để xác nhận token OK.

    python test_fb_connection.py
"""
import sys, os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

import requests
import config
from rich.console import Console
from rich.table import Table
from rich.panel import Panel

console = Console()


def test_app_credentials():
    """Test App ID + Secret hợp lệ."""
    console.print("[yellow]1️⃣  Kiểm tra App credentials...[/yellow]")
    url = f"{config.FB_BASE_URL}/app"
    resp = requests.get(url, params={
        "access_token": f"{config.FB_APP_ID}|{config.FB_APP_SECRET}"
    })
    if resp.status_code == 200:
        data = resp.json()
        console.print(f"[green]✅ App OK:[/green] {data.get('name')} (ID: {data.get('id')})")
        return True
    else:
        console.print(f"[red]❌ App credentials lỗi: {resp.text}[/red]")
        return False


def test_page_token():
    """Test Page Access Token và lấy thông tin Page."""
    console.print("[yellow]2️⃣  Kiểm tra Page Access Token...[/yellow]")

    if not config.FB_PAGE_ACCESS_TOKEN:
        console.print(
            "[red]❌ FB_PAGE_ACCESS_TOKEN chưa set![/red]\n"
            "→ Chạy: [bold]python agents/fb_token_helper.py[/bold]"
        )
        return False

    url = f"{config.FB_BASE_URL}/{config.FB_PAGE_ID}"
    resp = requests.get(url, params={
        "fields": "id,name,fan_count,category",
        "access_token": config.FB_PAGE_ACCESS_TOKEN,
    })

    if resp.status_code == 200:
        data = resp.json()
        table = Table(show_header=False, border_style="green")
        table.add_column("Key", style="bold")
        table.add_column("Value")
        table.add_row("Page ID", data.get("id", "N/A"))
        table.add_row("Tên Page", data.get("name", "N/A"))
        table.add_row("Followers", str(data.get("fan_count", 0)))
        table.add_row("Category", data.get("category", "N/A"))
        console.print("[green]✅ Page Token OK[/green]")
        console.print(table)
        return True
    else:
        err = resp.json().get("error", {})
        console.print(f"[red]❌ Page token lỗi ({err.get('code')}): {err.get('message')}[/red]")
        if err.get("code") == 190:
            console.print("→ Token hết hạn! Chạy lại [bold]python agents/fb_token_helper.py[/bold]")
        return False


def main():
    console.print(Panel.fit(
        "[bold cyan]🔌 Facebook Connection Test[/bold cyan]",
        border_style="cyan"
    ))
    console.print(f"[dim]App ID: {config.FB_APP_ID}[/dim]")
    console.print(f"[dim]Page ID: {config.FB_PAGE_ID}[/dim]\n")

    ok_app = test_app_credentials()
    ok_token = test_page_token()

    console.print()
    if ok_app and ok_token:
        console.print(Panel.fit(
            "[bold green]🎉 Tất cả OK! Sẵn sàng chạy pipeline.[/bold green]\n"
            "→ [bold]python main.py --test-post[/bold]",
            border_style="green"
        ))
    else:
        console.print(Panel.fit(
            "[bold red]⚠️  Có lỗi. Xem hướng dẫn bên trên để fix.[/bold red]",
            border_style="red"
        ))


if __name__ == "__main__":
    main()

FILE:agents/fb_publisher_agent.py
"""
agents/fb_publisher_agent.py
─────────────────────────────
Agent 4: Đăng bài lên Facebook Fanpage bằng Graph API.

Features:
- Đăng text post
- Đăng post kèm ảnh (từ URL hoặc file local)
- Retry với exponential backoff
- Rate limit safe
"""
import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import time
import requests
import config
from pathlib import Path
from rich.console import Console

console = Console()


class FacebookPublisherAgent:
    def __init__(self):
        self.page_token = config.FB_PAGE_ACCESS_TOKEN
        self.page_id = config.FB_PAGE_ID
        self.base_url = config.FB_BASE_URL

        if not self.page_token:
            console.print(
                "[bold red]⚠️  FB_PAGE_ACCESS_TOKEN chưa được set trong .env![/bold red]\n"
                "→ Chạy: python agents/fb_token_helper.py để lấy token."
            )

    # ── Internal: retry wrapper ─────────────────────────────
    def _request(self, method: str, url: str, max_retries: int = None, **kwargs) -> dict:
        """HTTP request với retry + exponential backoff."""
        max_retries = max_retries or config.MAX_POST_RETRIES
        for attempt in range(max_retries):
            try:
                resp = requests.request(method, url, timeout=30, **kwargs)

                # Rate limit
                if resp.status_code == 429:
                    wait_sec = 10 * (2 ** attempt)
                    console.print(f"[yellow]⏳ Rate limited. Chờ {wait_sec}s...[/yellow]")
                    time.sleep(wait_sec)
                    continue

                # Token expired / invalid
                if resp.status_code in (400, 401, 403):
                    data = resp.json()
                    err = data.get("error", {})
                    console.print(f"[red]❌ FB API Error {err.get('code')}: {err.get('message')}[/red]")
                    if err.get("code") in (190, 102):  # Token expired
                        console.print("[red]→ Token hết hạn! Chạy lại fb_token_helper.py[/red]")
                    raise RuntimeError(f"FB API Error: {err.get('message')}")

                resp.raise_for_status()
                return resp.json()

            except requests.exceptions.ConnectionError:
                console.print(f"[yellow]🔌 Mất kết nối (lần {attempt+1}/{max_retries}). Chờ 5s...[/yellow]")
                time.sleep(5)

        raise RuntimeError(f"Request thất bại sau {max_retries} lần thử: {url}")

    # ── Upload ảnh (unpublished) ────────────────────────────
    def _upload_photo_from_url(self, image_url: str) -> str:
        """Upload ảnh từ URL lên FB, trả về photo_id."""
        console.print("[yellow]📤 Đang upload ảnh lên Facebook...[/yellow]")
        url = f"{self.base_url}/{self.page_id}/photos"
        data = self._request("POST", url, data={
            "url": image_url,
            "published": "false",
            "access_token": self.page_token,
        })
        photo_id = data.get("id")
        console.print(f"[green]✅ Upload ảnh xong — photo_id: {photo_id}[/green]")
        return photo_id

    def _upload_photo_from_file(self, file_path: str) -> str:
        """Upload ảnh từ file local lên FB, trả về photo_id."""
        console.print("[yellow]📤 Đang upload ảnh từ file local...[/yellow]")
        url = f"{self.base_url}/{self.page_id}/photos"
        with open(file_path, "rb") as f:
            data = self._request("POST", url, data={
                "published": "false",
                "access_token": self.page_token,
            }, files={"source": (Path(file_path).name, f, "image/jpeg")})
        photo_id = data.get("id")
        console.print(f"[green]✅ Upload ảnh xong — photo_id: {photo_id}[/green]")
        return photo_id

    # ── Post text only ──────────────────────────────────────
    def post_text(self, message: str) -> dict:
        """Đăng bài text thuần lên Fanpage."""
        console.print("[bold cyan]📝 Đăng text post lên Fanpage...[/bold cyan]")
        url = f"{self.base_url}/{self.page_id}/feed"
        result = self._request("POST", url, data={
            "message": message,
            "access_token": self.page_token,
        })
        post_id = result.get("id", "unknown")
        console.print(f"[bold green]🎉 Đăng thành công! Post ID: {post_id}[/bold green]")
        console.print(f"[link]https://www.facebook.com/{post_id}[/link]")
        return result

    # ── Post ảnh + caption ──────────────────────────────────
    def post_with_image_url(self, message: str, image_url: str) -> dict:
        """Đăng bài kèm ảnh (từ URL công khai) lên Fanpage."""
        console.print("[bold cyan]🖼️  Đăng post kèm ảnh (URL) lên Fanpage...[/bold cyan]")
        photo_id = self._upload_photo_from_url(image_url)
        time.sleep(config.POST_DELAY_SECONDS)  # Chờ FB xử lý

        url = f"{self.base_url}/{self.page_id}/feed"
        result = self._request("POST", url, data={
            "message": message,
            "attached_media[0]": f'{{"media_fbid":"{photo_id}"}}',
            "access_token": self.page_token,
        })
        post_id = result.get("id", "unknown")
        console.print(f"[bold green]🎉 Đăng thành công! Post ID: {post_id}[/bold green]")
        console.print(f"[link]https://www.facebook.com/{post_id}[/link]")
        return result

    def post_with_image_file(self, message: str, file_path: str) -> dict:
        """Đăng bài kèm ảnh (từ file local) lên Fanpage."""
        console.print("[bold cyan]🖼️  Đăng post kèm ảnh (file) lên Fanpage...[/bold cyan]")
        photo_id = self._upload_photo_from_file(file_path)
        time.sleep(config.POST_DELAY_SECONDS)

        url = f"{self.base_url}/{self.page_id}/feed"
        result = self._request("POST", url, data={
            "message": message,
            "attached_media[0]": f'{{"media_fbid":"{photo_id}"}}',
            "access_token": self.page_token,
        })
        post_id = result.get("id", "unknown")
        console.print(f"[bold green]🎉 Đăng thành công! Post ID: {post_id}[/bold green]")
        console.print(f"[link]https://www.facebook.com/{post_id}[/link]")
        return result

    # ── Scheduled post ──────────────────────────────────────
    def post_scheduled(self, message: str, unix_timestamp: int) -> dict:
        """Đăng bài theo lịch."""
        console.print(f"[cyan]⏰ Lên lịch đăng bài lúc timestamp: {unix_timestamp}[/cyan]")
        url = f"{self.base_url}/{self.page_id}/feed"
        result = self._request("POST", url, data={
            "message": message,
            "scheduled_publish_time": str(unix_timestamp),
            "published": "false",
            "access_token": self.page_token,
        })
        console.print(f"[green]✅ Đã lên lịch! Post ID: {result.get('id')}[/green]")
        return result

    # ── Xem insights bài đăng ──────────────────────────────
    def get_post_insights(self, post_id: str) -> dict:
        """Lấy thống kê reactions, comments, shares của bài."""
        url = f"{self.base_url}/{post_id}"
        result = self._request("GET", url, params={
            "fields": "message,created_time,reactions.summary(true),comments.summary(true),shares",
            "access_token": self.page_token,
        })
        return result


# ── Test ────────────────────────────────────────────────────
if __name__ == "__main__":
    agent = FacebookPublisherAgent()
    # Test text post
    result = agent.post_text(
        "🤖 Test post từ AI Agent!\n\n"
        "Đây là bài test tự động bằng OpenClaw + Facebook Graph API.\n\n"
        "Bạn thấy bài này thì pipeline đang chạy đúng rồi! 🚀\n\n"
        "#AIAgent #Automation #Facebook"
    )
    print(result)

FILE:agents/fb_token_helper.py
"""
agents/fb_token_helper.py
──────────────────────────
Tiện ích lấy và exchange Facebook Access Token.

Chạy trực tiếp:
    python agents/fb_token_helper.py

Hướng dẫn:
  1. Chạy script này
  2. Làm theo hướng dẫn để lấy short-lived token từ Graph Explorer
  3. Script sẽ exchange → long-lived user token → page token
  4. Copy Page Access Token vào .env  (FB_PAGE_ACCESS_TOKEN)
"""
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import requests
import json
import config
from rich.console import Console
from rich.panel import Panel
from rich.table import Table

console = Console()


def exchange_to_long_lived(short_token: str) -> str:
    """
    Đổi Short-lived User Token → Long-lived User Token (60 ngày).
    """
    url = f"{config.FB_BASE_URL}/oauth/access_token"
    params = {
        "grant_type": "fb_exchange_token",
        "client_id": config.FB_APP_ID,
        "client_secret": config.FB_APP_SECRET,
        "fb_exchange_token": short_token,
    }
    resp = requests.get(url, params=params)
    resp.raise_for_status()
    data = resp.json()
    if "access_token" not in data:
        raise ValueError(f"Lỗi exchange token: {data}")
    return data["access_token"]


def get_page_tokens(long_lived_user_token: str) -> list[dict]:
    """
    Lấy danh sách Pages + Page Access Tokens từ long-lived user token.
    """
    url = f"{config.FB_BASE_URL}/me/accounts"
    params = {"access_token": long_lived_user_token}
    resp = requests.get(url, params=params)
    resp.raise_for_status()
    return resp.json().get("data", [])


def debug_token(token: str) -> dict:
    """Kiểm tra thông tin token."""
    url = f"{config.FB_BASE_URL}/debug_token"
    params = {
        "input_token": token,
        "access_token": f"{config.FB_APP_ID}|{config.FB_APP_SECRET}",
    }
    resp = requests.get(url, params=params)
    resp.raise_for_status()
    return resp.json().get("data", {})


def run_interactive():
    console.print(Panel.fit(
        "[bold cyan]Facebook Token Helper[/bold cyan]\n"
        "Giúp bạn lấy Page Access Token để đăng bài Fanpage",
        border_style="cyan"
    ))

    console.print("\n[bold yellow]BƯỚC 1:[/bold yellow] Lấy Short-lived User Token")
    console.print(
        "→ Mở link sau trong trình duyệt:\n"
        f"[link]https://developers.facebook.com/tools/explorer[/link]\n\n"
        "Chọn App [bold]4348763312075291[/bold], click [bold]Generate Access Token[/bold]\n"
        "Tích chọn permissions:\n"
        "  ✅ pages_manage_posts\n"
        "  ✅ pages_read_engagement\n"
        "  ✅ pages_show_list\n"
        "  ✅ public_profile\n"
    )

    short_token = console.input("[bold green]Paste Short-lived Token vào đây:[/bold green] ").strip()
    if not short_token:
        console.print("[red]Token rỗng — thoát.[/red]")
        return

    console.print("\n[yellow]⏳ Đang exchange sang Long-lived token...[/yellow]")
    try:
        long_token = exchange_to_long_lived(short_token)
        console.print(f"[green]✅ Long-lived User Token:[/green]\n{long_token}\n")
    except Exception as e:
        console.print(f"[red]❌ Exchange thất bại: {e}[/red]")
        return

    # Debug token
    info = debug_token(long_token)
    console.print(f"[dim]Token expires: {info.get('expires_at', 'N/A')} | valid: {info.get('is_valid')}[/dim]\n")

    console.print("[yellow]⏳ Đang lấy Page Access Tokens...[/yellow]")
    try:
        pages = get_page_tokens(long_token)
    except Exception as e:
        console.print(f"[red]❌ Lỗi lấy pages: {e}[/red]")
        return

    if not pages:
        console.print("[red]⚠️  Không tìm thấy Page nào. Đảm bảo tài khoản FB quản lý ít nhất 1 Page.[/red]")
        return

    table = Table(title="📄 Danh sách Pages", show_header=True, header_style="bold magenta")
    table.add_column("Page ID", style="cyan")
    table.add_column("Page Name", style="white")
    table.add_column("Page Token (50 ký tự đầu)", style="dim")

    for page in pages:
        table.add_row(
            page.get("id", ""),
            page.get("name", ""),
            page.get("access_token", "")[:50] + "..."
        )
    console.print(table)

    # Save to .env hint
    console.print("\n[bold green]✅ XONG! Cập nhật .env:[/bold green]")
    for page in pages:
        console.print(f"\nPage: [bold]{page.get('name')}[/bold] (ID: {page.get('id')})")
        console.print(f"FB_USER_ACCESS_TOKEN={long_token}")
        console.print(f"FB_PAGE_ACCESS_TOKEN={page.get('access_token')}")
        console.print(f"FB_PAGE_ID={page.get('id')}")

    # Save to file
    output_path = "fb_tokens_output.json"
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump({
            "long_lived_user_token": long_token,
            "pages": pages
        }, f, ensure_ascii=False, indent=2)
    console.print(f"\n[dim]Token đã lưu vào {output_path} (đừng commit file này!)[/dim]")


if __name__ == "__main__":
    run_interactive()

# Facebook Graph API Skill (Advanced)

## Purpose
Production-oriented guide for building Facebook Graph API workflows for Pages:
publishing posts (text + image), managing tokens, and operating Page content
safely using direct HTTPS calls.

## Best fit
- Page posting automation with images (DALL-E generated or external URL)
- Token management (short-lived → long-lived → page token)
- Retry-safe, rate-limit-aware production pipelines

## Not a fit
- Personal profile posting (not supported by Graph API for third-party apps)
- Ads / Marketing API workflows
- Browser-based OAuth flows

## Quick orientation
```
agents/fb_token_helper.py     ← Get & exchange tokens (run this first!)
agents/fb_publisher_agent.py  ← Post text / images to Page
config.py                     ← All env vars
test_fb_connection.py         ← Verify token is working
```

## Token Flow
```
Short-lived User Token (1-2h)
        ↓  GET /oauth/access_token?grant_type=fb_exchange_token
Long-lived User Token (60 days)
        ↓  GET /me/accounts
Page Access Token (never expires*)
```
*Until user changes password or revokes app.

## Required Environment Variables
```env
FB_APP_ID=...           # From Meta for Developers
FB_APP_SECRET=...       # App secret
FB_PAGE_ID=...          # Target Fanpage ID
FB_PAGE_ACCESS_TOKEN=... # From fb_token_helper.py
```

## Key API Endpoints

### Post text
```
POST /v21.0/{page_id}/feed
  message=...
  access_token={page_token}
```

### Upload photo (unpublished)
```
POST /v21.0/{page_id}/photos
  url={image_url}
  published=false
  access_token={page_token}
→ Returns: { "id": "PHOTO_ID" }
```

### Post with photo
```
POST /v21.0/{page_id}/feed
  message=...
  attached_media[0]={"media_fbid":"PHOTO_ID"}
  access_token={page_token}
```

### Scheduled post
```
POST /v21.0/{page_id}/feed
  message=...
  scheduled_publish_time={unix_timestamp}
  published=false
  access_token={page_token}
```

## Required Permissions
| Permission | Purpose |
|-----------|---------|
| `pages_manage_posts` | Create/edit posts |
| `pages_read_engagement` | Read reactions, comments |
| `pages_show_list` | List managed pages |
| `public_profile` | Basic user identity |

## Rate Limits
- 200 calls/hour/user token
- Implement retry with exponential backoff (see fb_publisher_agent.py)
- POST 4-5 times/day max per Page for safety

## Security
- Never log tokens or app secrets
- Store all secrets in .env (ignored by git)
- Validate webhook signatures if using webhooks
- Monitor token validity daily with a cron job