"""Scrapes a business directory into a clean Excel workbook + CSV. - Reads robots.txt first; never requests disallowed paths; waits Crawl-delay between requests (or --delay). - Follows category pagination via rel="next"; handles two listing-page templates. - Retries 429/5xx with backoff (honours Retry-After); logs every request. - De-duplicates listings that appear in more than one category; normalises phone numbers to +61 format. Usage: python3 scrape_directory.py BASE_URL [--delay SECONDS] Only use on sites whose terms allow it. Work sample: tested on a sample local site.""" import re, sys, time, csv, argparse, urllib.robotparser from urllib.parse import urljoin, urlparse from pathlib import Path import requests from bs4 import BeautifulSoup H = Path(__file__).parent; OUT = H / "output" UA = "PortfolioSampleScraper/1.0 (contact: owner of this sample)" class Scraper: def __init__(self, base, delay=None): self.base = base.rstrip("/") + "/"; self.s = requests.Session(); self.s.headers["User-Agent"] = UA self.rp = urllib.robotparser.RobotFileParser(); self.rp.set_url(urljoin(self.base, "robots.txt")) self.rp.parse(self.s.get(urljoin(self.base, "robots.txt"), timeout=10).text.splitlines()) self.delay = delay if delay is not None else float(self.rp.crawl_delay(UA) or 1) self.log = []; self.skipped = []; self.last = 0 def get(self, url, tries=4): if not self.rp.can_fetch(UA, url): self.skipped.append(url); return None for attempt in range(1, tries + 1): wait = self.delay - (time.time() - self.last) if wait > 0: time.sleep(wait) t0 = time.time(); self.last = t0 try: r = self.s.get(url, timeout=15); code = r.status_code except requests.RequestException as e: r, code = None, f"error: {type(e).__name__}" self.log.append({"url": urlparse(url).path, "attempt": attempt, "status": code, "ms": round((time.time() - t0) * 1000)}) if r is not None and r.status_code == 200: r.encoding = "utf-8"; return BeautifulSoup(r.text, "html.parser") if r is not None and r.status_code not in (429, 500, 502, 503, 504): return None retry_after = float(r.headers.get("Retry-After", 0)) if r is not None else 0 time.sleep(min(retry_after, 30) + 0.5 * 2 ** (attempt - 1)) # honour Retry-After, then exponential backoff return None def phone_e164(raw): d = re.sub(r"\D", "", raw or "") if d.startswith("61"): d = "0" + d[2:] return "+61" + d[1:] if len(d) == 10 and d.startswith("0") else "" def parse_listing(soup): if soup.select_one("article.biz"): # current template a = soup.select_one("article.biz"); g = lambda sel: (a.select_one(sel).get_text(" ", strip=True) if a.select_one(sel) else "") rec = {"listing_id": a["data-id"], "name": g(".biz-name"), "category": g(".cat"), "address": g(".addr"), "phone_raw": g(".phone"), "email": g(".email"), "website": g(".web"), "template": "current"} else: # older table template a = soup.select_one("#listing"); t = {th.get_text(strip=True): th.find_next("td").get_text(strip=True) for th in a.select("table.details th")} cat = re.search(r"Category:\s*(.+)", a.get_text("\n")) rec = {"listing_id": a["data-listing"], "name": a.h2.get_text(strip=True), "category": cat.group(1).strip() if cat else "", "address": t.get("Address", ""), "phone_raw": t.get("Phone", ""), "email": t.get("Email", ""), "website": t.get("Website", ""), "template": "older"} st = soup.select_one(".rating .stars"); rating = st.get("data-rating") if st else None cnt = soup.select_one(".rating .count") rec["rating"] = float(rating) if rating not in (None, "None") else None rec["reviews"] = int(re.sub(r"\D", "", cnt.get_text())) if cnt else 0 m = re.match(r"(.+),\s*(.+?)\s+VIC$", rec["address"]); rec["street"], rec["town"] = (m.group(1), m.group(2)) if m else (rec["address"], "") rec["phone"] = phone_e164(rec["phone_raw"]) return rec def run(base, delay=None): sc = Scraper(base, delay); home = sc.get(sc.base) cats = [urljoin(sc.base, a["href"]) for a in home.select("a.cat-link")] for a in home.find_all("a"): # follow every link on the home page; robots.txt decides what is allowed u = urljoin(sc.base, a["href"]) if u not in cats and "/category/" not in u: sc.get(u) listed_under, order, pages = {}, [], 0 for url in cats: while url: soup = sc.get(url); pages += 1 if soup is None: break cname = soup.h1.get_text(strip=True) for a in soup.select("a.biz-link"): u = urljoin(sc.base, a["href"]) if u not in listed_under: listed_under[u] = []; order.append(u) listed_under[u].append(cname) nxt = soup.select_one('a[rel="next"]'); url = urljoin(sc.base, nxt["href"]) if nxt else None rows, failed = [], [] for u in order: soup = sc.get(u) if soup is None: failed.append(u); continue r = parse_listing(soup); r["listed_under"] = "; ".join(listed_under[u]); r["source_url"] = urlparse(u).path flags = [f for f, bad in (("no email", not r["email"]), ("no website", not r["website"]), ("no rating", r["rating"] is None), ("phone not parsed", not r["phone"])) if bad] r["flags"] = ", ".join(flags); rows.append(r) return rows, sc, pages, failed, sum(len(v) for v in listed_under.values()) COLS = ["listing_id", "name", "category", "listed_under", "street", "town", "phone", "phone_raw", "email", "website", "rating", "reviews", "flags", "template", "source_url"] def save(rows, sc, pages, failed, seen): OUT.mkdir(exist_ok=True) with open(OUT / "directory_listings.csv", "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, COLS, extrasaction="ignore"); w.writeheader(); w.writerows(rows) with open(OUT / "scrape_log.csv", "w", newline="") as f: w = csv.DictWriter(f, ["url", "attempt", "status", "ms"]); w.writeheader(); w.writerows(sc.log) from build_workbook import build build(rows, sc, pages, failed, seen, COLS) if __name__ == "__main__": ap = argparse.ArgumentParser(); ap.add_argument("base"); ap.add_argument("--delay", type=float); a = ap.parse_args() rows, sc, pages, failed, seen = run(a.base, a.delay); save(rows, sc, pages, failed, seen) print(f"{len(rows)} unique listings from {seen} category entries on {pages} pages; {len(sc.log)} requests, " f"{sum(1 for l in sc.log if l['status'] != 200)} retried, {len(sc.skipped)} blocked by robots.txt, {len(failed)} failed")