Codex
109 lines
5.3 KiB
TypeScript
109 lines
5.3 KiB
TypeScript
export const remotePythonAnnouncementsScript = String.raw`
|
|
|
|
def announcements_payload():
|
|
return json.loads(base64.b64decode(PAYLOAD_B64).decode("utf-8"))
|
|
|
|
def announcements_list(token, payload):
|
|
query = [
|
|
"page=" + quote(str(payload.get("page") or 1), safe=""),
|
|
"page_size=" + quote(str(payload.get("pageSize") or 20), safe=""),
|
|
"sort_by=" + quote(str(payload.get("sortBy") or "created_at"), safe=""),
|
|
"sort_order=" + quote(str(payload.get("sortOrder") or "desc"), safe=""),
|
|
]
|
|
if payload.get("status"):
|
|
query.append("status=" + quote(str(payload.get("status")), safe=""))
|
|
if payload.get("search"):
|
|
query.append("search=" + quote(str(payload.get("search")), safe=""))
|
|
data = ensure_success(curl_api("GET", "/api/v1/admin/announcements?" + "&".join(query), bearer=token), "list announcements")
|
|
items = extract_items(data)
|
|
return {
|
|
"ok": True, "mode": "read-only", "mutation": False, "writeAttempted": 0,
|
|
"items": items,
|
|
"total": data.get("total", len(items)) if isinstance(data, dict) else len(items),
|
|
"page": data.get("page", payload.get("page")) if isinstance(data, dict) else payload.get("page"),
|
|
"pageSize": data.get("page_size", payload.get("pageSize")) if isinstance(data, dict) else payload.get("pageSize"),
|
|
"totalPages": data.get("total_pages") if isinstance(data, dict) else None,
|
|
"valuesPrinted": False,
|
|
}
|
|
|
|
def announcements_get(token, announcement_id):
|
|
item = ensure_success(curl_api("GET", f"/api/v1/admin/announcements/{announcement_id}", bearer=token), "get announcement")
|
|
return {"ok": isinstance(item, dict), "mode": "read-only", "mutation": False, "writeAttempted": 0, "announcement": item, "valuesPrinted": False}
|
|
|
|
def announcement_time_seconds(value):
|
|
if value is None or value == "":
|
|
return None
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
|
return int(value)
|
|
if isinstance(value, str):
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=timezone.utc)
|
|
return int(parsed.timestamp())
|
|
except ValueError:
|
|
return value
|
|
return value
|
|
|
|
def announcement_targeting(value):
|
|
if not isinstance(value, dict):
|
|
return value
|
|
normalized = dict(value)
|
|
if normalized.get("any_of") is None:
|
|
normalized["any_of"] = []
|
|
return normalized
|
|
|
|
def announcements_reconcile(expected, actual):
|
|
fields = ("title", "content", "status", "notify_mode", "targeting", "starts_at", "ends_at")
|
|
expected_values = {}
|
|
actual_values = {}
|
|
mismatched_fields = []
|
|
for field in fields:
|
|
expected_value = expected.get(field) if isinstance(expected, dict) else None
|
|
actual_value = actual.get(field) if isinstance(actual, dict) else None
|
|
if field in ("starts_at", "ends_at"):
|
|
expected_value = announcement_time_seconds(expected_value)
|
|
actual_value = announcement_time_seconds(actual_value)
|
|
elif field == "targeting":
|
|
expected_value = announcement_targeting(expected_value)
|
|
actual_value = announcement_targeting(actual_value)
|
|
expected_values[field] = expected_value
|
|
actual_values[field] = actual_value
|
|
if expected_value != actual_value:
|
|
mismatched_fields.append(field)
|
|
return {"expected": expected_values, "actual": actual_values, "mismatchedFields": mismatched_fields}
|
|
|
|
def announcements_create(token, payload):
|
|
announcement = payload.get("announcement") if isinstance(payload.get("announcement"), dict) else {}
|
|
created = ensure_success(curl_api("POST", "/api/v1/admin/announcements", bearer=token, payload=announcement), "create announcement")
|
|
announcement_id = created.get("id") if isinstance(created, dict) else None
|
|
if not isinstance(announcement_id, int) or announcement_id <= 0:
|
|
raise RuntimeError("create announcement response missing positive id")
|
|
actual = ensure_success(curl_api("GET", f"/api/v1/admin/announcements/{announcement_id}", bearer=token), "read announcement after create")
|
|
reconciliation = announcements_reconcile(announcement, actual)
|
|
id_matched = isinstance(actual, dict) and actual.get("id") == announcement_id
|
|
mismatched_fields = list(reconciliation.get("mismatchedFields") or [])
|
|
if not id_matched:
|
|
mismatched_fields.insert(0, "id")
|
|
reconciled = id_matched and len(mismatched_fields) == 0
|
|
return {
|
|
"ok": reconciled, "mode": "confirmed", "mutation": True, "writeAttempted": 1,
|
|
"writeSucceeded": 1, "writeFailed": 0, "reconciled": reconciled,
|
|
"mismatchedFields": mismatched_fields,
|
|
"reconciliation": {**reconciliation, "idMatched": id_matched},
|
|
"announcement": actual, "valuesPrinted": False,
|
|
}
|
|
|
|
def run_announcements():
|
|
payload = announcements_payload()
|
|
_, token, _ = login()
|
|
action = payload.get("action")
|
|
if action == "list":
|
|
return announcements_list(token, payload)
|
|
if action == "get":
|
|
return announcements_get(token, payload.get("id"))
|
|
if action == "create" and payload.get("confirm") is True:
|
|
return announcements_create(token, payload)
|
|
raise RuntimeError("unsupported announcements action")
|
|
`;
|