114 lines
5.5 KiB
Python
114 lines
5.5 KiB
Python
"""Always-on QR Master 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 font(name, size):
|
|
return ImageFont.truetype(f"/usr/share/fonts/truetype/dejavu/{name}", size)
|
|
|
|
|
|
def render_card(card):
|
|
"""Render the consented immutable scan snapshot; never invent trend data."""
|
|
image = Image.new("RGB", (1200, 630), "#f8fafc")
|
|
draw = ImageDraw.Draw(image)
|
|
navy, blue, slate, border, green = "#061b31", "#0256ff", "#64748b", "#e2e8f0", "#059669"
|
|
regular, medium, bold, display = font("DejaVuSans.ttf", 28), font("DejaVuSans-Bold.ttf", 28), font("DejaVuSans-Bold.ttf", 42), font("DejaVuSans-Bold.ttf", 116)
|
|
draw.rounded_rectangle((45, 42, 1155, 588), radius=24, fill="#ffffff", outline=border, width=2)
|
|
draw.rounded_rectangle((82, 79, 114, 111), radius=7, fill=blue)
|
|
draw.text((130, 81), "QR MASTER", font=medium, fill=navy)
|
|
draw.rounded_rectangle((932, 77, 1118, 114), radius=8, fill="#ecfdf5")
|
|
draw.text((954, 84), "Verified milestone", font=font("DejaVuSans-Bold.ttf", 17), fill=green)
|
|
draw.text((84, 158), "TOTAL UNIQUE SCANS", font=font("DejaVuSans-Bold.ttf", 18), fill="#94a3b8")
|
|
total = int(card.get("totalUniqueScans") or card.get("threshold") or 0)
|
|
draw.text((78, 184), f"{total:,}", font=display, fill=navy)
|
|
label = "eindeutige Scans" if card.get("language") == "de" else "unique scans"
|
|
draw.text((86, 325), label, font=regular, fill=slate)
|
|
trend = card.get("trend") or None
|
|
if trend and len(trend.get("series", [])) > 1:
|
|
series = trend["series"]
|
|
left, top, width, height = 84, 390, 1030, 88
|
|
max_value = max(series) or 1
|
|
points = [(left + round(index * width / (len(series) - 1)), top + height - round(value / max_value * height)) for index, value in enumerate(series)]
|
|
for y in (top, top + height // 2, top + height):
|
|
draw.line((left, y, left + width, y), fill="#edf2f7", width=2)
|
|
draw.line(points, fill=blue, width=6, joint="curve")
|
|
for x, y in (points[0], points[-1]):
|
|
draw.ellipse((x - 7, y - 7, x + 7, y + 7), fill=blue)
|
|
draw.text((84, 495), f"Last {trend.get('periodDays', 7)} days · {trend.get('recentTotal', 0)} unique scans", font=font("DejaVuSans.ttf", 18), fill=slate)
|
|
else:
|
|
draw.rounded_rectangle((84, 398, 357, 452), radius=10, fill="#eff6ff")
|
|
copy = "Erste Dynamik" if card.get("language") == "de" else "Early momentum"
|
|
draw.text((106, 411), copy, font=font("DejaVuSans-Bold.ttf", 20), fill=blue)
|
|
draw.text((84, 495), "Trend appears once enough real scan history exists." if card.get("language") != "de" else "Der Trend erscheint mit ausreichend echten Scan-Daten.", font=font("DejaVuSans.ttf", 18), fill=slate)
|
|
draw.line((84, 532, 1116, 532), fill=border, width=2)
|
|
draw.text((84, 549), card.get("qrTitle") or card.get("title") or "QR code", font=medium, fill=navy)
|
|
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"))
|
|
tweet_id = result.get("data", {}).get("id")
|
|
post_url = f"https://x.com/i/web/status/{tweet_id}" if tweet_id else None
|
|
api("PATCH", base, {"id": milestone["id"], "result": "posted", "postUrl": post_url})
|
|
print(json.dumps({"posted": milestone["id"], "x": result}), flush=True)
|
|
except Exception as error:
|
|
api("PATCH", base, {"id": milestone["id"], "result": "failed", "error": str(error)[:500]})
|
|
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)
|