Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 76 additions & 0 deletions database_helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -184,6 +184,82 @@ def fetch_and_insert_atcoder_contests(self):
except Exception as e:
logging.error(f"Error fetching or inserting AtCoder contests: {e}")

def fetch_and_insert_hackerrank_contests(self):
try:
r = requests.get("https://www.hackerrank.com/community/engage/events", headers={"User-Agent": self.ua.random})
response = r.json()

# Check if the response is valid
if response["error"] == "false":
raise ValueError("Failed to fetch contests from Hackerrank")

all_contests = response["data"]["events"]
upcoming_contests = []
ongoing_contests = all_contests["ongoing_events"]

# Process contests
for contest in ongoing_contests:
contest_attributes = contest["attributes"]
if contest_attributes["status"] != "completed":

# Start time
iso_dt = contest_attributes["start_time"]
dt = datetime.strptime(iso_dt, "%Y-%m-%dT%H:%M:%S.%fZ")
start_date = dt.strftime('%Y-%m-%d')
start_time = dt.strftime('%H:%M:%S')

# End time
iso_end_dt = contest_attributes["end_time"]
end_dt = datetime.strptime(iso_end_dt, "%Y-%m-%dT%H:%M:%S.%fZ")
end_date = end_dt.strftime('%Y-%m-%d')
end_time = end_dt.strftime('%H:%M:%S')

# Duration
duration = end_dt - dt
hours, remainder = divmod(duration.total_seconds(), 3600)
minutes, seconds = divmod(remainder, 60)

# Registration end time
iso_registration_dt = contest_attributes["registration_end_time"]
registration_dt = datetime.strptime(iso_registration_dt, "%Y-%m-%dT%H:%M:%S.%fZ")
registration_date = registration_dt.strftime('%Y-%m-%d')
registration_time = registration_dt.strftime('%H:%M:%S')

my_contest = {
"contest_provider": "Hackerrank",
"contest_type": contest["type"],
"contest_id": contest["id"],
"contest_name": contest_attributes["name"],
"start_date": start_date,
"start_time": start_time,
"registration_date": registration_date,
"registration_time": registration_time,
"company": contest_attributes["company_name"],
"url": contest_attributes["microsite_url"],
"notified": False,
"duration": f"{int(hours)}h {int(minutes)}m {int(seconds)}s"}
upcoming_contests.append(my_contest)
else:
break

# Insert contests into the database if they don't already exist
new_contests = []
existing_contests = self.supabase.table("contests").select("url").eq("contest_provider", "Hackerrank").execute()
existing_urls = set(contest['url'] for contest in existing_contests.data)

for contest in upcoming_contests:
if contest["url"] not in existing_urls:
new_contests.append(contest)

if new_contests:
self.supabase.table("contests").insert(new_contests).execute()
logging.info(f"Inserted {len(new_contests)} new Hackerrank contests.")
else:
logging.info("No new hackerrank contests to insert.")

except Exception as e:
logging.error(f"Error fetching or inserting Hackerrank contests: {e}")

def fetch_unnotified_contests(self):
try:
logging.info("Fetching list of unnotified contests from the database.")
Expand Down