#!/usr/bin/env python3 """Sequential Pixelixe certificate experiment. Python 3.9+ and curl required. PIXELIXE_API_KEY is used only in an HTTPS Authorization header. Returned storage URLs remain in memory and are never written to the report or request log. """ import argparse import csv import datetime as dt import hashlib import json import math import os from pathlib import Path import statistics import struct import subprocess import tempfile import time from urllib.parse import urlsplit ENDPOINT = 'https://studio.pixelixe.com/api/graphic/automation/v2' FIELDS = ('recipient_name', 'course_title', 'issue_date', 'issuer_name') def now(): return dt.datetime.now(dt.timezone.utc).isoformat(timespec='milliseconds') def curl(url, headers=None, payload=None, output=None): args = ['curl', '--silent', '--show-error', '--connect-timeout', '15', '--max-time', '90', '--proto', '=https', '--output', str(output), '--write-out', '%{http_code} %{time_total}'] if headers: args += ['--header', '@' + str(headers), '--header', 'Content-Type: application/x-www-form-urlencoded'] if payload is not None: args += ['--data-urlencode', 'json@-'] args.append(url) result = subprocess.run(args, input=json.dumps(payload, ensure_ascii=False).encode() if payload is not None else None, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) bits = result.stdout.decode().strip().split() status = int(bits[0]) if bits and bits[0].isdigit() else 0 elapsed = float(bits[1]) if len(bits) > 1 else 0 # Never print curl stderr: a URL in an error may contain an account identifier. return status, round(elapsed * 1000, 3), result.returncode def payload_for(row, document_uid): modifications = [{'element_name': key, 'type': 'text', 'text': str(row[key])} for key in FIELDS] modifications[0]['font-size'] = 'auto' return {'document_uid': document_uid, 'format': 'json', 'image_type': 'png', 'modifications': modifications} def main(): p = argparse.ArgumentParser(description=__doc__) p.add_argument('--csv', type=Path, required=True) p.add_argument('--document-uid', required=True) p.add_argument('--output', type=Path, required=True) p.add_argument('--count', type=int, default=100) p.add_argument('--auth-header-file', type=Path, help='Alternative to PIXELIXE_API_KEY; file containing one Authorization header.') a = p.parse_args() if not 1 <= a.count <= 100: p.error('--count must be between 1 and 100') if a.document_uid.startswith('YOUR_'): p.error('Set your own saved certificate template UID') with a.csv.open(newline='', encoding='utf-8-sig') as f: reader = csv.DictReader(f) if not set(FIELDS).issubset(reader.fieldnames or []): p.error('CSV must contain: ' + ', '.join(FIELDS)) rows = list(reader)[:a.count] if len(rows) != a.count or any(not str(r.get(k, '')).strip() for r in rows for k in FIELDS): p.error('Provide enough non-empty rows for the requested count') key = os.environ.get('PIXELIXE_API_KEY', '') if not a.auth_header_file and not key: p.error('Set PIXELIXE_API_KEY or --auth-header-file') a.output.mkdir(parents=True, exist_ok=False) images = a.output / 'images' images.mkdir() records, downloads = [], [] with tempfile.TemporaryDirectory(prefix='pixelixe-experiment-') as temporary: tmp = Path(temporary) header = a.auth_header_file or tmp / 'auth.txt' if not a.auth_header_file: header.write_text('Authorization: Bearer ' + key + '\n') header.chmod(0o600) started = now() batch_start = time.perf_counter() for index, row in enumerate(rows, 1): response = tmp / 'response.json' status, elapsed, code = curl(ENDPOINT, headers=header, payload=payload_for(row, a.document_uid), output=response) data = {} try: data = json.loads(response.read_text()) except (ValueError, OSError, UnicodeDecodeError): pass url = data.get('image_url', '') if isinstance(data, dict) else '' success = bool(code == 0 and status == 200 and data.get('status') == 'success' and url) record = {'row': index, 'recipient_name': row['recipient_name'], 'http_status': status, 'api_elapsed_ms': elapsed, 'api_success': success, 'download_success': False, 'width': None, 'height': None, 'png_bytes': 0, 'sha256': '', 'output_file': '', 'error': ''} if success and urlsplit(url).scheme == 'https' and urlsplit(url).hostname == 'studio.pixelixe.com': downloads.append((record, url)) else: record['api_success'] = False record['error'] = 'transport_or_api_response' records.append(record) if index % 10 == 0 or not success: print(f'API {index}/{len(rows)}; HTTP {status}; {elapsed:.0f} ms', flush=True) if status in (401, 403, 429): print('Stopped on authorization or rate-limit response; no automatic retry.', flush=True) break api_batch_seconds = time.perf_counter() - batch_start api_completed = now() download_start = time.perf_counter() for record, url in downloads: target = images / ('certificate-%03d.png' % record['row']) status, elapsed, code = curl(url, output=target) raw = target.read_bytes() if target.is_file() else b'' valid = code == 0 and status == 200 and raw[:8] == b'\x89PNG\r\n\x1a\n' and len(raw) >= 33 if valid: width, height = struct.unpack('>II', raw[16:24]) valid = (width, height) == (1754, 1240) if valid: record.update(download_success=True, width=width, height=height, png_bytes=len(raw), sha256=hashlib.sha256(raw).hexdigest(), output_file='images/' + target.name) else: record['error'] = 'download_or_png_dimensions' target.unlink(missing_ok=True) record['download_elapsed_ms'] = elapsed download_seconds = time.perf_counter() - download_start completed = now() latencies = sorted(r['api_elapsed_ms'] for r in records if r['api_success']) successful = sum(r['api_success'] and r['download_success'] for r in records) unique = len({r['sha256'] for r in records if r['download_success']}) summary = {'experiment': '100 sequential certificate PNG renders from a UTF-8 CSV', 'date_utc': started[:10], 'started_utc': started, 'api_completed_utc': api_completed, 'completed_utc': completed, 'endpoint': ENDPOINT, 'mode': 'sequential, concurrency=1, retries=0', 'requested_rows': a.count, 'attempted_rows': len(records), 'api_successes': sum(r['api_success'] for r in records), 'verified_pngs': successful, 'distinct_png_sha256': unique, 'dimensions': [1754, 1240], 'api_batch_seconds': round(api_batch_seconds, 3), 'downloads_seconds': round(download_seconds, 3), 'median_api_ms': round(statistics.median(latencies), 3) if latencies else None, 'p95_api_ms_nearest_rank': latencies[math.ceil(.95 * len(latencies)) - 1] if latencies else None, 'min_api_ms': min(latencies) if latencies else None, 'max_api_ms': max(latencies) if latencies else None, 'total_png_bytes': sum(r['png_bytes'] for r in records), 'source_csv_sha256': hashlib.sha256(a.csv.read_bytes()).hexdigest(), 'limits': ['One run from one client; not an SLA, capacity test or representative performance guarantee.', 'Request timing includes network and API response time. File download time is measured separately.', 'Template creation, manual preparation and human visual QA are not included in the API batch time.', 'No manual-time or financial-savings comparison was measured.', 'Programmatic PNG checks cover signature and dimensions; a separate decode and visual check is recommended.']} (a.output / 'results.json').write_text(json.dumps({'summary': summary, 'rows': records}, indent=2, ensure_ascii=False) + '\n') with (a.output / 'results.csv').open('w', encoding='utf-8', newline='') as f: writer = csv.DictWriter(f, fieldnames=list(records[0])) writer.writeheader() writer.writerows(records) print(json.dumps(summary, ensure_ascii=False, indent=2)) if successful != a.count or unique != a.count: raise SystemExit('Experiment incomplete: inspect results before publishing any conclusion.') if __name__ == '__main__': main()