| """ |
| Facebook video upload + CDN URL polling. |
| Adapted from the original Colab script into two standalone functions |
| so the scheduler can call them once per tick without blocking on a |
| long sleep loop. |
| """ |
| import json |
| import requests |
| from requests.adapters import HTTPAdapter |
| from urllib3.util.retry import Retry |
|
|
| API_VERSION = "v19.0" |
| GRAPH_VIDEO_BASE = f"https://graph-video.facebook.com/{API_VERSION}" |
| GRAPH_BASE = f"https://graph.facebook.com/{API_VERSION}" |
|
|
|
|
| class FacebookUploadError(Exception): |
| """A definite failure -- bad token, bad request, etc. Not worth retrying.""" |
| pass |
|
|
|
|
| class TransientNetworkError(Exception): |
| """A connection/TLS/timeout hiccup -- worth retrying on the next tick.""" |
| pass |
|
|
|
|
| def _session() -> requests.Session: |
| session = requests.Session() |
| |
| |
| |
| retry = Retry( |
| total=3, |
| backoff_factor=1.5, |
| status_forcelist=[502, 503, 504], |
| allowed_methods=["GET", "POST"], |
| ) |
| adapter = HTTPAdapter(max_retries=retry) |
| session.mount("https://", adapter) |
| return session |
|
|
|
|
| def start_upload(page_id: str, access_token: str, file_url: str, title: str, |
| description: str, privacy: str = "SELF") -> str: |
| """Kick off a hosted-file upload. Returns the new video_id.""" |
| url = f"{GRAPH_VIDEO_BASE}/{page_id}/videos" |
| data = { |
| "access_token": access_token, |
| "file_url": file_url, |
| "title": title, |
| "description": description, |
| "privacy": json.dumps({"value": privacy}), |
| "published": "false", |
| } |
| try: |
| resp = _session().post(url, data=data, timeout=120) |
| except (requests.exceptions.SSLError, requests.exceptions.ConnectionError, |
| requests.exceptions.Timeout) as e: |
| raise TransientNetworkError(str(e)) |
|
|
| try: |
| result = resp.json() |
| except ValueError: |
| raise TransientNetworkError(f"Non-JSON response (status {resp.status_code})") |
|
|
| if "id" in result: |
| return result["id"] |
|
|
| error = result.get("error", {}) |
| msg = error.get("message", "Unknown error") if isinstance(error, dict) else str(error) |
| code = error.get("code", "N/A") if isinstance(error, dict) else "N/A" |
| raise FacebookUploadError(f"{msg} (code: {code})") |
|
|
|
|
| def poll_video(page_id_unused: str, access_token: str, video_id: str): |
| """ |
| One status check. Returns a dict: |
| {"ready": True, "source": <cdn url>} |
| {"ready": False, "status": <fb status string>} |
| Raises FacebookUploadError on a hard API error, TransientNetworkError |
| on a connection/TLS/timeout hiccup that's worth retrying. |
| """ |
| url = f"{GRAPH_BASE}/{video_id}" |
| params = { |
| "fields": "source,permalink_url,status,privacy", |
| "access_token": access_token, |
| } |
| try: |
| resp = _session().get(url, params=params, timeout=30) |
| except (requests.exceptions.SSLError, requests.exceptions.ConnectionError, |
| requests.exceptions.Timeout) as e: |
| raise TransientNetworkError(str(e)) |
|
|
| if resp.status_code != 200: |
| return {"ready": False, "status": f"http_{resp.status_code}"} |
|
|
| try: |
| info = resp.json() |
| except ValueError: |
| return {"ready": False, "status": "non_json_response"} |
|
|
| if "error" in info: |
| error = info["error"] |
| msg = error.get("message", "Unknown error") if isinstance(error, dict) else str(error) |
| raise FacebookUploadError(msg) |
|
|
| if info.get("source"): |
| return {"ready": True, "source": info["source"]} |
|
|
| return {"ready": False, "status": info.get("status", "processing")} |
|
|
