87 lines
3.4 KiB
Python
87 lines
3.4 KiB
Python
"""Always-on QRMaster X milestone worker. The web app never receives X keys."""
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
from requests_oauthlib import OAuth1Session
|
|
|
|
|
|
def required(name):
|
|
value = os.getenv(name, "").strip()
|
|
if not value:
|
|
raise RuntimeError(f"Missing {name}")
|
|
return value
|
|
|
|
|
|
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 render_card(card):
|
|
image = Image.new("RGB", (1200, 675), "#061b31")
|
|
draw = ImageDraw.Draw(image)
|
|
fonts = Path("/usr/share/fonts/truetype/dejavu")
|
|
bold = ImageFont.truetype(str(fonts / "DejaVuSans-Bold.ttf"), 112)
|
|
regular = ImageFont.truetype(str(fonts / "DejaVuSans.ttf"), 34)
|
|
image_draw = draw
|
|
image_draw.rounded_rectangle((55, 55, 1145, 620), radius=28, outline="#304866", width=2)
|
|
image_draw.text((100, 105), "QR MASTER", font=regular, fill="#dce8f7")
|
|
image_draw.text((100, 210), f"{card['threshold']:,}", font=bold, fill="#ffffff")
|
|
scans = "eindeutige Scans" if card.get("language") == "de" else "unique scans"
|
|
image_draw.text((105, 350), scans, font=regular, fill="#b8c7da")
|
|
image_draw.line((100, 500, 1100, 500), fill="#304866", width=2)
|
|
image_draw.text((100, 535), f"{card['label']} · {card['title']}", font=regular, fill="#dce8f7")
|
|
path = Path(tempfile.mkstemp(suffix=".png")[1])
|
|
image.save(path, "PNG", optimize=True)
|
|
return path
|
|
|
|
|
|
def post_x(text, card):
|
|
oauth = 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"))
|
|
path = render_card(card) if card else None
|
|
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": 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()
|
|
return result.json()
|
|
finally:
|
|
if path:
|
|
path.unlink(missing_ok=True)
|
|
|
|
|
|
def run_once():
|
|
base = required("QRMASTER_API_BASE").rstrip("/") + "/api/internal/social-milestones"
|
|
milestone = api("GET", base).get("milestone")
|
|
if not milestone:
|
|
return
|
|
try:
|
|
result = post_x(milestone["text"], milestone.get("card"))
|
|
api("PATCH", base, {"id": milestone["id"], "result": "posted"})
|
|
print(json.dumps({"posted": milestone["id"], "x": result}), flush=True)
|
|
except Exception as error:
|
|
api("PATCH", base, {"id": milestone["id"], "result": "failed"})
|
|
print(f"Milestone post failed: {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")
|
|
while True:
|
|
run_once()
|
|
time.sleep(interval)
|