import json import os import sys import time import urllib.request import urllib.error import subprocess from datetime import datetime OUTPUT_FILE = os.path.join(os.path.dirname(__file__), "solved_ac_problems.json") CRAWLED_CONTESTS_FILE = os.path.join(os.path.dirname(__file__), "crawled_contests.json") HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8", "Accept-Language": "en-US,en;q=0.5" } # Group 90 Contests (UCIPC and NAIPC) GROUP_90_CONTESTS = { "428": {"name": "UCIPC 2012", "path": "University > UCIPC"}, "854": {"name": "UCIPC 2013", "path": "University > UCIPC"}, "1282": {"name": "NAIPC 2014", "path": "University > NAIPC"}, "1387": {"name": "NAIPC 2015", "path": "University > NAIPC"}, "1502": {"name": "NAIPC 2016", "path": "University > NAIPC"}, "1733": {"name": "NAIPC 2017", "path": "University > NAIPC"}, "1906": {"name": "NAIPC 2018", "path": "University > NAIPC"}, "2003": {"name": "NAIPC 2019", "path": "University > NAIPC"}, } def load_all_target_contests(): with open(CRAWLED_CONTESTS_FILE, "r", encoding="utf-8") as f: contests = json.load(f) # Merge Group 90 contests for cid, meta in GROUP_90_CONTESTS.items(): if cid not in contests: contests[cid] = meta else: if "Group 90" not in contests[cid]["path"]: contests[cid]["path"] += f" | {meta['path']}" return contests def fetch_contest_html(contest_id): url = f"https://solved.ac/en/problems/archive/contest/{contest_id}" req = urllib.request.Request(url, headers=HEADERS) while True: try: with urllib.request.urlopen(req, timeout=15) as response: return response.read().decode("utf-8") except urllib.error.HTTPError as e: if e.code == 404: print(f"Contest {contest_id}: HTTP 404 Not Found.") return None # For 429, 500, 502, 503, 504, or any throttling/server error: 15-minute backoff now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"\n[HTTP {e.code} Error] Throttled or server error on Contest {contest_id} at {now_str}.") print("Backing off and sleeping for 15 minutes (900 seconds) before retrying...") time.sleep(15 * 60) print(f"Resuming retry of Contest {contest_id} at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}...") continue except Exception as e: now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"\n[Connection Error: {e}] on Contest {contest_id} at {now_str}.") print("Backing off and sleeping for 15 minutes (900 seconds) before retrying...") time.sleep(15 * 60) print(f"Resuming retry of Contest {contest_id} at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}...") continue def save_output(scraped_contests): # Sort contests numerically by contest_id contest_list = sorted(scraped_contests.values(), key=lambda c: int(c["contest_id"])) all_problems = [] for c in contest_list: all_problems.extend(c.get("problems", [])) payload = { "metadata": { "total_contests": len(contest_list), "total_problems": len(all_problems), "last_updated": datetime.now().isoformat() }, "contests": contest_list, "problems": all_problems } with open(OUTPUT_FILE, "w", encoding="utf-8") as f: json.dump(payload, f, indent=2, ensure_ascii=False) def main(): target_contests = load_all_target_contests() all_target_ids = sorted(target_contests.keys(), key=lambda x: int(x)) total_targets = len(all_target_ids) # Load existing scraped data to resume seamlessly scraped_contests = {} if os.path.exists(OUTPUT_FILE): try: with open(OUTPUT_FILE, "r", encoding="utf-8") as f: prev_data = json.load(f) if isinstance(prev_data, dict) and "contests" in prev_data: for c in prev_data["contests"]: scraped_contests[str(c["contest_id"])] = c print(f"Resuming scrape: found {len(scraped_contests)} / {total_targets} already scraped in {OUTPUT_FILE}.") except Exception as e: print(f"Error reading existing {OUTPUT_FILE} ({e}). Starting fresh.") remaining_ids = [cid for cid in all_target_ids if cid not in scraped_contests] print(f"Remaining contests to scrape: {len(remaining_ids)} of {total_targets}.") if not remaining_ids: print("All target contests are already scraped!") return # Start persistent Node parser worker node_proc = subprocess.Popen( ["node", "node_worker.js"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True, bufsize=1 ) last_save_time = time.time() try: for idx, cid in enumerate(remaining_ids, len(scraped_contests) + 1): meta = target_contests[cid] html = fetch_contest_html(cid) if not html: contest_obj = { "contest_id": int(cid), "contest_name": meta.get("name", ""), "contest_full_name": meta.get("name", ""), "contest_year": None, "contest_path": meta.get("path", ""), "problem_count": 0, "problems": [], "fetch_error": True } scraped_contests[cid] = contest_obj print(f"[{idx}/{total_targets}] Contest {cid} ({meta.get('name')}): Failed / 404") else: # Send HTML to Node worker worker_input = json.dumps({"id": int(cid), "html": html}) + "\n" node_proc.stdin.write(worker_input) node_proc.stdin.flush() resp_line = node_proc.stdout.readline() parsed = json.loads(resp_line) data = parsed.get("data", {}) if data.get("error"): contest_obj = { "contest_id": int(cid), "contest_name": meta.get("name", ""), "contest_full_name": meta.get("name", ""), "contest_year": None, "contest_path": meta.get("path", ""), "problem_count": 0, "problems": [], "parse_error": data["error"] } scraped_contests[cid] = contest_obj print(f"[{idx}/{total_targets}] Contest {cid} ({meta.get('name')}): Parse error ({data['error']})") else: contest_obj = { "contest_id": int(cid), "contest_name": data.get("contest_name") or meta.get("name", ""), "contest_full_name": data.get("contest_full_name", ""), "contest_year": data.get("contest_year"), "contest_path": meta.get("path", ""), "problem_count": data.get("problem_count", 0), "problems": data.get("problems", []) } scraped_contests[cid] = contest_obj print(f"[{idx}/{total_targets}] Contest {cid} ({contest_obj['contest_name']}): {contest_obj['problem_count']} problems") # Checkpoint save every 5 contests or every 15 seconds if idx % 5 == 0 or (time.time() - last_save_time > 15): save_output(scraped_contests) last_save_time = time.time() # Pacing sleep time.sleep(0.5) except KeyboardInterrupt: print("\nInterrupted by user. Saving progress...") finally: node_proc.terminate() save_output(scraped_contests) print(f"\nSaved current progress: {len(scraped_contests)} contests in {OUTPUT_FILE}") if __name__ == "__main__": main()