Module adcp.migrate.v3_to_v4

v3 → v4 migration for the AdCP SDK.

The spec redesign in 4.0 renamed the 9 <Type>Asset payload classes to <Type>Content and removed several legacy types (BrandManifest, DeliverTo, Pricing, PromotedProducts, PromotedOfferings, FormatCategory, PackageStatus). This module does the mechanical rewrites and prints a structured report of everything that still needs human attention.

Two kinds of findings:

• Applied: direct name rewrites (AudioAsset → AudioContent etc). The 9 rename targets are distinctive enough that word-boundary regex is safe; sellers should still review the diff.
• Flagged: removed types, numbered Assets<N> imports, adcp.types.generated_poc imports. These don't rewrite — the seller has to choose the replacement (e.g. BrandManifest → BrandReference(domain=...) depends on call-site context).

Invocation::

python -m adcp.migrate v3-to-v4 ./src               # dry run, report only
python -m adcp.migrate v3-to-v4 ./src --apply       # rewrite files in place
python -m adcp.migrate v3-to-v4 ./src --auto-apply  # also rewrite safe imports
python -m adcp.migrate v3-to-v4 ./src --json        # structured report

The dry run is the default — you always see what would change before anything moves. --apply rewrites the 9 <Type>Asset renames in place. --auto-apply implies --apply and additionally rewrites flag_private findings whose target symbol is a known public alias in adcp.types, and flag_numbered findings with a documented semantic alias (Assets81 → VideoFormatAsset, etc.). flag_removed findings always require human review and remain flagged even with --auto-apply. Commit your tree before running either write mode so git diff is your review view.

Global variables

var REPORT_SCHEMA_VERSION

Version of the JSON report shape. CI scripts / editors parsing the migrate output key on this so a future shape change (adding a summary block, renaming fields) doesn't silently break them.

Bump the minor SDK version AND this constant when changing the JSON shape in a non-additive way. Additive changes (new optional keys) stay at the same version.

v1 shape:

.. code-block:: json

{
  "schema_version": 1,
  "scanned_files": int,
  "rewritten_files": int,
  "applied": [
    {"kind": "rename", "path": str, "line": int, "column": int,
     "before": str, "after": str, "hint": null, "migration_anchor": null}
  ],
  "auto_applied": [
    {"kind": "auto_applied", "path": str, "line": int, "column": int,
     "before": str, "after": str, "hint": str | null, "migration_anchor": null}
  ],
  "flagged": [
    {"kind": "flag_removed" | "flag_numbered" | "flag_private"
             | "flag_attribute" | "flag_enum_value",
     "path": str, "line": int, "column": int, "before": str,
     "after": str | null, "hint": str | null, "migration_anchor": str | null}
  ]
}

auto_applied is an additive field (v1, no version bump needed). Parsers that don't know about it receive an empty array in non---auto-apply runs and can safely ignore it. Entries in flagged always require human attention regardless of what auto_applied contains.

Functions

def main(argv: list[str] | None = None) ‑> int

Expand source code

def main(argv: list[str] | None = None) -> int:
    """CLI entry point for ``python -m adcp.migrate v3-to-v4``."""
    parser = argparse.ArgumentParser(
        prog="adcp.migrate v3-to-v4",
        description=(
            "Rewrite adcp 3.x → 4.0 ``<Type>Asset`` → ``<Type>Content`` renames "
            "and flag usages of removed types. "
            "Exits 0 when all findings are mechanical (or none); "
            "exits 1 when flag_removed findings remain for human review; "
            "exits 2 on usage errors."
        ),
    )
    parser.add_argument(
        "path",
        type=Path,
        help="File or directory to scan (source tree root in typical use).",
    )
    parser.add_argument(
        "--apply",
        action="store_true",
        help=(
            "Rewrite files in place. Default is dry-run (report only). "
            "Commit your tree first so `git diff` is your review view. "
            "See also --auto-apply to also mechanically fix flag_private "
            "and flag_numbered findings."
        ),
    )
    parser.add_argument(
        "--auto-apply",
        action="store_true",
        dest="auto_apply",
        help=(
            "Rewrite files in place (implies --apply) and additionally "
            "auto-apply safe import rewrites: flag_private findings "
            "whose target symbol exists on adcp.types, and flag_numbered "
            "findings with a documented semantic alias (Assets81 → "
            "VideoFormatAsset, etc.). flag_removed findings always "
            "require human review and remain flagged; exit code 1 when "
            "any remain."
        ),
    )
    parser.add_argument(
        "--allow-dirty",
        action="store_true",
        help=(
            "Allow --apply / --auto-apply even when the git working tree "
            "has uncommitted changes. Default is to refuse so `git diff` "
            "after the migration shows only the codemod's rewrites, "
            "not a mix of the seller's in-progress work and the "
            "codemod. Pass --allow-dirty when you know what you're "
            "doing (e.g. applying to a staged change deliberately)."
        ),
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Emit structured JSON report instead of the human-readable text.",
    )
    args = parser.parse_args(argv)

    # --auto-apply implies --apply; treat them uniformly downstream.
    if args.auto_apply:
        args.apply = True

    if not args.path.exists():
        print(f"error: path does not exist: {args.path}", file=sys.stderr)
        return 2

    if args.apply and not args.allow_dirty and _is_dirty_tree(args.path):
        flag_used = "--auto-apply" if args.auto_apply else "--apply"
        print(
            f"error: {flag_used} refused on a dirty git working tree.\n"
            "       Commit your changes first so `git diff` after the\n"
            "       migration shows only the codemod's rewrites. Pass\n"
            "       --allow-dirty to override (e.g. you're deliberately\n"
            "       applying on top of staged changes).",
            file=sys.stderr,
        )
        return 2

    report = run(args.path, apply_changes=args.apply, auto_apply=args.auto_apply)

    if args.json:
        print(_format_json_report(report))
    else:
        print(_format_text_report(report, apply_changes=args.apply, auto_apply=args.auto_apply))

    # Return non-zero when there are manual-review findings so CI can
    # gate on a clean report. Applied/auto-applied rewrites alone don't
    # trip the gate — they're mechanical and apply cleanly.
    return 1 if report.flagged else 0

CLI entry point for python -m adcp.migrate v3-to-v4.

def run(root: Path, *, apply_changes: bool = False, auto_apply: bool = False) ‑> Report

Expand source code

def run(root: Path, *, apply_changes: bool = False, auto_apply: bool = False) -> Report:
    """Execute the migration across ``root``. Returns a :class:`Report`."""
    report = Report()
    for path in _iter_python_files(root):
        report.scanned_files += 1
        findings, new_contents = scan_file(path, apply_changes=apply_changes, auto_apply=auto_apply)
        for f in findings:
            report.add(f)
        if new_contents is not None:
            # newline="" preserves whatever line endings were read
            # (including mixed — unusual but possible). Pair with the
            # ``open(..., newline="")`` read in ``scan_file``.
            with open(path, "w", encoding="utf-8", newline="") as fh:
                fh.write(new_contents)
            report.rewritten_files += 1
    return report

Execute the migration across root. Returns a :class:Report.

def scan_file(path: Path, *, apply_changes: bool, auto_apply: bool = False) ‑> tuple[list[Finding], str | None]

Expand source code

def scan_file(
    path: Path, *, apply_changes: bool, auto_apply: bool = False
) -> tuple[list[Finding], str | None]:
    """Scan one file. Returns (findings, new_contents_or_None).

    new_contents_or_None is None when apply_changes=False or when no
    renames fired; the caller uses it as the signal to rewrite.

    ``auto_apply=True`` promotes safe findings to ``kind="auto_applied"``
    in the returned list, but file rewrites only happen when
    ``apply_changes=True`` as well — ``auto_apply`` alone never writes.

    Reads with ``utf-8-sig`` so UTF-8-BOM-prefixed source files (legal
    Python, common on Windows) migrate correctly. Uses ``newline=""``
    on read and write so CRLF line endings are preserved verbatim —
    Windows sellers otherwise get a giant noise diff where every line
    flips to LF.
    """
    findings: list[Finding] = []
    try:
        # Use ``open(..., newline="")`` over ``Path.read_text(newline=)``
        # — the latter was added in 3.13 but the SDK supports 3.10+.
        with open(path, encoding="utf-8-sig", newline="") as fh:
            original = fh.read()
    except (UnicodeDecodeError, OSError):
        # Skip unreadable or non-UTF8 files; migration targets Python source.
        return findings, None

    # Detect renames per-line so the report carries column info and the
    # same pattern that matched detection also drives the rewrite.
    updated = original
    rename_hits = False
    auto_apply_hits = False  # any numbered or private-import rewrites queued

    for lineno, line in enumerate(original.splitlines(), start=1):
        # Pre-pass: when this line is a single-line ``generated_poc``
        # import, decide whether the line as a whole is auto-apply-safe.
        # An import is *unsafe* when at least one of its symbols (after
        # the hypothetical numbered substitution) isn't in
        # ``_AUTO_APPLY_PUBLIC_SYMBOLS``; rewriting one symbol while
        # leaving another behind would leave the line importing a
        # public name from a private module — guaranteed ImportError.
        # The rewrite block (`updated.splitlines()` later) skips
        # unsafe-mixed lines; the per-symbol Finding emission below
        # also treats numbered references on those lines as
        # ``flag_numbered`` rather than ``auto_applied`` so the report
        # matches the file content.
        line_is_mixed_unsafe_import = False
        if "adcp.types.generated_poc" in line:
            from_match = _GENERATED_POC_FROM_IMPORT.search(line)
            if from_match:
                raw_syms = [s.strip() for s in from_match.group(1).split(",")]
                pre_syms = [r.split(" as ")[0].strip() for r in raw_syms if r.strip()]
                post_syms = [NUMBERED_ASSETS_RENAMES.get(s, s) for s in pre_syms]
                if pre_syms and not all(s in _AUTO_APPLY_PUBLIC_SYMBOLS for s in post_syms):
                    line_is_mixed_unsafe_import = True

        for old, new in ASSET_CONTENT_RENAMES.items():
            for match in _RENAME_PATTERNS[old].finditer(line):
                findings.append(
                    Finding(
                        kind="rename",
                        path=str(path),
                        line=lineno,
                        column=match.start() + 1,
                        before=old,
                        after=new,
                    )
                )
                rename_hits = True

        # Removed types — flagged, not rewritten.
        for name, (hint, anchor) in REMOVED_TYPES.items():
            for match in _REMOVED_PATTERNS[name].finditer(line):
                findings.append(
                    Finding(
                        kind="flag_removed",
                        path=str(path),
                        line=lineno,
                        column=match.start() + 1,
                        before=name,
                        hint=hint,
                        migration_anchor=anchor,
                    )
                )

        # Numbered Assets imports / references.
        for match in NUMBERED_ASSETS_PATTERN.finditer(line):
            symbol = match.group(0)
            alias = NUMBERED_ASSETS_RENAMES.get(symbol)
            if auto_apply and alias is not None and not line_is_mixed_unsafe_import:
                findings.append(
                    Finding(
                        kind="auto_applied",
                        path=str(path),
                        line=lineno,
                        column=match.start() + 1,
                        before=symbol,
                        after=alias,
                    )
                )
                auto_apply_hits = True
            else:
                findings.append(
                    Finding(
                        kind="flag_numbered",
                        path=str(path),
                        line=lineno,
                        column=match.start() + 1,
                        before=symbol,
                        after=alias,  # hint toward the public alias even in flag mode
                        hint=(
                            "numbered Assets classes are unstable across spec revisions; "
                            "import the semantic alias from adcp.types instead"
                        ),
                        migration_anchor="numbered-discriminated-union-classes-shifted",
                    )
                )

        # adcp.types.generated_poc imports.
        #
        # When the line is a single-line
        #   ``from adcp.types.generated_poc.<path> import <symbols>``
        # emit one per-symbol Finding.  For symbols in
        # GENERATED_POC_SYMBOL_MAP the Finding carries the public alias;
        # unknown symbols get the generic "private module" flag so they
        # still surface (fixing the prior silent-drop on mixed lines).
        #
        # Under ``--auto-apply`` the per-symbol findings are promoted to
        # ``auto_applied`` only when ALL symbols on the line are in the
        # map — a mixed line cannot be safely rewritten without splitting
        # the import statement, so it remains flagged.
        #
        # Numbered-Assets imports (e.g. ``import Assets81``) appear here
        # too.  Those are handled by the numbered pass above and will be
        # fixed by the post-scan import-path rewrite; suppress the extra
        # generic flag so the report doesn't double-count them.
        for private_path, hint in PRIVATE_IMPORT_PATHS.items():
            if private_path not in line:
                continue
            col = line.index(private_path) + 1
            from_match = _GENERATED_POC_FROM_IMPORT.search(line)
            if from_match:
                raw_symbols = [s.strip() for s in from_match.group(1).split(",")]
                parsed: list[tuple[str, str | None]] = []
                for raw in raw_symbols:
                    raw = raw.strip()
                    if not raw:
                        continue
                    symbol = raw.split(" as ")[0].strip()
                    if not symbol:
                        continue
                    parsed.append((symbol, GENERATED_POC_SYMBOL_MAP.get(symbol)))

                if not parsed:
                    findings.append(
                        Finding(
                            kind="flag_private",
                            path=str(path),
                            line=lineno,
                            column=col,
                            before=private_path,
                            hint=hint,
                        )
                    )
                    continue

                all_known = all(
                    repl is not None or (auto_apply and symbol in NUMBERED_ASSETS_RENAMES)
                    for symbol, repl in parsed
                )

                for symbol, replacement in parsed:
                    sym_col = line.find(symbol, from_match.start(1)) + 1
                    if sym_col <= 0:
                        sym_col = col
                    if replacement is not None:
                        kind = "auto_applied" if (auto_apply and all_known) else "flag_private"
                        if kind == "auto_applied":
                            auto_apply_hits = True
                        findings.append(
                            Finding(
                                kind=kind,
                                path=str(path),
                                line=lineno,
                                column=sym_col,
                                before=symbol,
                                after=replacement,
                                hint=(
                                    f"private module — import {symbol} from "
                                    "adcp.types (stable public API) instead"
                                ),
                            )
                        )
                    else:
                        # Unknown symbol.  Suppress the generic flag when
                        # auto_apply is active and the symbol is a numbered
                        # asset that will be renamed by the other pass —
                        # the import-path fix covers it.
                        if auto_apply and symbol in NUMBERED_ASSETS_RENAMES:
                            continue
                        findings.append(
                            Finding(
                                kind="flag_private",
                                path=str(path),
                                line=lineno,
                                column=sym_col,
                                before=private_path,
                                hint=hint,
                            )
                        )
            else:
                # Multiline import, star import, or regex mismatch.
                findings.append(
                    Finding(
                        kind="flag_private",
                        path=str(path),
                        line=lineno,
                        column=col,
                        before=private_path,
                        hint=hint,
                    )
                )

        # Removed attribute accesses (.brand_manifest etc.). Regex with
        # trailing word boundary prevents false-positives on
        # ``.brand_manifest_v2``, ``.brand_manifest_override``, etc.
        for attr, hint in REMOVED_ATTRIBUTE_ACCESSES.items():
            for match in _REMOVED_ATTRIBUTE_PATTERNS[attr].finditer(line):
                findings.append(
                    Finding(
                        kind="flag_attribute",
                        path=str(path),
                        line=lineno,
                        column=match.start() + 1,
                        before=attr,
                        hint=hint,
                    )
                )

        # Removed enum values (e.g. MediaBuyStatus.pending_activation). The
        # class-qualified form is anchored tightly enough that false positives
        # are unlikely; trailing word boundary prevents suffix matches like
        # ``MediaBuyStatus.pending_activation_v2``.
        for enum_val, (enum_hint, enum_anchor) in REMOVED_ENUM_VALUES.items():
            for match in _REMOVED_ENUM_VALUE_PATTERNS[enum_val].finditer(line):
                findings.append(
                    Finding(
                        kind="flag_enum_value",
                        path=str(path),
                        line=lineno,
                        column=match.start() + 1,
                        before=enum_val,
                        hint=enum_hint,
                        migration_anchor=enum_anchor,
                    )
                )

    needs_write = False

    if apply_changes and rename_hits:
        for old, new in ASSET_CONTENT_RENAMES.items():
            updated = _RENAME_PATTERNS[old].sub(new, updated)
        needs_write = True

    if apply_changes and auto_apply and auto_apply_hits:
        # Process the file line-by-line so generated_poc imports get a
        # safety check against the post-numbered-substitution symbol set
        # before any rewrite happens. The earlier "Step 1: substitute
        # Assets<N> file-wide; Step 2: fix import paths only when safe"
        # ordering corrupted mixed lines like
        # ``from generated_poc.core.format import Assets81, Assets149``
        # — Assets81 became VideoFormatAsset while Assets149 stayed,
        # leaving VideoFormatAsset imported from a private module.
        new_lines: list[str] = []
        for text_line in updated.splitlines(keepends=True):
            is_generated_poc_import = (
                "adcp.types.generated_poc" in text_line
                and _GENERATED_POC_FROM_IMPORT.search(text_line) is not None
            )
            if is_generated_poc_import:
                m = _GENERATED_POC_FROM_IMPORT.search(text_line)
                assert m is not None  # narrowed above
                raw_syms = [s.strip() for s in m.group(1).split(",")]
                pre_syms = [r.split(" as ")[0].strip() for r in raw_syms if r.strip()]
                # Apply the hypothetical numbered rename to each symbol
                # so we can check if the *post-rename* symbol set is
                # safe.
                post_syms = [NUMBERED_ASSETS_RENAMES.get(s, s) for s in pre_syms]
                if post_syms and all(s in _AUTO_APPLY_PUBLIC_SYMBOLS for s in post_syms):
                    # Whole import is safe — substitute numbered names
                    # AND fix the module path.
                    for old, new in NUMBERED_ASSETS_RENAMES.items():
                        text_line = _NUMBERED_RENAME_PATTERNS[old].sub(new, text_line)
                    text_line = _GENERATED_POC_MODULE_RE.sub("from adcp.types import", text_line)
                # Mixed line — leave it alone. The findings list still
                # carries the per-symbol flag_private and flag_numbered
                # entries so the adopter sees the work to do.
                new_lines.append(text_line)
                continue
            # Non-import lines: substitute numbered names freely (the
            # semantic alias is already importable via adcp.types and
            # any local reference the line carries is a usage site).
            for old, new in NUMBERED_ASSETS_RENAMES.items():
                text_line = _NUMBERED_RENAME_PATTERNS[old].sub(new, text_line)
            new_lines.append(text_line)
        updated = "".join(new_lines)
        needs_write = True

    if needs_write:
        return findings, updated
    return findings, None

Scan one file. Returns (findings, new_contents_or_None).

new_contents_or_None is None when apply_changes=False or when no renames fired; the caller uses it as the signal to rewrite.

auto_apply=True promotes safe findings to kind="auto_applied" in the returned list, but file rewrites only happen when apply_changes=True as well — auto_apply alone never writes.

Reads with utf-8-sig so UTF-8-BOM-prefixed source files (legal Python, common on Windows) migrate correctly. Uses newline="" on read and write so CRLF line endings are preserved verbatim — Windows sellers otherwise get a giant noise diff where every line flips to LF.

Classes

class Finding (kind: str,
path: str,
line: int,
column: int,
before: str,
after: str | None = None,
hint: str | None = None,
migration_anchor: str | None = None)

Expand source code

@dataclass
class Finding:
    """One migration finding — either an applied rename or a manual TODO."""

    # Valid kind values: "rename" | "auto_applied" | "flag_removed" |
    #   "flag_private" | "flag_numbered" | "flag_attribute" |
    #   "flag_enum_value"
    kind: str
    path: str
    line: int
    column: int
    before: str
    after: str | None = None  # None for flag-only items
    hint: str | None = None
    migration_anchor: str | None = None

One migration finding — either an applied rename or a manual TODO.

Instance variables

var after : str | None

var before : str

var column : int

var hint : str | None

var kind : str

var line : int

var migration_anchor : str | None

var path : str

class Report (applied: list[Finding] = <factory>,
auto_applied: list[Finding] = <factory>,
flagged: list[Finding] = <factory>,
scanned_files: int = 0,
rewritten_files: int = 0)

Expand source code

@dataclass
class Report:
    """Structured migration report."""

    applied: list[Finding] = field(default_factory=list)
    auto_applied: list[Finding] = field(default_factory=list)
    flagged: list[Finding] = field(default_factory=list)
    scanned_files: int = 0
    rewritten_files: int = 0

    def add(self, finding: Finding) -> None:
        if finding.kind == "rename":
            self.applied.append(finding)
        elif finding.kind == "auto_applied":
            self.auto_applied.append(finding)
        else:
            self.flagged.append(finding)

Structured migration report.

Instance variables

var applied : list[Finding]

var auto_applied : list[Finding]

var flagged : list[Finding]

var rewritten_files : int

var scanned_files : int

Methods

def add(self,
finding: Finding) ‑> None

Expand source code

def add(self, finding: Finding) -> None:
    if finding.kind == "rename":
        self.applied.append(finding)
    elif finding.kind == "auto_applied":
        self.auto_applied.append(finding)
    else:
        self.flagged.append(finding)