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. The adcp.types.generated_poc → adcp.types.domains import-path rename is applied the same way (#1417): the generated tree moved there with its module stems unchanged, and the old path returns the same objects, so the rewrite is mechanical. The one stem that split, brand, is resolved per imported name against the two halves it became (domains.brand_discovery for brand.json's classes, domains.brand for the task-schema aggregator) and flagged when it cannot be.
• Flagged: removed types and numbered Assets<N> 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 and the generated_poc → domains import paths in place. --auto-apply implies --apply and additionally lifts a generated_poc import to adcp.types when every name on the line is a known public alias there (the flat surface is the first choice; a domain path is for a name it cannot bind), and rewrites 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" | "rename_import", "path": str, "line": int,
     "column": int, "before": str, "after": str, "hint": str | 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), and so is the rename_import kind in applied (#1417): the adcp.types.generated_poc → adcp.types.domains import-path rename, reported with the old module path in before and the new one in after. 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, "
            "rename adcp.types.generated_poc imports to adcp.types.domains, "
            "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: the <Type>Asset renames and the "
            "adcp.types.generated_poc → adcp.types.domains import paths. "
            "Default is dry-run (report only). Commit your tree first so "
            "`git diff` is your review view. See also --auto-apply."
        ),
    )
    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: a generated_poc import "
            "whose every symbol is bound on adcp.types is imported from "
            "there instead of the domain path, 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
    rename_import_hits = False  # any generated_poc → domains path renames 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:
                module = from_match.group("module") or ""
                raw_syms = [s.strip() for s in from_match.group("symbols").split(",")]
                pre_syms = [r.split(" as ")[0].strip() for r in raw_syms if r.strip()]
                if pre_syms and not all(
                    _generated_symbol_replacement(module, symbol) is not None
                    or symbol in NUMBERED_ASSETS_RENAMES
                    for symbol in pre_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 references (#1417).
        #
        # Under ``--auto-apply``, a single-line
        #   ``from adcp.types.generated_poc.<path> import <symbols>``
        # whose every symbol is a known public alias (or a mapped numbered
        # asset) is lifted to ``adcp.types`` — one ``auto_applied`` Finding
        # per symbol. Every other reference is a prefix rename to
        # ``adcp.types.domains`` (``rename_import``, applied by ``--apply``),
        # except a bare split stem (``generated_poc.brand``) whose names
        # cannot be placed, which is flagged with the two halves named.
        #
        # Numbered-Assets imports appear on these lines too; the numbered
        # pass above owns their Findings.
        if DEPRECATED_TYPES_ROOT in line:
            from_match = _GENERATED_POC_FROM_IMPORT.search(line)
            lifted = False
            if from_match is not None and auto_apply and not line_is_mixed_unsafe_import:
                module = from_match.group("module") or ""
                _raws, symbols = _import_symbols(from_match.group("symbols"))
                lifted = bool(symbols)
                for symbol in symbols:
                    replacement = _generated_symbol_replacement(module, symbol)
                    if replacement is None:
                        continue  # a mapped numbered asset; the numbered pass reported it
                    sym_col = line.find(symbol, from_match.start(1)) + 1
                    findings.append(
                        Finding(
                            kind="auto_applied",
                            path=str(path),
                            line=lineno,
                            column=(
                                sym_col if sym_col > 0 else line.index(DEPRECATED_TYPES_ROOT) + 1
                            ),
                            before=symbol,
                            after=replacement,
                            hint=(
                                "deprecated path — import "
                                f"{replacement.rsplit('.', 1)[-1]} from "
                                "adcp.types (stable public API) instead"
                            ),
                        )
                    )
                    auto_apply_hits = True
            if not lifted:
                for finding in _prefix_rename_findings(path, lineno, line):
                    findings.append(finding)
                    if finding.kind == "rename_import":
                        rename_import_hits = True

        # 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
                module = m.group("module") or ""
                raw_syms = [s.strip() for s in m.group("symbols").split(",")]
                pre_syms = [r.split(" as ")[0].strip() for r in raw_syms if r.strip()]
                replacements = [
                    _generated_symbol_replacement(module, symbol)
                    or (
                        f"adcp.types.{NUMBERED_ASSETS_RENAMES[symbol]}"
                        if symbol in NUMBERED_ASSETS_RENAMES
                        else None
                    )
                    for symbol in pre_syms
                ]
                if replacements and all(replacement is not None for replacement in replacements):
                    public_imports: list[str] = []
                    for raw, symbol, replacement in zip(
                        raw_syms, pre_syms, replacements, strict=True
                    ):
                        assert replacement is not None
                        public_name = replacement.rsplit(".", 1)[-1]
                        local_name = raw.split(" as ", 1)[1].strip() if " as " in raw else None
                        if local_name is not None:
                            public_imports.append(f"{public_name} as {local_name}")
                        elif symbol in NUMBERED_ASSETS_RENAMES:
                            public_imports.append(public_name)
                        elif public_name != symbol:
                            # Preserve existing use sites while moving the import
                            # to the semantic public name.
                            public_imports.append(f"{public_name} as {symbol}")
                        else:
                            public_imports.append(public_name)
                    replacement_import = "from adcp.types import " + ", ".join(public_imports)
                    text_line = text_line[: m.start()] + replacement_import + text_line[m.end() :]
                # 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 apply_changes and rename_import_hits and DEPRECATED_TYPES_ROOT in updated:
        # After any lift to ``adcp.types`` above, every remaining
        # ``generated_poc`` reference is a prefix rename (or a flagged split
        # stem, which ``_prefix_rename_line`` leaves alone).
        updated = "".join(
            _prefix_rename_line(text_line) if DEPRECATED_TYPES_ROOT in text_line else text_line
            for text_line in updated.splitlines(keepends=True)
        )
        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" | "rename_import" | "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 in ("rename", "rename_import"):
            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 in ("rename", "rename_import"):
        self.applied.append(finding)
    elif finding.kind == "auto_applied":
        self.auto_applied.append(finding)
    else:
        self.flagged.append(finding)