Helpers

zavod.helpers

Data cleaning and entity generation helpers.

This module contains a number of functions that are useful for parsing real-world data (like XML, CSV, date formats) and converting it into FollowTheMoney entity structures. Factory methods are provided for handling common entity patterns as a way to reduce boilerplate code and improve consistency across datasets.

A typical use might look like this:

from zavod import Context
from zavod import helpers as h

def crawl(context: Context) -> None:
    # ... fetch some data
    for row in data:
        entity = context.make("Person")
        entity.id = context.make_id(row.get("id"))
        # Using the helper guarantees a consistent handling of the
        # attributes, and in this case will also automatically
        # generate a full name for the entity:
        h.apply_name(
            entity,
            first_name=row.get("first_name"),
            patronymic=row.get("patronymic"),
            last_name=row.get("last_name"),
            title=row.get("title"),
        )
        context.emit(entity)

Any data wrangling code that is repeated in three or more crawlers should be considered for inclusion in the helper library.

apply_address(context, entity, address)

Link the given entity to the given address and emits the address.

Parameters:

Name	Type	Description	Default
context	Context	The runner context used for emitting entities.	required
entity	Entity	The thing located at the given address.	required
address	Entity | None	The address entity, usually constructed with make_address.	required

Source code in zavod/helpers/addresses.py

def apply_address(context: Context, entity: Entity, address: Entity | None) -> None:
    """Link the given entity to the given address and emits the address.

    Args:
        context: The runner context used for emitting entities.
        entity: The thing located at the given address.
        address: The address entity, usually constructed with `make_address`.
    """
    if address is None:
        return
    assert address.schema.is_a("Address"), "address must be an Address"
    assert entity.schema.get("addressEntity") is not None, (
        "Entity must have addressEntity"
    )
    entity.add("country", address.get("country"))
    if address.has("full"):
        entity.add("addressEntity", address)
        context.emit(address)
        entity.add("address", address.get("full"))

apply_date(entity, prop, text, formats=None, original_value=None)

Apply a date value to an entity, parsing it if necessary and cleaning it up.

Uses the dates configuration of the dataset to parse the date.

Parameters:

Name	Type	Description	Default
entity	Entity	The entity to which the date will be applied.	required
prop	str	The property to which the date will be applied.	required
text	DateValue	The date value to be applied.	required
formats	tuple[str] | None	A list of date formats to use for parsing, overriding dataset defaults.	None
original_value	str | None	If provided, recorded as the entity's original value for this property instead of text. Use when text has already been transformed.	None

Source code in zavod/helpers/dates.py

def apply_date(
    entity: Entity,
    prop: str,
    text: DateValue,
    formats: tuple[str] | None = None,
    original_value: str | None = None,
) -> None:
    """Apply a date value to an entity, parsing it if necessary and cleaning it up.

    Uses the `dates` configuration of the dataset to parse the date.

    Args:
        entity: The entity to which the date will be applied.
        prop: The property to which the date will be applied.
        text: The date value to be applied.
        formats: A list of date formats to use for parsing, overriding dataset defaults.
        original_value: If provided, recorded as the entity's original value for
            this property instead of ``text``. Use when ``text`` has already been
            transformed.
    """
    prop_ = entity.schema.get(prop)
    if prop_ is None or prop_.type != registry.date:
        log.warning(f"Property is not a date: {prop}", text=text)
        return

    if not isinstance(text, str):
        text = stringify(text)
    if text is None:
        return None

    if original_value is None:
        original_value = text
    dates = extract_date(entity.dataset, text, formats=formats)
    return entity.add(prop_, dates, original_value=original_value)

apply_dates(entity, prop, texts)

Apply a list of date values to an entity, parsing them if necessary and cleaning them up.

Parameters:

Name	Type	Description	Default
entity	Entity	The entity to which the date will be applied.	required
prop	str	The property to which the date will be applied.	required
texts	Iterable[DateValue]	The iterable of date values to be applied.	required

Source code in zavod/helpers/dates.py

def apply_dates(entity: Entity, prop: str, texts: Iterable[DateValue]) -> None:
    """Apply a list of date values to an entity, parsing them if necessary and cleaning them up.

    Args:
        entity: The entity to which the date will be applied.
        prop: The property to which the date will be applied.
        texts: The iterable of date values to be applied.
    """
    for text in texts:
        apply_date(entity, prop, text)

apply_name(entity, full=None, name1=None, first_name=None, given_name=None, name2=None, second_name=None, middle_name=None, name3=None, patronymic=None, matronymic=None, name4=None, name5=None, tail_name=None, last_name=None, maiden_name=None, prefix=None, suffix=None, alias=False, name_prop='name', is_weak=False, quiet=False, lang=None, origin=None)

A standardised way to set a name for a person or other entity, which handles normalising the categories of names found in source data to the correct properties (e.g. "family name" becomes "lastName").

Parameters:

Name	Type	Description	Default
entity	Entity	The entity to set the name on.	required
full	str | None	The full name if available (this will otherwise be generated).	None
name1	str | None	The first name if numeric parts are used.	None
first_name	str | None	The first name.	None
given_name	str | None	The given name (also first name).	None
name2	str | None	The second name if numeric parts are used.	None
second_name	str | None	The second name.	None
middle_name	str | None	The middle name.	None
name3	str | None	The third name if numeric parts are used.	None
patronymic	str | None	The patronymic (father-derived) name.	None
matronymic	str | None	The matronymic (mother-derived) name.	None
name4	str | None	The fourth name if numeric parts are used.	None
name5	str | None	The fifth name if numeric parts are used.	None
tail_name	str | None	A secondary last name.	None
last_name	str | None	The last/family name name.	None
maiden_name	str | None	The maiden name (before marriage).	None
prefix	str | None	A prefix to the name (e.g. "Mr").	None
suffix	str | None	A suffix to the name (e.g. "Jr").	None
alias	bool	If this is an alias name.	False
name_prop	str	The property to set the full name on.	'name'
is_weak	bool	If this is a weak alias name.	False
quiet	bool	If this should not raise errors on invalid properties.	False
lang	str | None	The language of the name.	None
origin	str | None	The origin of the name (e.g. a GPT model).	None

Source code in zavod/helpers/names.py

def apply_name(
    entity: Entity,
    full: str | None = None,
    name1: str | None = None,
    first_name: str | None = None,
    given_name: str | None = None,
    name2: str | None = None,
    second_name: str | None = None,
    middle_name: str | None = None,
    name3: str | None = None,
    patronymic: str | None = None,
    matronymic: str | None = None,
    name4: str | None = None,
    name5: str | None = None,
    tail_name: str | None = None,
    last_name: str | None = None,
    maiden_name: str | None = None,
    prefix: str | None = None,
    suffix: str | None = None,
    alias: bool = False,
    name_prop: str = "name",
    is_weak: bool = False,
    quiet: bool = False,
    lang: str | None = None,
    origin: str | None = None,
) -> None:
    """A standardised way to set a name for a person or other entity, which handles
    normalising the categories of names found in source data to the correct properties
    (e.g. "family name" becomes "lastName").

    Args:
        entity: The entity to set the name on.
        full: The full name if available (this will otherwise be generated).
        name1: The first name if numeric parts are used.
        first_name: The first name.
        given_name: The given name (also first name).
        name2: The second name if numeric parts are used.
        second_name: The second name.
        middle_name: The middle name.
        name3: The third name if numeric parts are used.
        patronymic: The patronymic (father-derived) name.
        matronymic: The matronymic (mother-derived) name.
        name4: The fourth name if numeric parts are used.
        name5: The fifth name if numeric parts are used.
        tail_name: A secondary last name.
        last_name: The last/family name name.
        maiden_name: The maiden name (before marriage).
        prefix: A prefix to the name (e.g. "Mr").
        suffix: A suffix to the name (e.g. "Jr").
        alias: If this is an alias name.
        name_prop: The property to set the full name on.
        is_weak: If this is a weak alias name.
        quiet: If this should not raise errors on invalid properties.
        lang: The language of the name.
        origin: The origin of the name (e.g. a GPT model).
    """
    if not is_weak:
        set_name_part(entity, "firstName", given_name, quiet, lang, origin)
        set_name_part(entity, "firstName", first_name, quiet, lang, origin)
        set_name_part(entity, "secondName", second_name, quiet, lang, origin)
        set_name_part(entity, "middleName", middle_name, quiet, lang, origin)
        set_name_part(entity, "fatherName", patronymic, quiet, lang, origin)
        set_name_part(entity, "motherName", matronymic, quiet, lang, origin)
        set_name_part(entity, "lastName", last_name, quiet, lang, origin)
        set_name_part(entity, "lastName", maiden_name, quiet, lang, origin)
        set_name_part(entity, "firstName", name1, quiet, lang, origin)
        set_name_part(entity, "secondName", name2, quiet, lang, origin)
        set_name_part(entity, "middleName", name3, quiet, lang, origin)
        set_name_part(entity, "middleName", name4, quiet, lang, origin)
        set_name_part(entity, "middleName", name5, quiet, lang, origin)
        set_name_part(entity, "lastName", tail_name, quiet, lang, origin)
    if alias:
        name_prop = "alias"
    if is_weak:
        name_prop = "weakAlias"

    # Provenance for full names created from parts
    full_origin = origin
    if full is None or len(full) == 0:
        full_origin = ORIGIN_INFERRED
    full = make_name(
        full=full,
        name1=name1,
        first_name=first_name,
        given_name=given_name,
        name2=name2,
        second_name=second_name,
        middle_name=middle_name,
        name3=name3,
        patronymic=patronymic,
        matronymic=matronymic,
        name4=name4,
        name5=name5,
        tail_name=tail_name,
        last_name=last_name,
        prefix=prefix,
        suffix=suffix,
    )
    if full is not None and len(full):
        entity.add(name_prop, full, quiet=quiet, lang=lang, origin=full_origin)

apply_number(entity, prop, value, origin=None)

Apply a numeric value to a property of an entity. This will try and parse the number, round it, and normalize the present unit specifier (e.g. km, tons) if present.

Parameters:

Name	Type	Description	Default
entity	Entity	The entity to which the property belongs.	required
prop	str	The property to which the value will be applied.	required
value	NumberValue	The numeric value to apply.	required
origin	str | None	An optional origin for the value.	None

Source code in zavod/helpers/numbers.py

def apply_number(
    entity: Entity, prop: str, value: NumberValue, origin: str | None = None
) -> None:
    """Apply a numeric value to a property of an entity. This will try and parse the
    number, round it, and normalize the present unit specifier (e.g. km, tons) if present.

    Args:
        entity: The entity to which the property belongs.
        prop: The property to which the value will be applied.
        value: The numeric value to apply.
        origin: An optional origin for the value.
    """
    prop_obj = entity.schema.get(prop)
    assert prop_obj and prop_obj.type == registry.number
    if isinstance(value, str):
        if not len(value.strip()):
            return
        num, unit = registry.number.parse(
            value,
            decimal=entity.dataset.numbers.decimal,
            separator=entity.dataset.numbers.separator,
        )
        if num is None:
            log.warning(f"Cannot parse number: {value}")
            return
        try:
            num = _float_str(float(num))
        except ValueError:
            log.warning("Cannot convert number to float: %s", num)
            return
        if unit is not None:
            unit = normalize_unit(unit)
            text = f"{num} {unit}"
        else:
            text = num
    elif isinstance(value, float):
        text = _float_str(value)
    elif isinstance(value, Decimal):
        text = f"{value:.2f}"
    else:
        text = str(value)
    entity.unsafe_add(
        prop_obj,
        text,
        cleaned=True,
        original_value=str(value),
        origin=origin,
    )

apply_reviewed_name_string(context, entity, *, string, original_prop='name', lang=None, llm_cleaning=False)

Clean the name(s) in the provided string if needed, then post them for review.

Cleaned names are applied to an entity if accepted, potentially to a different property from 'original_prop' if cleaning proposed an alternative which was accepted or modified in review.

Unaccepted reviews result in the name being applied to 'original_prop'.

Also falls back to 'original_prop' with a warning if llm_cleaning is True but the LLM service is not configured.

Parameters:

Name	Type	Description	Default
context	Context	The current context.	required
entity	Entity	The entity to apply names to.	required
string	str | None	The raw name(s) string.	required
original_prop	str	The original property for the name according to the data source. Must be one of the Names model fields, e.g. "name" or "alias".	'name'
lang	str | None	The language of the name, if known.	None
llm_cleaning	bool	Whether to use LLM-based name cleaning.	False

Source code in zavod/helpers/names.py

def apply_reviewed_name_string(
    context: Context,
    entity: Entity,
    *,
    string: str | None,
    original_prop: str = "name",
    lang: str | None = None,
    llm_cleaning: bool = False,
) -> None:
    """
    Clean the name(s) in the provided string if needed, then post them for review.

    Cleaned names are applied to an entity if accepted, potentially to a different
    property from 'original_prop' if cleaning proposed an alternative which was accepted
    or modified in review.

    Unaccepted reviews result in the name being applied to 'original_prop'.

    Also falls back to 'original_prop' with a warning if llm_cleaning is True
    but the LLM service is not configured.

    Args:
        context: The current context.
        entity: The entity to apply names to.
        string: The raw name(s) string.
        original_prop: The original property for the name according to the data source.
            Must be one of the ``Names`` model fields, e.g. "name" or "alias".
        lang: The language of the name, if known.
        llm_cleaning: Whether to use LLM-based name cleaning.
    """
    # Names tolerates unknown keys on validation (stored review payloads may carry
    # them), so a typo'd prop would otherwise silently produce an empty Names and
    # the entity would be emitted without any name.
    if original_prop not in Names.model_fields:
        raise ValueError(
            f"Invalid original_prop {original_prop!r}. "
            f"Expected one of: {', '.join(sorted(Names.model_fields))}"
        )
    original = Names(**{original_prop: string})

    apply_reviewed_names(
        context,
        entity,
        original=original,
        suggested=None,
        lang=lang,
        llm_cleaning=llm_cleaning,
    )

apply_reviewed_names(context, entity, *, original, suggested=None, is_irregular=False, lang=None, llm_cleaning=False, default_accepted=False)

Determines whether names need cleaning and if so, posts them for review.

If 'suggested' is not supplied, 'check_names_regularity' is used to determine if cleaning or review is needed, and potentially suggest categorisation.

Names are considered to have been pre-determined to need cleaning/review if 'is_irregular' is passed as True, or if 'suggested' is supplied and differs from 'original'. Crawlers that do their own suggestions should normally do those on the result of check_names_regularity, so that its suggestions don't override the crawler's suggestions.

If 'llm_cleaning' is True, an LLM-based cleaning step is additionally done on 'suggested' if provided, otherwise on 'original', before posting for review. Any categorisation in 'original' and 'suggested' is disregarded and left to the LLM to determine. This can not be used with crawler-supplied suggestions and, and heuristic suggestions are not passed to the LLM.

Parameters:

Name	Type	Description	Default
context	Context	The current context.	required
entity	Entity	The entity to apply names to.	required
original	Names	The original string(s) and their categorisation according to the data source.	required
suggested	Names | None	Optional suggestion of different categorisation of names.	None
lang	str | None	The language of the name, if known.	None
llm_cleaning	bool	Whether to use LLM-based name cleaning.	False
default_accepted	bool	Marks the review as accepted from the start, if one is created.	False

Source code in zavod/helpers/names.py

def apply_reviewed_names(
    context: Context,
    entity: Entity,
    *,
    original: Names,
    suggested: Names | None = None,
    is_irregular: bool = False,
    lang: str | None = None,
    llm_cleaning: bool = False,
    default_accepted: bool = False,
) -> None:
    """
    Determines whether names need cleaning and if so, posts them for review.

    If 'suggested' is not supplied, 'check_names_regularity' is used to determine
    if cleaning or review is needed, and potentially suggest categorisation.

    Names are considered to have been pre-determined to need cleaning/review if
    'is_irregular' is passed as True, or if 'suggested'
    is supplied and differs from 'original'. Crawlers that do their own suggestions
    should normally do those on the result of check_names_regularity, so that
    its suggestions don't override the crawler's suggestions.

    If 'llm_cleaning' is True, an LLM-based cleaning step is additionally done
    on 'suggested' if provided, otherwise on 'original', before posting for review.
    Any categorisation in 'original' and 'suggested' is disregarded and left to the LLM
    to determine. This can not be used with crawler-supplied suggestions and,
    and heuristic suggestions are not passed to the LLM.

    Args:
        context: The current context.
        entity: The entity to apply names to.
        original: The original string(s) and their categorisation according to the data source.
        suggested: Optional suggestion of different categorisation of names.
        lang: The language of the name, if known.
        llm_cleaning: Whether to use LLM-based name cleaning.
        default_accepted: Marks the review as accepted from the start, if one is created.
    """
    review = review_names(
        context,
        entity,
        original=original,
        suggested=suggested,
        is_irregular=is_irregular,
        llm_cleaning=llm_cleaning,
        default_accepted=default_accepted,
    )

    if review is None or not review.accepted:
        apply_names(entity, original=original, names=original, lang=lang)
        return

    apply_names(
        entity,
        original=original,
        names=review.extracted_data,
        lang=lang,
        origin=review.origin,
    )

assert_dom_hash(node, hash, raise_exc=False, text_only=False)

Assert that a DOM node has a given SHA1 hash.

Source code in zavod/helpers/change.py

def assert_dom_hash(
    node: ElementOrTree | None,
    hash: str,
    raise_exc: bool = False,
    text_only: bool = False,
) -> bool:
    """Assert that a DOM node has a given SHA1 hash."""
    actual = _compute_node_hash(node, text_only=text_only)
    if actual != hash:
        if raise_exc:
            msg = f"Expected hash {hash}, got {actual} for {node!r}"
            raise AssertionError(msg)
        else:
            log.warning(
                f"DOM hash changed: {node}",
                expected=hash,
                actual=actual,
                node=repr(node),
            )
        return False
    return True

assert_file_hash(path, hash, raise_exc=False)

Assert that a file has a given SHA1 hash.

Source code in zavod/helpers/change.py

def assert_file_hash(
    path: Path,
    hash: str,
    raise_exc: bool = False,
) -> bool:
    """Assert that a file has a given SHA1 hash."""
    digest = sha1()
    with open(path, "rb") as fh:
        digest.update(fh.read())
    actual = digest.hexdigest()
    if actual != hash:
        if raise_exc:
            msg = f"Expected hash {hash}, got {actual} for {path.name}"
            raise AssertionError(msg)
        else:
            log.warning(
                f"File hash changed: {path.name}",
                expected=hash,
                actual=actual,
                name=path.name,
            )
        return False
    return True

assert_html_url_hash(context, url, hash, path=None, raise_exc=False, text_only=False)

Assert that an HTML document located at the URL has a given SHA1 hash.

Source code in zavod/helpers/change.py

def assert_html_url_hash(
    context: Context,
    url: str,
    hash: str,
    path: str | None = None,
    raise_exc: bool = False,
    text_only: bool = False,
) -> bool:
    """Assert that an HTML document located at the URL has a given SHA1 hash."""
    doc = context.fetch_html(url)
    node = doc.find(path) if path is not None else doc
    return assert_dom_hash(node, hash, raise_exc=raise_exc, text_only=text_only)

assert_url_hash(context, url, hash, raise_exc=False, auth=None, headers=None)

Assert that a document located at the URL has a given SHA1 hash.

Source code in zavod/helpers/change.py

def assert_url_hash(
    context: Context,
    url: str,
    hash: str,
    raise_exc: bool = False,
    auth: Any | None = None,
    headers: Any | None = None,
) -> bool:
    """Assert that a document located at the URL has a given SHA1 hash."""
    digest = sha1()
    with context.http.get(url, auth=auth, headers=headers, stream=True) as res:
        res.raise_for_status()
        for chunk in res.iter_content(chunk_size=8192 * 10):
            digest.update(chunk)
    actual = digest.hexdigest()
    if actual != hash:
        if raise_exc:
            msg = f"Expected hash {hash}, got {actual} for {url}"
            raise AssertionError(msg)
        else:
            log.warning(
                f"URL hash changed: {url}",
                expected=hash,
                actual=actual,
                url=url,
            )
        return False
    return True

backdate(date, delta)

Return a partial ISO8601 date string backdated by the number of days provided

Source code in zavod/helpers/dates.py

def backdate(date: datetime, delta: timedelta) -> str:
    """Return a partial ISO8601 date string backdated by the number of days provided"""
    dt = date - delta
    return dt.isoformat()[:10]

cells_to_str(row)

Return the string value of each HtmlElement value in the passed dictionary

Useful when all you need is the string value of each cell in a table row.

Source code in zavod/helpers/html.py

def cells_to_str(row: dict[str, Element]) -> dict[str, str | None]:
    """
    Return the string value of each HtmlElement value in the passed dictionary

    Useful when all you need is the string value of each cell in a table row.
    """
    return {
        # Empty cells are None, not the empty string
        k: element_text(v) or None
        for k, v in row.items()
    }

check_name_regularity(entity, string)

Determine whether a name string potentially needs cleaning.

Source code in zavod/helpers/names.py

def check_name_regularity(entity: Entity, string: str | None) -> Regularity:
    """Determine whether a name string potentially needs cleaning."""
    string = squash_spaces(string or "")

    if not string:
        return Regularity(is_irregular=False)

    names_spec = entity.dataset.names
    spec = names_spec.get_spec(entity.schema)

    result = _check_suggesting_heuristics(entity, string, names_spec)
    if result is not None:
        return result

    if spec is not None:
        result = _check_schema_name_specs(string, spec)
        if result is not None:
            return result

    if contains_split_phrase(string):
        return Regularity(is_irregular=True)

    return Regularity(is_irregular=False)

check_names_regularity(entity, names)

Determine whether any name string in the given Names instance is irregular and needs cleaning.

Returns a tuple of a boolean indicating whether any name string is irregular, and a Names instance based on that supplied, with any heuristic-suggested categorisation adjustments applied (e.g. suggesting that a name be moved from "name" to "alias" or "weakAlias").

Source code in zavod/helpers/names.py

def check_names_regularity(entity: Entity, names: Names) -> tuple[bool, LangNames]:
    """
    Determine whether any name string in the given Names instance is irregular
    and needs cleaning.

    Returns a tuple of a boolean indicating whether any name string is irregular,
    and a Names instance based on that supplied, with any heuristic-suggested
    categorisation adjustments applied (e.g. suggesting that a name be moved
    from "name" to "alias" or "weakAlias").
    """
    is_irregular = False
    updated_suggested_data: dict[str, list[LangText]] = defaultdict(list)
    for key, names_values in names.as_langtexts():
        for name_val in names_values:
            regularity = check_name_regularity(entity, name_val.text)
            if regularity.is_irregular:
                is_irregular = True
            if regularity.suggested_prop is None:
                updated_suggested_data[key].append(name_val)
            else:
                updated_suggested_data[regularity.suggested_prop].append(name_val)
    updated_suggested = LangNames(**updated_suggested_data)
    return is_irregular, updated_suggested

clean_note(text)

Remove a set of specific text sections from notes supplied by sanctions data publishers. These include cross-references to the Security Council web site and the Interpol web site.

Parameters:

Name	Type	Description	Default
text	str | None | Sequence[str | None]	The note text from source	required

Returns:

Type	Description
list[str]	A cleaned version of the text.

Source code in zavod/helpers/text.py

def clean_note(text: str | None | Sequence[str | None]) -> list[str]:
    """Remove a set of specific text sections from notes supplied by sanctions data
    publishers. These include cross-references to the Security Council web site and
    the Interpol web site.

    Args:
        text: The note text from source

    Returns:
        A cleaned version of the text.
    """
    out: list[str] = []
    if text is None:
        return out
    if is_listish(text):
        for t in text:
            out.extend(clean_note(t))
        return out
    if isinstance(text, str):
        text = PREFIX.sub(" ", text)
        text = INTERPOL_URL.sub(" ", text)
        text = squash_spaces(text)
        if len(text) == 0:
            return out
        return [text]
    return out

convert_excel_cell(book, cell)

Convert an Excel cell to a string, handling different types.

Parameters:

Name	Type	Description	Default
book	Book	The Excel workbook.	required
cell	Cell	The Excel cell.	required

Returns:

Type	Description
str | None	The cell value as a string, or None if the cell is empty.

Source code in zavod/helpers/excel.py

def convert_excel_cell(book: Book, cell: Cell) -> str | None:
    """Convert an Excel cell to a string, handling different types.

    Args:
        book: The Excel workbook.
        cell: The Excel cell.

    Returns:
        The cell value as a string, or `None` if the cell is empty.
    """
    # https://xlrd.readthedocs.io/en/latest/api.html#xlrd.sheet.Cell
    if cell.ctype == XL_CELL_NUMBER:
        # Excel stores all numbers as floats; stringify keeps the fractional part.
        return stringify(cell.value)
    elif cell.ctype in (XL_CELL_EMPTY, XL_CELL_ERROR, XL_CELL_BLANK):
        return None
    if cell.ctype == XL_CELL_DATE:
        assert isinstance(cell.value, float)
        dt = xldate_as_datetime(cell.value, book.datemode)
        # Naive ISO 8601, e.g. "2023-07-26T00:00:00" — Excel cells carry no timezone.
        return dt.isoformat(sep="T", timespec="seconds")
    else:
        if cell.value is None:
            return None
        return str(cell.value)

convert_excel_date(value)

Convert an Excel date to a string.

Parameters:

Name	Type	Description	Default
value	str | int | float | None	The Excel date value (e.g. 44876).	required

Returns:

Type	Description
str | None	The date value as a string, or None if the value is empty.

Source code in zavod/helpers/excel.py

def convert_excel_date(value: str | int | float | None) -> str | None:
    """Convert an Excel date to a string.

    Args:
        value: The Excel date value (e.g. 44876).

    Returns:
        The date value as a string, or `None` if the value is empty.
    """
    if value is None:
        return None
    if isinstance(value, str):
        try:
            value = float(value)
        except ValueError:
            return None
    if isinstance(value, float):
        value = int(value)
    if value < 4_000 or value > 100_000:
        return None
    dt = datetime.fromordinal(datetime(1900, 1, 1).toordinal() + value - 2)
    # Naive ISO 8601, e.g. "2022-11-11T00:00:00" — Excel dates carry no timezone.
    return dt.isoformat(sep="T", timespec="seconds")

copy_address(entity, address)

Assign to full address text and country directly to the given entity.

This is an alternative to using apply_address when the address should be inlined into the entity, instead of emitting a separate address object.

Parameters:

Name	Type	Description	Default
entity	Entity	The entity to be assigned the address.	required
address	Entity | None	The address entity to be copied into the entity.	required

Source code in zavod/helpers/addresses.py

def copy_address(entity: Entity, address: Entity | None) -> None:
    """Assign to full address text and country directly to the given entity.

    This is an alternative to using `apply_address` when the address should
    be inlined into the entity, instead of emitting a separate address object.

    Args:
        entity: The entity to be assigned the address.
        address: The address entity to be copied into the entity.
    """
    if address is not None:
        for stmt in address.get_statements("full"):
            entity.adopt_statement(stmt, prop="address")
        for country in address.get("country"):
            if country not in entity.countries:
                for stmt in address.get_statements("country"):
                    entity.adopt_statement(stmt, prop="country")

deref_wikidata_id(context, qid, cache_days=60)

Check if a Wikidata QID is a redirect, and return the target QID if so.

This is used with static data sources that reference Wikidata items that may have been merged or redirected.

Parameters:

Name	Type	Description	Default
context	Context	The zavod context to use for fetching.	required
qid	str | None	The Wikidata QID to dereference.	required
cache_days	int	Number of days to cache the fetch result.	60

Returns:

Type	Description
str | None	The target QID if the input was a redirect, otherwise the original QID.

Source code in zavod/helpers/wikidata.py

def deref_wikidata_id(
    context: Context, qid: str | None, cache_days: int = 60
) -> str | None:
    """Check if a Wikidata QID is a redirect, and return the target QID if so.

    This is used with static data sources that reference Wikidata items that may have
    been merged or redirected.

    Args:
        context: The zavod context to use for fetching.
        qid: The Wikidata QID to dereference.
        cache_days: Number of days to cache the fetch result.

    Returns:
        The target QID if the input was a redirect, otherwise the original QID.
    """
    if qid is None or not is_qid(qid):
        return None

    try:
        params = {
            "format": "json",
            "ids": qid,
            "action": "wbgetentities",
            # "props": "info",
        }
        res = context.fetch_json(
            WikidataClient.WD_API,
            params=params,
            cache_days=cache_days,
        )
        entity = res.get("entities", {}).get(qid, {})
        target = entity.get("redirects", {}).get("to")
        if target is not None:
            context.log.info(f"Wikidata ID {qid} is a redirect to {target}")
            return str(target)
    except RequestException as exc:
        context.log.warning(f"Failed to dereference Wikidata ID {qid}: {exc}")
    return qid

earliest_term_start(topics=['gov.national'])

Returns a date that can be used as a cut-off date for parliamentary or government terms when crawling historical data. For example, if a dataset is known to include data from the inception of a country, but we only want to consider people as PEPs if they held a position within the last 20 years, we can use this function to determine the earliest term start date to consider when crawling.

The date is framed as a start date but should include sufficient slack to also be used as a filter on end dates if just those are available.

Parameters:

Name	Type	Description	Default
topics	list[str]	A list of topics to determine the earliest term start date for. For example, ["gov.national"] for national-level positions or ["gov.state"] for subnational positions. ["gov.diplo"] for international diplomatic positions. The default is ["gov.national"].	['gov.national']

Returns:

Type	Description
str	A date string in ISO format representing the earliest term start date to consider.

Source code in zavod/helpers/positions.py

def earliest_term_start(topics: list[str] = ["gov.national"]) -> str:
    """Returns a date that can be used as a cut-off date for parliamentary or government terms
    when crawling historical data. For example, if a dataset is known to include data from the
    inception of a country, but we only want to consider people as PEPs if they held a position
    within the last 20 years, we can use this function to determine the earliest term start date
    to consider when crawling.

    The date is framed as a start date but should include sufficient slack to also be used as a
    filter on end dates if just those are available.

    Args:
        topics: A list of topics to determine the earliest term start date for.
            For example, ["gov.national"] for national-level positions or ["gov.state"] for
            subnational positions. ["gov.diplo"] for international diplomatic positions.
            The default is ["gov.national"].

    Returns:
        A date string in ISO format representing the earliest term start date to consider.
    """
    after_office = get_after_office(topics)
    after_office = after_office + (DEFAULT_AFTER_OFFICE * 2)  # Add extra slack
    earliest_date = (settings.RUN_TIME - after_office).date().isoformat()
    return earliest_date

element_text(el, squash=True)

Return the text content of an HtmlElement, or an empty string if empty.

Parameters:

Name	Type	Description	Default
el	Element | None	The HTML element to extract text from.	required
squash	bool	Whether to squash whitespace and newlines in the text content.	True

Returns:

Type	Description
str	The text content of the element, or an empty string if empty.

Source code in zavod/helpers/html.py

def element_text(el: Element | None, squash: bool = True) -> str:
    """
    Return the text content of an HtmlElement, or an empty string if empty.

    Args:
        el: The HTML element to extract text from.
        squash: Whether to squash whitespace and newlines in the text content.

    Returns:
        The text content of the element, or an empty string if empty.
    """
    # Workaround because lxml-stubs doesn't yet support HtmlElement
    # https://github.com/lxml/lxml-stubs/pull/71
    if el is None:
        return ""
    try:
        text = str(cast(HtmlElement, el).text_content())
    except AttributeError:
        text = str(el.xpath("string()", smart_strings=False))

    if squash:
        text = squash_spaces(text)
    return text

element_text_hash(el)

Return a hash of the text content of an HtmlElement. Empty elements will return the sha1 of no data (da39a3ee5e6b4b0d3255bfef95601890afd80709).

Parameters:

Name	Type	Description	Default
el	Element	The HTML element to extract text from.	required

Returns:

Type	Description
str	A hash of the text content of the element

Source code in zavod/helpers/html.py

def element_text_hash(el: Element) -> str:
    """
    Return a hash of the text content of an HtmlElement. Empty elements will return the sha1
    of no data (`da39a3ee5e6b4b0d3255bfef95601890afd80709`).

    Args:
        el: The HTML element to extract text from.

    Returns:
        A hash of the text content of the element
    """
    text = element_text(el)
    return text_hash(text)

extract_cryptos(text)

Extract cryptocurrency addresses from text.

Parameters:

Name	Type	Description	Default
text	str | None	The text to extract from.	required

Returns:

Type	Description
dict[str, str]	A set of cryptocurrency IDs, with currency code.

Source code in zavod/helpers/crypto.py

def extract_cryptos(text: str | None) -> dict[str, str]:
    """Extract cryptocurrency addresses from text.

    Args:
        text: The text to extract from.

    Returns:
        A set of cryptocurrency IDs, with currency code.
    """
    out: dict[str, str] = {}
    if text is None:
        return out
    for currency, v in CRYPTOS_RE.items():
        for key in v.findall(text):
            out[key] = currency
    return out

extract_date(dataset, text, formats=None, fallback_to_original=True) cached

Extract a date from the provided text using predefined formats in the metadata. If the text doesn't match any format, returns the original text.

Source code in zavod/helpers/dates.py

@lru_cache(maxsize=5000)
def extract_date(
    dataset: Dataset,
    text: DateValue,
    formats: tuple[str] | None = None,
    fallback_to_original: bool = True,
) -> list[str]:
    """
    Extract a date from the provided text using predefined `formats` in the metadata.
    If the text doesn't match any format, returns the original text.
    """
    if text is None:
        return []
    if isinstance(text, date):
        return [text.isoformat()]
    elif isinstance(text, datetime):
        if text.tzinfo is not None:
            text = text.astimezone(UTC)
        iso = text.date().isoformat()
        return [iso]
    elif isinstance(text, str):
        text = text.strip()

    replaced_text = replace_months(dataset, text)
    dataset_formats_ = dataset.dates.formats + ALWAYS_FORMATS
    formats_ = dataset_formats_ if formats is None else list(formats)
    parsed = parse_formats(replaced_text, formats_)
    if parsed.text is not None:
        return [parsed.text]
    if dataset.dates.year_only:
        years = extract_years(text)
        if len(years):
            return years
    if fallback_to_original:
        return [text]
    raise ValueError(f"Invalid date: {text}")

extract_years(text)

Try to locate year numbers in a string such as 'circa 1990'. This will fail if any numbers that don't look like years are found in the string, a strong indicator that a more precise date is encoded (e.g. '1990 Mar 03').

This is bounded to years between 1800 and 2100.

Parameters:

Name	Type	Description	Default
text	str	a string to extract years from.	required

Returns:

Type	Description
list[str]	a set of year strings.

Source code in zavod/helpers/dates.py

def extract_years(text: str) -> list[str]:
    """Try to locate year numbers in a string such as 'circa 1990'. This will fail if
    any numbers that don't look like years are found in the string, a strong indicator
    that a more precise date is encoded (e.g. '1990 Mar 03').

    This is bounded to years between 1800 and 2100.

    Args:
        text: a string to extract years from.

    Returns:
        a set of year strings.
    """
    years: set[str] = set()
    for match in NUMBERS.finditer(text):
        year = match.group()
        number = int(year)
        if number < 1800 or number > 2100:
            continue
        years.add(year)
    return list(years)

format_address(summary=None, po_box=None, street=None, street2=None, street3=None, house=None, house_number=None, postal_code=None, city=None, county=None, state=None, state_district=None, state_code=None, country=None, country_code=None) cached

Given the components of a postal address, format it into a single line using some country-specific templating logic.

Parameters:

Name	Type	Description	Default
summary	str | None	A short description of the address.	None
po_box	str | None	The PO box/mailbox number.	None
street	str | None	The street or road name.	None
street2	str | None	The street or road name, line 2.	None
street3	str | None	The street or road name, line 3.	None
house	str | None	The descriptive name of the house.	None
house_number	str | None	The number of the house on the street.	None
postal_code	str | None	The postal code or ZIP code.	None
city	str | None	The city or town name.	None
county	str | None	The county or district name.	None
state	str | None	The state or province name.	None
state_district	str | None	The state or province district name.	None
state_code	str | None	The state or province code.	None
country	str | None	The name of the country (words, not ISO code).	None
country_code	str | None	A pre-normalized country code.	None

Returns:

Type	Description
str	A single-line string with the formatted address.

Source code in zavod/helpers/addresses.py

@lru_cache(maxsize=10000)
def format_address(
    summary: str | None = None,
    po_box: str | None = None,
    street: str | None = None,
    street2: str | None = None,
    street3: str | None = None,
    house: str | None = None,
    house_number: str | None = None,
    postal_code: str | None = None,
    city: str | None = None,
    county: str | None = None,
    state: str | None = None,
    state_district: str | None = None,
    state_code: str | None = None,
    country: str | None = None,
    country_code: str | None = None,
) -> str:
    """Given the components of a postal address, format it into a single line
    using some country-specific templating logic.

    Args:
        summary: A short description of the address.
        po_box: The PO box/mailbox number.
        street: The street or road name.
        street2: The street or road name, line 2.
        street3: The street or road name, line 3.
        house: The descriptive name of the house.
        house_number: The number of the house on the street.
        postal_code: The postal code or ZIP code.
        city: The city or town name.
        county: The county or district name.
        state: The state or province name.
        state_district: The state or province district name.
        state_code: The state or province code.
        country: The name of the country (words, not ISO code).
        country_code: A pre-normalized country code.

    Returns:
        A single-line string with the formatted address."""
    if country_code is None and country is not None:
        country_code = registry.country.clean_text(country)
    if country_code is not None:
        country_code = country_code.lower().strip()

    # Trim extended ZIP+4 codes to 5 digits for ease of comparison:
    if country_code == "us" and postal_code is not None:
        if len(postal_code) > 5:
            zip_code = postal_code[:5]
            if zip_code.isdigit():
                postal_code = zip_code

    street = join_text(street, street2, street3, sep=", ")
    data = {
        "attention": summary,
        "road": street,
        "house": po_box or house,
        "house_number": house_number,
        "postcode": postal_code,
        "city": city,
        "county": county,
        "state": state,
        "state_district": state_district,
        "state_code": state_code,
        "country": country,
    }
    return format_address_line(data, country=country_code)

is_active(sanction)

Check if a sanction is currently active.

A sanction is active if the current time is between its earliest start date and latest end date.

Parameters:

Name	Type	Description	Default
sanction	Entity	The sanction entity to check.	required

Source code in zavod/helpers/sanctions.py

def is_active(sanction: Entity) -> bool:
    """Check if a sanction is currently active.

    A sanction is active if the current time is between its earliest start date and latest end date.

    Args:
        sanction: The sanction entity to check.
    """
    iso_start_date = min(sanction.get("startDate"), default=None)
    iso_end_date = max(sanction.get("endDate"), default=None)
    is_active = (
        iso_start_date is None or not starts_after(iso_start_date, settings.RUN_TIME)
    ) and (iso_end_date is None or not ended_before(iso_end_date, settings.RUN_TIME))
    return is_active

is_empty(text)

Check if the given text is empty: it can either be null, or the stripped version of the string could have 0 length.

Parameters:

Name	Type	Description	Default
text	str | None	Text to be checked	required

Returns:

Type	Description
bool	Whether the text is empty or not.

Source code in zavod/helpers/text.py

def is_empty(text: str | None) -> bool:
    """Check if the given text is empty: it can either be null, or
    the stripped version of the string could have 0 length.

    Args:
        text: Text to be checked

    Returns:
        Whether the text is empty or not.
    """
    if text is None:
        return True
    if isinstance(text, str):
        text = text.strip()
        return len(text) == 0
    return False

is_name_irregular(entity, string)

Determine whether a name string is irregular and needs cleaning.

Source code in zavod/helpers/names.py

def is_name_irregular(entity: Entity, string: str | None) -> bool:
    """Determine whether a name string is irregular and needs cleaning."""
    return check_name_regularity(entity, string).is_irregular

links_to_dict(el)

Return a dictionary of the text content and href of each anchor element in the passed HtmlElement

Useful for when the link labels are consistent and can be used as keys

Source code in zavod/helpers/html.py

def links_to_dict(el: Element) -> dict[str | None, str | None]:
    """
    Return a dictionary of the text content and href of each anchor element in the
    passed HtmlElement

    Useful for when the link labels are consistent and can be used as keys
    """
    return {
        slugify(element_text(a), sep="_"): a.get("href") for a in el.findall(".//a")
    }

lookup_sanction_program_key(context, source_key)

Lookup the sanction program key based on the source key.

Source code in zavod/helpers/sanctions.py

def lookup_sanction_program_key(context: Context, source_key: str | None) -> str | None:
    """Lookup the sanction program key based on the source key."""
    res = context.lookup("sanction.program", source_key)
    if res is None:
        context.log.warn(f"Program key for source key {source_key!r} not found.")
        return None
    return res.value

make_address(context, full=None, remarks=None, summary=None, po_box=None, street=None, street2=None, street3=None, city=None, place=None, postal_code=None, state=None, region=None, country=None, country_code=None, key=None, lang=None, origin=None)

Generate an address schema object adjacent to the main entity.

Parameters:

Name	Type	Description	Default
context	Context	The runner context used for making and emitting entities.	required
full	str | None	The full address as a single string.	None
remarks	str | None	Delivery remarks for the address.	None
summary	str | None	A short description of the address.	None
po_box	str | None	The PO box/mailbox number.	None
street	str | None	The street or road name.	None
street2	str | None	The street or road name, line 2.	None
street3	str | None	The street or road name, line 3.	None
city	str | None	The city or town name.	None
place	str | None	The name of a smaller locality (same as city).	None
postal_code	str | None	The postal code or ZIP code.	None
state	str | None	The state or province name.	None
region	str | None	The region or district name.	None
country	str | None	The country name (words, not ISO code).	None
country_code	str | None	A pre-normalized country code.	None
key	str | None	An optional key to be included in the ID of the address.	None
lang	str | None	The language of the address details.	None

Returns:

Type	Description
Entity | None	A new entity of type Address.

Source code in zavod/helpers/addresses.py

def make_address(
    context: Context,
    full: str | None = None,
    remarks: str | None = None,
    summary: str | None = None,
    po_box: str | None = None,
    street: str | None = None,
    street2: str | None = None,
    street3: str | None = None,
    city: str | None = None,
    place: str | None = None,
    postal_code: str | None = None,
    state: str | None = None,
    region: str | None = None,
    country: str | None = None,
    country_code: str | None = None,
    key: str | None = None,
    lang: str | None = None,
    origin: str | None = None,
) -> Entity | None:
    """Generate an address schema object adjacent to the main entity.

    Args:
        context: The runner context used for making and emitting entities.
        full: The full address as a single string.
        remarks: Delivery remarks for the address.
        summary: A short description of the address.
        po_box: The PO box/mailbox number.
        street: The street or road name.
        street2: The street or road name, line 2.
        street3: The street or road name, line 3.
        city: The city or town name.
        place: The name of a smaller locality (same as city).
        postal_code: The postal code or ZIP code.
        state: The state or province name.
        region: The region or district name.
        country: The country name (words, not ISO code).
        country_code: A pre-normalized country code.
        key: An optional key to be included in the ID of the address.
        lang: The language of the address details.

    Returns:
        A new entity of type `Address`."""
    city = join_text(place, city, sep=", ")
    street = join_text(street, street2, street3, sep=", ")

    original_country_code = country_code
    # This is meant to handle cases where the country field contains a country code
    # in a subset of the given records:
    if country is not None and len(country.strip()) == 2:
        context.log.warn(
            "Country name looks like a country code",
            country=country,
            country_code=country_code,
        )
        if country_code is None:
            country_code = country
            country = None

    # Normalize casing (as format_address does internally) so that the
    # country code hashed into the address ID is stable across datasets
    # passing e.g. "US" vs "us":
    if country_code is not None:
        country_code = country_code.lower().strip()

    if country is not None:
        parsed_code = registry.country.clean(country)
        if parsed_code is not None:
            if country_code is not None and country_code != parsed_code:
                context.log.warn(
                    "Country code mismatch",
                    country=country,
                    country_code=country_code,
                )
            country_code = parsed_code

    if country_code is None:
        country_code = registry.country.clean(full)

    # If both fields carry the same value, keep only the state so that no
    # rendering path can duplicate it (e.g. "Aleppo, Aleppo"):
    if region is not None and state is not None and region == state:
        region = None

    full_origin = origin
    if not full:
        full = format_address(
            summary=summary,
            po_box=po_box,
            street=street,
            postal_code=postal_code,
            city=city,
            state=state,
            state_district=region,
            country=country,
            country_code=country_code,
        )
        if state is not None and state not in full:
            # Some country templates (e.g. ae, sa, sy) have a state_district
            # slot but no state slot, dropping the state from the rendered
            # line. Fold it into state_district instead. format_address is
            # cached, so the second render is cheap.
            full = format_address(
                summary=summary,
                po_box=po_box,
                street=street,
                postal_code=postal_code,
                city=city,
                state=state,
                state_district=join_text(region, state, sep=", "),
                country=country,
                country_code=country_code,
            )
        full_origin = ORIGIN_INFERRED

    if full == country:
        full = None

    address = context.make("Address")
    address.id = _make_id(address, full, country_code, key=key)
    if address.id is None:
        return None

    address.add("full", full, lang=lang, origin=full_origin)
    address.add("remarks", remarks, lang=lang, origin=origin)
    address.add("summary", summary, lang=lang, origin=origin)
    address.add("postOfficeBox", po_box, lang=lang, origin=origin)
    address.add("street", street, lang=lang, origin=origin)
    address.add("city", city, lang=lang, origin=origin)
    address.add("postalCode", postal_code, lang=lang, origin=origin)
    address.add("region", region, lang=lang, origin=origin)
    address.add("state", state, quiet=True, lang=lang, origin=origin)
    cov = country if original_country_code is None else original_country_code
    address.add("country", country_code, lang=lang, original_value=cov, origin=origin)
    return address

make_article(context, url, key_extra=None, title=None, published_at=None)

Create an article entity based on the URL where it was published.

Parameters:

Name	Type	Description	Default
context	Context	The runner context with dataset metadata.	required
url	str	The URL where the article was published.	required
key_extra	str | None	An optional value to be included in the generated Article ID hash.	None
title	str | None	The title the article.	None
published_at	str | None	The publication date of the article.	None

Source code in zavod/helpers/articles.py

def make_article(
    context: Context,
    url: str,
    key_extra: str | None = None,
    title: str | None = None,
    published_at: str | None = None,
) -> Entity:
    """
    Create an article entity based on the URL where it was published.

    Args:
        context: The runner context with dataset metadata.
        url: The URL where the article was published.
        key_extra: An optional value to be included in the generated Article ID hash.
        title: The title the article.
        published_at: The publication date of the article.
    """

    article = context.make("Article")
    article.id = context.make_id("Article", url, key_extra)
    article.add("sourceUrl", url)
    article.add("title", title)
    h.apply_date(article, "publishedAt", published_at)

    return article

make_documentation(context, entity, article, key_extra=None, date=None)

Creates a documentation entity to link an article to a related entity. The article's publishedAt date is added to the Documentation date property unless the date argument is provided.

This is useful to link one or more entities to an article they were mentioned in.

Create a distinct Documentation entity for each entity-article pair.

Parameters:

Name	Type	Description	Default
context	Context	The runner context with dataset metadata.	required
entity	Entity	The entity related to the article.	required
article	Entity	The related article.	required
key_extra	str | None	An optional value to be included in the generated Documentation ID hash.	None
date	str | None	The publication date of the article, added to the Documentation date property.	None

Source code in zavod/helpers/articles.py

def make_documentation(
    context: Context,
    entity: Entity,
    article: Entity,
    key_extra: str | None = None,
    date: str | None = None,
) -> Entity:
    """
    Creates a documentation entity to link an article to a related entity.
    The article's publishedAt date is added to the Documentation date property
    unless the date argument is provided.

    This is useful to link one or more entities to an article they were mentioned in.

    Create a distinct Documentation entity for each entity-article pair.

    Args:
        context: The runner context with dataset metadata.
        entity: The entity related to the article.
        article: The related article.
        key_extra: An optional value to be included in the generated Documentation ID hash.
        date: The publication date of the article, added to the Documentation date property.
    """

    documentation = context.make("Documentation")
    assert entity.id is not None
    assert article.id is not None
    documentation.id = context.make_id(
        "Documentation", entity.id, article.id, key_extra
    )
    documentation.add("entity", entity)
    documentation.add("document", article)

    if date:
        h.apply_date(documentation, "date", date)
    else:
        documentation.set("date", article.get("publishedAt"))
    return documentation

make_identification(context, entity, number, doc_type=None, country=None, summary=None, start_date=None, end_date=None, authority=None, key=None, passport=False, origin=None)

Create an Identification or Passport object linked to a passport holder.

Parameters:

Name	Type	Description	Default
context	Context	The context used for making entities.	required
entity	Entity	The entity that holds the passport.	required
number	str | None	The passport number.	required
doc_type	str | None	The type of document (e.g. "passport", "national id").	None
country	str | None	The country that issued the passport.	None
summary	str | None	A summary of the passport details.	None
start_date	str | None	The date the passport was issued.	None
end_date	str | None	The date the passport expires.	None
authority	str | None	The issuing authority.	None
key	str | None	An optional key to be included in the ID of the identification.	None
passport	bool	Whether the identification is a passport or not.	False
origin	str | None	An optional origin to attribute the emitted statements to, e.g. the model behind a reviewed extraction.	None

Returns:

Type	Description
Entity | None	A new entity of type Identification or Passport.

Source code in zavod/helpers/identification.py

def make_identification(
    context: Context,
    entity: Entity,
    number: str | None,
    doc_type: str | None = None,
    country: str | None = None,
    summary: str | None = None,
    start_date: str | None = None,
    end_date: str | None = None,
    authority: str | None = None,
    key: str | None = None,
    passport: bool = False,
    origin: str | None = None,
) -> Entity | None:
    """Create an `Identification` or `Passport` object linked to a passport holder.

    Args:
        context: The context used for making entities.
        entity: The entity that holds the passport.
        number: The passport number.
        doc_type: The type of document (e.g. "passport", "national id").
        country: The country that issued the passport.
        summary: A summary of the passport details.
        start_date: The date the passport was issued.
        end_date: The date the passport expires.
        authority: The issuing authority.
        key: An optional key to be included in the ID of the identification.
        passport: Whether the identification is a passport or not.
        origin: An optional origin to attribute the emitted statements to, e.g.
            the model behind a reviewed extraction.

    Returns:
        A new entity of type `Identification` or `Passport`.
    """
    schema = "Passport" if passport else "Identification"
    proxy = context.make(schema)
    holder_prop = proxy.schema.get("holder")
    assert holder_prop is not None
    assert holder_prop.range is not None
    if not entity.schema.is_a(holder_prop.range):
        log.warning(
            f"Holder is not a valid type for {schema}",
            entity_schema=entity.schema,
            entity_id=entity.id,
            number=number,
        )
        return None

    if number is None:
        return None
    # It is very unlikely that two countries issue the same person a document
    # with the same number.
    proxy.id = context.make_id(entity.id, number, doc_type, key)
    proxy.add("holder", entity.id, origin=origin)
    proxy.add("number", number, origin=origin)
    proxy.add("type", doc_type, origin=origin)
    proxy.add("country", country, origin=origin)
    proxy.add("authority", authority, origin=origin)
    proxy.add("summary", summary, origin=origin)
    apply_date(proxy, "startDate", start_date)
    apply_date(proxy, "endDate", end_date)
    # context.inspect(proxy.to_dict())
    if passport:
        entity.add("passportNumber", number, origin=origin)
    else:
        entity.add("idNumber", number, origin=origin)
    return proxy

make_name(full=None, name1=None, first_name=None, given_name=None, name2=None, second_name=None, middle_name=None, name3=None, patronymic=None, matronymic=None, name4=None, name5=None, tail_name=None, last_name=None, prefix=None, suffix=None)

Provides a standardised way of assembling the components of a human name. This does a whole lot of cultural ignorance work, so YMMV.

Parameters:

Name	Type	Description	Default
full	str | None	The full name if available (this will otherwise be generated).	None
name1	str | None	The first name if numeric parts are used.	None
first_name	str | None	The first name.	None
given_name	str | None	The given name (also first name).	None
name2	str | None	The second name if numeric parts are used.	None
second_name	str | None	The second name.	None
middle_name	str | None	The middle name.	None
name3	str | None	The third name if numeric parts are used.	None
patronymic	str | None	The patronymic (father-derived) name.	None
matronymic	str | None	The matronymic (mother-derived) name.	None
name4	str | None	The fourth name if numeric parts are used.	None
name5	str | None	The fifth name if numeric parts are used.	None
tail_name	str | None	A secondary last name.	None
last_name	str | None	The last/family name name.	None
prefix	str | None	A prefix to the name (e.g. "Mr").	None
suffix	str | None	A suffix to the name (e.g. "Jr").	None

Returns:

Type	Description
str | None	The full name.

Source code in zavod/helpers/names.py

def make_name(
    full: str | None = None,
    name1: str | None = None,
    first_name: str | None = None,
    given_name: str | None = None,
    name2: str | None = None,
    second_name: str | None = None,
    middle_name: str | None = None,
    name3: str | None = None,
    patronymic: str | None = None,
    matronymic: str | None = None,
    name4: str | None = None,
    name5: str | None = None,
    tail_name: str | None = None,
    last_name: str | None = None,
    prefix: str | None = None,
    suffix: str | None = None,
) -> str | None:
    """Provides a standardised way of assembling the components of a human name.
    This does a whole lot of cultural ignorance work, so YMMV.

    Args:
        full: The full name if available (this will otherwise be generated).
        name1: The first name if numeric parts are used.
        first_name: The first name.
        given_name: The given name (also first name).
        name2: The second name if numeric parts are used.
        second_name: The second name.
        middle_name: The middle name.
        name3: The third name if numeric parts are used.
        patronymic: The patronymic (father-derived) name.
        matronymic: The matronymic (mother-derived) name.
        name4: The fourth name if numeric parts are used.
        name5: The fifth name if numeric parts are used.
        tail_name: A secondary last name.
        last_name: The last/family name name.
        prefix: A prefix to the name (e.g. "Mr").
        suffix: A suffix to the name (e.g. "Jr").

    Returns:
        The full name.
    """
    if full is not None:
        full = squash_spaces(full)
        if len(full) > 0:
            return full
    return join_text(
        prefix,
        name1,
        first_name,
        given_name,
        name2,
        second_name,
        middle_name,
        name3,
        patronymic,
        matronymic,
        name4,
        name5,
        tail_name,
        last_name,
        suffix,
    )

make_occupancy(context, person, position, no_end_implies_current=True, current_time=settings.RUN_TIME, start_date=None, end_date=None, period_start=None, period_end=None, election_date=None, categorisation=None, status=None, key_prefix=None)

Creates and returns an Occupancy entity if the arguments meet our criteria for PEP position occupancy, otherwise returns None. Also adds the role.pep topic to the person if an Occupancy is returned. Emit the person after calling this to include this change.

Unless status is overridden, Occupancies are only returned if end_date is None or less than the after-office period after current_time.

current_time defaults to the process start date and time.

The after-office threshold is determined based on the position topics.

Occupancy.status is set to

• current if end_date is None and no_end_implies_current is True, otherwise status will be unknown
• current if end_date is some date in the future, unless the dataset coverage.end is a date in the past, in which case status will be unknown
• ended if end_date is some date in the past.

Parameters:

Name	Type	Description	Default
context	Context	The context to create the entity in.	required
person	Entity	The person holding the position. They will be added to the holder property.	required
position	Entity	The position held by the person. This will be added to the post property.	required
no_end_implies_current	bool	Set this to True if a dataset is regularly maintained and it can be assumed that no end date implies the person is currently occupying this position. In this case, status will be set to current. Otherwise, status will be set to unknown.	True
current_time	datetime	Defaults to the run time of the current crawl.	RUN_TIME
start_date	str | None	Set if the date the person started occupying the position is known.	None
end_date	str | None	Set if the date the person left the position is known.	None
status	OccupancyStatus | None	Overrides determining PEP occupancy status	None

Source code in zavod/helpers/positions.py

def make_occupancy(
    context: Context,
    person: Entity,
    position: Entity,
    no_end_implies_current: bool = True,
    current_time: datetime = settings.RUN_TIME,
    start_date: str | None = None,
    end_date: str | None = None,
    period_start: str | None = None,
    period_end: str | None = None,
    election_date: str | None = None,
    categorisation: PositionCategorisation | None = None,
    status: OccupancyStatus | None = None,
    key_prefix: str | None = None,
) -> Entity | None:
    """Creates and returns an Occupancy entity if the arguments meet our criteria
    for PEP position occupancy, otherwise returns None. Also adds the `role.pep` topic
    to the person if an Occupancy is returned.
    **Emit the person after calling this to include this change.**

    Unless `status` is overridden, Occupancies are only returned if end_date is None or
    less than the after-office period after current_time.

    current_time defaults to the process start date and time.

    The after-office threshold is determined based on the position topics.

    Occupancy.status is set to

    - `current` if `end_date` is `None` and `no_end_implies_current` is `True`,
      otherwise `status` will be `unknown`
    - `current` if `end_date` is some date in the future, unless the dataset
      `coverage.end` is a date in the past, in which case `status` will be `unknown`
    - `ended` if `end_date` is some date in the past.

    Args:
        context: The context to create the entity in.
        person: The person holding the position. They will be added to the
            `holder` property.
        position: The position held by the person. This will be added to the
            `post` property.
        no_end_implies_current: Set this to True if a dataset is regularly maintained
            and it can be assumed that no end date implies the person is currently
            occupying this position. In this case, `status` will be set to `current`.
            Otherwise, `status` will be set to `unknown`.
        current_time: Defaults to the run time of the current crawl.
        start_date: Set if the date the person started occupying the position is known.
        end_date: Set if the date the person left the position is known.
        status: Overrides determining PEP occupancy status
    """
    assert person.schema.is_a("Person")
    assert position.schema.is_a("Position")

    occupancy = context.make("Occupancy")
    # Include started and ended strings so that two occupancies, one missing start
    # and and one missing end, don't get normalisted to the same ID
    parts = [
        person.id,
        position.id,
        "started",
        start_date or "unknown",
        "ended",
        end_date or "unknown",
        "period_start" if period_start else None,
        period_start,
        "period_end" if period_end else None,
        period_end,
    ]
    occupancy.id = context.make_id(*parts, hash_prefix=key_prefix)
    occupancy.add("holder", person)
    occupancy.add("post", position)

    h.apply_date(occupancy, "startDate", start_date)
    h.apply_date(occupancy, "endDate", end_date)
    h.apply_date(occupancy, "periodStart", period_start)
    h.apply_date(occupancy, "periodEnd", period_end)
    h.apply_date(occupancy, "electionDate", election_date)

    if categorisation is not None and not categorisation.is_pep:
        context.log.warning(
            "Position is not categorized as a PEP, but was passed to make_occupancy",
            person=person.id,
            position=position.id,
            categorisation=categorisation,
        )
        return None

    if status is None:
        status = occupancy_status(
            context,
            person=person,
            position=position,
            occupancy=occupancy,
            no_end_implies_current=no_end_implies_current,
            current_time=current_time,
            birth_date=max(person.get("birthDate"), default=None),
            death_date=max(person.get("deathDate"), default=None),
            categorisation=categorisation,
        )
    if status is None:
        return None

    if status != OccupancyStatus.UNKNOWN:
        occupancy.add("status", status.value)

    person.add("topics", "role.pep", origin=ORIGIN_INFERRED)

    return occupancy

make_org_imo_id(value)

Build a stable entity id for an organisation from its IMO company number.

The maritime equivalent of a company register id: registered owners, managers and other shipping companies carry an IMO company number. Use this to key those organisations the same way make_vessel_imo_id keys ships. Returns None when no IMO text is supplied.

Source code in zavod/helpers/vessels.py

def make_org_imo_id(value: str | None) -> str | None:
    """Build a stable entity id for an organisation from its IMO company number.

    The maritime equivalent of a company register id: registered owners, managers and other
    shipping companies carry an IMO company number. Use this to key those organisations the
    same way [make_vessel_imo_id][zavod.helpers.make_vessel_imo_id] keys ships. Returns None
    when no IMO text is supplied.
    """
    key = _imo_id_key(value)
    return None if key is None else f"imo-org-{key}"

make_pdf_page_images(pdf_path)

Split a PDF file into PNG images of its pages.

This requires pdftoppm to be installed on the system, which is part of the poppler-utils package on Debian-based systems.

Source code in zavod/helpers/pdf.py

def make_pdf_page_images(pdf_path: Path) -> list[Path]:
    """Split a PDF file into PNG images of its pages.

    This requires `pdftoppm` to be installed on the system, which is
    part of the `poppler-utils` package on Debian-based systems.
    """
    output_path = Path(mkdtemp())
    output_prefix = output_path / pdf_path.stem
    command = [
        "pdftoppm",
        "-png",
        "-r",
        "150",
        pdf_path.as_posix(),
        output_prefix.as_posix(),
    ]
    subprocess.run(command, check=True)
    return sorted(output_path.glob("*.png"))

make_position(context, name, summary=None, description=None, country=None, topics=None, subnational_area=None, organization=None, inception_date=None, dissolution_date=None, number_of_seats=None, wikidata_id=None, source_url=None, lang=None, id_hash_prefix=None, translate_name=False)

Creates a Position entity.

Position categorisation should then be fetched using zavod.logic.pep.categorise and the result's is_pep checked.

Parameters:

Name	Type	Description	Default
context	Context	The context to create the entity in.	required
name	str	The name of the position.	required
summary	str | None	A short summary of the position.	None
description	str | None	A longer description of the position.	None
country	str | Iterable[str] | None	The country or countries the position is in.	None
topics	list[str] | None	The scope and role of the position, e.g. ["gov.national", "gov.legislative"]. Pass these for positions the crawler names itself; omit them for positions read out of the source data, where the review and classification system decides.	None
subnational_area	str | None	The state or district the position is in.	None
organization	Entity | None	The organization the position is a part of.	None
inception_date	Iterable[str] | None	The date the position was created.	None
dissolution_date	Iterable[str] | None	The date the position was dissolved.	None
number_of_seats	str | None	The number of seats that can hold the position.	None
wikidata_id	str | None	The Wikidata QID of the position.	None
source_url	str | None	The URL of the source the position was found in.	None
lang	str | None	Override the dataset language when the position details are in a non-default language.	None
id_hash_prefix	str | None	Namespace the generated entity ID, so that positions built from the same name in different contexts do not collide.	None
translate_name	bool	If True and the resolved source language is non-English, the position name is translated to English via an LLM and stored as the name (with the original kept as the value's original_value). The entity id is always derived from the untranslated name, so it stays stable and independent of the (LLM-produced) translation.	False

Returns:

Type	Description
Entity	A new entity of type Position.

Source code in zavod/helpers/positions.py

def make_position(
    context: Context,
    name: str,
    summary: str | None = None,
    description: str | None = None,
    country: str | Iterable[str] | None = None,
    topics: list[str] | None = None,
    subnational_area: str | None = None,
    organization: Entity | None = None,
    inception_date: Iterable[str] | None = None,
    dissolution_date: Iterable[str] | None = None,
    number_of_seats: str | None = None,
    wikidata_id: str | None = None,
    source_url: str | None = None,
    lang: str | None = None,
    id_hash_prefix: str | None = None,
    translate_name: bool = False,
) -> Entity:
    """Creates a Position entity.

    Position categorisation should then be fetched using zavod.logic.pep.categorise
    and the result's is_pep checked.

    Args:
        context: The context to create the entity in.
        name: The name of the position.
        summary: A short summary of the position.
        description: A longer description of the position.
        country: The country or countries the position is in.
        topics: The scope and role of the position, e.g. `["gov.national",
            "gov.legislative"]`. Pass these for positions the crawler names itself;
            omit them for positions read out of the source data, where the review and
            classification system decides.
        subnational_area: The state or district the position is in.
        organization: The organization the position is a part of.
        inception_date: The date the position was created.
        dissolution_date: The date the position was dissolved.
        number_of_seats: The number of seats that can hold the position.
        wikidata_id: The Wikidata QID of the position.
        source_url: The URL of the source the position was found in.
        lang: Override the dataset language when the position details are in a
            non-default language.
        id_hash_prefix: Namespace the generated entity ID, so that positions built
            from the same name in different contexts do not collide.
        translate_name: If True and the resolved source language is non-English,
            the position name is translated to English via an LLM and stored as
            the `name` (with the original kept as the value's original_value).
            The entity id is always derived from the untranslated `name`, so it
            stays stable and independent of the (LLM-produced) translation.

    Returns:
        A new entity of type `Position`."""

    position = context.make("Position")

    parts: list[str] = [name]
    if country is not None:
        parts.extend(ensure_list(country))
    if inception_date is not None:
        parts.extend(ensure_list(inception_date))
    if dissolution_date is not None:
        parts.extend(ensure_list(dissolution_date))
    if subnational_area is not None:
        parts.extend(ensure_list(subnational_area))

    if wikidata_id is not None:
        position.id = wikidata_id
    else:
        position.id = context.make_id(*parts, hash_prefix=id_hash_prefix)

    source_lang = lang or context.lang

    # Optionally translate the name to English. The id above is keyed on the
    # untranslated name, so it stays stable regardless of the LLM output.
    if translate_name and source_lang is not None and source_lang != "eng":
        # Local import to break the cycle: zavod.shed.trans imports the helpers
        # package, which imports this module. TODO: move the translation core to
        # a helpers-free zavod.helpers.translate in a followup so this can become
        # a top-level import.
        from zavod.shed.trans import translate_position_name

        result = translate_position_name(context, LangText(text=name, lang=source_lang))
        translated = result.get_preferred_language()
        if translated is not None:
            position.add(
                "name",
                translated.text,
                lang=translated.lang,
                original_value=name,
                origin=result.origin,
            )
        else:
            position.add("name", name, lang=lang)
    else:
        position.add("name", name, lang=lang)

    position.add("summary", summary, lang=lang)
    position.add("description", description, lang=lang)
    position.add("country", country)
    position.add("topics", topics)
    position.add("organization", organization, lang=lang)
    position.add("subnationalArea", subnational_area, lang=lang)
    position.add("inceptionDate", inception_date)
    position.add("dissolutionDate", dissolution_date)
    position.add("numberOfSeats", number_of_seats)
    position.add("wikidataId", wikidata_id)
    position.add("sourceUrl", source_url)

    return position

make_sanction(context, entity, key=None, program_name=None, source_program_key=None, program_key=None, start_date=None, end_date=None)

Create and return a sanctions object derived from the dataset metadata.

The country, authority, sourceUrl, and subject entity properties are automatically set.

If an end_date is given, a status of "active" or "inactive" is derived using the same semantics as is_active. Note that the status is only computed at construction time: dates applied to the sanction afterwards (e.g. via h.apply_date) do not update it.

Parameters:

Name	Type	Description	Default
context	Context	The runner context with dataset metadata.	required
entity	Entity	The entity to which the sanctions object will be linked.	required
key	str | None	An optional key to be included in the ID of the sanction.	None
program_name	str | None	An optional program name.	None
program_key	str | None	An optional OpenSanction program key.	None
source_program_key	str | None	Program key at the source, will be set as the original value for programId.	None
start_date	str | None	An optional start date for the sanction.	None
end_date	str | None	An optional end date for the sanction.	None

Returns:

Type	Description
Entity	A new entity of type Sanction.

Source code in zavod/helpers/sanctions.py

def make_sanction(
    context: Context,
    entity: Entity,
    key: str | None = None,
    program_name: str | None = None,
    source_program_key: str | None = None,
    program_key: str | None = None,
    start_date: str | None = None,
    end_date: str | None = None,
) -> Entity:
    """Create and return a sanctions object derived from the dataset metadata.

    The country, authority, sourceUrl, and subject entity properties
    are automatically set.

    If an ``end_date`` is given, a ``status`` of "active" or "inactive" is
    derived using the same semantics as `is_active`. Note that the status is
    only computed at construction time: dates applied to the sanction
    afterwards (e.g. via `h.apply_date`) do not update it.

    Args:
        context: The runner context with dataset metadata.
        entity: The entity to which the sanctions object will be linked.
        key: An optional key to be included in the ID of the sanction.
        program_name: An optional program name.
        program_key: An optional OpenSanction program key.
        source_program_key: Program key at the source, will be set as the original value for programId.
        start_date: An optional start date for the sanction.
        end_date: An optional end date for the sanction.

    Returns:
        A new entity of type Sanction.
    """
    assert entity.schema.is_a("Thing"), entity.schema
    assert entity.id is not None, entity.id
    dataset = context.dataset.model
    assert dataset.publisher is not None
    sanction = context.make("Sanction")
    sanction.id = context.make_id("Sanction", entity.id, key)
    sanction.add("entity", entity)
    if dataset.publisher.country != "zz":
        sanction.add("country", dataset.publisher.country, origin=ORIGIN_METADATA)
    sanction.add("authority", dataset.publisher.name, origin=ORIGIN_METADATA)
    sanction.add("sourceUrl", dataset.url, origin=ORIGIN_METADATA)
    sanction.set("program", program_name)

    if program_key is not None:
        program = programs.get_program_by_key(program_key)
        if program:
            sanction.set(
                "programId",
                program_key,
                original_value=source_program_key,
                origin=ORIGIN_METADATA,
            )
            entity.add("programId", program_key, origin=ORIGIN_METADATA)
            sanction.add("programUrl", program.url, origin=ORIGIN_METADATA)
        else:
            context.log.warn(
                f"Program with key {program_key!r} not found.",
                entity_id=entity.id,
            )

    if start_date:
        h.apply_date(sanction, "startDate", start_date)
    if end_date:
        h.apply_date(sanction, "endDate", end_date)
        if not sanction.get("endDate"):
            raise ValueError(
                f"Sanction end_date {end_date!r} could not be parsed as a date "
                f"(entity {entity.id!r}). Add a datepatterns entry or a lookup "
                "to clean the value."
            )
        sanction.add("status", "active" if is_active(sanction) else "inactive")

    return sanction

make_security(context, isin)

Make a security entity.

Source code in zavod/helpers/securities.py

def make_security(context: Context, isin: str) -> Entity:
    """Make a security entity."""
    isin = isin.upper()
    entity = context.make("Security")
    entity.id = f"isin-{isin}"
    entity.add("isin", isin)
    cc = isin[:2]
    if cc not in ISIN_NON_COUNTRY:
        entity.add("country", cc, origin=ORIGIN_INFERRED)
    return entity

make_vessel_imo_id(value)

Build a stable entity id for a vessel from its IMO number.

Reach for this when keying vessels so that records describing the same ship across sources converge on one entity without depending on any source's internal numbering. Valid IMOs collapse to their seven digits; malformed ones fall back to a slug of the raw value so a faulty IMO keeps the vessel rather than dropping it. Returns None when no IMO text is supplied — the caller should then key the entity another way.

Source code in zavod/helpers/vessels.py

def make_vessel_imo_id(value: str | None) -> str | None:
    """Build a stable entity id for a vessel from its IMO number.

    Reach for this when keying vessels so that records describing the same ship across
    sources converge on one entity without depending on any source's internal numbering.
    Valid IMOs collapse to their seven digits; malformed ones fall back to a slug of the raw
    value so a faulty IMO keeps the vessel rather than dropping it. Returns None when no IMO
    text is supplied — the caller should then key the entity another way.
    """
    key = _imo_id_key(value)
    return None if key is None else f"imo-vsl-{key}"

multi_split(text, splitters)

Sequentially attempt to split a text based on an array of splitting criteria. This is useful for strings where multiple separators are used to separate values, e.g.: test,other/misc. A special case of this is itemised lists like a) test b) other c) misc which sanction-makers seem to love.

Parameters:

Name	Type	Description	Default
text	str | Iterable[str | None] | None	A text or list of texts to be split up further.	required
splitters	Iterable[str]	A sequence of text splitting criteria to be applied to the text.	required

Returns:

Type	Description
list[str]	Fully subdivided text snippets.

Source code in zavod/helpers/text.py

def multi_split(
    text: str | Iterable[str | None] | None, splitters: Iterable[str]
) -> list[str]:
    """Sequentially attempt to split a text based on an array of splitting criteria.
    This is useful for strings where multiple separators are used to separate values,
    e.g.: `test,other/misc`. A special case of this is itemised lists like `a) test
    b) other c) misc` which sanction-makers seem to love.

    Args:
        text: A text or list of texts to be split up further.
        splitters: A sequence of text splitting criteria to be applied to the text.

    Returns:
        Fully subdivided text snippets.
    """
    if text is None:
        return []
    fragments = ensure_list(text)
    sorted_splitters = tuple(sorted(splitters, key=len, reverse=True))

    for splitter in sorted_splitters:
        out: list[str | None] = []
        for fragment in fragments:
            if fragment is None:
                continue
            for frag in fragment.split(splitter):
                frag = frag.strip()
                if len(frag):
                    out.append(frag)
        fragments = out
    result = [f for f in fragments if f is not None]

    return result

parse_html_table(table, header_tag='th', skiprows=0, ignore_colspan=None, slugify_headers=True, index_empty_headers=False)

Parse an HTML table into a generator yielding a dict for each row.

Parameters:

Name	Type	Description	Default
table	Element	The table HTML element to parse	required
header_tag	str	Default th, allows treating td as header	'th'
skiprows	int	Number of rows to skip before expecting the header row.	0
ignore_colspan	set[str] | None	colspans to ignore, e.g. when a full span means a subheading	None

Returns:

Type	Description
None	Generator of dict per row, where the keys are the _-slugified table headings and the values are the HtmlElement of the cell.

See also

• zavod.helpers.cells_to_str
• zavod.helpers.links_to_dict

Source code in zavod/helpers/html.py

def parse_html_table(
    table: Element,
    header_tag: str = "th",
    skiprows: int = 0,
    ignore_colspan: set[str] | None = None,
    slugify_headers: bool = True,
    index_empty_headers: bool = False,
) -> Generator[dict[str, Element], None, None]:
    """
    Parse an HTML table into a generator yielding a dict for each row.

    Args:
        table: The table HTML element to parse
        header_tag: Default th, allows treating td as header
        skiprows: Number of rows to skip before expecting the header row.
        ignore_colspan: colspans to ignore, e.g. when a full span means a subheading

    Returns:
        Generator of dict per row, where the keys are the _-slugified table headings
            and the values are the HtmlElement of the cell.

    See also:
      - `zavod.helpers.cells_to_str`
      - `zavod.helpers.links_to_dict`
    """
    headers = None
    rows = table.findall(".//tr")
    child_rows = [
        row
        for row in rows
        # A descendant search also matches rows of tables nested inside a
        # cell; only rows whose nearest <table> ancestor is the target table
        # belong to it.
        if next(row.iterancestors("table"), None) in (table, None)
    ]
    if len(rows) != len(child_rows):
        # TODO: Turn warning into just ignoring a week or so after releasing
        # the warning. https://github.com/opensanctions/opensanctions/issues/5322
        log.warning("Nested table rows to be dropped.")
    for rownum, row in enumerate(rows):
        if rownum < skiprows:
            continue

        if headers is None:
            headers = []
            for colnum, el in enumerate(row.findall(f"./{header_tag}")):
                header_text: str | None = element_text(el)
                if slugify_headers:
                    header_text = slugify(header_text, sep="_")
                if index_empty_headers and not header_text:
                    header_text = f"column_{colnum}"
                assert header_text is not None, "No table header text"
                headers.append(header_text)
            duplicates = {hdr for hdr in headers if headers.count(hdr) > 1}
            # Rows are built with dict(zip(headers, cells)), so a duplicate
            # header would silently drop the earlier column's cell.
            assert not duplicates, f"Duplicate headers: {sorted(duplicates)}"
            continue

        cells = row.findall("./td")
        if len(headers) != len(cells):
            str_cells = [element_text(c) for c in cells]
            colspans = set([c.get("colspan") for c in cells])
            if ignore_colspan and colspans == set(ignore_colspan):
                log.info(f"Ignoring row {rownum} with colspan: {str_cells}")
                continue
            else:
                msg = f"Expected {len(headers)} cells, found {len(cells)} on row {rownum} {str_cells}"
                assert len(headers) == len(cells), msg

        yield {hdr: c for hdr, c in zip(headers, cells)}

parse_pdf_table(context, path, headers_per_page=False, preserve_header_newlines=False, start_page=None, end_page=None, skiprows=0, page_settings=None)

Parse the largest table on each page of a PDF file and yield their rows as dictionaries.

Parameters:

Name	Type	Description	Default
path	Path	Path to the PDF file.	required
headers_per_page	bool	Set to true if the headers are repeated on each page.	False
preserve_header_newlines	bool	Don't slugify newlines in headers - e.g. for when the line breaks are meaningful.	False
start_page	int | None	The first page to process. 1-indexed.	None
end_page	int | None	The last page to process. 1-indexed.	None
skiprows	int	The number of rows to skip before processing table headers.	0
page_settings	Callable[[Page], tuple[Page, dict[str, Any]]] | None	A function that takes a pdfplumber.page.Page object and returns a tuple of a Page that will be used to extract a table, and a dictionary of settings for extract_table. The page could be e.g. a cropped version of the original.	None

Pro tip

Save debug images in the page settings function to help with debugging.

• https://github.com/jsvine/pdfplumber?tab=readme-ov-file#drawing-methods
• https://github.com/jsvine/pdfplumber?tab=readme-ov-file#visually-debugging-the-table-finder

def settings_func(page):
    cropped = page.crop((0, 93, page.width, page.height))
    im = cropped.to_image()
    im.save(f"page-{cropped.page_number}.png")
    return (cropped, PAGE_SETTINGS)

Source code in zavod/helpers/pdf.py

def parse_pdf_table(
    context: Context,
    path: Path,
    headers_per_page: bool = False,
    preserve_header_newlines: bool = False,
    start_page: int | None = None,
    end_page: int | None = None,
    skiprows: int = 0,
    page_settings: Callable[[Page], tuple[Page, dict[str, Any]]] | None = None,
) -> Generator[dict[str, str | None], None, None]:
    """
    Parse the largest table on each page of a PDF file and yield their rows as dictionaries.

    Arguments:
        path: Path to the PDF file.
        headers_per_page: Set to true if the headers are repeated on each page.
        preserve_header_newlines: Don't slugify newlines in headers -
            e.g. for when the line breaks are meaningful.
        start_page: The first page to process. 1-indexed.
        end_page: The last page to process. 1-indexed.
        skiprows: The number of rows to skip before processing table headers.
        page_settings: A function that takes a `pdfplumber.page.Page` object and returns
            a tuple of a Page that will be used to extract a table, and a dictionary of
            settings for `extract_table`. The page could be e.g. a cropped version of the
            original.

    Pro tip:
        Save debug images in the page settings function to help with debugging.

        - https://github.com/jsvine/pdfplumber?tab=readme-ov-file#drawing-methods
        - https://github.com/jsvine/pdfplumber?tab=readme-ov-file#visually-debugging-the-table-finder

        ```
        def settings_func(page):
            cropped = page.crop((0, 93, page.width, page.height))
            im = cropped.to_image()
            im.save(f"page-{cropped.page_number}.png")
            return (cropped, PAGE_SETTINGS)
        ```
    """
    start_page_idx = start_page - 1 if isinstance(start_page, int) else None
    end_page_idx = end_page if isinstance(end_page, int) else None
    pdf = pdfplumber.open(path)
    headers = None
    for page in pdf.pages[start_page_idx:end_page_idx]:
        if page.page_number % 100 == 0:
            context.log.info(f"Processing page {page.page_number}...")

        if headers_per_page:
            headers = None

        if page_settings is not None:
            page, settings = page_settings(page)
        else:
            settings = {}

        rows = page.extract_table(settings)
        if rows is None:
            raise Exception(f"No table found on page {page.page_number} of {path}")
        for row_num, row in enumerate(rows):
            if headers is None:
                if row_num < skiprows:
                    continue
                headers = [
                    header_slug(cell or "", preserve_header_newlines) for cell in row
                ]
                duplicates = {hdr for hdr in headers if headers.count(hdr) > 1}
                # Rows are built with dict(zip(headers, row)), so a duplicate
                # header (e.g. two empty header cells, which both slugify to
                # "") would silently drop the earlier column's cell.
                assert not duplicates, f"Duplicate headers: {sorted(duplicates)}"
                continue
            assert len(headers) == len(row), (headers, row)
            row_slugs = [
                header_slug(cell or "", preserve_header_newlines) for cell in row
            ]
            if row_slugs == headers:
                # Tables that repeat their header row on every page would
                # otherwise emit the repeated headers as a data row. Warn
                # rather than skip silently: the table probably wants
                # headers_per_page (and skiprows) so that any rows above the
                # repeated header, e.g. comments, are skipped too.
                context.log.warning(
                    (
                        "Skipping repeated header row. Consider headers_per_page "
                        "in case comment rows need skipping on each page."
                    ),
                    page=page.page_number,
                    row=row,
                )
                continue
            yield dict(zip(headers, row))

        page.close()
    pdf.close()

parse_xls_sheet(context, sheet, skiprows=0, join_header_rows=0)

Parse an Excel sheet into a sequence of dictionaries.

Keys are the column headings slugified with _ as separator.

Cells with links are included as keys with _url appended to the original key.

Source code in zavod/helpers/excel.py

def parse_xls_sheet(
    context: Context,
    sheet: Sheet,
    skiprows: int = 0,
    join_header_rows: int = 0,
) -> Generator[dict[str, str | None], None, None]:
    """
    Parse an Excel sheet into a sequence of dictionaries.

    Keys are the column headings slugified with _ as separator.

    Cells with links are included as keys with _url appended to the original key.
    """
    headers: list[str] | None = None
    headers_validated = False
    for row_ix, row in enumerate(sheet):
        if row_ix < skiprows:
            continue
        cells: list[str | None] = []
        record: dict[str, str | None] = {}
        for cell_ix, xl_cell in enumerate(row):
            if xl_cell.ctype == XL_CELL_DATE:
                # Convert Excel date format to zavod date
                assert isinstance(xl_cell.value, float)
                assert sheet.book is not None
                date_value = xldate_as_datetime(xl_cell.value, sheet.book.datemode)
                cells.append(date_value.date().isoformat())
            else:
                cells.append(stringify(xl_cell.value))

            # Add link to key ..._url
            if url := sheet.hyperlink_map.get((row_ix, cell_ix)):
                assert headers is not None, ("URLs not supported in headers yet.", row)
                key = f"{headers[cell_ix]}_url"
                record[key] = str(url.url_or_path)

        if headers is None or join_header_rows > 0:
            if headers:
                # Append row of split-headers to current headers
                for col_idx, cell in enumerate(cells):
                    if not cell:
                        continue
                    headers[col_idx] += f"_{slugify_text(cell, sep='_')}"
                join_header_rows -= 1
            else:
                # Initialise first row of headers
                headers = []
                for idx, cell in enumerate(cells):
                    if not cell:
                        cell = f"column_{idx}"
                    headers.append(slugify_text(cell, "_") or "")
            continue

        if not headers_validated:
            # Headers are final once the first data row is reached (they may
            # span several rows via join_header_rows). Records are built by
            # zipping headers with cells, so a duplicate header would silently
            # drop the earlier column's cell.
            duplicates = {hdr for hdr in headers if headers.count(hdr) > 1}
            assert not duplicates, f"Duplicate headers: {sorted(duplicates)}"
            headers_validated = True

        for header, value in zip(headers, cells):
            record[header] = stringify(value)

        if len(record) == 0:
            continue
        if all(v is None for v in record.values()):
            continue
        yield record

parse_xlsx_sheet(context, sheet, skiprows=0, header_lookup=None, extract_links=False)

Parse an Excel sheet into a sequence of dictionaries.

Parameters:

Name	Type	Description	Default
context	Context	Crawler context.	required
sheet	Worksheet	The Excel sheet.	required
skiprows	int	The number of rows to skip.	0
header_lookup	Lookup | None	The lookup key for translating headers.	None
extract_links	bool	Whether to extract hyperlinks. Only works when read_only=False	False

Source code in zavod/helpers/excel.py

def parse_xlsx_sheet(
    context: Context,
    sheet: Worksheet,
    skiprows: int = 0,
    header_lookup: Lookup | None = None,
    extract_links: bool = False,
) -> Iterator[dict[str, str | None]]:
    """
    Parse an Excel sheet into a sequence of dictionaries.

    Args:
        context: Crawler context.
        sheet: The Excel sheet.
        skiprows: The number of rows to skip.
        header_lookup: The lookup key for translating headers.
        extract_links: Whether to extract hyperlinks. Only works when read_only=False
    """
    headers: list[str] | None = None
    row_counter = 0

    for row in sheet.iter_rows():
        # Increment row counter
        row_counter += 1

        # Skip the desired number of rows
        if row_counter <= skiprows:
            continue
        cells = [c.value for c in row]
        if headers is None:
            headers = []
            for idx, header in enumerate(cells):
                header = stringify(header)
                if header is None:
                    header = f"column_{idx}"
                if header_lookup:
                    header = header_lookup.get_value(header) or header
                header_slug = slugify_text(header, sep="_")
                if header_slug is None and header is not None:
                    header_slug = f"column_{idx}"
                headers.append(header_slug)
            duplicates = {hdr for hdr in headers if headers.count(hdr) > 1}
            # Records are built by zipping headers with cells, so a duplicate
            # header would silently drop the earlier column's cell.
            assert not duplicates, f"Duplicate headers: {sorted(duplicates)}"
            continue

        record: dict[str, str | None] = {}
        for cell_ix, (header, cell) in enumerate(zip(headers, row)):
            value = cell.value
            if isinstance(value, datetime):
                value = value.date()
            record[header or ""] = stringify(value)

            if extract_links:
                # Check if the cell has a hyperlink
                if cell.hyperlink:
                    key = f"{header}_url"
                    record[key] = str(cell.hyperlink.target)

        if len(record) == 0:
            continue
        if all(v is None for v in record.values()):
            continue
        for header in headers:
            if header not in record:
                record[header] = None
        yield record

postcode_pobox(text)

For when PO Box is stuffed into postcode, sometimes.

Returns:

Type	Description
tuple[str | None, str | None]	Tuple of (postcode, po_box)

Source code in zavod/helpers/addresses.py

def postcode_pobox(text: str | None) -> tuple[str | None, str | None]:
    """
    For when PO Box is stuffed into postcode, sometimes.

    Returns:
        Tuple of (postcode, po_box)
    """
    if text is None:
        return None, None
    if match := REGEX_POBOX.match(text):
        return None, match.group(0)
    return text, None

remove_bracketed(text)

Helps to deal with property values where additional info has been supplied in brackets that makes it harder to parse the value. Examples:

• Russia (former USSR)
• 1977 (as Muhammad Da'ud Salman)

It's probably not useful in all of these cases to try and parse and derive meaning from the bracketed bit, so we'll just discard it.

Parameters:

Name	Type	Description	Default
text	str | None	Text with sub-text in brackets	required

Returns:

Type	Description
str | None	Text that was not in brackets.

Source code in zavod/helpers/text.py

def remove_bracketed(text: str | None) -> str | None:
    """Helps to deal with property values where additional info has been supplied in
    brackets that makes it harder to parse the value. Examples:

    - Russia (former USSR)
    - 1977 (as Muhammad Da'ud Salman)

    It's probably not useful in all of these cases to try and parse and derive meaning
    from the bracketed bit, so we'll just discard it.

    Args:
        text: Text with sub-text in brackets

    Returns:
        Text that was not in brackets.
    """
    if text is None:
        return None
    return BRACKETED.sub(" ", text)

remove_namespace(el)

Remove namespace in the passed XML/HTML document in place and return an updated element tree.

If the namespaces in a document define multiple tags with the same local tag name, this will create ambiguity and lead to errors. Most XML documents, however, only actively use one namespace.

Parameters:

Name	Type	Description	Default
el	ElementOrTree	The root element or tree to remove namespaces from.	required

Returns:

Type	Description
ElementOrTree	An updated element tree with the namespaces removed.

Source code in zavod/helpers/xml.py

def remove_namespace(el: ElementOrTree) -> ElementOrTree:
    """Remove namespace in the passed XML/HTML document in place and
    return an updated element tree.

    If the namespaces in a document define multiple tags with the same
    local tag name, this will create ambiguity and lead to errors. Most
    XML documents, however, only actively use one namespace.

    Args:
        el: The root element or tree to remove namespaces from.

    Returns:
        An updated element tree with the namespaces removed.
    """
    for elem in el.iter():
        # https://stackoverflow.com/a/47233934
        if elem.tag is etree.Comment:  # type: ignore
            # Can't make a QName from a comment
            continue
        elem.tag = etree.QName(elem).localname
        for key, value in list(elem.attrib.items()):
            local_key = etree.QName(key).localname
            if key != local_key:
                elem.attrib[local_key] = value
    etree.cleanup_namespaces(el)
    return el

replace_months(dataset, text)

Re-write month names to the latin form to get a date string ready for parsing.

Parameters:

Name	Type	Description	Default
dataset	Dataset	The dataset which contains a date format specification.	required
text	str	The string inside of which month names will be replaced.	required

Returns:

Type	Description
str	A string in which month names are normalized.

Source code in zavod/helpers/dates.py

def replace_months(dataset: Dataset, text: str) -> str:
    """Re-write month names to the latin form to get a date string ready for parsing.

    Args:
        dataset: The dataset which contains a date format specification.
        text: The string inside of which month names will be replaced.

    Returns:
        A string in which month names are normalized.
    """
    spec = dataset.dates
    if spec.months_re is None:
        return text
    return spec.months_re.sub(lambda m: spec.mappings[m.group().lower()], text)

review_names(context, entity, *, original, suggested=None, is_irregular=False, llm_cleaning=False, default_accepted=False)

Determines whether names need cleaning and if so, posts them for review.

If 'suggested' is not supplied, 'check_names_regularity' is used to determine if cleaning or review is needed, and potentially suggest categorisation.

Names are considered to have been pre-determined to need cleaning/review if 'is_irregular' is passed as True, or if 'suggested' is supplied and differs from 'original'. Crawlers that do their own suggestions should normally do those on the result of check_names_regularity, so that its suggestions don't override the crawler's suggestions.

If 'llm_cleaning' is True, an LLM-based cleaning step is additionally done on 'suggested' if provided, otherwise on 'original', before posting for review. Any categorisation in 'original' and 'suggested' is disregarded and left to the LLM to determine. This can not be used with crawler-supplied suggestions and, and heuristic suggestions are not passed to the LLM.

Returns None if no cleaning/review is needed and the original can be applied as-is.

Parameters:

Name	Type	Description	Default
context	Context	The current context.	required
entity	Entity	The entity to apply names to.	required
original	Names	The original categorisation of names. This is to convey to the analyst how the source data categorised the name string(s).	required
suggested	Names | None	The suggested categorisation of names. This contains an initial categorisation where the source dataset might have adjusted the categorisation based on heuristics specific to that dataset.	None
llm_cleaning	bool	Whether to use LLM-based name cleaning.	False
default_accepted	bool	Whether to mark the review as accepted from the start, if one is created.	False

Source code in zavod/helpers/names.py

def review_names(
    context: Context,
    entity: Entity,
    *,
    original: Names,
    suggested: Names | None = None,
    is_irregular: bool = False,
    llm_cleaning: bool = False,
    default_accepted: bool = False,
) -> Review[Names] | None:
    """
    Determines whether names need cleaning and if so, posts them for review.

    If 'suggested' is not supplied, 'check_names_regularity' is used to determine
    if cleaning or review is needed, and potentially suggest categorisation.

    Names are considered to have been pre-determined to need cleaning/review if
    'is_irregular' is passed as True, or if 'suggested'
    is supplied and differs from 'original'. Crawlers that do their own suggestions
    should normally do those on the result of check_names_regularity, so that
    its suggestions don't override the crawler's suggestions.

    If 'llm_cleaning' is True, an LLM-based cleaning step is additionally done
    on 'suggested' if provided, otherwise on 'original', before posting for review.
    Any categorisation in 'original' and 'suggested' is disregarded and left to the LLM
    to determine. This can not be used with crawler-supplied suggestions and,
    and heuristic suggestions are not passed to the LLM.

    Returns None if no cleaning/review is needed and the original can be applied as-is.

    Args:
        context: The current context.
        entity: The entity to apply names to.
        original: The original categorisation of names. This is to convey to the
            analyst how the source data categorised the name string(s).
        suggested: The suggested categorisation of names. This contains an initial
            categorisation where the source dataset might have adjusted the categorisation
            based on heuristics specific to that dataset.
        llm_cleaning: Whether to use LLM-based name cleaning.
        default_accepted: Whether to mark the review as accepted from the start, if one is created.
    """

    if original.is_empty():
        return None

    if llm_cleaning:
        assert suggested is None, (
            "Suggested names can't be supplied if LLM cleaning is enabled"
        )
        if _original_has_lang(original):
            # LLM cleaning returns plain strings, so per-value language will be dropped.
            # Use a separate review_names, apply_reviewed_names or apply_reviewed_name_string call
            # with the lang argument for each language instead.
            context.log.warning(
                "Names with LangText language values and llm_cleaning=True are not supported together.",
                original=original,
            )

    # heuristic-based review unless suggestion was supplied
    if suggested is None:
        is_irregular_, suggested = check_names_regularity(entity, original)
        is_irregular = is_irregular or is_irregular_

    # heuristics didn't identify irregularity, and the crawler didn't suggest
    # re-categorisation, there's nothing to review.
    if not is_irregular and suggested == original:
        return None

    # human and optionally LLM-based review
    return _review_names(
        context,
        entity,
        original=original,
        suggested=suggested,
        llm_cleaning=llm_cleaning,
        default_accepted=default_accepted,
    )

split_comma_names(context, text)

Split a string of multiple names that may contain company and individual names, some including commas, into individual names without breaking partnership names like "A, B and C Inc" or individuals like "Smith, Jane".

To make life easier, commas are stripped from company type suffixes like "Blue, LLC"

If the string can't be split into whole names reliably, a datapatch is looked up under the comma_names key, which should contain a list of names in the names attribute. If no match is found, the name is returned as a single item list, and a warning emitted.

Source code in zavod/helpers/names.py

def split_comma_names(context: Context, text: str) -> list[str]:
    """Split a string of multiple names that may contain company and individual names,
    some including commas, into individual names without breaking partnership names
    like "A, B and C Inc" or individuals like "Smith, Jane".

    To make life easier, commas are stripped from company type suffixes like "Blue, LLC"

    If the string can't be split into whole names reliably, a datapatch is looked up
    under the `comma_names` key, which should contain a list of names in the `names`
    attribute. If no match is found, the name is returned as a single item list,
    and a warning emitted.
    """
    text = squash_spaces(text)
    if len(text) == 0:
        return []

    # Check early for overrides of cases where splitting on comma is a mistake.
    res = context.lookup("comma_names", text)
    if res:
        return cast(list[str], res.names)

    text = REGEX_CLEAN_COMMA.sub(r" \1", text)
    # If the string ends in a comma, the last comma is unnecessary (e.g. Goldman Sachs & Co. LLC,)
    if text.endswith(","):
        text = text[:-1]

    if not REGEX_AND.search(text) and not REGEX_LNAME_FNAME.match(text):
        names = [n.strip() for n in text.split(",")]
        return names
    else:
        if ("," in text) or (" and " in text):
            res = context.lookup("comma_names", text)
            if res:
                return cast("list[str]", res.names)
            else:
                context.log.warning("Not sure how to split on comma or and.", text=text)
                return [text]
        else:
            return [text]

split_html_newline_tags(string)

Split a string on HTML
and

tags, returning a list of strings.

Empty and whitespace-only strings are dropped from the result.

Source code in zavod/helpers/html.py

def split_html_newline_tags(string: str) -> list[str]:
    """
    Split a string on HTML <br> and <p> tags, returning a list of strings.

    Empty and whitespace-only strings are dropped from the result.
    """
    return [s for s in BR_RE.split(string) if s.strip()]

strip_name_titles(context, name)

Strip configured title affixes from a source name.

Use when a dataset stores honorific prefixes or post-nominals as part of a name and declares those terms under names.prefixes_strip or names.suffixes_strip. Configure the exact affixes used by the source, such as "Dr.", "(Dr.)", or ", CS", rather than relying on punctuation variants. When storing the result directly, preserve provenance by passing the raw titled value as original_value if it differs from the cleaned name.

A term is only stripped at a word boundary: terms delimited by their own punctuation ("Hon.", "(Dr.)") match directly, while bare word terms ("Hon") must be followed (prefixes) or preceded (suffixes) by whitespace, so they never truncate names like "Honorata". If stripping consumes the entire name, a warning is emitted and None is returned so the affected record surfaces in the issue log instead of silently losing its name.

Source code in zavod/helpers/names.py

def strip_name_titles(context: Context, name: str | None) -> str | None:
    """Strip configured title affixes from a source name.

    Use when a dataset stores honorific prefixes or post-nominals as part of a
    name and declares those terms under `names.prefixes_strip` or
    `names.suffixes_strip`. Configure the exact affixes used by the source, such
    as `"Dr."`, `"(Dr.)"`, or `", CS"`, rather than relying on punctuation
    variants. When storing the result directly, preserve provenance by passing
    the raw titled value as `original_value` if it differs from the cleaned name.

    A term is only stripped at a word boundary: terms delimited by their own
    punctuation (`"Hon."`, `"(Dr.)"`) match directly, while bare word terms
    (`"Hon"`) must be followed (prefixes) or preceded (suffixes) by whitespace,
    so they never truncate names like "Honorata". If stripping consumes the
    entire name, a warning is emitted and `None` is returned so the affected
    record surfaces in the issue log instead of silently losing its name.
    """
    if name is None:
        return None

    name = squash_spaces(name)
    stripped = _strip_title_prefixes(name, context.dataset.names.prefixes_strip)
    stripped = _strip_title_suffixes(stripped, context.dataset.names.suffixes_strip)
    if len(stripped) == 0 and len(name) > 0:
        context.log.warning("Name consists only of title affixes", name=name)
        return None
    return stripped

within_max_age(context, date, max_age_days=MAX_ENFORCEMENT_DAYS)

Check if a the given date is within a specified maximum age, defaulting to MAX_ENFORCEMENT_DAYS.

This is useful for filtering out all but the most recent items, e.g. sanctions announcements or enforcement actions.

Parameters:

Name	Type	Description	Default
context	Context	The runner context with dataset metadata.	required
date	datetime | str	The date to check.	required
max_age_days	int	The maximum allowable age in days, if different from the default.	MAX_ENFORCEMENT_DAYS

Source code in zavod/helpers/dates.py

def within_max_age(
    context: "Context",
    date: datetime | str,
    max_age_days: int = MAX_ENFORCEMENT_DAYS,
) -> bool:
    """
    Check if a the given date is within a specified maximum age, defaulting to `MAX_ENFORCEMENT_DAYS`.

    This is useful for filtering out all but the most recent items, e.g. sanctions announcements
    or enforcement actions.

    Args:
        context: The runner context with dataset metadata.
        date: The date to check.
        max_age_days: The maximum allowable age in days, if different from the default.
    """
    if isinstance(date, str):
        date = date.strip()
    cleaned_date = extract_date(context.dataset, date, fallback_to_original=False)[0]
    return not ended_before(cleaned_date, RUN_TIME - timedelta(days=max_age_days))

xpath_element(el, xpath)

Evaluate an XPath expression and return the single matching element.

Use this when exactly one match is expected — e.g. selecting the main content table on a page. Raises if there are zero or more than one matches, catching unexpected duplication or removal at the call site.

Source code in zavod/helpers/html.py

def xpath_element(el: Element, xpath: str) -> Element:
    """
    Evaluate an XPath expression and return the single matching element.

    Use this when exactly one match is expected — e.g. selecting the main
    content table on a page. Raises if there are zero or more than one matches,
    catching unexpected duplication or removal at the call site.
    """
    return xpath_elements(el, xpath, expect_exactly=1)[0]

xpath_elements(el, xpath, *, expect_exactly=None)

Evaluate an XPath expression and return matching elements as a typed list.

Prefer this over el.xpath(...) when the expression returns elements: lxml's .xpath() is typed as Any because a single call can return elements, strings, numbers, or booleans depending on the expression. This helper asserts the result is a list of elements, so a mismatched expression or upstream HTML change fails at the call site rather than further down the crawler.

Parameters:

Name	Type	Description	Default
expect_exactly	int | None	If set, raise unless exactly this many elements match.	None

Source code in zavod/helpers/html.py

def xpath_elements(
    el: Element, xpath: str, *, expect_exactly: int | None = None
) -> list[Element]:
    """
    Evaluate an XPath expression and return matching elements as a typed list.

    Prefer this over `el.xpath(...)` when the expression returns elements:
    `lxml`'s `.xpath()` is typed as `Any` because a single call can return
    elements, strings, numbers, or booleans depending on the expression. This
    helper asserts the result is a list of elements, so a mismatched expression
    or upstream HTML change fails at the call site rather than further down
    the crawler.

    Args:
        expect_exactly: If set, raise unless exactly this many elements match.
    """
    result = el.xpath(xpath)
    assert isinstance(result, list), (
        f"Expected list as result of xpath, got {type(result)}"
    )
    element_types = [type(r) for r in result]
    if not all(isinstance(r, Element) for r in result):
        raise ValueError(
            f"Expected list[Element] as result of xpath, got {element_types}"
        )
    if expect_exactly is not None and len(result) != expect_exactly:
        raise ValueError(
            f"Expected {expect_exactly} elements, got {len(result)} for xpath {xpath!r}"
        )
    return [cast(Element, r) for r in result]

xpath_string(el, xpath)

Evaluate an XPath expression and return the single matching string.

Use for text-returning expressions where exactly one result is expected, such as string(.//h1) or .//meta[@name='title']/@content. Raises if there are zero or more than one matches.

Source code in zavod/helpers/html.py

def xpath_string(el: Element, xpath: str) -> str:
    """
    Evaluate an XPath expression and return the single matching string.

    Use for text-returning expressions where exactly one result is expected,
    such as `string(.//h1)` or `.//meta[@name='title']/@content`. Raises if
    there are zero or more than one matches.
    """
    return xpath_strings(el, xpath, expect_exactly=1)[0]

xpath_strings(el, xpath, *, expect_exactly=None)

Evaluate an XPath expression and return matching strings as a typed list.

Use this for expressions that return text rather than elements, such as .//td/text() or attribute selectors like .//@href. Like xpath_elements, this guards against .xpath()'s Any return type by asserting the result is a list of strings.

Parameters:

Name	Type	Description	Default
expect_exactly	int | None	If set, raise unless exactly this many strings match.	None

Source code in zavod/helpers/html.py

def xpath_strings(
    el: Element, xpath: str, *, expect_exactly: int | None = None
) -> list[str]:
    """
    Evaluate an XPath expression and return matching strings as a typed list.

    Use this for expressions that return text rather than elements, such as
    `.//td/text()` or attribute selectors like `.//@href`. Like `xpath_elements`,
    this guards against `.xpath()`'s `Any` return type by asserting the result
    is a list of strings.

    Args:
        expect_exactly: If set, raise unless exactly this many strings match.
    """
    result = el.xpath(xpath)
    if not isinstance(result, list) or not all(isinstance(r, str) for r in result):
        raise ValueError(f"Expected list[str] as result of xpath, got {type(result)}")
    if expect_exactly is not None and len(result) != expect_exactly:
        raise ValueError(
            f"Expected {expect_exactly} elements, got {len(result)} for xpath {xpath!r}"
        )
    return [cast(str, r) for r in result]