| """Tool interfaces: search, fetch, extract. |
| |
| - MockTools: scenario-driven, with failure injection (eval harness; no network). |
| - NetTools: optional real fetch of public pages via an network client; search is a fixed seed-URL stub |
| (no scraping of search engines; wire official source APIs later if needed). |
| - extract_claims: LLM-free regex extractor with plausibility guards (default). |
| - Optional LLM extraction is wired through interfaces.LLMExtractor + providers.json (SQ_LLM_* env). |
| """ |
|
|
| from __future__ import annotations |
| import json |
| import math |
| import os |
| import re |
|
|
| from bounds import area_lower_bound, trivial_upper_bound |
|
|
|
|
| class ToolError(Exception): |
| pass |
|
|
|
|
| class MockTools: |
| """tools spec (from scenario JSON): |
| {"search": [ {url, date, source_type, source_key?}, ... ] | "error", |
| "search_fail_times": int, # transient failures before success (-1 = always) |
| "pages": {url: {"text": str, "fail_times": int, "error": str}}}""" |
|
|
| def __init__(self, spec): |
| self.spec = spec |
| self.calls = {} |
|
|
| def _maybe_fail(self, key, fail_times, err): |
| k = self.calls.get(key, 0) |
| self.calls[key] = k + 1 |
| if fail_times == -1 or k < fail_times: |
| raise ToolError(f"{key}: {err}") |
|
|
| def search(self, query): |
| if self.spec.get("search") == "error": |
| raise ToolError("search: status 503") |
| self._maybe_fail("search", self.spec.get("search_fail_times", 0), "rate limited (429)") |
| return list(self.spec.get("search", [])) |
|
|
| def fetch(self, url): |
| page = self.spec.get("pages", {}).get(url) |
| if page is None: |
| raise ToolError(f"fetch {url}: 404") |
| self._maybe_fail(url, page.get("fail_times", 0), page.get("error", "timeout")) |
| if "text_file" in page: |
| with open(page["text_file"], encoding="utf-8") as f: |
| return f.read() |
| return page.get("text", "") |
|
|
|
|
| class NetTools: |
| """Real-network tools for manual runs. Search returns configured seed URLs only.""" |
|
|
| def __init__(self, seeds, timeout=20.0): |
| self.seeds = seeds |
| self.timeout = timeout |
|
|
| def search(self, query): |
| return list(self.seeds) |
|
|
| def fetch(self, url): |
| import httpx |
| try: |
| r = httpx.get(url, timeout=self.timeout, follow_redirects=True, |
| headers={"User-Agent": "squares-progress-agent/0.1"}) |
| r.raise_for_status() |
| except Exception as e: |
| raise ToolError(f"fetch {url}: {e}") from e |
| text = re.sub(r"<script.*?</script>|<style.*?</style>", " ", r.text, flags=re.S | re.I) |
| return re.sub(r"<[^>]+>", " ", text) |
|
|
|
|
| |
| NUM = r"([0-9]+(?:[.,][0-9]+)?)" |
| REL = r"\s*(?:=|≈|~|:|is|≤|<=)\s*" |
| PATTERNS = [ |
| re.compile(r"\bs\s*\(\s*(\d{1,4})\s*\)" + REL + NUM), |
| re.compile(r"\bn\s*=\s*(\d{1,4})\b[^\n]{0,60}?\b(?:s|L|side(?: length)?|container side)" + REL + NUM), |
| re.compile(r"\b(\d{1,4})\s+(?:unit\s+)?squares?\b[^\n]{0,60}?\b(?:side|container)[^\n0-9]{0,30}?" + NUM), |
| ] |
| TABLE_HEADER = re.compile(r"^\s*\|\s*n\s*\|\s*(?:s|side|L|best s)\s*\|", re.I) |
| TABLE_ROW = re.compile(r"^\s*\|\s*(\d{1,4})\s*\|\s*" + NUM + r"\s*\|") |
| SKIP_LINE = re.compile(r"\b(retract\w*|withdrawn|was wrong|was a bug|target|goal|hope|if we|would be)\b", re.I) |
| SCORE = re.compile(r"score\s*[=:]?\s*(-?[0-9]+\.[0-9]+)", re.I) |
| BLOCK = re.compile(r"```packing\s*\n(.*?)(?:```|\Z)", re.S) |
|
|
|
|
| def _num(v): |
| return float(v.replace(",", ".")) |
|
|
|
|
| TRUNC_AFTER = re.compile(r"^\s*(?:…|\.\.\.)") |
| APPROX_BEFORE = re.compile(r"(?:≈|~|approx\w*\.?)\s*$", re.I) |
|
|
|
|
| def display_uncertainty(line, start, end, raw): |
| """Absolute uncertainty (container-side units) implied by how a number is printed. |
| '2.707...' / '2.707…' (truncated) or '≈ 2.707' (rounded) -> 10**-decimals; otherwise 0.""" |
| dec = len(raw.replace(",", ".").split(".")[1]) if re.search(r"[.,]", raw) else 0 |
| if TRUNC_AFTER.match(line[end:]) or APPROX_BEFORE.search(line[max(0, start - 16):start]): |
| return 10.0 ** -dec |
| return 0.0 |
|
|
|
|
| def extract_claims(text, meta): |
| """Return (claims, notes). Each claim: n, s, coordinates_available, packing, score, |
| source_url, source_type, source_key, date, extracted_text.""" |
| claims, notes = [], [] |
| base = {"source_url": meta["url"], "source_type": meta.get("source_type", "unknown"), |
| "source_key": meta.get("source_key", meta["url"]), "date": meta.get("date", "")} |
|
|
| def add(n, s, quote, packing=None, score=None, unc=0.0): |
| if not 1 <= n <= 1000: |
| notes.append(f"ignored n={n} (out of range)"); return |
| if s is not None: |
| if s < area_lower_bound(n) - 1e-9: |
| notes.append(f"rejected n={n} s={s}: below area bound sqrt(n)={area_lower_bound(n):.6f}") |
| claims.append({**base, "n": n, "s": s, "rejected": "impossible (area bound)", |
| "coordinates_available": packing is not None, "packing": packing, |
| "score": score, "extracted_text": quote[:160]}) |
| return |
| if s > trivial_upper_bound(n) + 1e-9: |
| notes.append(f"ignored n={n} s={s}: worse than trivial grid (likely noise)"); return |
| claims.append({**base, "n": n, "s": s, "coordinates_available": packing is not None, |
| "packing": packing, "score": score, "s_uncertainty": unc, "extracted_text": quote[:160]}) |
|
|
| |
| for m in BLOCK.finditer(text): |
| raw = m.group(1) |
| try: |
| d = json.loads(raw) |
| n, side = int(d["n"]), d["side"] |
| s = float(eval_side(side)) |
| add(n, s, raw[:160], packing=d) |
| except Exception as e: |
| notes.append(f"unparseable packing block ({type(e).__name__})") |
| text_wo_blocks = BLOCK.sub(" ", text) |
|
|
| |
| in_table = False |
| seen = set() |
| for line in text_wo_blocks.splitlines(): |
| if TABLE_HEADER.match(line): |
| in_table = True; continue |
| if in_table: |
| m = TABLE_ROW.match(line) |
| if m: |
| key = (int(m.group(1)), _num(m.group(2))) |
| if key not in seen: |
| seen.add(key); add(key[0], key[1], line.strip()) |
| continue |
| if not line.strip().startswith("|"): |
| in_table = False |
| if SKIP_LINE.search(line): |
| if any(p.search(line) for p in PATTERNS): |
| notes.append(f"skipped hedged/retracted line: {line.strip()[:80]}") |
| continue |
| sc = SCORE.search(line) |
| for p in PATTERNS: |
| for m in p.finditer(line): |
| key = (int(m.group(1)), _num(m.group(2))) |
| if key in seen: |
| continue |
| seen.add(key) |
| add(key[0], key[1], line.strip(), score=float(sc.group(1)) if sc else None, |
| unc=display_uncertainty(line, m.start(2), m.end(2), m.group(2))) |
| return claims, notes |
|
|
|
|
| def eval_side(side): |
| """side may be a number or an exact string like '7/5'. Returns Fraction or float.""" |
| from fractions import Fraction |
| if isinstance(side, str) and re.fullmatch(r"\s*-?\d+(?:/\d+)?\s*", side): |
| return Fraction(side.strip()) |
| return float(side) |
|
|
|
|
| |
|
|
