Recipes for the Web and Your Desktop

1# scrape_table.py - copy a table from a web page into a CSV file
2import csv
3import sys
4from pathlib import Path
5
6import requests
7from bs4 import BeautifulSoup
8
9HERE = Path(__file__).parent
10SOURCE = HERE / "sample_page.html"  # or a web address, as text
11OUTPUT = HERE / "new_books.csv"
12HEADERS = {"User-Agent": "learning-python-scraper (personal use)"}
13
14
15def get_html(source):
16    """Read a page from the web, or from a file on your computer."""
17    if str(source).startswith(("http://", "https://")):
18        response = requests.get(
19            source, headers=HEADERS, timeout=10
20        )
21        response.raise_for_status()
22        return response.text
23    return Path(source).read_text(encoding="utf-8")
24
25
26def table_rows(html, table_id):
27    """Return the header and the rows of the table with this id."""
28    soup = BeautifulSoup(html, "html.parser")
29    table = soup.find("table", id=table_id)
30    if table is None:
31        raise ValueError(
32            f"no table with id '{table_id}' on the page"
33        )
34    header = [
35        th.get_text(strip=True) for th in table.find_all("th")
36    ]
37    rows = []
38    for tr in table.find("tbody").find_all("tr"):
39        rows.append(
40            [td.get_text(strip=True) for td in tr.find_all("td")]
41        )
42    return header, rows
43
44
45def main():
46    source = sys.argv[1] if len(sys.argv) > 1 else SOURCE
47    try:
48        header, rows = table_rows(get_html(source), "new-books")
49    except (
50        requests.RequestException,
51        ValueError,
52        OSError,
53    ) as error:
54        print(f"Couldn't get the table: {error}")
55        return
56    with open(OUTPUT, "w", newline="", encoding="utf-8") as file:
57        writer = csv.writer(file)
58        writer.writerow(header)
59        writer.writerows(rows)
60    print(f"Saved {len(rows)} row(s) to {OUTPUT.name}")
61    for row in rows[:3]:
62        print("  " + " | ".join(row))
63
64
65if __name__ == "__main__":
66    main()

1# watch_page.py - tell me when part of a web page changes
2import hashlib
3import json
4import logging
5from datetime import datetime
6from pathlib import Path
7
8import requests
9from bs4 import BeautifulSoup
10
11HERE = Path(__file__).parent
12STATE = HERE / "watch_state.json"
13logging.basicConfig(
14    filename=HERE / "watch_page.log",
15    level=logging.INFO,
16    format="%(asctime)s %(levelname)s %(message)s",
17)
18log = logging.getLogger("watch")
19
20# What to watch: a page, and the part of it you care about,
21# picked out with a CSS selector (see "How it works").
22WATCHES = {
23    "library hours": (HERE / "sample_page.html", "#hours"),
24    "new books": (HERE / "sample_page.html", "#new-books tbody"),
25}
26
27
28def get_html(source):
29    if str(source).startswith(("http://", "https://")):
30        headers = {"User-Agent": "learning-python-watcher"}
31        response = requests.get(source, timeout=10, headers=headers)
32        response.raise_for_status()
33        return response.text
34    return Path(source).read_text(encoding="utf-8")
35
36
37def current_text(source, selector):
38    """The text of the watched part, or None if it's missing."""
39    soup = BeautifulSoup(get_html(source), "html.parser")
40    part = soup.select_one(selector)
41    return part.get_text(" ", strip=True) if part else None
42
43
44def main():
45    state = (
46        json.loads(STATE.read_text()) if STATE.exists() else {}
47    )
48    changes = []
49    for name, (source, selector) in WATCHES.items():
50        try:
51            text = current_text(source, selector)
52        except (requests.RequestException, OSError) as error:
53            log.warning("%s: couldn't check (%s)", name, error)
54            continue
55        if text is None:
56            log.warning(
57                "%s: '%s' not found; has the page changed "
58                "layout?",
59                name,
60                selector,
61            )
62            continue
63        fingerprint = hashlib.sha256(text.encode()).hexdigest()
64        before = state.get(name)
65        if before is None:
66            log.info("%s: now watching", name)
67        elif before["hash"] != fingerprint:
68            changes.append((name, before["text"], text))
69            log.info("%s: CHANGED", name)
70        state[name] = {
71            "hash": fingerprint,
72            "text": text,
73            "checked": datetime.now().isoformat(),
74        }
75    STATE.write_text(
76        json.dumps(state, indent=2), encoding="utf-8"
77    )
78
79    if not changes:
80        print("No changes.")
81    for name, old, new in changes:
82        print(
83            f"{name} has changed!\n"
84            f"  was: {old[:60]}\n"
85            f"  now: {new[:60]}"
86        )
87
88
89if __name__ == "__main__":
90    main()

1# make_practice_files.py - files for your own little web server
2from pathlib import Path
3
4HERE = Path(__file__).parent
5FOLDER = HERE / "practice_server"
6FOLDER.mkdir(exist_ok=True)
7for month in ["january", "february", "march"]:
8    (FOLDER / f"report_{month}.txt").write_text(
9        f"Monthly report for {month.title()}\n"
10        + "Lots of figures.\n" * 2000,
11        encoding="utf-8",
12    )
13print(
14    f"Made 3 files in {FOLDER.name}/. Now, in a second terminal:"
15)
16print(f"  cd {FOLDER}")
17print("  python -m http.server 8000")

1# download_files.py - download a list of files, safely and politely
2import time
3from pathlib import Path
4from urllib.parse import urlparse
5
6import requests
7
8HERE = Path(__file__).parent
9URL_LIST = HERE / "urls.txt"
10FOLDER = HERE / "downloaded"
11TRIES = 3
12PAUSE = 1  # seconds between downloads
13
14
15def file_name_for(url):
16    """The last part of the address, like report_january.pdf."""
17    name = Path(urlparse(url).path).name
18    return name or "index.html"
19
20
21def download(url, target):
22    """Download url to target a chunk at a time, with retries."""
23    wait = 2
24    for attempt in range(1, TRIES + 1):
25        try:
26            with requests.get(
27                url, stream=True, timeout=20
28            ) as response:
29                response.raise_for_status()
30                partial = target.with_suffix(
31                    target.suffix + ".part"
32                )
33                with open(partial, "wb") as file:
34                    for chunk in response.iter_content(
35                        chunk_size=65536
36                    ):
37                        file.write(chunk)
38                partial.replace(target)  # only now is it complete
39                return True
40        except requests.HTTPError as error:
41            print(
42                f"  {error.response.status_code} for {url}:"
43                " not retrying"
44            )
45            return False
46        except requests.RequestException as error:
47            print(
48                f"  attempt {attempt} failed"
49                f" ({type(error).__name__})"
50            )
51            if attempt < TRIES:
52                time.sleep(wait)
53                wait *= 2
54    return False
55
56
57def main():
58    FOLDER.mkdir(exist_ok=True)
59    urls = [
60        line.strip()
61        for line in URL_LIST.read_text().splitlines()
62        if line.strip() and not line.startswith("#")
63    ]
64    got = skipped = failed = 0
65    for url in urls:
66        target = FOLDER / file_name_for(url)
67        if target.exists():
68            print(f"already have {target.name}")
69            skipped += 1
70            continue
71        print(f"downloading {target.name}")
72        if download(url, target):
73            got += 1
74        else:
75            failed += 1
76        time.sleep(PAUSE)
77    print(
78        f"Done: {got} downloaded, {skipped} already here, "
79        f"{failed} failed."
80    )
81
82
83if __name__ == "__main__":
84    main()

1# fill_form.py - fill in a web form for every row of a spreadsheet
2import csv
3from pathlib import Path
4
5from playwright.sync_api import sync_playwright
6
7HERE = Path(__file__).parent
8FORM = (
9    (HERE / "signup_form.html").resolve().as_uri()
10)  # a local page
11PEOPLE = HERE / "volunteers.csv"
12SHOTS = HERE / "screenshots"
13SHOW_BROWSER = False  # True lets you watch it work
14
15
16def sign_up(page, person):
17    page.goto(FORM)
18    page.fill("#name", person["name"])
19    page.fill("#email", person["email"])
20    page.select_option("#shift", person["shift"])
21    if person["gloves"] == "yes":
22        page.check("#gloves")
23    page.click("button[type=submit]")
24    page.wait_for_selector("#result:not(:empty)")
25    return page.inner_text("#result")
26
27
28def main():
29    SHOTS.mkdir(exist_ok=True)
30    with open(PEOPLE, newline="", encoding="utf-8") as file:
31        people = list(csv.DictReader(file))
32    with sync_playwright() as p:
33        browser = p.chromium.launch(headless=not SHOW_BROWSER)
34        page = browser.new_page()
35        for person in people:
36            try:
37                message = sign_up(page, person)
38            except Exception as error:  # one failure, not all
39                print(f"  {person['name']}: FAILED ({error})")
40                continue
41            shot = SHOTS / f"{person['email'].split('@')[0]}.png"
42            page.screenshot(path=shot)  # proof it worked
43            print(f"  {message}")
44        browser.close()
45
46
47if __name__ == "__main__":
48    main()

1# backup_rotate.py - zip a folder daily; keep the last few zips
2import logging
3import shutil
4import zipfile
5from datetime import datetime
6from pathlib import Path
7
8HERE = Path(__file__).parent
9SOURCE = HERE / "important"  # the folder to protect
10BACKUPS = HERE / "backups"  # better: a different drive!
11KEEP = 7  # how many zips to keep
12
13logging.basicConfig(
14    filename=HERE / "backup.log",
15    level=logging.INFO,
16    format="%(asctime)s %(levelname)s %(message)s",
17)
18log = logging.getLogger("backup")
19
20
21def make_backup():
22    BACKUPS.mkdir(exist_ok=True)
23    stamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")
24    base = BACKUPS / f"{SOURCE.name}_{stamp}"
25    archive = Path(
26        shutil.make_archive(str(base), "zip", root_dir=SOURCE)
27    )
28    with zipfile.ZipFile(archive) as z:  # check it really works
29        bad = z.testzip()
30        count = len(z.namelist())
31    if bad:
32        raise RuntimeError(f"{archive.name} is damaged at {bad}")
33    log.info(
34        "made %s (%d files, %.1f KB)",
35        archive.name,
36        count,
37        archive.stat().st_size / 1024,
38    )
39    return archive
40
41
42def remove_old_backups():
43    """Delete all but the newest KEEP zips made by this script."""
44    zips = sorted(
45        BACKUPS.glob(f"{SOURCE.name}_*.zip")
46    )  # oldest first
47    for old in zips[:-KEEP]:
48        old.unlink()
49        log.info("removed old backup %s", old.name)
50    return len(zips[:-KEEP])
51
52
53def main():
54    if not SOURCE.is_dir():
55        log.error("folder to back up is missing: %s", SOURCE)
56        print(f"{SOURCE} doesn't exist. Nothing backed up.")
57        return
58    try:
59        archive = make_backup()
60    except Exception:
61        log.exception("backup failed")
62        print("Backup FAILED. Details are in backup.log.")
63        return
64    removed = remove_old_backups()  # only after a good backup
65    print(
66        f"Backed up to {archive.name};"
67        f" removed {removed} old zip(s)."
68    )
69
70
71if __name__ == "__main__":
72    main()

1# track_rates.py - record an exchange rate every day, and chart it
2import csv
3import sys
4from datetime import date, timedelta
5from pathlib import Path
6
7import matplotlib
8
9matplotlib.use("Agg")  # draw to a file, not a window
10import matplotlib.pyplot as plt
11import requests
12
13HERE = Path(__file__).parent
14HISTORY = HERE / "rates.csv"
15CHART = HERE / "rates_chart.png"
16BASE, TARGET = "AUD", "USD"
17API = "https://api.frankfurter.dev/v1"
18
19
20def fetch(start=None):
21    """Return {date: rate}, for today or from start until today."""
22    period = f"{start.isoformat()}.." if start else "latest"
23    response = requests.get(
24        f"{API}/{period}",
25        timeout=10,
26        params={"from": BASE, "to": TARGET},
27    )
28    response.raise_for_status()
29    data = response.json()
30    if start:
31        return {
32            day: r[TARGET] for day, r in data["rates"].items()
33        }
34    return {data["date"]: data["rates"][TARGET]}
35
36
37def load_history():
38    if not HISTORY.exists():
39        return {}
40    with open(HISTORY, newline="", encoding="utf-8") as file:
41        return {
42            row["date"]: float(row["rate"])
43            for row in csv.DictReader(file)
44        }
45
46
47def save_history(history):
48    with open(HISTORY, "w", newline="", encoding="utf-8") as file:
49        writer = csv.writer(file)
50        writer.writerow(["date", "rate"])
51        for day in sorted(history):
52            writer.writerow([day, history[day]])
53
54
55def draw_chart(history):
56    days = sorted(history)
57    dates = [date.fromisoformat(d) for d in days]
58    rates = [history[d] for d in days]
59    fig, ax = plt.subplots(figsize=(8, 4))
60    ax.plot(dates, rates, color="#2f5d8a", linewidth=2)
61    ax.set_title(f"1 {BASE} in {TARGET}")
62    ax.set_ylabel(TARGET)
63    ax.grid(alpha=0.3)
64    fig.autofmt_xdate()  # tilt the dates so they fit
65    fig.tight_layout()
66    fig.savefig(CHART, dpi=150)
67    plt.close(fig)
68
69
70def main():
71    history = load_history()
72    try:
73        if "--backfill" in sys.argv:  # fill the last 60 days
74            new = fetch(start=date.today() - timedelta(days=60))
75        else:
76            new = fetch()
77        added = len(set(new) - set(history))
78        history.update(new)
79        save_history(history)
80        print(
81            f"Added {added} new day(s); {len(history)} in total."
82        )
83    except requests.RequestException as error:
84        print(
85            "Couldn't reach the rates service"
86            f" ({type(error).__name__})."
87        )
88        print("Charting the history I already have.")
89    if len(history) >= 2:
90        draw_chart(history)
91        print(f"Chart saved to {CHART.name}")
92    else:
93        print("Not enough history to chart yet. Try --backfill.")
94
95
96if __name__ == "__main__":
97    main()