feat(seed): add API test-data seeding script and make testdata target
- scripts/seed.py: stdlib HTTP client, JPEG+EXIF via optional Pillow (PNG fallback), additive-only, dry-run - Makefile: testdata target with ALBUMS/PHOTOS/SEED/TIMEOUT and boolean flags - README: usage section; new dev-seed-data capability spec
This commit is contained in:
+480
@@ -0,0 +1,480 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Populate a running galo instance with test data through its public API.
|
||||
|
||||
The script logs in as an existing owner account and adds albums, generated
|
||||
photos and (optionally) tags, comments, likes, share links and invites. It is
|
||||
strictly additive: it never deletes or modifies existing data. Repeat runs add
|
||||
new data that is distinguishable by a run tag baked into album titles.
|
||||
|
||||
Core functionality uses only the Python standard library. Pillow is an optional
|
||||
dependency: when present, photos are JPEG with EXIF (capture time and GPS);
|
||||
otherwise they are PNG without EXIF.
|
||||
|
||||
Configuration:
|
||||
GALO_BASE_URL base URL of the instance (default: http://localhost:8080)
|
||||
GALO_EMAIL owner email (required)
|
||||
GALO_PASSWORD owner password (required)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
import struct
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
import uuid
|
||||
import zlib
|
||||
from fractions import Fraction
|
||||
|
||||
try:
|
||||
from PIL import Image, ImageDraw
|
||||
|
||||
HAVE_PIL = True
|
||||
except ImportError: # pragma: no cover - exercised only without Pillow
|
||||
HAVE_PIL = False
|
||||
|
||||
DEFAULT_BASE_URL = "http://localhost:8080"
|
||||
|
||||
COMMENT_TEXTS = [
|
||||
"Какое место!",
|
||||
"Отличный кадр",
|
||||
"Нужно повторить эту поездку",
|
||||
"Здесь был потрясающий свет",
|
||||
"Любимое фото из альбома",
|
||||
]
|
||||
|
||||
TAG_NAMES = ["travel", "summer", "family", "friends", "nature", "city", "sunset"]
|
||||
|
||||
|
||||
class ApiError(Exception):
|
||||
"""Error returned by the galo API or the transport layer."""
|
||||
|
||||
def __init__(self, status: int, code: str, message: str):
|
||||
super().__init__(f"{status} {code}: {message}")
|
||||
self.status = status
|
||||
self.code = code
|
||||
self.message = message
|
||||
|
||||
|
||||
def encode_multipart(fields, files, boundary):
|
||||
"""Build a multipart/form-data body.
|
||||
|
||||
fields: iterable of (name, value). files: iterable of
|
||||
(name, filename, content_type, data).
|
||||
"""
|
||||
body = bytearray()
|
||||
for name, value in fields:
|
||||
body += f"--{boundary}\r\n".encode()
|
||||
body += f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode()
|
||||
body += str(value).encode()
|
||||
body += b"\r\n"
|
||||
for name, filename, content_type, data in files:
|
||||
body += f"--{boundary}\r\n".encode()
|
||||
body += (
|
||||
f'Content-Disposition: form-data; name="{name}"; '
|
||||
f'filename="{filename}"\r\n'
|
||||
).encode()
|
||||
body += f"Content-Type: {content_type}\r\n\r\n".encode()
|
||||
body += data
|
||||
body += b"\r\n"
|
||||
body += f"--{boundary}--\r\n".encode()
|
||||
return bytes(body)
|
||||
|
||||
|
||||
class Client:
|
||||
"""Minimal JSON/multipart HTTP client over urllib."""
|
||||
|
||||
def __init__(self, base_url, timeout=30):
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.timeout = timeout
|
||||
self.token = None
|
||||
|
||||
def request(self, method, path, *, json_body=None, files=None, fields=None):
|
||||
url = self.base_url + path
|
||||
headers = {"Accept": "application/json"}
|
||||
data = None
|
||||
if json_body is not None:
|
||||
data = json.dumps(json_body).encode("utf-8")
|
||||
headers["Content-Type"] = "application/json"
|
||||
elif files is not None:
|
||||
boundary = "----galoseed" + uuid.uuid4().hex
|
||||
data = encode_multipart(fields or [], files, boundary)
|
||||
headers["Content-Type"] = (
|
||||
"multipart/form-data; boundary=" + boundary
|
||||
)
|
||||
if self.token:
|
||||
headers["Authorization"] = "Bearer " + self.token
|
||||
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
|
||||
raw = resp.read()
|
||||
except urllib.error.HTTPError as exc:
|
||||
raise _http_error(exc)
|
||||
except urllib.error.URLError as exc:
|
||||
raise ApiError(0, "network_error", str(exc.reason))
|
||||
if not raw:
|
||||
return None
|
||||
try:
|
||||
return json.loads(raw)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _http_error(exc):
|
||||
code, message = "http_error", str(exc.reason)
|
||||
try:
|
||||
payload = json.loads(exc.read())
|
||||
err = payload.get("error") or {}
|
||||
code = err.get("code", code)
|
||||
message = err.get("message", message)
|
||||
except Exception:
|
||||
pass
|
||||
return ApiError(exc.code, code, message)
|
||||
|
||||
|
||||
# --- image generation -------------------------------------------------------
|
||||
|
||||
|
||||
def _rational(value):
|
||||
deg = int(value)
|
||||
minutes = int((value - deg) * 60)
|
||||
seconds = (value - deg - minutes / 60.0) * 3600.0
|
||||
return (Fraction(deg), Fraction(minutes), Fraction(round(seconds * 100), 100))
|
||||
|
||||
|
||||
def generate_jpeg(rng):
|
||||
"""Return JPEG bytes with EXIF capture time (and GPS when accepted)."""
|
||||
width = rng.randint(640, 1600)
|
||||
height = rng.randint(480, 1200)
|
||||
base = (rng.randint(20, 220), rng.randint(20, 220), rng.randint(20, 220))
|
||||
img = Image.new("RGB", (width, height))
|
||||
draw = ImageDraw.Draw(img)
|
||||
for step in range(0, height, 4):
|
||||
shade = (base[0] + step) % 256
|
||||
draw.rectangle(
|
||||
[0, step, width, step + 4],
|
||||
fill=(shade, (base[1] + step // 2) % 256, base[2]),
|
||||
)
|
||||
draw.ellipse(
|
||||
[
|
||||
rng.randint(0, width // 2),
|
||||
rng.randint(0, height // 2),
|
||||
rng.randint(width // 2, width),
|
||||
rng.randint(height // 2, height),
|
||||
],
|
||||
fill=(rng.randint(0, 255), rng.randint(0, 255), rng.randint(0, 255)),
|
||||
)
|
||||
|
||||
taken = "20{:02d}:{:02d}:{:02d} {:02d}:{:02d}:{:02d}".format(
|
||||
rng.randint(10, 24),
|
||||
rng.randint(1, 12),
|
||||
rng.randint(1, 28),
|
||||
rng.randint(0, 23),
|
||||
rng.randint(0, 59),
|
||||
rng.randint(0, 59),
|
||||
)
|
||||
exif = Image.Exif()
|
||||
exif[36867] = taken # DateTimeOriginal
|
||||
exif[36868] = taken # DateTimeDigitized
|
||||
exif[306] = taken # DateTime
|
||||
try:
|
||||
lat = rng.uniform(-60.0, 60.0)
|
||||
lon = rng.uniform(-120.0, 120.0)
|
||||
exif[0x8825] = {
|
||||
1: "N" if lat >= 0 else "S",
|
||||
2: _rational(abs(lat)),
|
||||
3: "E" if lon >= 0 else "W",
|
||||
4: _rational(abs(lon)),
|
||||
}
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
buf = io.BytesIO()
|
||||
img.save(buf, format="JPEG", quality=85, exif=exif)
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def _png_chunk(tag, data):
|
||||
return (
|
||||
struct.pack(">I", len(data))
|
||||
+ tag
|
||||
+ data
|
||||
+ struct.pack(">I", zlib.crc32(tag + data) & 0xFFFFFFFF)
|
||||
)
|
||||
|
||||
|
||||
def generate_png(rng):
|
||||
"""Return PNG bytes built with the standard library (no EXIF)."""
|
||||
width, height = 320, 240
|
||||
base = (rng.randint(0, 255), rng.randint(0, 255), rng.randint(0, 255))
|
||||
raw = bytearray()
|
||||
for y in range(height):
|
||||
raw.append(0) # filter type: none
|
||||
row = bytearray()
|
||||
for x in range(width):
|
||||
row += bytes(
|
||||
((base[0] + x) % 256, (base[1] + y) % 256, base[2])
|
||||
)
|
||||
raw += row
|
||||
ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)
|
||||
return (
|
||||
b"\x89PNG\r\n\x1a\n"
|
||||
+ _png_chunk(b"IHDR", ihdr)
|
||||
+ _png_chunk(b"IDAT", zlib.compress(bytes(raw), 6))
|
||||
+ _png_chunk(b"IEND", b"")
|
||||
)
|
||||
|
||||
|
||||
def make_image(rng):
|
||||
"""Return (filename, content_type, data) for one generated photo."""
|
||||
if HAVE_PIL:
|
||||
return "seed-{}.jpg".format(uuid.uuid4().hex[:8]), "image/jpeg", generate_jpeg(rng)
|
||||
return "seed-{}.png".format(uuid.uuid4().hex[:8]), "image/png", generate_png(rng)
|
||||
|
||||
|
||||
# --- seeding ----------------------------------------------------------------
|
||||
|
||||
|
||||
class Stats:
|
||||
def __init__(self):
|
||||
self.albums = 0
|
||||
self.photos = 0
|
||||
self.tags = 0
|
||||
self.comments = 0
|
||||
self.likes = 0
|
||||
self.share_links = 0
|
||||
self.invites = 0
|
||||
|
||||
def report(self):
|
||||
return (
|
||||
"создано: альбомов={albums}, фото={photos}, тегов={tags}, "
|
||||
"комментариев={comments}, лайков={likes}, "
|
||||
"share-ссылок={share_links}, инвайтов={invites}".format(
|
||||
**self.__dict__
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def login(client, email, password):
|
||||
resp = client.request(
|
||||
"POST", "/api/auth/login", json_body={"email": email, "password": password}
|
||||
)
|
||||
client.token = resp["tokens"]["access_token"]
|
||||
|
||||
|
||||
def create_album(client, title):
|
||||
resp = client.request(
|
||||
"POST",
|
||||
"/api/albums",
|
||||
json_body={"title": title, "description": "Тестовые данные (seed)"},
|
||||
)
|
||||
return resp["album"]["id"]
|
||||
|
||||
|
||||
def upload_photo(client, album_id, rng):
|
||||
filename, content_type, data = make_image(rng)
|
||||
resp = client.request(
|
||||
"POST",
|
||||
"/api/albums/{}/photos".format(album_id),
|
||||
files=[("files", filename, content_type, data)],
|
||||
)
|
||||
return resp["photos"][0]["id"]
|
||||
|
||||
|
||||
def wait_for_variants(client, photo_id, timeout, poll=1.0):
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
resp = client.request("GET", "/api/photos/{}".format(photo_id))
|
||||
photo = resp.get("photo") if resp else None
|
||||
if photo and photo.get("thumb_url"):
|
||||
return True
|
||||
time.sleep(poll)
|
||||
return False
|
||||
|
||||
|
||||
def add_tag(client, photo_id, name):
|
||||
client.request(
|
||||
"POST", "/api/photos/{}/tags".format(photo_id), json_body={"name": name}
|
||||
)
|
||||
|
||||
|
||||
def add_comment(client, photo_id, text):
|
||||
client.request(
|
||||
"POST", "/api/photos/{}/comments".format(photo_id), json_body={"text": text}
|
||||
)
|
||||
|
||||
|
||||
def add_like(client, photo_id):
|
||||
client.request("POST", "/api/photos/{}/likes".format(photo_id))
|
||||
|
||||
|
||||
def create_share_link(client, album_id):
|
||||
client.request("POST", "/api/albums/{}/share".format(album_id), json_body={})
|
||||
|
||||
|
||||
def create_invite(client, album_id, role):
|
||||
client.request(
|
||||
"POST",
|
||||
"/api/albums/{}/invites".format(album_id),
|
||||
json_body={"role": role},
|
||||
)
|
||||
|
||||
|
||||
def build_plan(args):
|
||||
photo_style = "JPEG+EXIF (Pillow)" if HAVE_PIL else "PNG без EXIF (stdlib)"
|
||||
lines = [
|
||||
"План наполнения (сухой прогон, запросы не отправляются):",
|
||||
" альбомов: {}".format(args.albums),
|
||||
" фото на альбом: {}".format(args.photos_per_album),
|
||||
" формат фото: {}".format(photo_style),
|
||||
]
|
||||
extras = []
|
||||
if args.tags:
|
||||
extras.append("теги")
|
||||
if args.comments:
|
||||
extras.append("комментарии")
|
||||
if args.likes:
|
||||
extras.append("лайки")
|
||||
if args.share_links:
|
||||
extras.append("share-ссылки")
|
||||
if args.invites:
|
||||
extras.append("инвайты")
|
||||
lines.append(" сопутствующие сущности: {}".format(", ".join(extras) or "нет"))
|
||||
lines.append(" ожидание вариантов: {}".format("да" if args.wait_variants else "нет"))
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def run(args, base_url, email, password):
|
||||
if args.dry_run:
|
||||
print(build_plan(args))
|
||||
return 0
|
||||
|
||||
rng = random.Random(args.seed)
|
||||
run_tag = "seed-{}-{}".format(int(time.time()), args.seed)
|
||||
client = Client(base_url, timeout=args.timeout)
|
||||
stats = Stats()
|
||||
|
||||
login(client, email, password)
|
||||
print("Вошел как {} @ {}".format(email, base_url))
|
||||
|
||||
for album_index in range(1, args.albums + 1):
|
||||
album_id = create_album(
|
||||
client, "{} #{}".format(run_tag, album_index)
|
||||
)
|
||||
stats.albums += 1
|
||||
print("Альбом {}: {}".format(album_index, album_id))
|
||||
|
||||
for photo_index in range(1, args.photos_per_album + 1):
|
||||
photo_id = upload_photo(client, album_id, rng)
|
||||
stats.photos += 1
|
||||
|
||||
if args.wait_variants:
|
||||
if wait_for_variants(client, photo_id, args.timeout):
|
||||
pass
|
||||
else:
|
||||
print(
|
||||
" ! Таймаут ожидания вариантов для фото {} "
|
||||
"(продолжаю)".format(photo_id),
|
||||
file=sys.stderr,
|
||||
)
|
||||
|
||||
if args.tags:
|
||||
add_tag(client, photo_id, rng.choice(TAG_NAMES))
|
||||
stats.tags += 1
|
||||
if args.comments and rng.random() < 0.5:
|
||||
add_comment(client, photo_id, rng.choice(COMMENT_TEXTS))
|
||||
stats.comments += 1
|
||||
if args.likes and rng.random() < 0.5:
|
||||
add_like(client, photo_id)
|
||||
stats.likes += 1
|
||||
|
||||
if args.share_links:
|
||||
create_share_link(client, album_id)
|
||||
stats.share_links += 1
|
||||
if args.invites:
|
||||
create_invite(client, album_id, "member")
|
||||
stats.invites += 1
|
||||
|
||||
print(stats.report())
|
||||
return 0
|
||||
|
||||
|
||||
def parse_args(argv=None):
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Наполнить инстанс galo тестовыми данными через API (только добавление)."
|
||||
)
|
||||
parser.add_argument("--albums", type=int, default=2, help="число альбомов (по умолчанию 2)")
|
||||
parser.add_argument(
|
||||
"--photos-per-album",
|
||||
type=int,
|
||||
default=10,
|
||||
help="фото на альбом (по умолчанию 10)",
|
||||
)
|
||||
parser.add_argument("--seed", type=int, default=0, help="seed генератора (по умолчанию 0)")
|
||||
parser.add_argument("--tags", action="store_true", help="добавлять теги")
|
||||
parser.add_argument("--comments", action="store_true", help="добавлять комментарии")
|
||||
parser.add_argument("--likes", action="store_true", help="добавлять лайки")
|
||||
parser.add_argument("--share-links", action="store_true", help="создавать share-ссылки")
|
||||
parser.add_argument("--invites", action="store_true", help="создавать инвайты (роль member)")
|
||||
parser.add_argument(
|
||||
"--wait-variants",
|
||||
action="store_true",
|
||||
help="дожидаться генерации превью каждого фото",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--timeout",
|
||||
type=float,
|
||||
default=30.0,
|
||||
help="таймаут ожидания вариантов и сетевых запросов, сек (по умолчанию 30)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
help="показать план и не отправлять изменяющих запросов",
|
||||
)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
args = parse_args(argv)
|
||||
base_url = os.environ.get("GALO_BASE_URL", DEFAULT_BASE_URL)
|
||||
email = os.environ.get("GALO_EMAIL")
|
||||
password = os.environ.get("GALO_PASSWORD")
|
||||
|
||||
if not args.dry_run:
|
||||
missing = [
|
||||
name
|
||||
for name, value in (("GALO_EMAIL", email), ("GALO_PASSWORD", password))
|
||||
if not value
|
||||
]
|
||||
if missing:
|
||||
print(
|
||||
"Не заданы переменные окружения: {}".format(", ".join(missing)),
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 2
|
||||
|
||||
try:
|
||||
return run(args, base_url, email or "", password or "")
|
||||
except ApiError as exc:
|
||||
if exc.status == 401:
|
||||
message = "Ошибка аутентификации: проверьте GALO_EMAIL и GALO_PASSWORD"
|
||||
else:
|
||||
message = "Ошибка API: {}".format(exc)
|
||||
print(message, file=sys.stderr)
|
||||
return 1
|
||||
except KeyboardInterrupt:
|
||||
print("Прервано пользователем", file=sys.stderr)
|
||||
return 130
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user