bot.py

#!/usr/bin/env python3
"""
Bot
"""

import urllib.request
import random
import json
import traceback
from pathlib import Path
from time import sleep

WEBHOOK_URL = json.loads(Path("conf.json").read_text())["webhook_url"]
PING_ERRORS = "<@818564860484780083>"
PING_NEW_VERSION = "<@&1342761186353090590>"
CHECK_IT = False

class Response:
	def __init__(self, url, status, body, headers):
		self.url = url
		self.status = status
		self.body = body
		self.headers = headers

class Date():
	def __init__(self, date):
		self.string = date
		self.date_set = set()
		
		for sub in date.split():
			self.date_set.add(sub.strip(" \t\r\n\v\f,.<>;'/:?{}[]-=_+\\|()*&^%$#@!`~"))
	
	def __eq__(self, other):
		if (len(self.date_set) == len(other.date_set)):
			for v in self.date_set:
				if v not in other.date_set:
					return False
			
			return True
		else:
			return False
	
	def __repr__(self):
		return self.string

def http_request(url, method='GET', data=None, headers={}):
	resp = urllib.request.urlopen(urllib.request.Request(url, data, method=method, headers={
		"User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:135.0) Gecko/20100101 Firefox/135.0"
	} | headers))
	data = resp.read()
	r = Response(resp.url, resp.status, data, resp.headers)
	resp.close()
	
	return r

def post(url, data):
	data = json.dumps(data).encode("utf-8")
	return http_request(url, 'POST', data, {"Content-Type": "application/json"})

def send_message(content):
	print("[Message]\n" + content + "\n[/Message]")
	
	try:
		post(WEBHOOK_URL, {
			"content": content,
		})
	except:
		pass

def get_app_updated_date_play(appid):
	if CHECK_IT: return random.choice(["17 Feb 2025", "Feb 17, 2025", "Mar 6, 2025", None])
	
	try:
		r = http_request(f"https://play.google.com/store/apps/details?id={appid}")
		r = r.body.decode("utf-8")
		i = r.index("Updated on")
		
		if (i < 0):
			print("Failed to parse response: String 'Updated on' not found")
			return None
		
		d = r[i:i+100]
		d = d.partition('>')[2].partition('>')[2].partition('<')[0]
		
		return d
	except:
		traceback.print_exc()
		return None

def get_app_version_ios(appurl):
	if CHECK_IT: return random.choice(["1.5.5", "1.0.4", "1.0.3", None])
	
	try:
		r = http_request(appurl)
		r = r.body.decode("utf-8")
		i = r.index('whats-new__latest__version">Version ')
		
		if (i < 0):
			print("Failed to parse App Store response: Could not find version string start")
			return None
		
		d = r[i:i+100]
		d = d.partition('>')[2].partition('<')[0][8:]
		
		return d
	except:
		traceback.print_exc()
		return None

class UpdateTracker:
	def __init__(self, display_name, identifier):
		self.display_name = display_name
		self.identifier = identifier
		self.last = None
		self.consec_errors = 0
		self.check()

class PlayTracker(UpdateTracker):
	def check(self):
		date = get_app_updated_date_play(self.identifier)
		
		if not date:
			self.consec_errors += 1
			
			if self.consec_errors == 4:
				return f"{PING_ERRORS} Failed to fetch google play update date for {self.display_name} four times in a row!"
			else:
				return None
		else:
			self.consec_errors = 0
		
		date = Date(date)
		
		if not self.last:
			print(f"Setting {self.display_name} (Google Play) date for first time")
			self.last = date
		else:
			prev = self.last
			self.last = date
			
			if prev != date:
				print(f"{prev} != {date}")
				return f"{PING_NEW_VERSION}\n## {self.display_name} updated on Google Play!\nOld date: {prev}\nNew date: {date}"
			else:
				print(f"{prev} == {date}")
	
	def table(self):
		return f"{self.display_name} (Android): {self.last}"

class AppleTracker(UpdateTracker):
	def check(self):
		ver = get_app_version_ios(self.identifier)
		
		if not ver:
			self.consec_errors += 1
			
			if self.consec_errors == 4:
				return f"{PING_ERRORS} Failed to fetch apple app store version for {self.display_name} four times in a row!"
			else:
				return None
		else:
			self.consec_errors = 0
		
		if not self.last:
			print(f"Setting {self.display_name} (App Store) version for first time")
			self.last = ver
		else:
			prev = self.last
			self.last = ver
			
			if prev != ver:
				print(f"{prev} != {ver}")
				return f"{PING_NEW_VERSION}\n## {self.display_name} updated on App Store!\nOld version: {prev}\nNew version: {ver}"
			else:
				print(f"{prev} == {ver}")
	
	def table(self):
		return f"{self.display_name} (iOS): {self.last}"

def rand_wait(base, variance):
	sleep(base + (2 * variance * (random.random() - 0.5)))

LAST_DATE = None
LAST_IOS_VER = None
CONSEC_ERRORS = 0

def generate_report():
	global LAST_DATE
	global LAST_IOS_VER
	global CONSEC_ERRORS
	
	date = get_app_updated_date_play("com.mediocre.smashhit")
	ios_ver = get_app_version_ios("https://apps.apple.com/us/app/smash-hit/id603527166")
	
	if not date or not ios_ver:
		CONSEC_ERRORS += 1
		
		if CONSEC_ERRORS == 4:
			send_message(f"{PING_ERRORS} Failed to fetch data four times in a row! Check that the bot is still working.")
		
		return
	else:
		CONSEC_ERRORS = 0
	
	date = Date(date)
	
	if not LAST_DATE:
		print("Setting date for first time")
		LAST_DATE = date
		LAST_IOS_VER = ios_ver
		send_message(f"## Starting up!\nRecorded current Play Store update date as: {date}\nRecorded current app store version as: {ios_ver}")
	else:
		if LAST_DATE != date:
			print(f"{repr(LAST_DATE)} != {repr(date)}")
			send_message(f"{PING_NEW_VERSION}\n## Smash Hit updated on Google Play!\nOld update date: {LAST_DATE}\nNew update date: {date}")
			
			LAST_DATE = date
		else:
			print(f"{repr(LAST_DATE)} == {repr(date)}")
		
		if LAST_IOS_VER != ios_ver:
			send_message(f"{PING_NEW_VERSION}\n## Smash Hit updated on App Store!\nOld version: {LAST_IOS_VER}\nNew version: {ios_ver}")
			LAST_IOS_VER = ios_ver

def main():
	while True:
		generate_report()
		rand_wait(60 * 30, 60 * 4)

def do_checking():
	trackers = [
		PlayTracker("Smash Hit", "com.mediocre.smashhit"),
		AppleTracker("Smash Hit", "https://apps.apple.com/us/app/smash-hit/id603527166"),
		AppleTracker("Smash Hit+", "https://apps.apple.com/us/app/smash-hit/id6503247519"),
	]
	
	# Startup message
	msg = "## Starting up!\nCurrent versions recorded as:\n"
	
	for t in trackers:
		msg += t.table() + "\n"
	
	send_message(msg.rstrip())
	
	while True:
		if CHECK_IT:
			sleep(10)
		else:
			rand_wait(60 * 30, 60 * 5)
		
		for t in trackers:
			msg = t.check()
			
			if msg:
				send_message(msg)

def main_new():
	try:
		do_checking()
	except KeyboardInterrupt:
		print("Bye!")

if __name__ == "__main__":
	main_new()