thepraandClaude Opus 5.5
02a05d7a9a
Iceshrimp.NET 2026.1.2-beta runs in the pasture with AuthorizedFetch on; its driver registers through the native API and takes Mastodon tokens from its OAuth form. The pair passes 225 cells. A new Iceshrimp note reaches its author's followers only, never an account it mentions or answers, until it is edited (G-0004, peer): delivery cells now say when a server holds a post only because it was addressed. The checker credits a seeder's fetch to the object fetched, not to the reply that needed it; the seeder skips the wait for an accept whose follow was already accepted; the PrivaPub driver signs its roots in again when a run reuses saved sessions. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LsXgEaXee4GCU1hwYgPJXw
539 lines
25 KiB
Python
539 lines
25 KiB
Python
"""The plan: a fake community drawn from a spec and its seed, the same on every run (tests/test_gen.py pins its hash).
|
|
|
|
A plan is a list of steps, each {n, phase, round, actor, verb, args, ref}. Accounts are named "<platform>/<username>",
|
|
objects by a ref ("p12"); the seeder turns refs into URIs as it goes. Phases run in order:
|
|
accounts every account, its profile, PrivaPub's roots and personas
|
|
groups PrivaPub circles and communities, Lemmy communities
|
|
graph follows (requests to locked accounts accepted, left pending or rejected), circle and community joins
|
|
content top-level posts, then reply rounds, each waiting for the round before to arrive
|
|
interact likes, boosts, reactions, votes, bookmarks
|
|
mutate edits, deletes, blocks, mutes, reports
|
|
The planner only asks a platform for what its driver can do (`caps`) and only lets an account touch what it could see.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
from core.rng import Rng
|
|
|
|
GENERATOR_VERSION = 1
|
|
HOSTS = {"privapub": "privapub.test", "gts": "gts.test", "mastodon": "mastodon.test", "misskey": "misskey.test",
|
|
"sharkey": "sharkey.test", "akkoma": "akkoma.test", "lemmy": "lemmy.test", "hollo": "hollo.test",
|
|
"iceshrimp": "iceshrimp.test"}
|
|
SHORT = {"privapub": "pp", "gts": "gt", "mastodon": "ms", "misskey": "mk", "sharkey": "sk", "akkoma": "ak", "lemmy": "lm",
|
|
"hollo": "ho", "iceshrimp": "is"}
|
|
MICRO = ("privapub", "gts", "mastodon", "misskey", "sharkey", "akkoma", "hollo", "iceshrimp")
|
|
CAPS = {
|
|
"privapub": {"post", "reply", "cw", "media", "poll", "like", "boost", "bookmark", "follow", "block", "mute", "report",
|
|
"dm", "delete", "edit", "vote", "react", "quote", "profile", "circle", "community", "located"},
|
|
"gts": {"post", "reply", "cw", "media", "poll", "like", "boost", "bookmark", "follow", "block", "mute", "report",
|
|
"dm", "delete", "edit", "vote", "profile"},
|
|
"mastodon": {"post", "reply", "cw", "media", "poll", "like", "boost", "bookmark", "follow", "block", "mute", "report",
|
|
"dm", "delete", "edit", "vote", "quote", "profile"},
|
|
"misskey": {"post", "reply", "cw", "media", "poll", "like", "react", "boost", "bookmark", "follow", "block", "mute",
|
|
"report", "dm", "delete", "vote", "quote", "profile"},
|
|
"sharkey": {"post", "reply", "cw", "media", "poll", "like", "react", "boost", "bookmark", "follow", "block", "mute",
|
|
"report", "dm", "delete", "vote", "quote", "profile", "edit"},
|
|
"akkoma": {"post", "reply", "cw", "media", "poll", "like", "boost", "bookmark", "follow", "block", "mute", "report",
|
|
"dm", "delete", "edit", "vote", "react", "quote", "profile"},
|
|
"lemmy": {"thread", "comment", "upvote", "downvote", "dm", "delete", "community", "edit", "block"},
|
|
"hollo": {"post", "reply", "cw", "media", "poll", "like", "boost", "bookmark", "follow", "block", "mute", "report",
|
|
"dm", "delete", "edit", "vote", "quote", "react", "profile"},
|
|
"iceshrimp": {"post", "reply", "cw", "media", "poll", "like", "boost", "bookmark", "follow", "block", "mute", "report",
|
|
"dm", "delete", "edit", "vote", "quote", "react", "profile"},
|
|
}
|
|
|
|
FIRST = ["ada", "bo", "cleo", "dario", "elif", "femi", "gaia", "hiro", "ines", "jun", "kai", "lia", "milo", "nadia",
|
|
"omar", "pia", "quinn", "rosa", "sami", "tea", "umbe", "vera", "wren", "xavi", "yuki", "zoe", "aldo", "bea",
|
|
"cosimo", "dora", "enzo", "flor", "gigi", "hana", "ivo", "jade", "kira", "luca", "mara", "nico"]
|
|
WORDS = {
|
|
"en": "the garden river morning bread market neighbour bicycle library rain coffee window festival map bridge "
|
|
"lantern harvest tide mountain postcard quiet bright small warm early late again together tomorrow".split(),
|
|
"it": "il giardino fiume mattina pane mercato vicino bicicletta biblioteca pioggia caffè finestra festa mappa ponte "
|
|
"lanterna raccolto marea montagna cartolina tranquillo luminoso piccolo caldo presto tardi insieme domani".split(),
|
|
"ja": "庭 川 朝 パン 市場 隣人 自転車 図書館 雨 コーヒー 窓 祭り 地図 橋 提灯 収穫 山 葉書 静か 明るい".split(),
|
|
"ar": "حديقة نهر صباح خبز سوق جار دراجة مكتبة مطر قهوة نافذة مهرجان خريطة جسر فانوس حصاد جبل هادئ".split(),
|
|
}
|
|
EMOJI = ["🌱", "🚲", "☕", "🌧", "🎉", "📚", "🌉", "🍞", "🗺", "🏔"]
|
|
TAGS = ["town", "pasture", "garden", "market", "rain", "bikes", "books", "festival"]
|
|
REACTIONS = ["🎉", "👍", "❤", "🔥", "😂"]
|
|
PLACES = [(45.4642, 9.19, "Milano"), (45.07, 7.6869, "Torino"), (41.9028, 12.4964, "Roma"), (48.8566, 2.3522, "Paris"),
|
|
(52.52, 13.405, "Berlin"), (35.6762, 139.6503, "Tokyo")]
|
|
|
|
|
|
def load(path):
|
|
if not os.path.isabs(path) and not os.path.exists(path):
|
|
path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "specs", path if path.endswith(".json") else f"{path}.json")
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
|
|
|
|
class Planner:
|
|
def __init__(self, spec):
|
|
self.spec = spec
|
|
self.rng = Rng(spec["seed"], f"{spec['name']}/v{GENERATOR_VERSION}")
|
|
self.steps = []
|
|
self.accounts = {} # key -> account dict
|
|
self.follows = {} # follower key -> set of followee keys (accepted)
|
|
self.objects = {} # ref -> object dict
|
|
self.groups = {} # ref -> group dict
|
|
self.n = 0
|
|
|
|
def step(self, phase, actor, verb, args, ref=None, round_=0):
|
|
self.n += 1
|
|
s = {"n": self.n, "phase": phase, "round": round_, "actor": actor, "verb": verb, "args": args}
|
|
if ref:
|
|
s["ref"] = ref
|
|
self.steps.append(s)
|
|
return s
|
|
|
|
@staticmethod
|
|
def acct(key):
|
|
platform, user = key.split("/", 1)
|
|
return f"{user}@{HOSTS[platform]}"
|
|
|
|
def text(self, rng, lang, n_words, ref):
|
|
words = WORDS.get(lang, WORDS["en"])
|
|
body = " ".join(rng.choice(words) for _ in range(n_words))
|
|
if rng.chance(0.3):
|
|
body += " " + rng.choice(EMOJI)
|
|
return f"{body} ({ref})"
|
|
|
|
# -- accounts
|
|
def plan_accounts(self):
|
|
rng = self.rng.sub("accounts")
|
|
prof = self.spec.get("profiles", {})
|
|
langs = prof.get("langs", {"en": 70, "it": 15, "ja": 8, "ar": 7})
|
|
used = set()
|
|
for platform, conf in self.spec["peers"].items():
|
|
if platform == "privapub":
|
|
for r in range(conf.get("roots", 1)):
|
|
# root logins and passwords are canaries: neither may ever appear on another server or in a log line
|
|
canary = hashlib.sha256(f"{self.spec['seed']}/root{r}".encode()).hexdigest()
|
|
root = f"rootcanary{r + 1:02d}{canary[:8]}"
|
|
password = f"Canary-{canary[8:28]}-9!"
|
|
for _ in range(rng.span(conf.get("personas", [2, 3]))):
|
|
self._account(rng, platform, used, prof, langs, root=root, password=password)
|
|
else:
|
|
for _ in range(conf.get("accounts", 3)):
|
|
self._account(rng, platform, used, prof, langs)
|
|
|
|
def _account(self, rng, platform, used, prof, langs, root=None, password=None):
|
|
while True:
|
|
name = f"{rng.choice(FIRST)}_{SHORT[platform]}{rng.between(1, 99):02d}"
|
|
if name not in used:
|
|
used.add(name)
|
|
break
|
|
key = f"{platform}/{name}"
|
|
lang = rng.weighted(langs)
|
|
locked = platform != "lemmy" and rng.chance(prof.get("locked", 0.2))
|
|
account = {
|
|
"platform": platform, "username": name,
|
|
"password": password or f"Town-{hashlib.sha256(name.encode()).hexdigest()[:12]}-7!",
|
|
"name": name.split("_")[0].capitalize() + f" of {platform}", "lang": lang,
|
|
"bio": self.text(rng, lang, rng.span(prof.get("bioWords", [5, 20])), f"bio-{name}"),
|
|
"fields": [[f"field{i}", rng.choice(WORDS["en"])] for i in range(rng.span(prof.get("fields", [0, 3])))],
|
|
"locked": locked, "bot": rng.chance(prof.get("bot", 0.05)),
|
|
"avatar_seed": f"avatar-{name}" if rng.chance(prof.get("avatar", 0.9)) else None,
|
|
"header_seed": f"header-{name}" if rng.chance(prof.get("header", 0.5)) else None,
|
|
"root": root,
|
|
}
|
|
self.accounts[key] = account
|
|
self.step("accounts", key, "account", account)
|
|
|
|
def keys(self, platform=None, micro=False):
|
|
return [k for k, a in self.accounts.items()
|
|
if (platform is None or a["platform"] == platform) and (not micro or a["platform"] in MICRO)]
|
|
|
|
# -- groups
|
|
def plan_groups(self):
|
|
rng = self.rng.sub("groups")
|
|
conf = self.spec["peers"]
|
|
personas = self.keys("privapub")
|
|
for i in range(conf.get("privapub", {}).get("circles", 0)):
|
|
owner = rng.choice(personas)
|
|
ref = f"circle{i + 1}"
|
|
self.groups[ref] = {"kind": "circle", "owner": owner, "members": {owner}, "name": f"{ref}_{rng.between(100, 999)}"}
|
|
self.step("groups", owner, "group", {"kind": "circle", "name": self.groups[ref]["name"]}, ref)
|
|
for i in range(conf.get("privapub", {}).get("communities", 0)):
|
|
owner = rng.choice(personas)
|
|
ref = f"ppcomm{i + 1}"
|
|
self.groups[ref] = {"kind": "community", "host": "privapub", "owner": owner, "members": set(),
|
|
"name": f"square{i + 1}_{rng.between(100, 999)}"}
|
|
self.step("groups", owner, "group", {"kind": "community", "name": self.groups[ref]["name"]}, ref)
|
|
for i in range(conf.get("lemmy", {}).get("communities", 0)):
|
|
lemmings = self.keys("lemmy")
|
|
if not lemmings:
|
|
break
|
|
owner = rng.choice(lemmings)
|
|
ref = f"lmcomm{i + 1}"
|
|
self.groups[ref] = {"kind": "community", "host": "lemmy", "owner": owner, "members": {owner},
|
|
"name": f"plaza{i + 1}_{rng.between(100, 999)}"}
|
|
self.step("groups", owner, "group", {"kind": "community", "name": self.groups[ref]["name"]}, ref)
|
|
|
|
# -- the graph
|
|
def plan_graph(self):
|
|
rng = self.rng.sub("graph")
|
|
g = self.spec.get("graph", {})
|
|
micro = self.keys(micro=True)
|
|
popularity = {k: 1 + rng.below(5) for k in micro}
|
|
for follower in micro:
|
|
n = rng.span(g.get("follows", [3, 8]))
|
|
candidates = [k for k in micro if k != follower]
|
|
weights = {k: popularity[k] * (3 if k.split("/")[0] == "privapub" or follower.startswith("privapub/") else 1)
|
|
for k in candidates}
|
|
chosen = set()
|
|
while len(chosen) < min(n, len(candidates)):
|
|
chosen.add(rng.weighted({k: w for k, w in weights.items() if k not in chosen}))
|
|
for target in sorted(chosen):
|
|
self._follow(rng, g, follower, target)
|
|
# mutuals: some of those followed follow back
|
|
for target in sorted(chosen):
|
|
if rng.chance(g.get("mutual", 0.4)) and follower not in self.follows.get(target, set()):
|
|
self._follow(rng, g, target, follower)
|
|
# circles: members are local siblings' neighbours and remote followers of the owner
|
|
for ref, grp in sorted(self.groups.items()):
|
|
if grp["kind"] == "circle":
|
|
pool = sorted(k for k in micro if k != grp["owner"] and k.split("/")[0] in ("privapub", "gts", "mastodon", "akkoma", "hollo", "iceshrimp"))
|
|
for member in rng.sample(pool, min(len(pool), self.spec.get("circleMembers", 4))):
|
|
grp["members"].add(member)
|
|
self.step("graph", member, "join", {"group": ref})
|
|
self.step("graph", grp["owner"], "approve", {"group": ref, "member": member})
|
|
elif grp["host"] == "privapub":
|
|
pool = sorted(k for k in self.accounts if k != grp["owner"])
|
|
for member in rng.sample(pool, min(len(pool), self.spec.get("communityMembers", 6))):
|
|
grp["members"].add(member)
|
|
self.step("graph", member, "join", {"group": ref})
|
|
else:
|
|
pool = sorted(k for k in self.accounts if k != grp["owner"] and self.accounts[k]["platform"] in ("privapub", "lemmy", "mastodon", "gts"))
|
|
for member in rng.sample(pool, min(len(pool), self.spec.get("communityMembers", 6))):
|
|
grp["members"].add(member)
|
|
self.step("graph", member, "join", {"group": ref})
|
|
|
|
def _follow(self, rng, g, follower, target):
|
|
self.step("graph", follower, "follow", {"target": target})
|
|
if self.accounts[target]["locked"]:
|
|
fate = rng.weighted({"accept": 1 - g.get("pending", 0.1) - g.get("rejected", 0.05),
|
|
"pending": g.get("pending", 0.1), "reject": g.get("rejected", 0.05)})
|
|
if fate == "accept":
|
|
self.step("graph", target, "accept", {"target": follower})
|
|
elif fate == "reject":
|
|
self.step("graph", target, "reject", {"target": follower})
|
|
return
|
|
else:
|
|
return
|
|
self.follows.setdefault(follower, set()).add(target)
|
|
|
|
def followers(self, key):
|
|
return {f for f, targets in self.follows.items() if key in targets}
|
|
|
|
def can_see(self, viewer, obj):
|
|
if viewer == obj["author"]:
|
|
return True
|
|
v = obj["visibility"]
|
|
if v in ("public", "unlisted"):
|
|
return True
|
|
if v == "followers":
|
|
return viewer in obj["followers"] or viewer in obj["mentions"]
|
|
if v == "direct":
|
|
return viewer in obj["mentions"]
|
|
if v == "circle":
|
|
return viewer in self.groups[obj["group"]]["members"]
|
|
if v == "community":
|
|
return True
|
|
return False
|
|
|
|
# -- content
|
|
def plan_content(self):
|
|
rng = self.rng.sub("content")
|
|
c = self.spec.get("content", {})
|
|
mix = c.get("mix", {"text": 40, "cw": 8, "tags": 10, "mention": 10, "image": 8, "poll": 6, "quote": 5})
|
|
vis = c.get("visibility", {"public": 55, "unlisted": 15, "followers": 15, "direct": 10, "circle": 5})
|
|
micro = self.keys(micro=True)
|
|
top = []
|
|
for i in range(c.get("posts", 60)):
|
|
author = rng.choice(micro)
|
|
platform = author.split("/")[0]
|
|
obj = self._post(rng, f"p{len(self.objects) + 1}", author, platform, mix, vis, c, top)
|
|
if obj:
|
|
top.append(obj)
|
|
# community posts: Lemmy threads, posts into the PrivaPub community
|
|
for ref, grp in sorted(self.groups.items()):
|
|
if grp["kind"] != "community":
|
|
continue
|
|
posters = sorted(m for m in grp["members"] | {grp["owner"]}
|
|
if (grp["host"] == "lemmy" and m.startswith("lemmy/")) or (grp["host"] == "privapub" and m.startswith(("privapub/", "lemmy/"))))
|
|
for _ in range(c.get("communityPosts", 4)):
|
|
if not posters:
|
|
break
|
|
author = rng.choice(posters)
|
|
oref = f"p{len(self.objects) + 1}"
|
|
obj = {"ref": oref, "author": author, "visibility": "community", "group": ref, "mentions": set(),
|
|
"followers": set(), "kind": "thread", "round": 0}
|
|
self.objects[oref] = obj
|
|
lang = self.accounts[author]["lang"]
|
|
self.step("content", author, "post", {"visibility": "community", "group": ref,
|
|
"title": f"{rng.choice(WORDS['en']).capitalize()} thread {oref}",
|
|
"text": self.text(rng, lang, rng.between(8, 30), oref)}, oref)
|
|
top.append(obj)
|
|
# reply rounds
|
|
rounds = c.get("replyRounds", 3)
|
|
previous = top
|
|
for rnd in range(1, rounds + 1):
|
|
current = []
|
|
for parent in previous:
|
|
if not rng.chance(c.get("replies", 0.45) if rnd == 1 else c.get("deeperReplies", 0.35)):
|
|
continue
|
|
repliers = [k for k in self.accounts if k != parent["author"] and self.can_see(k, parent)
|
|
and self._can_reply(k, parent)]
|
|
if not repliers:
|
|
continue
|
|
for _ in range(1 if rnd > 1 else rng.between(1, 2)):
|
|
replier = rng.choice(repliers)
|
|
obj = self._reply(rng, replier, parent, rnd)
|
|
if obj:
|
|
current.append(obj)
|
|
previous = current
|
|
|
|
def _can_reply(self, key, parent):
|
|
platform = key.split("/")[0]
|
|
if parent["visibility"] == "community":
|
|
grp = self.groups[parent["group"]]
|
|
return key in grp["members"] or key == grp["owner"]
|
|
if platform == "lemmy":
|
|
return False
|
|
if parent["visibility"] == "circle":
|
|
return platform == "privapub" or key in self.groups[parent["group"]]["members"]
|
|
if parent["visibility"] == "located":
|
|
return False
|
|
return "reply" in CAPS[platform]
|
|
|
|
def _post(self, rng, ref, author, platform, mix, vis, c, top):
|
|
caps = CAPS[platform]
|
|
vweights = {k: w for k, w in vis.items() if k in ("public", "unlisted", "followers", "direct")
|
|
or (k in ("circle", "located") and k in caps)}
|
|
visibility = rng.weighted(vweights)
|
|
circles = []
|
|
if visibility == "circle":
|
|
circles = [r for r, g in sorted(self.groups.items()) if g["kind"] == "circle" and g["owner"] == author]
|
|
if not circles:
|
|
visibility = "followers"
|
|
# circle and located posts go through /clientapi, which takes text, a warning and a title only
|
|
plain = visibility in ("circle", "located")
|
|
kind = rng.weighted({k: w for k, w in mix.items()
|
|
if (k in ("text", "tags", "mention") or k in caps or (k == "image" and "media" in caps))
|
|
and not (k == "quote" and not top)
|
|
and not (plain and k in ("image", "poll", "quote", "mention", "tags"))})
|
|
lang = self.accounts[author]["lang"]
|
|
args = {"visibility": visibility, "text": self.text(rng, lang, rng.between(4, 40), ref), "language": lang}
|
|
obj = {"ref": ref, "author": author, "visibility": visibility, "mentions": set(), "kind": kind, "round": 0,
|
|
"followers": set(self.followers(author)), "group": None}
|
|
if visibility == "circle":
|
|
obj["group"] = circles[0]
|
|
args["group"] = circles[0]
|
|
if visibility == "located":
|
|
lat, lng, place = rng.choice(PLACES)
|
|
args["location"] = {"lat": lat, "lng": lng, "range_km": rng.choice([2, 5, 20])}
|
|
args["text"] += f" near {place}"
|
|
if visibility == "direct" or kind == "mention":
|
|
others = [k for k in self.keys(micro=True) if k != author]
|
|
count = rng.between(1, 3) if visibility == "direct" else 1
|
|
obj["mentions"] = set(rng.sample(others, count))
|
|
args["mentions"] = sorted(obj["mentions"])
|
|
if kind == "cw":
|
|
args["cw"] = f"{rng.choice(WORDS['en'])} {rng.choice(['spoilers', 'food', 'weather', 'politics'])}"
|
|
obj["cw"] = args["cw"]
|
|
elif kind == "tags":
|
|
args["tags"] = rng.sample(TAGS, rng.between(1, 3))
|
|
elif kind == "image":
|
|
args["media"] = [{"kind": "image", "seed": f"{ref}-{i}", "alt": f"picture {i + 1} of {ref}",
|
|
"sensitive": rng.chance(0.15)} for i in range(rng.between(1, 3))]
|
|
elif kind == "poll":
|
|
opts = rng.sample(WORDS["en"], rng.between(2, 4))
|
|
args["poll"] = {"options": opts, "multiple": rng.chance(0.3), "expires_in": 86400}
|
|
obj["options"] = len(opts)
|
|
elif kind == "quote":
|
|
candidates = [o for o in top if o["visibility"] in ("public", "unlisted") and o["kind"] != "quote"]
|
|
if not candidates or "quote" not in caps:
|
|
kind = obj["kind"] = "text"
|
|
else:
|
|
target = rng.choice(candidates)
|
|
args["quote"] = target["ref"]
|
|
obj["quote"] = target["ref"]
|
|
self.objects[ref] = obj
|
|
self.step("content", author, "post", args, ref)
|
|
return obj
|
|
|
|
def _reply(self, rng, replier, parent, rnd):
|
|
ref = f"p{len(self.objects) + 1}"
|
|
platform = replier.split("/")[0]
|
|
if parent["visibility"] == "community" and platform == "lemmy":
|
|
kind = "comment"
|
|
elif platform == "lemmy":
|
|
return None
|
|
else:
|
|
kind = "reply"
|
|
visibility = parent["visibility"] if parent["visibility"] in ("direct", "followers", "circle", "community") else \
|
|
rng.weighted({"public": 70, "unlisted": 30})
|
|
if visibility == "circle" and platform != "privapub":
|
|
visibility = "direct" # a remote member answers a circle post as the post's own audience
|
|
mentions = set()
|
|
if visibility == "direct":
|
|
mentions = (parent["mentions"] | {parent["author"]}) - {replier}
|
|
elif visibility in ("public", "unlisted", "followers") and parent["author"].split("/")[0] in MICRO \
|
|
and parent["author"] != replier:
|
|
mentions = {parent["author"]} # clients mention the account replied to, and servers deliver by mention
|
|
lang = self.accounts[replier]["lang"]
|
|
args = {"visibility": visibility, "text": self.text(rng, lang, rng.between(3, 25), ref), "reply_to": parent["ref"],
|
|
"language": lang}
|
|
if mentions:
|
|
args["mentions"] = sorted(mentions)
|
|
if visibility == "circle":
|
|
args["group"] = parent["group"]
|
|
if visibility == "community":
|
|
grp = self.groups[parent["group"]]
|
|
if platform == "privapub" and grp["host"] == "privapub":
|
|
args["group"] = parent["group"]
|
|
elif platform != "lemmy":
|
|
args["visibility"] = "public" # a microblog reply to a community thread is a public reply the group relays
|
|
obj = {"ref": ref, "author": replier, "visibility": visibility, "mentions": mentions, "kind": kind,
|
|
"round": rnd, "parent": parent["ref"], "followers": set(self.followers(replier)), "group": parent.get("group")}
|
|
self.objects[ref] = obj
|
|
self.step("content", replier, "post", args, ref, rnd)
|
|
return obj
|
|
|
|
# -- interactions
|
|
def plan_interactions(self):
|
|
rng = self.rng.sub("interact")
|
|
conf = self.spec.get("interactions", {})
|
|
for ref, obj in sorted(self.objects.items(), key=lambda kv: int(kv[0][1:])):
|
|
if obj["visibility"] == "located":
|
|
continue
|
|
viewers = [k for k in self.accounts if k != obj["author"] and self.can_see(k, obj)
|
|
and self._can_interact(k, obj)]
|
|
if not viewers:
|
|
continue
|
|
for k in rng.sample(viewers, rng.span(conf.get("likes", [0, 4]))):
|
|
platform = k.split("/")[0]
|
|
if platform == "lemmy":
|
|
self.step("interact", k, "like" if rng.chance(0.75) else "downvote", {"target": ref})
|
|
elif "react" in CAPS[platform] and rng.chance(conf.get("react", 0.3)):
|
|
self.step("interact", k, "react", {"target": ref, "emoji": rng.choice(REACTIONS)})
|
|
else:
|
|
self.step("interact", k, "like", {"target": ref})
|
|
if obj["visibility"] in ("public", "unlisted"):
|
|
for k in rng.sample([v for v in viewers if "boost" in CAPS[v.split("/")[0]]], rng.span(conf.get("boosts", [0, 2]))):
|
|
self.step("interact", k, "boost", {"target": ref})
|
|
if obj.get("options"):
|
|
for k in rng.sample([v for v in viewers if "vote" in CAPS[v.split("/")[0]]],
|
|
round(len(viewers) * conf.get("vote", 0.6))):
|
|
self.step("interact", k, "vote", {"target": ref, "choices": [rng.below(obj["options"])]})
|
|
if rng.chance(conf.get("bookmark", 0.1)):
|
|
k = rng.choice([v for v in viewers if "bookmark" in CAPS[v.split("/")[0]]] or [None])
|
|
if k:
|
|
self.step("interact", k, "bookmark", {"target": ref})
|
|
|
|
def _can_interact(self, key, obj):
|
|
platform = key.split("/")[0]
|
|
if obj["visibility"] == "community":
|
|
grp = self.groups[obj["group"]]
|
|
return key in grp["members"] or key == grp["owner"]
|
|
if platform == "lemmy":
|
|
return False
|
|
if obj["visibility"] == "circle":
|
|
return key in self.groups[obj["group"]]["members"]
|
|
return True
|
|
|
|
# -- mutations
|
|
def plan_mutations(self):
|
|
rng = self.rng.sub("mutate")
|
|
m = self.spec.get("mutations", {})
|
|
for ref, obj in sorted(self.objects.items(), key=lambda kv: int(kv[0][1:])):
|
|
platform = obj["author"].split("/")[0]
|
|
if "edit" in CAPS[platform] and rng.chance(m.get("edit", 0.08)) and obj["visibility"] not in ("circle", "community", "located"):
|
|
lang = self.accounts[obj["author"]]["lang"]
|
|
# an edit keeps the names the post had, as a client's edit box does; dropping them changes who it is for
|
|
self.step("mutate", obj["author"], "edit", {"target": ref, "text": self.text(rng, lang, 6, ref) + " (edited)",
|
|
"mentions": sorted(obj["mentions"]), "cw": obj.get("cw")})
|
|
obj["edited"] = True
|
|
for ref, obj in sorted(self.objects.items(), key=lambda kv: int(kv[0][1:])):
|
|
if obj["round"] > 0 and rng.chance(m.get("delete", 0.04)) and obj["visibility"] not in ("circle", "located"):
|
|
self.step("mutate", obj["author"], "delete", {"target": ref})
|
|
obj["deleted"] = True
|
|
micro = self.keys(micro=True)
|
|
for verb, count in (("block", m.get("blocks", 2)), ("mute", m.get("mutes", 2))):
|
|
for _ in range(count):
|
|
a, b = rng.sample(micro, 2)
|
|
self.step("mutate", a, verb, {"target": b})
|
|
for _ in range(m.get("reports", 1)):
|
|
reporter = rng.choice(micro)
|
|
theirs = [o for o in self.objects.values() if o["author"] != reporter and o["visibility"] == "public"
|
|
and o["author"].split("/")[0] in MICRO]
|
|
if theirs:
|
|
o = rng.choice(theirs)
|
|
self.step("mutate", reporter, "report", {"target": o["author"], "refs": [o["ref"]], "comment": "town report"})
|
|
|
|
def plan(self):
|
|
self.plan_accounts()
|
|
self.plan_groups()
|
|
self.plan_graph()
|
|
self.plan_content()
|
|
self.plan_interactions()
|
|
self.plan_mutations()
|
|
return self.steps
|
|
|
|
|
|
def plan(spec):
|
|
return Planner(spec).plan()
|
|
|
|
|
|
def snapshot(planner):
|
|
"""The planner's world as JSON (sets as sorted lists), stored with a run so later generator changes never re-read it."""
|
|
def plain(v):
|
|
if isinstance(v, set):
|
|
return sorted(v)
|
|
if isinstance(v, dict):
|
|
return {k: plain(x) for k, x in v.items()}
|
|
if isinstance(v, list):
|
|
return [plain(x) for x in v]
|
|
return v
|
|
return {"version": GENERATOR_VERSION, "steps": planner.steps, "accounts": planner.accounts,
|
|
"objects": plain(planner.objects), "groups": plain(planner.groups), "follows": plain(planner.follows)}
|
|
|
|
|
|
class Stored(Planner):
|
|
"""A Planner rebuilt from a run's plan.json instead of from the spec."""
|
|
|
|
def __init__(self, snap):
|
|
self.spec = None
|
|
self.steps = snap["steps"]
|
|
self.accounts = snap["accounts"]
|
|
self.objects = {r: {k: (set(v) if k in ("mentions", "followers") else v) for k, v in o.items()} for r, o in snap["objects"].items()}
|
|
self.groups = {r: {k: (set(v) if k == "members" else v) for k, v in g.items()} for r, g in snap["groups"].items()}
|
|
self.follows = {k: set(v) for k, v in snap["follows"].items()}
|
|
|
|
def plan(self):
|
|
return self.steps
|
|
|
|
|
|
def digest(steps):
|
|
return hashlib.sha256(json.dumps(steps, sort_keys=True, ensure_ascii=False).encode()).hexdigest()
|
|
|
|
|
|
def main(args):
|
|
spec = load(args[0] if args else "village")
|
|
steps = plan(spec)
|
|
if "--summary" in args:
|
|
from collections import Counter
|
|
print(json.dumps({"steps": len(steps), "sha": digest(steps),
|
|
"by": dict(Counter(f"{s['phase']}:{s['verb']}" for s in steps))}, indent=1))
|
|
return 0
|
|
for s in steps:
|
|
print(json.dumps(s, ensure_ascii=False))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:]))
|