"""Always-on QR Master milestone publisher. Polls the app's internal queue per channel and publishes what a customer has explicitly approved for that channel. The web app never receives X or Meta credentials; this worker never composes text of its own. """ import base64 import io import json import os import tempfile import time from pathlib import Path import requests from PIL import Image from requests_oauthlib import OAuth1Session CHANNELS = ("x", "instagram") def required(name): value = os.getenv(name, "").strip() if not value: raise RuntimeError(f"Missing {name}") return value def enabled_channels(): configured = [value.strip().lower() for value in os.getenv("SOCIAL_WORKER_CHANNELS", "x").split(",")] return [channel for channel in configured if channel in CHANNELS] or ["x"] def api(method, url, payload=None): response = requests.request(method, url, json=payload, headers={"Authorization": f"Bearer {required('INTERNAL_API_SECRET')}"}, timeout=30) response.raise_for_status() return response.json() def graph(path): return f"https://graph.facebook.com/{os.getenv('GRAPH_API_VERSION', 'v22.0')}/{path}" def graph_error(response): """Meta answers with a JSON error body that says far more than the status.""" try: error = response.json().get("error") or {} detail = error.get("error_user_msg") or error.get("message") if detail: return f"{detail} (code {error.get('code')})" except ValueError: pass return f"HTTP {response.status_code}" def milestone_image(milestone, image_format="landscape"): """Download the card the app renders at /s/m//og. The popup, the link preview and the published post therefore show the exact same image: one renderer, one source of truth, nothing to keep in sync here. The internal base is used on purpose - the public host is not necessarily reachable from inside the worker network. """ token = str(milestone.get("shareToken") or "").strip() if not token: return None base = required("QRMASTER_API_BASE").rstrip("/") response = requests.get(f"{base}/s/m/{token}/og", params={"format": image_format}, timeout=60) response.raise_for_status() path = Path(tempfile.mkstemp(suffix=".png")[1]) path.write_bytes(response.content) return path # --- X --------------------------------------------------------------------- def oauth_client(): return OAuth1Session(required("X_API_KEY"), client_secret=required("X_API_SECRET"), resource_owner_key=required("X_ACCESS_TOKEN"), resource_owner_secret=required("X_ACCESS_TOKEN_SECRET")) def find_existing_x_post(oauth, milestone): """Reconcile an uncertain prior attempt before creating another X post.""" token = str(milestone.get("shareToken") or "").strip() if not token: raise RuntimeError("Milestone has no share token for duplicate-safe publishing") identity = oauth.get("https://api.x.com/2/users/me", timeout=30) identity.raise_for_status() user_id = identity.json().get("data", {}).get("id") if not user_id: raise RuntimeError("X did not return the authenticated user id") timeline = oauth.get( f"https://api.x.com/2/users/{user_id}/tweets", params={"max_results": 100, "tweet.fields": "created_at,entities", "exclude": "retweets,replies"}, timeout=30, ) timeline.raise_for_status() for post in timeline.json().get("data") or []: urls = (post.get("entities") or {}).get("urls") or [] expanded = " ".join(str(url.get("expanded_url") or url.get("unwound_url") or "") for url in urls) if token in expanded: return f"https://x.com/i/web/status/{post.get('id')}" return None def publish_x(milestone): oauth = oauth_client() # Reading the timeline costs far more X quota than writing a post, so # reconcile only when this post was already attempted before. if milestone.get("attempts"): existing = find_existing_x_post(oauth, milestone) if existing: return existing, True path = milestone_image(milestone) try: media_id = None if path: with path.open("rb") as image: upload = oauth.post("https://upload.x.com/1.1/media/upload.json", files={"media": image}, timeout=60) upload.raise_for_status() media_id = upload.json()["media_id_string"] payload = {"text": milestone["text"]} if media_id: payload["media"] = {"media_ids": [media_id]} result = oauth.post("https://api.x.com/2/tweets", json=payload, timeout=30) result.raise_for_status() tweet_id = result.json().get("data", {}).get("id") return (f"https://x.com/i/web/status/{tweet_id}" if tweet_id else None), False finally: if path: path.unlink(missing_ok=True) # --- Instagram ------------------------------------------------------------- def instagram_jpeg(path): """Instagram accepts JPEG only, so the rendered PNG is flattened onto white. The card is drawn at 1080x1350 (4:5), the tallest ratio Instagram allows. """ with Image.open(path) as image: rgba = image.convert("RGBA") canvas = Image.new("RGB", rgba.size, "white") canvas.paste(rgba, mask=rgba.split()[3]) buffer = io.BytesIO() canvas.save(buffer, "JPEG", quality=92, optimize=True) target = Path(tempfile.mkstemp(suffix=".jpg")[1]) target.write_bytes(buffer.getvalue()) return target def host_asset(path): """Meta downloads `image_url` itself, so the file must be publicly readable. /api/social-assets stores it in Postgres and serves it from the verified qrmaster.net domain - no object storage and no deploy needed per post. """ base = required("QRMASTER_API_BASE").rstrip("/") response = requests.post( f"{base}/api/social-assets", headers={"x-admin-key": required("SOCIAL_ASSET_ADMIN_KEY")}, json={"files": [{"filename": path.name, "mimeType": "image/jpeg", "dataBase64": base64.b64encode(path.read_bytes()).decode()}]}, timeout=60, ) if not response.ok: raise RuntimeError(f"Asset upload failed: {response.text[:200]}") url = ((response.json().get("assets") or [{}])[0]).get("url") if not url: raise RuntimeError("Asset upload returned no URL") return url def find_existing_instagram_post(caption): """Reconcile by caption: a repeated attempt must not post twice. Instagram has nothing like a client-side idempotency key, and a share token in the caption would only be visible clutter - the caption itself is the identifying detail. """ response = requests.get( graph(f"{required('INSTAGRAM_USER_ID')}/media"), params={"fields": "id,caption,permalink", "limit": 25, "access_token": required("INSTAGRAM_ACCESS_TOKEN")}, timeout=30, ) if not response.ok: raise RuntimeError(graph_error(response)) for media in response.json().get("data") or []: if (media.get("caption") or "").strip() == caption.strip(): return media.get("permalink") return None def publish_instagram(milestone): caption = milestone["text"] if milestone.get("attempts"): existing = find_existing_instagram_post(caption) if existing: return existing, True user_id = required("INSTAGRAM_USER_ID") token = required("INSTAGRAM_ACCESS_TOKEN") png = milestone_image(milestone, "portrait") if not png: raise RuntimeError("Milestone has no share token, so no image can be published") jpeg = None try: jpeg = instagram_jpeg(png) image_url = host_asset(jpeg) container = requests.post(graph(f"{user_id}/media"), data={"image_url": image_url, "caption": caption, "access_token": token}, timeout=60) if not container.ok: raise RuntimeError(graph_error(container)) creation_id = container.json().get("id") if not creation_id: raise RuntimeError("Instagram did not return a container id") # Meta fetches and processes the image asynchronously. deadline = time.time() + 120 while True: status = requests.get(graph(creation_id), params={"fields": "status_code,status", "access_token": token}, timeout=30) if not status.ok: raise RuntimeError(graph_error(status)) code = status.json().get("status_code") if code == "FINISHED": break if code in {"ERROR", "EXPIRED"}: raise RuntimeError(f"Instagram rejected the media container: {status.json().get('status') or code}") if time.time() > deadline: raise RuntimeError("Instagram did not finish processing the image within 120s") time.sleep(5) published = requests.post(graph(f"{user_id}/media_publish"), data={"creation_id": creation_id, "access_token": token}, timeout=60) if not published.ok: raise RuntimeError(graph_error(published)) media_id = published.json().get("id") permalink = None if media_id: link = requests.get(graph(media_id), params={"fields": "permalink", "access_token": token}, timeout=30) permalink = link.json().get("permalink") if link.ok else None return permalink, False finally: png.unlink(missing_ok=True) if jpeg: jpeg.unlink(missing_ok=True) PUBLISHERS = {"x": publish_x, "instagram": publish_instagram} def run_once(channel): base = required("QRMASTER_API_BASE").rstrip("/") + "/api/internal/social-milestones" milestone = api("GET", f"{base}?channel={channel}").get("milestone") if not milestone: return try: post_url, reconciled = PUBLISHERS[channel](milestone) api("PATCH", base, {"id": milestone["id"], "result": "posted", "postUrl": post_url}) print(json.dumps({"posted": milestone["id"], "channel": channel, "reconciled": reconciled, "url": post_url}), flush=True) except Exception as error: api("PATCH", base, {"id": milestone["id"], "result": "failed", "error": str(error)[:500]}) print(f"Milestone post failed ({channel}): {error}", flush=True) if __name__ == "__main__": interval = max(5, int(os.getenv("SOCIAL_WORKER_INTERVAL_SECONDS", "10"))) if os.getenv("SOCIAL_MILESTONE_POSTING_ENABLED", "").lower() not in {"true", "1", "yes"}: raise RuntimeError("Set SOCIAL_MILESTONE_POSTING_ENABLED=true to run this worker") channels = enabled_channels() print(json.dumps({"worker": "social-milestones", "status": "started", "intervalSeconds": interval, "channels": channels}), flush=True) while True: for channel in channels: try: run_once(channel) except Exception as error: # Stay alive and make configuration/network errors visible in the # container logs instead of entering a silent restart loop. print(f"Worker cycle failed ({channel}): {error}", flush=True) time.sleep(interval)