repl.it
@Scoder12/

forker async

Python

No description

fork
loading
Files
  • main.py
  • replit_client.py
  • repls.json
  • Packager files
  • poetry.lock
  • pyproject.toml
  • requirements.txt
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
import replit; replit.clear()
import asyncio
import aiohttp
from replit_client import Client
from getpass import getpass
import os
import json

username = "Scoder12"
crosis_key = os.environ["crosis"]
BATCHES = 4
BATCH_SIZE = 10
tasks = []
created = []


async def check_crosis_request(r):
    try:
        if r.status != 200:
            raise ValueError("Bad status")
        j = await r.json()
        if "error" in j:
            raise ValueError(j["error"])
        assert j["success"]
    except:
        print("Request text:", await r.text(), sep="\n")
        raise


async def crosis_run(fork_data):
    fuser = username
    fslug = fork_data["slug"]
    fid = fork_data["id"]
    
    async with aiohttp.ClientSession() as s:
        print(f"[{fork_data.get('slug')}] Installing packages...")
        async with s.post(f"https://crosis-http-api.scoder12.repl.co/api/v1/{fid}/packager/install", 
            cookies={
                'crosis': crosis_key
            }) as r:
            await check_crosis_request(r)

        print(f"[{fork_data.get('slug')}] Running...")
        async with s.post(
            f"https://crosis-http-api.scoder12.repl.co/api/v1/{fid}/interp2/run",
            cookies={
                'crosis': crosis_key
            }) as r:
            await check_crosis_request(r)

    co_url = f"https://{fslug.lower()}.{fuser.lower()}.repl.co"
    print(co_url)
    created.append({
        'id': fid,
        'url': co_url
    })


async def fork_and_run(replit, user, repl, rid=None):
    if not rid:
        repl_data = await replit.get_repl_data(
            user, 
            repl
        )
        rid = repl_data['id']
    fork_data = await replit.fork(rid)
    fslug = fork_data['slug']
    print(f"Created @{username}/{fslug}")
    tasks.append(asyncio.create_task(crosis_run(fork_data)))


def find_filename():
    num = 0
    fname = "repls.json"
    while os.path.exists(fname):
        num += 1
        fname = f"repls-{num}.json"
    return fname


async def main():
    global tasks
    async with Client() as replit:
        await replit.login(
            username, 
            getpass(f"Enter password for account {username!r}: ")
        )

        for _ in range(BATCHES):
            print("\nNew Batch\n")
            tasks = []
            for _ in range(BATCH_SIZE):
                await fork_and_run(replit, "@Scoder12", "libtasks-client")
            print("Finished a batch")
            for task in tasks:
                await task
        
        print("\nCreated repls:")
        #print("\n".join(created))
        with open(find_filename(), "w+") as f:
            f.write(json.dumps(created, indent=4))


if __name__ == "__main__":
    asyncio.get_event_loop().run_until_complete(main())
Fetching token
?