Ingest Reference

philanthropy.ingest

On-ramps from an upstream donor system to a PhilanthroPy donor-level feature table.

UniSchema: read_constituent_events loads UniSchema's JSON / NDJSON egress files; constituent_events_to_features aggregates them into the one-row-per-donor feature frame the estimators consume.

CiviCRM: read_civicrm_contributions loads a contribution export CSV; civicrm_contributions_to_features aggregates it the same way, dropping test-mode and non-Completed rows first.

civicrm_contributions_to_features(contributions, *, reference_date=None, statuses=('Completed',))

Aggregate a CiviCRM contribution log into donor-level features.

Parameters:

Name	Type	Description	Default
contributions	iterable of mapping, or DataFrame	CiviCRM contribution rows. Accepts the output of :func:read_civicrm_contributions, an APIv4 Contribution.get result, or a DataFrame of the same fields, under either the export labels or the APIv4 column names. contact_id, receive_date and total_amount are required; id, currency, contribution_status, financial_type, is_test, email, first_name and last_name are used when present.	required
reference_date	str or datetime - like	Anchor for the recency features (years_active, recency_days). If None, the latest receive_date in the batch is used; this keeps the aggregation reproducible and free of "now" leakage. Naive timestamps are interpreted as UTC.	None
statuses	sequence of str or None	Contribution statuses to count, matched case-insensitively against contribution_status. None disables the filter and counts every row, including Failed and Refunded ones. Test-mode rows are dropped either way.	``("Completed",)``

Returns:

Name	Type	Description
features	DataFrame	One row per donor, indexed by contact_id, with the columns declared in _FEATURE_DTYPES. recency_days, gift_count and total_gift_amount are the R, F and M of an RFM model. Rows are sorted by contact_id for determinism.

Raises:

Type	Description
KeyError	If contact_id, receive_date or total_amount is absent. An export missing one of them cannot be aggregated at all, and failing here names the field instead of surfacing a bare column lookup later.

Warns:

Type	Description
UserWarning	If statuses was requested but the batch carries no contribution_status column: the filter silently counting refunded and failed gifts is exactly the error this bridge exists to prevent.
UserWarning	If the batch mixes currencies. total_gift_amount is a plain sum with no FX conversion, so a single-currency export is assumed.

Examples:

>>> rows = [
...     {"Contact ID": "101", "Contribution Date": "2025-01-15",
...      "Total Amount": "250.00", "Contribution Status": "Completed"},
...     {"Contact ID": "101", "Contribution Date": "2025-06-01",
...      "Total Amount": "1,000.00", "Contribution Status": "Completed"},
...     {"Contact ID": "101", "Contribution Date": "2025-06-02",
...      "Total Amount": "99.00", "Contribution Status": "Failed"},
... ]
>>> feats = civicrm_contributions_to_features(rows)
>>> float(feats.loc["101", "total_gift_amount"])
1250.0
>>> int(feats.loc["101", "gift_count"])
2

Source code in philanthropy/ingest/_civicrm.py

def civicrm_contributions_to_features(
    contributions: Union[Iterable[Mapping], pd.DataFrame],
    *,
    reference_date: Optional[Union[str, pd.Timestamp]] = None,
    statuses: Optional[Sequence[str]] = ("Completed",),
) -> pd.DataFrame:
    """Aggregate a CiviCRM contribution log into donor-level features.

    Parameters
    ----------
    contributions : iterable of mapping, or DataFrame
        CiviCRM contribution rows. Accepts the output of
        :func:`read_civicrm_contributions`, an APIv4 ``Contribution.get``
        result, or a DataFrame of the same fields, under either the export
        labels or the APIv4 column names. ``contact_id``, ``receive_date`` and
        ``total_amount`` are required; ``id``, ``currency``,
        ``contribution_status``, ``financial_type``, ``is_test``, ``email``,
        ``first_name`` and ``last_name`` are used when present.
    reference_date : str or datetime-like, optional
        Anchor for the recency features (``years_active``, ``recency_days``).
        If ``None``, the latest ``receive_date`` in the batch is used; this
        keeps the aggregation reproducible and free of "now" leakage. Naive
        timestamps are interpreted as UTC.
    statuses : sequence of str or None, default=``("Completed",)``
        Contribution statuses to count, matched case-insensitively against
        ``contribution_status``. ``None`` disables the filter and counts every
        row, including ``Failed`` and ``Refunded`` ones. Test-mode rows are
        dropped either way.

    Returns
    -------
    features : pandas.DataFrame
        One row per donor, indexed by ``contact_id``, with the columns declared
        in ``_FEATURE_DTYPES``. ``recency_days``, ``gift_count`` and
        ``total_gift_amount`` are the R, F and M of an RFM model. Rows are
        sorted by ``contact_id`` for determinism.

    Raises
    ------
    KeyError
        If ``contact_id``, ``receive_date`` or ``total_amount`` is absent. An
        export missing one of them cannot be aggregated at all, and failing
        here names the field instead of surfacing a bare column lookup later.

    Warns
    -----
    UserWarning
        If ``statuses`` was requested but the batch carries no
        ``contribution_status`` column: the filter silently counting refunded
        and failed gifts is exactly the error this bridge exists to prevent.
    UserWarning
        If the batch mixes currencies. ``total_gift_amount`` is a plain sum with
        no FX conversion, so a single-currency export is assumed.

    Examples
    --------
    >>> rows = [
    ...     {"Contact ID": "101", "Contribution Date": "2025-01-15",
    ...      "Total Amount": "250.00", "Contribution Status": "Completed"},
    ...     {"Contact ID": "101", "Contribution Date": "2025-06-01",
    ...      "Total Amount": "1,000.00", "Contribution Status": "Completed"},
    ...     {"Contact ID": "101", "Contribution Date": "2025-06-02",
    ...      "Total Amount": "99.00", "Contribution Status": "Failed"},
    ... ]
    >>> feats = civicrm_contributions_to_features(rows)
    >>> float(feats.loc["101", "total_gift_amount"])
    1250.0
    >>> int(feats.loc["101", "gift_count"])
    2
    """
    df = _normalise_headers(_to_frame(contributions))
    if df.empty:
        return _empty_feature_frame()

    missing = [col for col in _REQUIRED if col not in df.columns]
    if missing:
        raise KeyError(
            f"CiviCRM contribution log is missing {missing}. Export the "
            f"'Contact ID', 'Contribution Date' and 'Total Amount' fields "
            f"(APIv4: contact_id, receive_date, total_amount); got "
            f"{sorted(df.columns)}."
        )

    df = df.copy()

    # Two rows sharing a CiviCRM contribution id are the same contribution;
    # concatenating overlapping monthly exports is how that happens. Collapse
    # only rows with a real id: pandas' duplicated() treats NaN == NaN, which
    # would drop every id-less gift once any row carries one.
    if "id" in df.columns:
        df = df[~(df["id"].notna() & df.duplicated(subset="id"))]

    if "is_test" in df.columns:
        df = df[~_to_bool(df["is_test"])]

    if statuses is not None:
        if "contribution_status" in df.columns:
            wanted = {str(s).strip().casefold() for s in statuses}
            status = df["contribution_status"].astype("string").str.strip().str.casefold()
            df = df[status.isin(wanted)]
        else:
            warnings.warn(
                f"CiviCRM contribution log has no contribution_status column, so "
                f"statuses={tuple(statuses)!r} could not be applied: Pending, "
                f"Failed, Refunded and Cancelled gifts (if any) are counted in "
                f"total_gift_amount. Add the 'Contribution Status' field to the "
                f"export, or pass statuses=None to accept every row.",
                stacklevel=2,
            )

    # total_gift_amount is a plain sum; CiviCRM carries a per-row `currency` but
    # the export has no FX rates, so a mixed-currency batch would add apples to
    # oranges. Warn rather than convert (rates aren't there) or crash.
    if "currency" in df.columns and df["currency"].dropna().nunique() > 1:
        warnings.warn(
            "CiviCRM contribution log mixes currencies "
            f"({sorted(df['currency'].dropna().unique())}); total_gift_amount is "
            "summed without FX conversion. Normalise to one currency first.",
            stacklevel=2,
        )

    df["_contact_id"] = df["contact_id"].astype("string").str.strip()
    df["_ts"] = _to_datetime(df["receive_date"])
    df["_amount"] = _to_amount(df["total_amount"]).fillna(0.0)
    # A gift we cannot attribute to a donor, or place in time, contributes to no
    # feature; drop it rather than let a NaT poison a donor's recency.
    df = df[
        df["_contact_id"].notna()
        & (df["_contact_id"].str.len() > 0)
        & df["_ts"].notna()
    ]
    if df.empty:
        return _empty_feature_frame()

    ref = _resolve_reference_date(reference_date, df["_ts"])

    grouped = df.groupby("_contact_id", sort=True)
    out = pd.DataFrame(index=grouped.size().index)
    # Optional identity fields: carried through when the export supplies them
    # (this is a donor-level table keyed by CRM id, not a de-identified feature
    # store). ``.first()`` skips nulls, so a donor whose name rode in on only
    # some rows still resolves. Absent column -> None.
    for out_col, src in (
        ("constituent_email", "email"),
        ("first_name", "first_name"),
        ("last_name", "last_name"),
    ):
        out[out_col] = grouped[src].first() if src in df.columns else None
    out["total_gift_amount"] = grouped["_amount"].sum()
    out["gift_count"] = grouped.size()
    out["largest_gift_amount"] = grouped["_amount"].max()
    out["first_gift_date"] = grouped["_ts"].min()
    out["last_gift_date"] = grouped["_ts"].max()
    out["years_active"] = (
        (ref - out["first_gift_date"]).dt.days / DAYS_PER_YEAR
    ).clip(lower=0.0)
    out["recency_days"] = (ref - out["last_gift_date"]).dt.days.clip(lower=0)

    # Financial Type is CiviCRM's gift classification (Donation, Member Dues,
    # Event Fee, ...); breadth across them is the CiviCRM analogue of the
    # UniSchema bridge's distinct_source_systems.
    type_col = next(
        (c for c in ("financial_type", "financial_type_id") if c in df.columns), None
    )
    out["distinct_financial_types"] = grouped[type_col].nunique() if type_col else 0

    out.index.name = "contact_id"
    return _coerce_schema(out)

read_civicrm_contributions(path)

Read CiviCRM contribution export CSV(s) into one normalised frame.

Accepts either a single .csv file or a directory, which is walked recursively and whose *.csv files are concatenated in sorted relative-path order, the shape you get from keeping a folder of monthly exports. Symlinks are not followed.

Every column is read as text and the headers are normalised to the APIv4 spelling, so "Total Amount", "Contact ID" and "Contribution Date" arrive as total_amount, contact_id and receive_date. Nothing else is done to the rows: test-mode and non-Completed contributions are still present, and it is :func:civicrm_contributions_to_features that drops them and types the values. Reading is deliberately lossless so the raw export stays inspectable.

Parameters:

Name	Type	Description	Default
path	str or Path	CSV file, or a directory of them.	required

Returns:

Name	Type	Description
contributions	DataFrame	The export as written, with normalised column names and text values. A directory holding no CSV returns an empty frame.

Raises:

Type	Description
FileNotFoundError	If path does not exist. Without this an absent directory falls through to the single-file branch and surfaces as an opaque OSError.

Source code in philanthropy/ingest/_civicrm.py

def read_civicrm_contributions(path: Union[str, Path]) -> pd.DataFrame:
    """Read CiviCRM contribution export CSV(s) into one normalised frame.

    Accepts either a single ``.csv`` file or a directory, which is walked
    **recursively** and whose ``*.csv`` files are concatenated in sorted
    relative-path order, the shape you get from keeping a folder of monthly
    exports. Symlinks are not followed.

    Every column is read as text and the headers are normalised to the APIv4
    spelling, so ``"Total Amount"``, ``"Contact ID"`` and ``"Contribution Date"``
    arrive as ``total_amount``, ``contact_id`` and ``receive_date``. Nothing else
    is done to the rows: **test-mode and non-``Completed`` contributions are
    still present**, and it is
    :func:`civicrm_contributions_to_features` that drops them and types the
    values. Reading is deliberately lossless so the raw export stays inspectable.

    Parameters
    ----------
    path : str or pathlib.Path
        CSV file, or a directory of them.

    Returns
    -------
    contributions : pandas.DataFrame
        The export as written, with normalised column names and text values.
        A directory holding no CSV returns an empty frame.

    Raises
    ------
    FileNotFoundError
        If ``path`` does not exist. Without this an absent directory falls
        through to the single-file branch and surfaces as an opaque OSError.
    """
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(f"No such file or directory: {p}")
    if p.is_dir():
        files = [
            f
            for f in p.rglob("*.csv")
            # Skip symlinks: one inside the export folder pointing outside it
            # must not be followed and read (path-traversal hardening).
            if f.is_file() and not f.is_symlink()
        ]
        if not files:
            return pd.DataFrame()
        # Sort by relative path so ordering is deterministic across platforms.
        frames = [
            _read_csv(f) for f in sorted(files, key=lambda f: f.relative_to(p).as_posix())
        ]
        return pd.concat(frames, ignore_index=True)
    return _read_csv(p)

constituent_events_to_features(events, *, reference_date=None, deduplicate=True)

Aggregate a UniSchema ConstituentEvent stream into donor features.

Parameters:

Name	Type	Description	Default
events	iterable of mapping, or DataFrame	Records following UniSchema's ConstituentEvent schema. Each event carries at least constituentEmail, eventType, sourceSystem, and createdAt (ISO-8601); amount, externalConstituentId, eventId, firstName, and lastName are optional. Accepts the output of :func:read_constituent_events, a list of dicts, or a DataFrame of the same fields.	required
reference_date	str or datetime - like	Anchor for the recency features (years_active, recency_days). If None, the latest createdAt in the batch is used; this keeps the aggregation reproducible and free of "now" leakage. Naive timestamps are interpreted as UTC.	None
deduplicate	bool	Drop repeated eventId values before aggregating. Advancement webhooks are at-least-once, so this prevents a redelivered donation from being counted (and its dollars summed) twice.	True

Returns:

Name	Type	Description
features	DataFrame	One row per constituent, indexed by constituent_id (the externalConstituentId when present, else constituentEmail), with the columns declared in _FEATURE_DTYPES (first_name / last_name are populated from the feed when available, else null). Rows are sorted by constituent_id for determinism.

Warns:

Type	Description
UserWarning	If the batch mixes currencies (more than one distinct currency). total_gift_amount is a plain sum with no FX conversion, so a single-currency feed is assumed; normalise upstream if it isn't.

Examples:

>>> events = [
...     {"constituentEmail": "a@x.edu", "eventType": "DONATION",
...      "sourceSystem": "GIVECAMPUS", "amount": 250.0,
...      "createdAt": "2025-03-01T12:00:00Z"},
...     {"constituentEmail": "a@x.edu", "eventType": "EVENT_REGISTRATION",
...      "sourceSystem": "CVENT", "createdAt": "2025-06-01T09:00:00Z"},
... ]
>>> feats = constituent_events_to_features(events)
>>> float(feats.loc["a@x.edu", "total_gift_amount"])
250.0
>>> int(feats.loc["a@x.edu", "event_attendance_count"])
1

Source code in philanthropy/ingest/_constituent_events.py

def constituent_events_to_features(
    events: Union[Iterable[Mapping], pd.DataFrame],
    *,
    reference_date: Optional[Union[str, pd.Timestamp]] = None,
    deduplicate: bool = True,
) -> pd.DataFrame:
    """Aggregate a UniSchema ``ConstituentEvent`` stream into donor features.

    Parameters
    ----------
    events : iterable of mapping, or DataFrame
        Records following UniSchema's ``ConstituentEvent`` schema.  Each event
        carries at least ``constituentEmail``, ``eventType``, ``sourceSystem``,
        and ``createdAt`` (ISO-8601); ``amount``, ``externalConstituentId``,
        ``eventId``, ``firstName``, and ``lastName`` are optional.  Accepts the
        output of
        :func:`read_constituent_events`, a list of dicts, or a DataFrame of the
        same fields.
    reference_date : str or datetime-like, optional
        Anchor for the recency features (``years_active``, ``recency_days``).
        If ``None``, the latest ``createdAt`` in the batch is used; this keeps
        the aggregation reproducible and free of "now" leakage.  Naive
        timestamps are interpreted as UTC.
    deduplicate : bool, default=True
        Drop repeated ``eventId`` values before aggregating.  Advancement
        webhooks are at-least-once, so this prevents a redelivered donation from
        being counted (and its dollars summed) twice.

    Returns
    -------
    features : pandas.DataFrame
        One row per constituent, indexed by ``constituent_id`` (the
        ``externalConstituentId`` when present, else ``constituentEmail``), with
        the columns declared in ``_FEATURE_DTYPES`` (``first_name`` /
        ``last_name`` are populated from the feed when available, else null).
        Rows are sorted by ``constituent_id`` for determinism.

    Warns
    -----
    UserWarning
        If the batch mixes currencies (more than one distinct ``currency``).
        ``total_gift_amount`` is a plain sum with no FX conversion, so a
        single-currency feed is assumed; normalise upstream if it isn't.

    Examples
    --------
    >>> events = [
    ...     {"constituentEmail": "a@x.edu", "eventType": "DONATION",
    ...      "sourceSystem": "GIVECAMPUS", "amount": 250.0,
    ...      "createdAt": "2025-03-01T12:00:00Z"},
    ...     {"constituentEmail": "a@x.edu", "eventType": "EVENT_REGISTRATION",
    ...      "sourceSystem": "CVENT", "createdAt": "2025-06-01T09:00:00Z"},
    ... ]
    >>> feats = constituent_events_to_features(events)
    >>> float(feats.loc["a@x.edu", "total_gift_amount"])
    250.0
    >>> int(feats.loc["a@x.edu", "event_attendance_count"])
    1
    """
    df = _to_frame(events)
    if df.empty:
        return _empty_feature_frame()

    # total_gift_amount sums raw amounts; UniSchema carries a per-event
    # `currency` but no FX rates, so a mixed-currency feed would sum apples and
    # oranges. Warn rather than convert (rates aren't in the stream) or crash.
    if "currency" in df.columns and df["currency"].dropna().nunique() > 1:
        warnings.warn(
            "ConstituentEvent feed mixes currencies "
            f"({sorted(df['currency'].dropna().unique())}); total_gift_amount "
            "is summed without FX conversion. Normalise to one currency first.",
            stacklevel=2,
        )

    if deduplicate and "eventId" in df.columns:
        # Collapse only rows that share a real eventId.  A missing eventId is not
        # "equal" to another missing one, but pandas' drop_duplicates treats
        # NaN == NaN, which would silently drop every id-less donation (and even
        # whole donors) once any event carries an id.
        is_dup = df["eventId"].notna() & df.duplicated(subset="eventId")
        df = df[~is_dup]

    df = df.copy()
    df["_constituent_id"] = _constituent_id(df)
    df["_ts"] = _to_utc_naive(df["createdAt"])
    # An event we can't place in time can't contribute to time-aware features;
    # drop it rather than let a NaT poison a donor's recency to a crash.
    df = df[df["_ts"].notna()]
    if df.empty:
        return _empty_feature_frame()

    # fillna("") keeps a missing/None eventType from becoming pd.NA in the
    # comparisons below (np.where on a NA-bearing mask raises); such an event
    # simply matches no type and contributes to no typed count.
    event_type = (
        df.get("eventType", pd.Series(index=df.index, dtype="object"))
        .astype("string")
        .fillna("")
    )
    raw_amount = df["amount"] if "amount" in df.columns else pd.Series(np.nan, index=df.index)
    # ponytail: UniSchema guarantees `amount` is a JSON number, so a missing
    # amount coerces to 0 (a real absent gift); an unparseable string would too.
    # Parse currency strings ('$250', '1,250.00') here only if a non-conforming
    # feed is ever fed in directly, bypassing UniSchema's validation.
    amount = pd.to_numeric(raw_amount, errors="coerce")

    is_donation = (event_type == _DONATION).to_numpy()
    df["_gift_amount"] = np.where(is_donation, amount.fillna(0.0).to_numpy(), 0.0)
    df["_is_donation"] = is_donation.astype("int64")
    df["_is_event"] = (event_type == _EVENT_REGISTRATION).astype("int64")
    df["_is_click"] = (event_type == _EMAIL_CLICK).astype("int64")
    df["_donation_ts"] = df["_ts"].where(is_donation)

    ref = _resolve_reference_date(reference_date, df["_ts"])

    grouped = df.groupby("_constituent_id", sort=True)
    out = pd.DataFrame(index=grouped.size().index)
    out["constituent_email"] = grouped["constituentEmail"].first()
    # Optional identity fields: carried through when the feed supplies them
    # (the output is a donor-level table, not a de-identified feature store, so
    # it already holds constituent_email). ``.first()`` skips nulls, so a donor
    # whose name rode in on only some events still resolves. Absent column -> None.
    for out_col, src in (("first_name", "firstName"), ("last_name", "lastName")):
        out[out_col] = grouped[src].first() if src in df.columns else None
    out["total_gift_amount"] = grouped["_gift_amount"].sum()
    out["gift_count"] = grouped["_is_donation"].sum()
    out["event_attendance_count"] = grouped["_is_event"].sum()
    out["email_click_count"] = grouped["_is_click"].sum()
    out["first_gift_date"] = grouped["_donation_ts"].min()
    out["last_gift_date"] = grouped["_donation_ts"].max()

    first_seen = grouped["_ts"].min()
    last_seen = grouped["_ts"].max()
    out["years_active"] = ((ref - first_seen).dt.days / DAYS_PER_YEAR).clip(lower=0.0)
    out["recency_days"] = (ref - last_seen).dt.days.clip(lower=0)

    if "sourceSystem" in df.columns:
        out["distinct_source_systems"] = grouped["sourceSystem"].nunique()
    else:
        out["distinct_source_systems"] = 0

    out.index.name = "constituent_id"
    return _coerce_schema(out)

read_constituent_events(path)

Read UniSchema egress files into a list of ConstituentEvent dicts.

Handles the shapes UniSchema's egress writes:

• a single .json file holding one event (object) or many (array);
• a .ndjson / .jsonl batch, one event per line;
• a directory, which is walked recursively: every *.json, *.ndjson, and *.jsonl file at any depth is read and concatenated, sorted by relative path. This handles UniSchema's date-partitioned egress ({prefix}/{vendor}/{yyyy}/{mm}/{dd}/{eventId}.json); a flat directory still works too. *.manifest.json batch sidecars are skipped.

Parameters:

Name	Type	Description	Default
path	str or Path	File or directory to read.	required

Returns:

Name	Type	Description
events	list of dict	Parsed events, ready to pass to :func:constituent_events_to_features.

Raises:

Type	Description
FileNotFoundError	If path does not exist. Without this an absent directory falls through to the single-file branch and surfaces as an opaque OSError.

Source code in philanthropy/ingest/_constituent_events.py

def read_constituent_events(
    path: Union[str, Path],
) -> "list[dict]":
    """Read UniSchema egress files into a list of ``ConstituentEvent`` dicts.

    Handles the shapes UniSchema's egress writes:

    * a single ``.json`` file holding one event (object) or many (array);
    * a ``.ndjson`` / ``.jsonl`` batch, one event per line;
    * a directory, which is walked **recursively**: every ``*.json``,
      ``*.ndjson``, and ``*.jsonl`` file at any depth is read and concatenated,
      sorted by relative path.  This handles UniSchema's date-partitioned egress
      (``{prefix}/{vendor}/{yyyy}/{mm}/{dd}/{eventId}.json``); a flat directory
      still works too.  ``*.manifest.json`` batch sidecars are skipped.

    Parameters
    ----------
    path : str or pathlib.Path
        File or directory to read.

    Returns
    -------
    events : list of dict
        Parsed events, ready to pass to :func:`constituent_events_to_features`.

    Raises
    ------
    FileNotFoundError
        If ``path`` does not exist.  Without this an absent directory falls
        through to the single-file branch and surfaces as an opaque OSError.
    """
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(f"No such file or directory: {p}")
    if p.is_dir():
        # UniSchema's local egress is date-partitioned: it writes each event to
        # {prefix}/{vendor}/{yyyy}/{mm}/{dd}/{eventId}.json (see UniSchema
        # src/egress/objectKey.ts), so the files sit several levels down and a
        # non-recursive scan of the top dir finds nothing. Walk the whole tree.
        events: "list[dict]" = []
        files = [
            f
            for f in p.rglob("*")
            if f.is_file()
            # Skip symlinks: a symlink inside the egress tree pointing outside
            # it must not be followed and read (path-traversal hardening).
            and not f.is_symlink()
            and f.suffix.lower() in {".json", ".ndjson", ".jsonl"}
            # Skip S3 batch sidecars: batch metadata, not ConstituentEvents.
            and not f.name.lower().endswith(".manifest.json")
        ]
        # Sort by relative path so ordering is deterministic across platforms.
        for child in sorted(files, key=lambda f: f.relative_to(p).as_posix()):
            events.extend(_read_events_file(child))
        return events
    return _read_events_file(p)