Spaces:
Sleeping
Sleeping
File size: 5,010 Bytes
ca2863a afb566d ca2863a 5f2de2a ca2863a a6801bf ca2863a afb566d ca2863a afb566d 5f2de2a ca2863a afb566d 5f2de2a afb566d ca2863a a6801bf ca2863a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | import os
import argparse
import logging
import requests
import socket
import time
from requests.exceptions import RequestException
from dotenv import load_dotenv
from retry_queue import enqueue as enqueue_retry
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)
def _append_log(entry: str):
try:
with open("log.txt", "a", encoding="utf-8") as lf:
lf.write(entry + "\n")
except Exception:
logger.exception("Failed to write to log.txt")
def post_to_facebook(page_id: str, access_token: str, message: str, link: str | None = None):
url = f"https://graph.facebook.com/{page_id}/feed"
payload = {"message": message, "access_token": access_token}
if link:
payload["link"] = link
logger.info("Posting to Facebook page %s", page_id)
# Ensure graph.facebook.com resolves before attempting to POST to avoid long, noisy stack traces
try:
socket.getaddrinfo("graph.facebook.com", 443)
except Exception as e:
logger.error("DNS resolution failed for graph.facebook.com: %s", e)
_append_log(f"[{__import__('time').strftime('%Y-%m-%d %H:%M:%S')}] FB_DNS_RESOLUTION_FAILED {e}")
# enqueue for later retry
try:
enqueue_retry({"type": "message", "page_id": page_id, "access_token": access_token, "message": message, "link": link})
except Exception:
logger.exception("Failed to enqueue message after DNS failure")
raise
# Retry with exponential backoff for transient network issues
attempts = 4
backoff = 2
last_exc = None
for attempt in range(1, attempts + 1):
try:
resp = requests.post(url, data=payload, timeout=15)
resp.raise_for_status()
last_exc = None
break
except RequestException as e:
last_exc = e
logger.warning("Attempt %s/%s: Facebook POST failed: %s", attempt, attempts, e)
if attempt < attempts:
time.sleep(backoff)
backoff *= 2
else:
logger.error("All Facebook POST attempts failed")
_append_log(f"[{__import__('time').strftime('%Y-%m-%d %H:%M:%S')}] FB_POST_ERROR page={page_id} message={message[:120]!r} exception={e}")
# enqueue for retry
try:
enqueue_retry({"type": "message", "page_id": page_id, "access_token": access_token, "message": message, "link": link})
except Exception:
logger.exception("Failed to enqueue failed message post")
raise
data = resp.json()
logger.info("Post successful: %s", data)
# append a concise entry to log.txt for auditing
try:
_append_log(f"[{__import__('time').strftime('%Y-%m-%d %H:%M:%S')}] FB_POST_SUCCESS page={page_id} id={data.get('id')} post_id={data.get('post_id')}")
except Exception:
logger.exception("Failed to append FB success to log.txt")
return data
def main():
load_dotenv()
parser = argparse.ArgumentParser(description="Post a message to a Facebook Page using Page Access Token from .env")
parser.add_argument("-m", "--message", help="Message text to post (or set MESSAGE in .env)")
parser.add_argument("-l", "--link", help="Optional link to include")
parser.add_argument("--use-app-access", action="store_true", help="Attempt to post using the app access token (APP_ID|APP_SECRET)")
args = parser.parse_args()
# Prefer explicit env names already present in the repo
page_id = os.getenv("FB_PAGE_ID") or os.getenv("PAGE_ID")
access_token = os.getenv("FB_PAGE_ACCESS_TOKEN") or os.getenv("PAGE_ACCESS_TOKEN")
message = args.message or os.getenv("MESSAGE")
link = args.link or os.getenv("LINK")
use_app_access = args.use_app_access
# build app access token if requested
if use_app_access:
app_id = os.getenv("FB_APP_ID")
app_secret = os.getenv("FB_APP_SECRET")
if not app_id or not app_secret:
print("ERROR: FB_APP_ID and FB_APP_SECRET required to build app access token")
raise SystemExit(1)
access_token = f"{app_id}|{app_secret}"
if not page_id or not access_token:
logger.error("Missing FB_PAGE_ID or FB_PAGE_ACCESS_TOKEN in environment (.env)")
raise SystemExit(1)
if not message:
logger.error("No message provided (use --message or set MESSAGE in .env)")
raise SystemExit(1)
try:
result = post_to_facebook(page_id, access_token, message, link)
except requests.HTTPError as e:
details = e.response.text if e.response is not None else str(e)
logger.error("Facebook API request failed: %s", details)
raise SystemExit(1)
logger.info("Posted to Facebook page successfully: %s", result)
print("Posted to Facebook page successfully:", result)
if __name__ == "__main__":
main()
|