#!/usr/bin/env python3 from __future__ import annotations import argparse import json import re import sys import urllib.parse import urllib.request import zipfile from html.parser import HTMLParser from pathlib import Path, PurePosixPath EXAMPLE_DIR = Path(__file__).resolve().parent SRC_DIR = EXAMPLE_DIR / "src" sys.path.insert(0, str(SRC_DIR)) from euler_common import Problem, write_jsonl # noqa: E402 DATASET_URL = "https://www.kaggle.com/datasets/dheerajmpai/projecteuler" DATASET_LICENSE = "CC BY-NC-SA 4.0" PROJECT_EULER_BASE = "https://projecteuler.net/" class ProblemHTMLParser(HTMLParser): BLOCK_TAGS = {"p", "div", "br", "li", "tr", "table", "blockquote"} SKIP_TAGS = {"script", "style", "svg"} def __init__(self) -> None: super().__init__(convert_charrefs=True) self.parts: list[str] = [] self.skip_depth = 0 def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: if tag in self.SKIP_TAGS: self.skip_depth += 1 return if self.skip_depth: return if tag in self.BLOCK_TAGS: self.parts.append("\n") elif tag in {"td", "th"}: self.parts.append("\t") elif tag == "sup": self.parts.append("^(") elif tag == "img": alt = dict(attrs).get("alt") if alt: self.parts.append(f" [image: {alt}] ") def handle_endtag(self, tag: str) -> None: if tag in self.SKIP_TAGS and self.skip_depth: self.skip_depth -= 1 return if self.skip_depth: return if tag == "sup": self.parts.append(")") elif tag in self.BLOCK_TAGS: self.parts.append("\n") def handle_data(self, data: str) -> None: if not self.skip_depth: self.parts.append(data) def text(self) -> str: lines = [] for line in "".join(self.parts).splitlines(): compact = re.sub(r"[ \t]+", " ", line).strip() if compact: lines.append(compact) return "\n".join(lines) def parse_answers(text: str) -> dict[int, str]: answers: dict[int, str] = {} for line in text.splitlines(): match = re.match(r"^(\d+)\.\s+(.+?)\s*$", line) if match: answers[int(match.group(1))] = match.group(2) return answers def safe_resource_ref(value: str) -> str | None: clean = urllib.parse.urlsplit(value).path.lstrip("/") path = PurePosixPath(clean) if not clean.startswith("resources/") or ".." in path.parts: return None return str(path) def extract_refs(html: str, attribute: str) -> tuple[str, ...]: pattern = rf"{attribute}=[\"']([^\"']+)[\"']" refs = [] for value in re.findall(pattern, html, flags=re.IGNORECASE): clean = safe_resource_ref(value) if clean and clean not in refs: refs.append(clean) return tuple(refs) def fetch_resource(ref: str, destination: Path) -> bool: url = urllib.parse.urljoin(PROJECT_EULER_BASE, ref) request = urllib.request.Request(url, headers={"User-Agent": "machine-project-euler-example/1.0"}) try: with urllib.request.urlopen(request, timeout=30) as response: payload = response.read() except Exception as error: # noqa: BLE001 print(f"warning: could not fetch {url}: {error}", file=sys.stderr) return False destination.parent.mkdir(parents=True, exist_ok=True) destination.write_bytes(payload) return True def prepare(archive: Path, output_dir: Path, fetch_resources: bool) -> dict[str, object]: output_dir.mkdir(parents=True, exist_ok=True) resources_dir = output_dir / "resources" with zipfile.ZipFile(archive) as bundle: names = set(bundle.namelist()) if "Solutions.md.txt" not in names: raise ValueError("archive does not contain Solutions.md.txt") answers = parse_answers(bundle.read("Solutions.md.txt").decode("utf-8", errors="replace")) problems: list[Problem] = [] missing_resources: set[str] = set() for name in sorted(names): match = re.fullmatch(r"euler/(\d+)\.html", name) if not match: continue problem_id = int(match.group(1)) answer = answers.get(problem_id) if answer is None: continue html = bundle.read(name).decode("utf-8", errors="replace") parser = ProblemHTMLParser() parser.feed(html) prompt = parser.text() document_refs = tuple( ref for ref in extract_refs(html, "href") if Path(ref).suffix.lower() in {".txt", ".csv", ".zip", ".dat"} ) image_refs = extract_refs(html, "src") for ref in document_refs: destination = output_dir / ref if not destination.exists() and fetch_resources: fetch_resource(ref, destination) if not destination.exists(): missing_resources.add(ref) available = bool(prompt) and all((output_dir / ref).exists() for ref in document_refs) problems.append( Problem( problem_id=problem_id, prompt=prompt, answer=answer, available=available, resource_refs=document_refs, image_refs=image_refs, source_url=f"https://projecteuler.net/problem={problem_id}", ) ) write_jsonl(output_dir / "problems.jsonl", (problem.to_dict() for problem in problems)) manifest = { "archive": str(archive.resolve()), "dataset_url": DATASET_URL, "dataset_license": DATASET_LICENSE, "problem_count": len(problems), "available_count": sum(problem.available for problem in problems), "missing_resources": sorted(missing_resources), "warning": "Answer-bearing generated data. Do not expose this directory to evaluated agents.", } (output_dir / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8") return manifest def main() -> None: parser = argparse.ArgumentParser(description="Import the local Kaggle Project Euler archive.") parser.add_argument("--archive", type=Path, required=True, help="Path to euler-questions.zip") parser.add_argument("--output-dir", type=Path, default=EXAMPLE_DIR / "data") parser.add_argument("--fetch-resources", action="store_true", help="Fetch referenced Project Euler data files") args = parser.parse_args() manifest = prepare(args.archive.expanduser(), args.output_dir.expanduser(), args.fetch_resources) print(json.dumps(manifest, indent=2)) if __name__ == "__main__": main()