#!/usr/bin/env python3 """Publish ONE local video as an Instagram story, through the official Graph API. This is the reference implementation for porthole's Swift `GraphPublisher`. It is deliberately dependency-free (stdlib only) so it runs on a bare Mac mini with no uv, no venv and no persona install — the project this logic was extracted from is being deprecated, so nothing here may import it. THE FLOW, and it cannot be collapsed: create a container, push the bytes to Meta's upload host, wait for the container to say FINISHED, then publish it. Four calls, in that order. WHY VIDEO AND NOT IMAGES: Meta cURLs images from a public URL and offers no image upload endpoint at all. A local JPEG is not a slow path, it is an impossible one. Video has the resumable host, which takes bytes directly — so a local .mp4 works from a machine with no public address. The Mini binds to the tailnet only, so this distinction is the whole reason tonight's test is a video. Extracted 2026-08-03 from digital-presence's `persona/publish.py`, whose `_upload_resumable` carried the note "UNPROVEN against this account: proving it requires an actual upload, which is a write." This run is that proof. """ from __future__ import annotations import json import mimetypes import sys import time import urllib.error import urllib.parse import urllib.request from pathlib import Path GRAPH = "https://graph.instagram.com" UPLOAD_HOST = "https://rupload.facebook.com/ig-api-upload" VERSION = "v23.0" TOKEN_FILE = Path.home() / ".config/porthole/instagram/token.json" LOG_FILE = Path.home() / ".config/porthole/publish.log" # Meta: a container not FINISHED within five minutes is stuck, not slow. CONTAINER_POLL_SECONDS = 6 CONTAINER_TIMEOUT_SECONDS = 300 STORY_MAX_SECONDS = 60 class PublishError(RuntimeError): pass def log(message: str) -> None: """Every line timestamped, to stdout AND the log, because launchd swallows stdout unless it is told not to and a silent overnight failure is the exact thing this test exists to catch.""" line = f"{time.strftime('%Y-%m-%d %H:%M:%S %Z')} {message}" print(line, flush=True) LOG_FILE.parent.mkdir(parents=True, exist_ok=True) with LOG_FILE.open("a") as handle: handle.write(line + "\n") def token() -> str: if not TOKEN_FILE.exists(): raise PublishError(f"no token at {TOKEN_FILE}") data = json.loads(TOKEN_FILE.read_text()) expires_at = data.get("expires_at", 0) days = (expires_at - time.time()) / 86400 if days < 0: raise PublishError(f"token expired {-days:.0f} days ago — refresh it before publishing") if days < 7: log(f"WARNING: token expires in {days:.1f} days. Refresh it.") return data["access_token"] def _request(method: str, url: str, *, data: bytes | None = None, headers: dict | None = None, timeout: int = 60) -> dict: request = urllib.request.Request(url, data=data, method=method, headers=headers or {}) try: with urllib.request.urlopen(request, timeout=timeout) as response: body = response.read().decode() except urllib.error.HTTPError as error: detail = error.read().decode()[:500] raise PublishError(f"{method} {url.split('?')[0]} -> {error.code}: {detail}") from None return json.loads(body) if body else {} def graph_post(path: str, **params) -> dict: params["access_token"] = token() return _request("POST", f"{GRAPH}/{VERSION}/{path}", data=urllib.parse.urlencode(params).encode()) def graph_get(path: str, **params) -> dict: params["access_token"] = token() query = urllib.parse.urlencode(params) return _request("GET", f"{GRAPH}/{VERSION}/{path}?{query}") def duration_seconds(path: Path) -> float | None: """ffprobe if it is here, None if it is not. A missing ffprobe must not stop a publish — it only costs us the pre-flight length check.""" import shutil import subprocess probe = shutil.which("ffprobe") if not probe: return None result = subprocess.run( [probe, "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nk=1", str(path)], capture_output=True, text=True, timeout=60) try: return float(result.stdout.strip()) except ValueError: return None def preflight(path: Path) -> None: """Refuse what Meta cannot take, BEFORE a container exists. A container created for a file that was never publishable is litter on his account.""" if not path.exists(): raise PublishError(f"{path} does not exist") if path.stat().st_size == 0: raise PublishError(f"{path} is empty") kind = (mimetypes.guess_type(str(path))[0] or "") if not kind.startswith("video/"): raise PublishError( f"{path} is {kind or 'an unknown type'}. Stories from a local file must be " "VIDEO — Meta fetches images by URL and has no image upload endpoint.") seconds = duration_seconds(path) if seconds is None: log("ffprobe not found — skipping the length check, Meta will enforce it") elif seconds > STORY_MAX_SECONDS: raise PublishError( f"{seconds:.1f}s is longer than a story's {STORY_MAX_SECONDS}s ceiling") else: log(f"pre-flight ok: {seconds:.1f}s, {path.stat().st_size / 1e6:.1f} MB") def publish_story(path: Path, dry_run: bool = True) -> dict: preflight(path) who = graph_get("me", fields="username,account_type") log(f"account: {who.get('username')} ({who.get('account_type')})") if dry_run: log("DRY RUN — every check above really ran; no container was created.") return {"ok": True, "dry_run": True, "file": str(path)} container = graph_post("me/media", media_type="STORIES", upload_type="resumable") container_id = container["id"] log(f"container {container_id} created") payload = path.read_bytes() _request("POST", f"{UPLOAD_HOST}/{VERSION}/{container_id}", headers={"Authorization": f"OAuth {token()}", "offset": "0", "file_size": str(len(payload))}, data=payload, timeout=600) log(f"uploaded {len(payload) / 1e6:.1f} MB") deadline = time.time() + CONTAINER_TIMEOUT_SECONDS status = "IN_PROGRESS" while time.time() < deadline: body = graph_get(container_id, fields="status_code,status") status = body.get("status_code", "IN_PROGRESS") if status == "FINISHED": break if status in ("ERROR", "EXPIRED"): raise PublishError(f"container is {status}: {body.get('status')}") log(f"container {status}, waiting") time.sleep(CONTAINER_POLL_SECONDS) else: raise PublishError(f"container still {status} after {CONTAINER_TIMEOUT_SECONDS}s") published = graph_post("me/media_publish", creation_id=container_id) log(f"PUBLISHED: media id {published.get('id')}") return {"ok": True, "dry_run": False, "media_id": published.get("id")} def _self_check() -> None: """Offline. Touches no network, reads no token, publishes nothing.""" import tempfile with tempfile.TemporaryDirectory() as tmp: missing = Path(tmp) / "nope.mp4" try: preflight(missing) raise AssertionError("a missing file must be refused") except PublishError as error: assert "does not exist" in str(error) empty = Path(tmp) / "empty.mp4" empty.write_bytes(b"") try: preflight(empty) raise AssertionError("an empty file must be refused") except PublishError as error: assert "empty" in str(error) # The one that matters: a local image can never be a story, and it must # fail here rather than after a container exists on his account. image = Path(tmp) / "photo.jpg" image.write_bytes(b"\xff\xd8\xff\xe0not really a jpeg") try: preflight(image) raise AssertionError("a local image must be refused with the reason") except PublishError as error: assert "no image upload endpoint" in str(error), error print("publish-story self-check ok") if __name__ == "__main__": if "--self-check" in sys.argv: _self_check() raise SystemExit(0) arguments = [a for a in sys.argv[1:] if not a.startswith("--")] if not arguments: raise SystemExit("usage: publish-story.py [--apply]") try: result = publish_story(Path(arguments[0]).expanduser(), dry_run="--apply" not in sys.argv) except PublishError as error: log(f"FAILED: {error}") raise SystemExit(1) print(json.dumps(result, indent=2))