152 lines
6.4 KiB
Python
152 lines
6.4 KiB
Python
"""Always-on QR Master X milestone worker. The web app never receives X keys."""
|
|
import io
|
|
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):
|
|
for candidate in (f"/usr/share/fonts/truetype/dejavu/{name}", f"C:/Windows/Fonts/{'arialbd.ttf' if 'Bold' in name else 'arial.ttf'}"):
|
|
if Path(candidate).exists():
|
|
return ImageFont.truetype(candidate, size)
|
|
return ImageFont.load_default()
|
|
|
|
|
|
def logo():
|
|
"""Use the actual deployed QR Master favicon, not an invented icon."""
|
|
try:
|
|
base = required("QRMASTER_API_BASE").rstrip("/")
|
|
response = requests.get(f"{base}/favicon.ico", timeout=10)
|
|
response.raise_for_status()
|
|
mark = Image.open(io.BytesIO(response.content)).convert("RGBA")
|
|
mark.thumbnail((56, 56))
|
|
return mark
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def render_card(card):
|
|
"""Render an immutable cumulative scan timeline from the stored snapshot."""
|
|
image = Image.new("RGB", (1200, 630), "#f8f7f4")
|
|
draw = ImageDraw.Draw(image)
|
|
navy, blue, slate, mint = "#061b31", "#0256ff", "#64748b", "#059669"
|
|
regular, medium, display = font("DejaVuSans.ttf", 27), font("DejaVuSans-Bold.ttf", 27), font("DejaVuSans.ttf", 142)
|
|
mark = logo()
|
|
if mark:
|
|
image.paste(mark, (68, 58), mark)
|
|
logo_x = 140
|
|
else:
|
|
logo_x = 68
|
|
draw.text((logo_x, 70), "QR MASTER", font=medium, fill=navy)
|
|
|
|
total = int(card.get("totalUniqueScans") or card.get("threshold") or 0)
|
|
total_scans = int(card.get("totalScans") or total)
|
|
draw.text((66, 212), f"{total:,}", font=display, fill=navy)
|
|
draw.text((73, 374), "UNIQUE SCANS", font=medium, fill=navy)
|
|
draw.text((74, 410), f"{total_scans:,} total scans", font=font("DejaVuSans.ttf", 20), fill=slate)
|
|
draw.line((68, 548, 108, 548), fill=blue, width=4)
|
|
draw.text((126, 530), "QR code milestone", font=regular, fill=slate)
|
|
|
|
trend = card.get("trend") or {}
|
|
raw_points = trend.get("points") or []
|
|
left, top, width, height = 590, 140, 540, 330
|
|
if raw_points:
|
|
start_ms = min(_timestamp(point.get("at")) for point in raw_points)
|
|
end_ms = max(_timestamp(point.get("at")) for point in raw_points)
|
|
span = max(1, end_ms - start_ms)
|
|
ceiling = float(trend.get("ceiling") or max(total * 1.25, 1.25))
|
|
points = [(left + round((_timestamp(point.get("at")) - start_ms) / span * width), top + height - round(float(point.get("total", 0)) / ceiling * height)) for point in raw_points]
|
|
target_y = top + height - round(float(trend.get("target") or total) / ceiling * height)
|
|
# Exactly five levels: for 20 scans, 5 / 10 / 15 / 20 / 25.
|
|
for index in range(1, 6):
|
|
value = total * index / 4
|
|
y = top + height - round(value / ceiling * height)
|
|
is_target = index == 4
|
|
draw.line((left, y, left + width, y), fill="#bfdbfe" if is_target else "#dde5ef", width=2 if is_target else 1)
|
|
draw.text((left - 15, y - 12), _format_axis(value), font=font("DejaVuSans.ttf", 18), fill=blue if is_target else slate, anchor="ra")
|
|
draw.line(points, fill=blue, width=5, joint="curve")
|
|
x, y = points[-1]
|
|
draw.ellipse((x - 8, y - 8, x + 8, y + 8), fill="#ffffff", outline=blue, width=5)
|
|
draw.text((left, top + height + 27), trend.get("startLabel") or "Created", font=font("DejaVuSans.ttf", 19), fill=slate)
|
|
draw.text((left + width, top + height + 27), trend.get("endLabel") or "Reached", font=font("DejaVuSans.ttf", 19), fill=slate, anchor="ra")
|
|
draw.text((870, 530), "VERIFIED SCAN DATA", font=font("DejaVuSans-Bold.ttf", 18), fill=mint)
|
|
path = Path(tempfile.mkstemp(suffix=".png")[1])
|
|
image.save(path, "PNG", optimize=True)
|
|
return path
|
|
|
|
|
|
def _timestamp(value):
|
|
try:
|
|
return int(__import__("datetime").datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() * 1000)
|
|
except Exception:
|
|
return 0
|
|
|
|
|
|
def _format_axis(value):
|
|
return f"{value:g}" if value < 1000 else f"{value:,.0f}"
|
|
|
|
|
|
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)
|