118 lines
3.6 KiB
Python
118 lines
3.6 KiB
Python
import aiohttp
|
|
import asyncio
|
|
import requests
|
|
import sys
|
|
import datetime
|
|
|
|
|
|
class Post:
|
|
post_id = None
|
|
poster = None
|
|
response = None
|
|
likes = None
|
|
image = None
|
|
|
|
def __init__(self, post_id):
|
|
self.post_id = post_id
|
|
|
|
async def get_post(self, session):
|
|
async with session.get(f"https://sketchersunited.org/posts/{self.post_id}") as response:
|
|
self.response = await response.json()
|
|
self.poster = self.response['post']['profile_id']
|
|
self.likes = self.response['post']['reactions']
|
|
self.image = self.response['post']['resource_url']
|
|
return self
|
|
|
|
def __str__(self):
|
|
if self.image:
|
|
return (f"<hr>\n"
|
|
f"<div>\n"
|
|
f" <b>Post id:</b> {self.post_id}\n<br>"
|
|
f" <b>Profile id:</b> <a href = 'https://sketchersunited.org/users/{self.poster}'>{self.poster}</a>\n<br>"
|
|
f" <b>Likes: {self.likes}</b>\n<br>"
|
|
f" <img src = '{self.image}' style='width:300px;height:auto;'>\n<br>"
|
|
f" </div>\n")
|
|
else:
|
|
return (f"<hr>\n"
|
|
f"<div>\n"
|
|
f" <b>Post id:</b> {self.post_id}\n<br>"
|
|
f" <b>Profile id:</b> <a href = 'https://sketchersunited.org/users/{self.poster}'>{self.poster}</a>\n<br>"
|
|
f" <b>Likes: {self.likes}</b>\n<br>"
|
|
f" This is a text post<br>"
|
|
f" </div>\n")
|
|
|
|
|
|
async def get_next_posts(tag, post_id):
|
|
headers = {
|
|
'User-Agent': 'tagtop by @username',
|
|
'Accept': 'application/json, text/plain, */*',
|
|
}
|
|
|
|
if post_id == 0:
|
|
response = requests.get(f'https://sketchersunited.org/json/tags/{tag}', headers=headers)
|
|
else:
|
|
response = requests.get(f'https://sketchersunited.org/json/tags/{tag}?p={post_id}', headers=headers)
|
|
|
|
return response.json()
|
|
|
|
|
|
async def get_all_posts(tag):
|
|
all_posts = []
|
|
next_post_id = 0
|
|
|
|
while next_post_id is not None:
|
|
print(f"loading posts from post {next_post_id}")
|
|
response = await get_next_posts(tag, next_post_id)
|
|
|
|
for x in response['data']['posts']:
|
|
post = Post(x['id'])
|
|
all_posts.append(post)
|
|
next_post_id = response['next']
|
|
|
|
return all_posts
|
|
|
|
|
|
def write_to_file(all_posts):
|
|
file_name = f'leaderboard/leaderboard{datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}.html'
|
|
|
|
with open(file_name, "w") as file:
|
|
file.write("""<html>\n
|
|
<body>\n
|
|
""")
|
|
|
|
sorted_posts = sorted(all_posts, key=lambda post_to_check: post_to_check.likes, reverse=True)
|
|
|
|
for post in sorted_posts:
|
|
print(f"Set up {post.post_id}")
|
|
with open(file_name, "a") as file:
|
|
file.write(str(post))
|
|
|
|
with open(file_name, "a") as file:
|
|
file.write("""</body>\n
|
|
</html>
|
|
""")
|
|
|
|
|
|
async def download_likes():
|
|
all_posts = await get_all_posts(sys.argv[1].replace("#",""))
|
|
|
|
headers = {
|
|
'User-Agent': 'tagtop by @username',
|
|
'Accept': 'application/json, text/plain, */*',
|
|
}
|
|
|
|
async with aiohttp.ClientSession(headers=headers) as session:
|
|
tasks = []
|
|
for post in all_posts:
|
|
tasks.append(asyncio.ensure_future(post.get_post(session)))
|
|
|
|
likes = await asyncio.gather(*tasks)
|
|
write_to_file(likes)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("Specify the tag you want to search! e.g. python3 tagtop.py funtag")
|
|
else:
|
|
asyncio.run(download_likes())
|