QR-master/scripts/social-worker/worker.py

"""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/<token>/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)