Something went wrong. Try again.
Parametric learning in your code harness.
Something went wrong. Try again.
6.9 kB · 189 lines
Python
at letta-code
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190#!/usr/bin/env python3from __future__ import annotations
import argparseimport jsonimport reimport sysimport urllib.parseimport urllib.requestimport zipfilefrom html.parser import HTMLParserfrom pathlib import Path, PurePosixPath
EXAMPLE_DIR = Path(__file__).resolve().parentSRC_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()