{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Audit `arcpy.da` cursor scope before upgrading to ArcGIS Pro 3.7\n",
    "\n",
    "**The situation this is for.** You are planning the move to Pro 3.7. Somewhere in your\n",
    "organization there is a folder of `.py` scripts, `.pyt` toolboxes and notebooks written over\n",
    "several years by several people, and nobody can tell you which of them still run.\n",
    "\n",
    "Pro 3.7 changed cursor lifetime: an `arcpy.da` cursor is **invalidated when its `with` block\n",
    "exits**. Code that called `reset()` or re-iterated the cursor afterwards worked for years and\n",
    "now raises `ValueError: I/O operation on closed file`.\n",
    "\n",
    "This notebook finds those places. Set `SCAN_PATH` below and run all cells. It parses your\n",
    "source with the standard-library `ast` module and **executes nothing**, so it is safe to point\n",
    "at code you have not read.\n",
    "\n",
    "**Findings are a review list, not a verdict.** See the limits at the end.\n",
    "\n",
    "Standard library only - runs in `arcgispro-py3` or any plain Python 3.8+. MIT licensed.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## CONFIG\n",
    "\n",
    "The only cell you need to edit.\n"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {},
   "outputs": [],
   "execution_count": null,
   "source": [
    "# ---------------------------------------------------------------------------------\n",
    "# EDIT THE VALUES, NOT THE COMMENTS.\n",
    "# Leaving SCAN_PATH at None is the most common reason this notebook returns nothing.\n",
    "# The commented lines below are examples - copy one, uncomment it, change the path.\n",
    "# ---------------------------------------------------------------------------------\n",
    "\n",
    "# Folder or single file to scan. A whole scripts share is a reasonable target.\n",
    "SCAN_PATH = None\n",
    "# SCAN_PATH = r\"C:\\gis\\scripts\"\n",
    "\n",
    "RECURSE = True                 # descend into subfolders\n",
    "\n",
    "# Optional report outputs. Each must be a FILE path - but if you give a folder, the\n",
    "# notebook will write a default filename into it rather than fail.\n",
    "# JSON is the one to keep if you want to re-run after fixing and diff the two.\n",
    "REPORT_TXT = None\n",
    "REPORT_JSON = None\n",
    "# REPORT_TXT  = r\"C:\\gis\\cursor-audit.txt\"\n",
    "# REPORT_JSON = r\"C:\\gis\\cursor-audit.json\"\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## The scanner\n",
    "\n",
    "Reads `.py`, `.pyt` and `.ipynb`. Notebook code cells are joined into one parse unit, because\n",
    "a notebook shares one namespace - a cursor opened in one cell and reset in the next is the\n",
    "most common real form of this bug and exists only across cells.\n"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {},
   "outputs": [],
   "execution_count": null,
   "source": [
    "\n",
    "import ast\n",
    "import json\n",
    "from pathlib import Path\n",
    "\n",
    "CURSOR_CLASSES = {\"SearchCursor\", \"UpdateCursor\", \"InsertCursor\"}\n",
    "SCAN_SUFFIXES = {\".py\", \".pyt\", \".ipynb\"}\n",
    "SCOPE_NODES = (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda, ast.ClassDef)\n",
    "\n",
    "\n",
    "# --------------------------------------------------------------------------- alias discovery\n",
    "class _Aliases(ast.NodeVisitor):\n",
    "    \"\"\"Names that refer to arcpy.da, and names bound directly to a cursor class.\"\"\"\n",
    "\n",
    "    def __init__(self) -> None:\n",
    "        self.da_names = {\"da\"}\n",
    "        self.direct: dict[str, str] = {}\n",
    "\n",
    "    def visit_Import(self, node: ast.Import) -> None:\n",
    "        for a in node.names:\n",
    "            if a.name in (\"arcpy.da\",) and a.asname:\n",
    "                self.da_names.add(a.asname)\n",
    "        self.generic_visit(node)\n",
    "\n",
    "    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:\n",
    "        mod = node.module or \"\"\n",
    "        for a in node.names:\n",
    "            if mod == \"arcpy\" and a.name == \"da\":\n",
    "                self.da_names.add(a.asname or \"da\")\n",
    "            elif mod in (\"arcpy.da\", \"da\") and a.name in CURSOR_CLASSES:\n",
    "                self.direct[a.asname or a.name] = a.name\n",
    "        self.generic_visit(node)\n",
    "\n",
    "\n",
    "def _cursor_class(call: ast.AST, al: _Aliases) -> str | None:\n",
    "    if not isinstance(call, ast.Call):\n",
    "        return None\n",
    "    fn = call.func\n",
    "    if isinstance(fn, ast.Attribute) and fn.attr in CURSOR_CLASSES:\n",
    "        owner = fn.value\n",
    "        if isinstance(owner, ast.Attribute) and owner.attr == \"da\":\n",
    "            return fn.attr                                  # arcpy.da.SearchCursor\n",
    "        if isinstance(owner, ast.Name) and owner.id in al.da_names:\n",
    "            return fn.attr                                  # da.SearchCursor / da2.SearchCursor\n",
    "    if isinstance(fn, ast.Name) and fn.id in al.direct:\n",
    "        return al.direct[fn.id]                             # SearchCursor / SC\n",
    "    return None\n",
    "\n",
    "\n",
    "# --------------------------------------------------------------------------- scope splitting\n",
    "def _own_nodes(scope: ast.AST):\n",
    "    \"\"\"Walk a scope without descending into nested scopes, which have their own names.\"\"\"\n",
    "    stack = list(ast.iter_child_nodes(scope))\n",
    "    while stack:\n",
    "        node = stack.pop()\n",
    "        yield node\n",
    "        if not isinstance(node, SCOPE_NODES):\n",
    "            stack.extend(ast.iter_child_nodes(node))\n",
    "\n",
    "\n",
    "def _bindings(scope: ast.AST) -> dict[str, list[int]]:\n",
    "    \"\"\"Every line in this scope where a name is bound. Missing a binding form causes a false\n",
    "    positive, so this covers the ones that actually appear in scripts.\"\"\"\n",
    "    out: dict[str, list[int]] = {}\n",
    "\n",
    "    def add(name: str, line: int) -> None:\n",
    "        out.setdefault(name, []).append(line)\n",
    "\n",
    "    # parameters exist from the first line of the scope\n",
    "    args = getattr(scope, \"args\", None)\n",
    "    if args is not None:\n",
    "        for a in (*args.posonlyargs, *args.args, *args.kwonlyargs,\n",
    "                  args.vararg, args.kwarg):\n",
    "            if a is not None:\n",
    "                add(a.arg, getattr(scope, \"lineno\", 0))\n",
    "\n",
    "    for node in _own_nodes(scope):\n",
    "        if isinstance(node, (ast.Assign, ast.AugAssign, ast.AnnAssign)):\n",
    "            targets = node.targets if isinstance(node, ast.Assign) else [node.target]\n",
    "            for t in targets:\n",
    "                for n in ast.walk(t):\n",
    "                    if isinstance(n, ast.Name):\n",
    "                        add(n.id, node.lineno)\n",
    "        elif isinstance(node, (ast.For, ast.AsyncFor)):   # NOT ast.comprehension - it does not leak\n",
    "            for n in ast.walk(node.target):\n",
    "                if isinstance(n, ast.Name):\n",
    "                    add(n.id, getattr(node, \"lineno\", getattr(node.target, \"lineno\", 0)))\n",
    "        elif isinstance(node, (ast.With, ast.AsyncWith)):\n",
    "            for item in node.items:\n",
    "                if item.optional_vars is not None:\n",
    "                    for n in ast.walk(item.optional_vars):\n",
    "                        if isinstance(n, ast.Name):\n",
    "                            add(n.id, node.lineno)\n",
    "        elif isinstance(node, ast.NamedExpr):\n",
    "            if isinstance(node.target, ast.Name):\n",
    "                add(node.target.id, node.lineno)\n",
    "        elif isinstance(node, ast.ExceptHandler) and node.name:\n",
    "            add(node.name, node.lineno)\n",
    "        elif isinstance(node, (ast.Import, ast.ImportFrom)):\n",
    "            for a in node.names:\n",
    "                add((a.asname or a.name).split(\".\")[0], node.lineno)\n",
    "        elif isinstance(node, SCOPE_NODES) and hasattr(node, \"name\"):\n",
    "            add(node.name, node.lineno)\n",
    "    return out\n",
    "\n",
    "\n",
    "def _end_line(node: ast.AST) -> int:\n",
    "    end = getattr(node, \"end_lineno\", None)\n",
    "    return end or max((getattr(n, \"lineno\", 0) for n in ast.walk(node)), default=0)\n",
    "\n",
    "\n",
    "def _scan_scope(scope: ast.AST, al: _Aliases, origin: str, locate) -> list[dict]:\n",
    "    binds = _bindings(scope)\n",
    "    loads = [(n.id, n.lineno) for n in _own_nodes(scope)\n",
    "             if isinstance(n, ast.Name) and isinstance(n.ctx, ast.Load)]\n",
    "    found: list[dict] = []\n",
    "    for node in _own_nodes(scope):\n",
    "        if not isinstance(node, (ast.With, ast.AsyncWith)):\n",
    "            continue\n",
    "        for item in node.items:\n",
    "            cls = _cursor_class(item.context_expr, al)\n",
    "            if not cls or not isinstance(item.optional_vars, ast.Name):\n",
    "                continue\n",
    "            name = item.optional_vars.id\n",
    "            end = _end_line(node)\n",
    "            nxt = min((ln for ln in binds.get(name, []) if ln > end), default=None)\n",
    "            for used, line in loads:\n",
    "                if used != name or line <= end:\n",
    "                    continue\n",
    "                if nxt is not None and line >= nxt:\n",
    "                    continue\n",
    "                where, wline = locate(line)\n",
    "                endwhere, endline = locate(end)\n",
    "                # In a notebook the block can close in a DIFFERENT cell from the use - naming only\n",
    "                # a line number there would send the reader to the wrong place.\n",
    "                closed = f\"line {endline}\" if endwhere in (\"\", where) else f\"{endwhere} line {endline}\"\n",
    "                found.append({\"origin\": origin, \"where\": where, \"line\": wline,\n",
    "                              \"name\": name, \"kind\": cls,\n",
    "                              \"detail\": f\"used at line {wline}; its `with` block closed at {closed}\"})\n",
    "    return found\n",
    "\n",
    "\n",
    "def scan_source(src: str, origin: str, locate=None) -> list[dict]:\n",
    "    \"\"\"locate maps a line in `src` to (label, line) - used to report notebook cells.\"\"\"\n",
    "    locate = locate or (lambda ln: (\"\", ln))\n",
    "    try:\n",
    "        tree = ast.parse(src)\n",
    "    except SyntaxError as exc:\n",
    "        where, wline = locate(exc.lineno or 1)\n",
    "        return [{\"origin\": origin, \"where\": where, \"line\": wline, \"name\": \"\", \"kind\": \"SYNTAX\",\n",
    "                 \"detail\": f\"could not parse: {exc.msg}\"}]\n",
    "    al = _Aliases()\n",
    "    al.visit(tree)\n",
    "    scopes = [tree] + [n for n in ast.walk(tree) if isinstance(n, SCOPE_NODES)]\n",
    "    out: list[dict] = []\n",
    "    for sc in scopes:\n",
    "        out += _scan_scope(sc, al, origin, locate)\n",
    "    return out\n",
    "\n",
    "\n",
    "# --------------------------------------------------------------------------- notebooks\n",
    "def _notebook_unit(path: Path):\n",
    "    \"\"\"Join code cells into ONE parse unit. A notebook shares a namespace across cells, so a cursor\n",
    "    opened in one cell and reset in the next is a real breakage - and it is invisible if each cell\n",
    "    is parsed alone. Returns (source, locate) where locate maps a joined line to (cell label, line).\"\"\"\n",
    "    try:\n",
    "        nb = json.loads(path.read_text(encoding=\"utf-8\"))\n",
    "    except (json.JSONDecodeError, UnicodeDecodeError, OSError) as exc:\n",
    "        return None, str(exc)\n",
    "    if not isinstance(nb, dict) or not isinstance(nb.get(\"cells\"), list):\n",
    "        return None, \"not a notebook: no top-level cells array\"\n",
    "\n",
    "    lines: list[str] = []\n",
    "    index: list[tuple[str, int]] = []\n",
    "    for pos, cell in enumerate(nb.get(\"cells\", []), start=1):\n",
    "        if cell.get(\"cell_type\") != \"code\":\n",
    "            continue\n",
    "        src = cell.get(\"source\", \"\")\n",
    "        if isinstance(src, list):\n",
    "            src = \"\".join(src)\n",
    "        label = f\"cell {pos}\"\n",
    "        cell_lines = src.splitlines()\n",
    "        # A %%cell-magic makes the ENTIRE cell something other than Python, so blank all of it.\n",
    "        # Blanking only the first line would leave a shell body to be parsed as Python, and because\n",
    "        # cells are joined into one unit that single cell would fail the parse for the whole\n",
    "        # notebook and hide every real finding in it.\n",
    "        whole_cell_magic = bool(cell_lines) and cell_lines[0].lstrip().startswith(\"%%\")\n",
    "        for i, ln in enumerate(cell_lines, start=1):\n",
    "            if whole_cell_magic or ln.lstrip().startswith((\"%\", \"!\")):\n",
    "                lines.append(\"\")          # keep the line count aligned with Jupyter's gutter\n",
    "            else:\n",
    "                lines.append(ln)\n",
    "            index.append((label, i))\n",
    "        lines.append(\"\")            # separator so the last statement of a cell terminates\n",
    "        index.append((label, len(src.splitlines()) + 1))\n",
    "\n",
    "    def locate(ln: int):\n",
    "        return index[ln - 1] if 1 <= ln <= len(index) else (\"\", ln)\n",
    "\n",
    "    return \"\\n\".join(lines), locate\n",
    "\n",
    "\n",
    "def scan_path(root: Path, recurse: bool = True) -> list[dict]:\n",
    "    root = Path(root)\n",
    "    findings: list[dict] = []\n",
    "    paths = [root] if root.is_file() else sorted(\n",
    "        p for p in (root.rglob(\"*\") if recurse else root.glob(\"*\"))\n",
    "        if p.suffix.lower() in SCAN_SUFFIXES)\n",
    "    for p in paths:\n",
    "        origin = str(p)                       # full path: two files can share a basename\n",
    "        # A short form for the printed report. On a real scripts share the absolute paths are\n",
    "        # long enough to push the finding off the right of the screen; the JSON keeps `origin`.\n",
    "        try:\n",
    "            rel = p.name if root.is_file() else str(p.relative_to(root))\n",
    "        except ValueError:                    # not under root; should not happen, but do not crash\n",
    "            rel = origin\n",
    "        before = len(findings)\n",
    "        if p.suffix.lower() == \".ipynb\":\n",
    "            src, locate = _notebook_unit(p)\n",
    "            if src is None:\n",
    "                findings.append({\"origin\": origin, \"where\": \"\", \"line\": 0, \"name\": \"\",\n",
    "                                 \"kind\": \"READ\", \"detail\": f\"unreadable notebook: {locate}\"})\n",
    "            else:\n",
    "                findings += scan_source(src, origin, locate)\n",
    "        else:\n",
    "            try:\n",
    "                findings += scan_source(p.read_text(encoding=\"utf-8\", errors=\"replace\"), origin)\n",
    "            except OSError as exc:\n",
    "                findings.append({\"origin\": origin, \"where\": \"\", \"line\": 0, \"name\": \"\",\n",
    "                                 \"kind\": \"READ\", \"detail\": str(exc)})\n",
    "        for f in findings[before:]:\n",
    "            f[\"rel\"] = rel\n",
    "    return findings\n",
    "\n",
    "\n",
    "def report(findings: list[dict], scanned: int | None = None, root=None) -> str:\n",
    "    head = f\"Scanned {scanned} file(s).\" if scanned is not None else \"\"\n",
    "    if not findings:\n",
    "        return (f\"{head} No cursor-after-block usage found in the shapes this tool understands.\\n\"\n",
    "                \"That is not a clean bill of health: cursors held in containers or attributes, and \"\n",
    "                \"Python embedded in .atbx/.tbx toolboxes, are outside what it can see.\").strip()\n",
    "    out = [f\"{head} {len(findings)} place(s) to review:\".strip()]\n",
    "    if root is not None:\n",
    "        out.append(f\"Paths below are relative to {root}\")\n",
    "    out.append(\"\")\n",
    "    def shown(f):\n",
    "        return f.get(\"rel\") or f[\"origin\"]\n",
    "    w = max(len(shown(f)) for f in findings)\n",
    "    for f in sorted(findings, key=lambda f: (shown(f), f.get(\"where\", \"\"), f[\"line\"])):\n",
    "        tag = f[\"kind\"] if f[\"kind\"] in CURSOR_CLASSES else f\"[{f['kind']}]\"\n",
    "        where = f\" {f['where']}\" if f.get(\"where\") else \"\"\n",
    "        out.append(f\"  {shown(f):<{w}}{where}  line {f['line']:>5}  {tag:<13} {f['detail']}\")\n",
    "    out += [\"\",\n",
    "            \"Each of these is a place to LOOK, not a confirmed bug - the check follows a name, so a\",\n",
    "            \"name reused for something this tool cannot trace will appear here. Where the finding is\",\n",
    "            \"real, at Pro 3.7 it raises ValueError: I/O operation on closed file. The fix is a fresh\",\n",
    "            \"cursor for the second pass, or finishing the work inside the original `with` block.\"]\n",
    "    return \"\\n\".join(out)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Run it\n"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {},
   "outputs": [],
   "execution_count": null,
   "source": [
    "from pathlib import Path\n",
    "\n",
    "findings = None\n",
    "n = 0\n",
    "\n",
    "if not SCAN_PATH:\n",
    "    raise ValueError(\n",
    "        \"SCAN_PATH is still None, so there is nothing to scan.\\n\\n\"\n",
    "        \"Set the VALUE in the CONFIG cell above - editing the example path inside a\\n\"\n",
    "        \"comment has no effect. It should end up looking like this, with no leading #:\\n\\n\"\n",
    "        '    SCAN_PATH = r\"C:\\\\gis\\\\scripts\"\\n\\n'\n",
    "        \"Then run all cells again.\")\n",
    "\n",
    "target = Path(SCAN_PATH).expanduser()\n",
    "if not target.exists():\n",
    "    raise FileNotFoundError(\n",
    "        f\"SCAN_PATH does not exist: {target}\\n\"\n",
    "        \"Check for a typo, and that the r'' prefix is present so backslashes are literal.\")\n",
    "\n",
    "n = 1 if target.is_file() else sum(\n",
    "    1 for p in (target.rglob('*') if RECURSE else target.glob('*'))\n",
    "    if p.suffix.lower() in SCAN_SUFFIXES)\n",
    "\n",
    "if n == 0:\n",
    "    raise ValueError(\n",
    "        f\"No .py, .pyt or .ipynb files found under: {target}\\n\"\n",
    "        + (\"Set RECURSE = True if they are in subfolders.\" if not RECURSE\n",
    "           else \"Check you are pointing at the right folder.\"))\n",
    "\n",
    "findings = scan_path(target, recurse=RECURSE)\n",
    "print(report(findings, scanned=n, root=target))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Optional: write the report out\n"
   ]
  },
  {
   "cell_type": "code",
   "metadata": {},
   "outputs": [],
   "execution_count": null,
   "source": [
    "# `findings is None` means the run cell did not complete - guard so Run All cannot NameError.\n",
    "\n",
    "def _as_file(path, default_name):\n",
    "    \"\"\"Accept a folder as well as a file path, and name the file sensibly if given one.\"\"\"\n",
    "    p = Path(path).expanduser()\n",
    "    return p / default_name if p.is_dir() else p\n",
    "\n",
    "if findings is not None and REPORT_TXT:\n",
    "    out = _as_file(REPORT_TXT, 'cursor-audit.txt')\n",
    "    out.parent.mkdir(parents=True, exist_ok=True)\n",
    "    out.write_text(report(findings, scanned=n, root=target) + '\\n', encoding='utf-8')\n",
    "    print('wrote', out)\n",
    "\n",
    "if findings is not None and REPORT_JSON:\n",
    "    out = _as_file(REPORT_JSON, 'cursor-audit.json')\n",
    "    out.parent.mkdir(parents=True, exist_ok=True)\n",
    "    out.write_text(json.dumps(findings, indent=2), encoding='utf-8')\n",
    "    print('wrote', out)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Reading the results, and what this cannot see\n",
    "\n",
    "Each line is a file, a cell where relevant, a line number and the cursor class. For a notebook\n",
    "the line number is relative to the cell, matching Jupyter's own gutter.\n",
    "\n",
    "**The fix is almost always one of two things.** Create a fresh cursor for the second pass, or\n",
    "finish the work inside the original `with` block - most second passes exist only because the\n",
    "first pass did not collect enough.\n",
    "\n",
    "**Blind spots, before you trust an empty result:**\n",
    "\n",
    "* A cursor stored in a list, dict or attribute, passed into a function, or returned from one.\n",
    "* A cursor referenced from a *different scope* than the one that opened it - a nested function,\n",
    "  a lambda, or a `global`/`nonlocal` reference. Scope awareness removes false positives and buys\n",
    "  these false negatives; that is the trade.\n",
    "* Python embedded in `.atbx` or `.tbx` toolboxes - binary containers this does not open.\n",
    "* Anything depending on runtime values, since nothing here runs.\n",
    "\n",
    "An empty result means nothing was found *in the shapes this understands*. It is not a\n",
    "certification.\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "ArcGIS Pro",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "name": "python",
   "version": "3.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}