#!/usr/bin/env python3 """Measure two real public-data retrieval procedures, not human task time or customer savings. Only public organisational columns are selected from the official workbook. Full source downloads and selected rows remain in ignored out/; publish the aggregate receipt after review. Uses Python's standard library and no account or payment. """ import argparse import csv import hashlib import html import io import json import re import time import urllib.parse import urllib.request import xml.etree.ElementTree as ET import zipfile from datetime import datetime, timezone from pathlib import Path SOURCE_PAGE = "https://www.cqc.org.uk/about-us/transparency/using-cqc-data" NS = {"table": "urn:oasis:names:tc:opendocument:xmlns:table:1.0", "text": "urn:oasis:names:tc:opendocument:xmlns:text:1.0"} TAG = lambda local: "{" + NS["table"] + "}" + local FIELDS = {"locationId": "Location ID", "locationName": "Location Name", "numberOfBeds": "Care homes beds", "localAuthority": "Location Local Authority", "postcode": "Location Postal Code", "overallRating": "Location Latest Overall Rating", "website": "Location Web Address"} ATTRIBUTION = "Contains public sector information licensed under the Open Government Licence v3.0. Source: Care Quality Commission (CQC). DataScoop is independent of CQC." def get(url): request = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 (compatible; DataScoop/1.0; +https://datascoop.io)", "Accept": "*/*"}) started = time.perf_counter() with urllib.request.urlopen(request, timeout=60) as response: data = response.read() return data, {"url": response.url, "status": response.status, "bytes": len(data), "elapsedSeconds": round(time.perf_counter() - started, 3), "sha256": hashlib.sha256(data).hexdigest()} def cell_text(cell): return " ".join("".join(p.itertext()) for p in cell.findall("text:p", NS)).strip() def selected_cells(row, indices=None): result, column = {}, 0 for cell in row: if cell.tag not in [TAG("table-cell"), TAG("covered-table-cell")]: continue repeated = int(cell.attrib.get(TAG("number-columns-repeated"), "1")) if indices is None: value = cell_text(cell) if value: if repeated != 1: raise ValueError("Unexpected repeated populated header") result[column] = value else: wanted = indices.intersection(range(column, column + repeated)) if wanted: value = cell_text(cell) for index in wanted: result[index] = value column += repeated return result def workbook_rows(path): records, seen, total_rows, source_date, columns = [], set(), 0, None, None with zipfile.ZipFile(path) as archive, archive.open("content.xml") as source: sheet = "" for event, element in ET.iterparse(source, events=["start", "end"]): if event == "start" and element.tag == TAG("table"): sheet = element.attrib.get(TAG("name"), "") if event != "end" or element.tag != TAG("table-row"): continue if sheet == "README": # Only the source-date sentence is retained from the README. for value in selected_cells(element).values(): match = re.search(r"Source: CQC database as at (\d{1,2} \w+ \d{4})", value) if match: source_date = datetime.strptime(match.group(1), "%d %B %Y").date().isoformat() elif sheet == "HSCA_Active_Locations": if columns is None: header = selected_cells(element) inverse = {name: index for index, name in header.items()} needed = set(FIELDS.values()) | {"Care home?"} if not needed.issubset(inverse): raise ValueError("Required public workbook columns are missing") columns = {name: inverse[name] for name in needed} else: cells = selected_cells(element, set(columns.values())) value = lambda name: cells.get(columns[name], "") if value("Location ID"): if int(element.attrib.get(TAG("number-rows-repeated"), "1")) != 1: raise ValueError("Repeated populated location row") total_rows += 1 beds = value("Care homes beds") if value("Care home?") == "Y" and value("Location Local Authority") == "Surrey" and re.fullmatch(r"\d+", beds) and int(beds) >= 40: row = {key: value(name) for key, name in FIELDS.items()} if row["locationId"] in seen: raise ValueError("Duplicate selected official location ID") seen.add(row["locationId"]) records.append(row) element.clear() if not columns or not source_date or not records: raise ValueError("Official workbook selection/source date could not be established") return sorted(records, key=lambda row: row["locationId"]), source_date, total_rows def write_csv(path, records): with path.open("w", newline="", encoding="utf-8") as target: writer = csv.DictWriter(target, fieldnames=list(FIELDS)) writer.writeheader() writer.writerows(records) with path.open(newline="", encoding="utf-8") as target: checked = list(csv.DictReader(target)) if checked != records: raise ValueError("CSV round trip changed rows or IDs") return {"rows": len(records), "bytes": path.stat().st_size, "sha256": hashlib.sha256(path.read_bytes()).hexdigest()} def datascoop_rows(base): rows, requests, imported_at, total = [], [], None, None filters = {"careHome": "true", "localAuthorityExact": "Surrey", "minBeds": "40"} for _ in range(200): params = urllib.parse.urlencode({**filters, "strict_filters": "1", "limit": "25", "offset": len(rows), "internal": "1"}) data, receipt = get(base.rstrip("/") + "/v1/datasets/cqc/query?" + params) requests.append(receipt) result = json.loads(data) count = result.get("total") imported = result.get("provenance", {}).get("imported_at") if type(count) is not int or not 0 < count <= 5000 or not isinstance(imported, str): raise ValueError("Missing or excessive DataScoop total/date") if result.get("dataset") != "cqc" or result.get("ignored") != []: raise ValueError("Unexpected DataScoop selection") for key, expected in filters.items(): actual = result.get("applied_filters", {}).get(key) if str(actual).lower() != expected.lower(): raise ValueError("DataScoop filter was not applied") if total is None: total, imported_at = count, imported if count != total or imported != imported_at or result.get("offset") != len(rows): raise ValueError("Source import or pagination changed during the comparison") page = result.get("rows") if not isinstance(page, list) or not page or len(page) > 25 or result.get("returned") != len(page): raise ValueError("Incomplete DataScoop page") for record in page: if not isinstance(record, dict): raise ValueError("Invalid DataScoop row") rows.append({key: "" if record.get(key) is None else str(record.get(key, "")).strip() for key in FIELDS}) if result.get("truncated") is not (len(rows) < total) or len(rows) > total: raise ValueError("Untruthful pagination") if len(rows) == total: if len({row["locationId"] for row in rows}) != total or any(not row["locationId"] for row in rows): raise ValueError("DataScoop IDs are not unique") return sorted(rows, key=lambda row: row["locationId"]), imported_at, requests raise ValueError("Pagination budget exhausted") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--out", default="out/care-growth/task-comparison") parser.add_argument("--base", default="https://datascoop.io") args = parser.parse_args() target = Path(args.out) target.mkdir(parents=True, exist_ok=True) started_at = datetime.now(timezone.utc).isoformat() started = time.perf_counter() page, page_receipt = get(SOURCE_PAGE) links = re.findall(r'href="([^"]+HSCA_Active_Locations\.ods[^"]*)"', page.decode(), re.I) if len(links) != 1: raise ValueError("Expected one current active-location workbook on CQC's source page") url = urllib.parse.urljoin(SOURCE_PAGE, html.unescape(links[0])) if urllib.parse.urlparse(url).hostname != "www.cqc.org.uk": raise ValueError("Unexpected official download host") source, source_receipt = get(url) workbook = target / "official-source.ods" workbook.write_bytes(source) del source official, source_date, source_count = workbook_rows(workbook) official_csv = write_csv(target / "official-surrey-40.csv", official) official_seconds = round(time.perf_counter() - started, 3) started = time.perf_counter() ds, imported_at, ds_requests = datascoop_rows(args.base) ds_csv = write_csv(target / "datascoop-surrey-40.csv", ds) ds_seconds = round(time.perf_counter() - started, 3) a, b = {row["locationId"]: row for row in official}, {row["locationId"]: row for row in ds} common = sorted(a.keys() & b.keys()) changed = {field: sum(a[key][field] != b[key][field] for key in common) for field in FIELDS if field != "locationId"} receipt = {"measuredAt": started_at, "task": "Care-home locations in the exact Surrey local authority with at least 40 recorded beds", "selection": {"careHome": True, "localAuthorityExact": "Surrey", "minBeds": 40}, "columns": list(FIELDS), "method": "One sequential scripted retrieval per route on the same machine; prepared Python standard-library code. Time from each route's first request through validated UTF-8 CSV. No browser, spreadsheet UI, development time, signup or checkout time is included.", "official": {"sourcePage": SOURCE_PAGE, "sourceDate": source_date, "workbookRows": source_count, "sourcePageRequest": page_receipt, "download": source_receipt, "output": official_csv, "elapsedSeconds": official_seconds, "steps": ["Open the CQC data page", "Download Care directory with filters", "Open HSCA_Active_Locations", "Select Care home? = Y, Location Local Authority = Surrey, Care homes beds >= 40", "Keep the seven public columns; preserve Location IDs as text", "Export and validate the CSV"]}, "datascoop": {"importedAt": imported_at, "publicJsonPages": len(ds_requests), "downloadedBytes": sum(r["bytes"] for r in ds_requests), "requests": ds_requests, "output": ds_csv, "elapsedSeconds": ds_seconds, "steps": ["Request the exact careHome/localAuthorityExact/minBeds predicates with strict validation", "Read every free JSON page, checking applied filters, total and import date", "Keep the same seven public columns; preserve Location IDs as text", "Write and validate the CSV"], "access": "Anonymous REST pagination, 25 rows/page; no payment or key. This measures the API route, not the paid one-click CSV or browser checkout."}, "reconciliation": {"commonLocationIds": len(common), "officialOnlyLocationIds": len(a.keys() - b.keys()), "datascoopOnlyLocationIds": len(b.keys() - a.keys()), "differentValuesByFieldAmongCommonIds": changed}, "limitations": ["Different source dates: membership/value differences do not establish error, opening or closure.", "Source dates and DataScoop import dates are different concepts; neither proves when each record was last checked.", "These are single-run technical timings with prepared code, not human preparation time, typical performance or measured customer time savings.", "Full official workbook transfer includes columns not used here. No personal columns are retained in selected output or aggregate evidence.", "Missing values stay blank; no capacity, rating, email or personal contact values are inferred."], "attribution": ATTRIBUTION, "licenceUrl": "https://www.nationalarchives.gov.uk/doc/open-government-licence/version/3/"} (target / "receipt.json").write_text(json.dumps(receipt, indent=2) + "\n") print(json.dumps({"receipt": str(target / "receipt.json"), "official": official_csv, "officialSeconds": official_seconds, "datascoop": ds_csv, "datascoopSeconds": ds_seconds, "reconciliation": receipt["reconciliation"]})) if __name__ == "__main__": main()