add current oneshot scripts

Author: nepeat

oneshots/redis_hit_all.py

@@ -0,0 +1,17 @@
+import ujson
+from placedump.common import get_redis
+
+redis = get_redis()
+
+for x in range(0, 1000):
+    for y in range(0, 1000):
+        redis.sadd(
+            "queue:pixels",
+            ujson.dumps(
+                {
+                    "x": str(x),
+                    "y": str(y),
+                    "board": "0",
+                }
+            ),
+        )

oneshots/url_table_backfill.py

@@ -0,0 +1,69 @@
+import sys
+from concurrent.futures import ThreadPoolExecutor
+
+import ujson as json
+from placedump.common import get_redis
+from placedump.constants import socket_key
+from placedump.model import URL, sm
+from placedump.tasks.pixels import download_url
+from sqlalchemy import func, select
+
+db = sm()
+redis = get_redis()
+sql_url = set()
+to_backfill = set()
+last_id = "0"
+pool = ThreadPoolExecutor(max_workers=8)
+
+for url in db.execute(select(URL)).scalars():
+    sql_url.add(url.url)
+
+print(len(sql_url), "in database")
+
+while True:
+    results = redis.xread({socket_key: last_id}, count=10000)
+    if not results:
+        break
+    messages = results[0][1]
+    for message_id, message in messages:
+        last_id = message_id
+        decoded = json.loads(message["message"])
+
+        # get canvas id
+
+        # get url
+        try:
+            if decoded["type"] in ["ka", "connection_ack"]:
+                continue
+
+            if (
+                decoded["payload"]["data"]["subscribe"]["data"]["__typename"]
+                == "ConfigurationMessageData"
+            ):
+                continue
+        except KeyError:
+            continue
+
+        try:
+            url = decoded["subscribe"]["data"]["name"]
+            canvas_id = decoded.get("canvas_id", int(decoded.get("id", 2)) - 2)
+        except KeyError:
+            try:
+                url = decoded["payload"]["data"]["subscribe"]["data"]["name"]
+                canvas_id = int(decoded.get("id", 2)) - 2
+            except KeyError:
+                continue
+
+        if url not in sql_url:
+            to_backfill.add((canvas_id, url))
+
+print(len(to_backfill), "to backfill")
+
+
+for item in to_backfill:
+    download_url.apply_async(
+        args=(item[0], item[1]),
+        priority=10,
+    )
+    sys.stdout.write(".")
+    sys.stdout.flush()

oneshots/url_table_backfill_b2.py

@@ -0,0 +1,56 @@
+import datetime
+import sys
+from concurrent.futures import ThreadPoolExecutor
+
+import ujson as json
+from placedump.common import get_b2_api, get_redis
+from placedump.constants import socket_key
+from placedump.model import URL, Pixel, sm
+from placedump.tasks.pixels import download_url
+from sqlalchemy import func, select
+
+db = sm()
+b2 = get_b2_api()
+sql_url = set()
+bucket = b2.get_bucket_by_name("erin-reddit-afd2022")
+
+for url in db.execute(select(URL)).scalars():
+    sql_url.add(url.url)
+
+print(len(sql_url), "in database")
+
+urls_to_insert = []
+
+with sm() as db_insert:
+    for file_version, folder_name in bucket.ls(
+        folder_to_list="hot-potato.reddit.com/media/canvas-images", latest_only=True
+    ):
+        fixed_url = "https://" + file_version.file_name
+        if fixed_url in sql_url:
+            continue
+
+        timestamp = float(file_version.upload_timestamp) / 1000.0
+        timestamp = datetime.datetime.fromtimestamp(timestamp)
+
+        urls_to_insert.append(
+            {
+                "url": fixed_url,
+                "fetched": timestamp,
+            }
+        )
+
+        if len(urls_to_insert) > 1024:
+            db_insert.bulk_insert_mappings(
+                URL,
+                urls_to_insert,
+            )
+            db_insert.commit()
+            print("Added 1024 URLs.")
+            urls_to_insert.clear()
+
+    db_insert.bulk_insert_mappings(
+        URL,
+        urls_to_insert,
+    )
+    db_insert.commit()
+    print(f"Added {len(urls_to_insert)} URLs.")

scripts/dump-redis.py

@@ -0,0 +1,14 @@
+from placedump.common import get_redis
+from placedump.constants import socket_key
+
+redis = get_redis()
+last_id = "0"
+
+while True:
+    results = redis.xread({socket_key: last_id}, count=10000)
+    if not results:
+        break
+    messages = results[0][1]
+    for message_id, message in messages:
+        last_id = message_id
+        print(message["message"], end="")