Spain (NAP)

NAPDownloader

NAPDownloader(api_key: Optional[str] = None)

Bases: BaseGTFSDownloader

Client for searching and downloading feeds from Spain's NAP.

Requires a NAP API key, available after registering at https://nap.transportes.gob.es.

Initialize the client.

Parameters:

Name	Type	Description	Default
api_key	Optional[str]	NAP API key. If not provided, it is read from the NAP_API_KEY environment variable.	None

Raises:

Type	Description
ValueError	If no API key is available.

Source code in pyGTFSHandler/downloaders/spain/nap.py

def __init__(self, api_key: Optional[str] = None):
    """Initialize the client.

    Args:
        api_key: NAP API key. If not provided, it is read from the
            `NAP_API_KEY` environment variable.

    Raises:
        ValueError: If no API key is available.
    """
    super().__init__(api_key=api_key)
    if not self.api_key:
        raise ValueError(
            "A NAP API key is required (pass api_key or set "
            f"{self.API_KEY_ENV_VAR})."
        )
    self.headers = {"ApiKey": self.api_key, "accept": "application/json"}

get_region_id

get_region_id(name: str, region_type: Union[int, str] = 3) -> Optional[int]

Resolve a region name to its NAP region id via fuzzy matching.

Parameters:

Name	Type	Description	Default
name	str	Region name to look up (e.g. a municipality name).	required
region_type	Union[int, str]	Which kind of region to search among: a NAP category name/alias (e.g. "municipio", "province") or its integer index (0=Provincia, 1=ComunidadAutonoma, 2=AreaUrbana, 3=Municipio).	3

Returns:

Type	Description
Optional[int]	The matched region's NAP id, or None if no region of that
Optional[int]	type was found or none matched closely enough.

Source code in pyGTFSHandler/downloaders/spain/nap.py

def get_region_id(self, name: str, region_type: Union[int, str] = 3) -> Optional[int]:
    """Resolve a region name to its NAP region id via fuzzy matching.

    Args:
        name: Region name to look up (e.g. a municipality name).
        region_type: Which kind of region to search among: a NAP
            category name/alias (e.g. "municipio", "province") or its
            integer index (0=Provincia, 1=ComunidadAutonoma,
            2=AreaUrbana, 3=Municipio).

    Returns:
        The matched region's NAP id, or `None` if no region of that
        type was found or none matched closely enough.
    """
    region_type = _resolve_alias(region_type, _REGION_TYPE_ALIASES)
    name = normalize_text(name)

    response = self._get("/Region")
    if response.status_code != 200:
        logger.error(f"Error fetching regions: {response.status_code} - {response.text}")
        return None

    regions = [
        r for r in response.json() if normalize_text(r.get("tipoNombre")) == normalize_text(region_type)
    ]
    if not regions:
        logger.warning(f"No regions found of type '{region_type}'.")
        return None

    region_names = [normalize_text(r["nombre"]) for r in regions]
    match = process.extractOne(name, region_names, scorer=fuzz.token_sort_ratio)
    if not match:
        logger.warning(f"No close match found for {region_type} '{name}'.")
        return None

    _, _, index = match
    return regions[index]["regionId"]

get_transport_type_id

get_transport_type_id(transport_name: str) -> Optional[int]

Resolve a transport mode name to its NAP transport type id.

Parameters:

Name	Type	Description	Default
transport_name	str	Transport mode name or alias (e.g. "bus", "train", "ferry").	required

Returns:

Type	Description
Optional[int]	The matched transport type's NAP id, or None if not found.

Source code in pyGTFSHandler/downloaders/spain/nap.py

def get_transport_type_id(self, transport_name: str) -> Optional[int]:
    """Resolve a transport mode name to its NAP transport type id.

    Args:
        transport_name: Transport mode name or alias (e.g. "bus",
            "train", "ferry").

    Returns:
        The matched transport type's NAP id, or `None` if not found.
    """
    try:
        canonical = _resolve_alias(transport_name, _TRANSPORT_TYPE_ALIASES)
    except ValueError:
        canonical = normalize_text(transport_name)

    response = self._get("/TipoTransporte")
    if response.status_code != 200:
        logger.error(f"Error fetching transport types: {response.status_code}")
        return None

    for transport in response.json():
        if normalize_text(transport["nombre"]) == canonical:
            return transport["tipoTransporteId"]

    logger.warning(f"Transport type '{transport_name}' not found.")
    return None

get_file_type_id

get_file_type_id(file_type: str = 'GTFS') -> Optional[int]

Resolve a file format name (e.g. "GTFS") to its NAP file type id.

Parameters:

Name	Type	Description	Default
file_type	str	File format name, e.g. "GTFS" or "NeTEx".	'GTFS'

Returns:

Type	Description
Optional[int]	The matched file type's NAP id, or None if not found.

Source code in pyGTFSHandler/downloaders/spain/nap.py

def get_file_type_id(self, file_type: str = "GTFS") -> Optional[int]:
    """Resolve a file format name (e.g. "GTFS") to its NAP file type id.

    Args:
        file_type: File format name, e.g. "GTFS" or "NeTEx".

    Returns:
        The matched file type's NAP id, or `None` if not found.
    """
    file_type = normalize_text(file_type)
    response = self._get("/TipoFichero")
    if response.status_code != 200:
        logger.error(f"Error fetching file types: {response.status_code}")
        return None

    for f in response.json():
        if normalize_text(f["nombre"]) == file_type:
            return f["tipoFicheroId"]

    logger.warning(f"File type '{file_type}' not found.")
    return None

get_organization_id

get_organization_id(organization_name: str) -> Optional[int]

Resolve an organization/operator name to its NAP organization id.

Parameters:

Name	Type	Description	Default
organization_name	str	Organization name to look up.	required

Returns:

Type	Description
Optional[int]	The matched organization's NAP id, or None if not found.

Source code in pyGTFSHandler/downloaders/spain/nap.py

def get_organization_id(self, organization_name: str) -> Optional[int]:
    """Resolve an organization/operator name to its NAP organization id.

    Args:
        organization_name: Organization name to look up.

    Returns:
        The matched organization's NAP id, or `None` if not found.
    """
    organization_name = normalize_text(organization_name)
    response = self._get(f"/Organizacion/GetByName/{organization_name}")
    if response.status_code != 200:
        logger.error(f"Error fetching organizations: {response.status_code} - {response.text}")
        return None

    organizations = response.json()
    org_names = [normalize_text(o["nombre"]) for o in organizations]
    match = process.extractOne(organization_name, org_names, scorer=fuzz.token_sort_ratio)
    if not match:
        logger.warning(f"No close match found for organization '{organization_name}'.")
        return None

    _, _, index = match
    return organizations[index]["organizacionId"]

find_datasets

find_datasets(region: Union[int, str, List[Union[int, str]]] = (), transport_type: Union[int, str, List[Union[int, str]]] = (), organization: Union[int, str, List[Union[int, str]]] = (), file_type: Union[int, str] = 'GTFS', region_type: Union[int, str, List[Union[int, str]]] = 3, start_date: Optional[Union[str, datetime]] = None, end_date: Optional[Union[str, datetime]] = None, file_description: Union[str, List[str]] = (), keep: str = 'newest') -> List[Dict[str, Any]]

Find NAP datasets ("conjuntos de datos") matching filters.

Parameters:

Name	Type	Description	Default
region	Union[int, str, List[Union[int, str]]]	Region name(s) or NAP region id(s) to filter by.	()
transport_type	Union[int, str, List[Union[int, str]]]	Transport mode name(s) or NAP transport type id(s) to filter by.	()
organization	Union[int, str, List[Union[int, str]]]	Organization name(s) or NAP organization id(s) to filter by.	()
file_type	Union[int, str]	File format name or NAP file type id. Defaults to "GTFS".	'GTFS'
region_type	Union[int, str, List[Union[int, str]]]	Region category name(s)/id(s) that region values are resolved against (see get_region_id).	3
start_date	Optional[Union[str, datetime]]	If set (with end_date), only keep datasets with a published file whose validity interval covers this range. Accepts "today", date/datetime, or "%d-%m-%Y"/"%d%m%Y" strings.	None
end_date	Optional[Union[str, datetime]]	End of the validity range; see start_date.	None
file_description	Union[str, List[str]]	Substring(s) to match against each dataset's name/description.	()
keep	str	When filtering by date and a dataset has multiple matching files, which to keep: "newest", "oldest", or "all".	'newest'

Returns:

Type	Description
List[Dict[str, Any]]	A list of raw NAP dataset dictionaries (each with a
List[Dict[str, Any]]	ficherosDto list of downloadable files).

Source code in pyGTFSHandler/downloaders/spain/nap.py

def find_datasets(
    self,
    region: Union[int, str, List[Union[int, str]]] = (),
    transport_type: Union[int, str, List[Union[int, str]]] = (),
    organization: Union[int, str, List[Union[int, str]]] = (),
    file_type: Union[int, str] = "GTFS",
    region_type: Union[int, str, List[Union[int, str]]] = 3,
    start_date: Optional[Union[str, datetime]] = None,
    end_date: Optional[Union[str, datetime]] = None,
    file_description: Union[str, List[str]] = (),
    keep: str = "newest",
) -> List[Dict[str, Any]]:
    """Find NAP datasets ("conjuntos de datos") matching filters.

    Args:
        region: Region name(s) or NAP region id(s) to filter by.
        transport_type: Transport mode name(s) or NAP transport type
            id(s) to filter by.
        organization: Organization name(s) or NAP organization id(s)
            to filter by.
        file_type: File format name or NAP file type id. Defaults to
            "GTFS".
        region_type: Region category name(s)/id(s) that `region`
            values are resolved against (see `get_region_id`).
        start_date: If set (with `end_date`), only keep datasets with
            a published file whose validity interval covers this
            range. Accepts "today", `date`/`datetime`, or
            `"%d-%m-%Y"`/`"%d%m%Y"` strings.
        end_date: End of the validity range; see `start_date`.
        file_description: Substring(s) to match against each
            dataset's name/description.
        keep: When filtering by date and a dataset has multiple
            matching files, which to keep: "newest", "oldest", or
            "all".

    Returns:
        A list of raw NAP dataset dictionaries (each with a
        `ficherosDto` list of downloadable files).
    """
    start_date, end_date = input_date(start_date, end_date)

    organization = organization if isinstance(organization, list) else [organization]
    region_type = region_type if isinstance(region_type, list) else [region_type]
    region = region if isinstance(region, list) else [region]
    transport_type = transport_type if isinstance(transport_type, list) else [transport_type]
    file_description = (
        file_description if isinstance(file_description, list) else [file_description]
    )

    organization_ids = [
        self.get_organization_id(o) if isinstance(o, str) else o for o in organization
    ]
    region_ids = [
        self.get_region_id(r, rt) if isinstance(r, str) else r
        for r in region
        for rt in region_type
    ]
    transport_type_ids = [
        self.get_transport_type_id(t) if isinstance(t, str) else t for t in transport_type
    ]
    file_type_id = self.get_file_type_id(file_type) if isinstance(file_type, str) else file_type

    payload = {
        "provincias": region_ids,
        "comunidades": region_ids,
        "areasurbanas": region_ids,
        "municipios": region_ids,
        "tipotransportes": transport_type_ids,
        "tipoficheros": [file_type_id],
        "organizaciones": organization_ids,
    }
    response = requests.post(
        f"{self.BASE_URL}/Fichero/Filter",
        headers={**self.headers, "Content-Type": "application/json"},
        json=payload,
    )
    if response.status_code != 200:
        logger.error(f"Error filtering datasets: {response.status_code} - {response.text}")
        return []

    result = response.json()
    if result.get("filesNum", 0) <= 0:
        return []
    datasets = result["conjuntosDatoDto"]

    if start_date is not None:
        datasets = self._filter_by_dates(datasets, start_date, end_date, keep=keep)

    if file_description:
        needles = [normalize_text(d) for d in file_description]
        datasets = [
            d
            for d in datasets
            if any(
                n in normalize_text(d["descripcion"]) or n in normalize_text(d["nombre"])
                for n in needles
            )
        ]

    return datasets

search_feeds

search_feeds(region: Union[int, str, List[Union[int, str]]] = (), transport_type: Union[int, str, List[Union[int, str]]] = (), organization: Union[int, str, List[Union[int, str]]] = (), file_type: Union[int, str] = 'GTFS', region_type: Union[int, str, List[Union[int, str]]] = 3, start_date: Optional[Union[str, datetime]] = None, end_date: Optional[Union[str, datetime]] = None, file_description: Union[str, List[str]] = (), keep: str = 'newest') -> List[GTFSFeedMetadata]

Search NAP for GTFS feeds matching filters.

See find_datasets for the filter arguments. Each NAP dataset can bundle several downloadable files; this returns one GTFSFeedMetadata per file.

Returns:

Type	Description
List[GTFSFeedMetadata]	A list of GTFSFeedMetadata, one per matching file.

Source code in pyGTFSHandler/downloaders/spain/nap.py

def search_feeds(
    self,
    region: Union[int, str, List[Union[int, str]]] = (),
    transport_type: Union[int, str, List[Union[int, str]]] = (),
    organization: Union[int, str, List[Union[int, str]]] = (),
    file_type: Union[int, str] = "GTFS",
    region_type: Union[int, str, List[Union[int, str]]] = 3,
    start_date: Optional[Union[str, datetime]] = None,
    end_date: Optional[Union[str, datetime]] = None,
    file_description: Union[str, List[str]] = (),
    keep: str = "newest",
) -> List[GTFSFeedMetadata]:
    """Search NAP for GTFS feeds matching filters.

    See `find_datasets` for the filter arguments. Each NAP dataset can
    bundle several downloadable files; this returns one
    `GTFSFeedMetadata` per file.

    Returns:
        A list of `GTFSFeedMetadata`, one per matching file.
    """
    datasets = self.find_datasets(
        region=region,
        transport_type=transport_type,
        organization=organization,
        file_type=file_type,
        region_type=region_type,
        start_date=start_date,
        end_date=end_date,
        file_description=file_description,
        keep=keep,
    )

    feeds: List[GTFSFeedMetadata] = []
    for dataset in datasets:
        main_name = dataset.get("nombre", "")
        for i, file_info in enumerate(dataset.get("ficherosDto", [])):
            suffix = f"_{i + 1}" if i > 0 else ""
            feeds.append(
                GTFSFeedMetadata(
                    id=str(file_info["ficheroId"]),
                    download_url=f"{self.BASE_URL}/Fichero/download/{file_info['ficheroId']}",
                    name=f"{main_name}{suffix}",
                    provider=None,
                    country_code="ES",
                    source=self.SOURCE_NAME,
                    raw={"dataset": dataset, "file": file_info},
                )
            )
    return feeds

download_feeds

download_feeds(feeds: List[GTFSFeedMetadata], download_folder: str, overwrite: bool = False, unzip: bool = True) -> List[str]

Download previously found NAP feeds to a local folder.

Overrides the base implementation to send the ApiKey header required by NAP's download endpoint.

Parameters:

Name	Type	Description	Default
feeds	List[GTFSFeedMetadata]	Feeds to download, as returned by search_feeds().	required
download_folder	str	Directory to store (and, if unzip, extract) the downloaded feeds into.	required
overwrite	bool	If True, re-download and replace files that already exist on disk.	False
unzip	bool	If True, extract each ZIP after downloading and delete the ZIP file.	True

Returns:

Type	Description
List[str]	Absolute paths to the downloaded feeds.

Source code in pyGTFSHandler/downloaders/spain/nap.py

def download_feeds(
    self,
    feeds: List[GTFSFeedMetadata],
    download_folder: str,
    overwrite: bool = False,
    unzip: bool = True,
) -> List[str]:
    """Download previously found NAP feeds to a local folder.

    Overrides the base implementation to send the `ApiKey` header
    required by NAP's download endpoint.

    Args:
        feeds: Feeds to download, as returned by `search_feeds()`.
        download_folder: Directory to store (and, if `unzip`,
            extract) the downloaded feeds into.
        overwrite: If True, re-download and replace files that
            already exist on disk.
        unzip: If True, extract each ZIP after downloading and delete
            the ZIP file.

    Returns:
        Absolute paths to the downloaded feeds.
    """
    return download_feeds(
        feeds, download_folder, overwrite=overwrite, unzip=unzip, headers=self.headers
    )

download_historic_stack

download_historic_stack(output_path: str, datasets: Union[Dict[str, Any], List[Dict[str, Any]]], start_date: Union[str, datetime], end_date: Union[str, datetime], day_separation: int = 1, overwrite: bool = False, aoi: Optional[AOIType] = None) -> List[str]

Download and stitch a dataset's publication history into one feed.

For each dataset, downloads every historic publication whose date falls within [start_date, end_date] (plus the publication immediately preceding it, to cover the start of the range), trims each publication's calendar.txt/calendar_dates.txt so consecutive publications don't overlap, and merges them via pyGTFSHandler.utils.stack_gtfs.historic_stack.

Parameters:

Name	Type	Description	Default
output_path	str	Directory to assemble the stitched feed(s) in.	required
datasets	Union[Dict[str, Any], List[Dict[str, Any]]]	One or more raw NAP dataset dictionaries, as returned by find_datasets.	required
start_date	Union[str, datetime]	Start of the date range to cover. Accepts "today", date/datetime, or "%d-%m-%Y"/"%d%m%Y" strings.	required
end_date	Union[str, datetime]	End of the date range to cover.	required
day_separation	int	Minimum number of days a publication is assumed to stay valid for, if service periods don't force it shorter.	1
overwrite	bool	If True, redo datasets whose stitched output already exists.	False
aoi	Optional[AOIType]	Optional AOI passed through to pyGTFSHandler.utils.stack_gtfs.historic_stack to restrict stops.	None

Returns:

Type	Description
List[str]	Paths to each dataset's stitched
List[str]	{SOURCE_NAME}_{dataset_name}_{start}_{end}.zip.

Source code in pyGTFSHandler/downloaders/spain/nap.py

def download_historic_stack(
    self,
    output_path: str,
    datasets: Union[Dict[str, Any], List[Dict[str, Any]]],
    start_date: Union[str, datetime],
    end_date: Union[str, datetime],
    day_separation: int = 1,
    overwrite: bool = False,
    aoi: Optional[AOIType] = None,
) -> List[str]:
    """Download and stitch a dataset's publication history into one feed.

    For each dataset, downloads every historic publication whose date
    falls within `[start_date, end_date]` (plus the publication
    immediately preceding it, to cover the start of the range), trims
    each publication's `calendar.txt`/`calendar_dates.txt` so
    consecutive publications don't overlap, and merges them via
    `pyGTFSHandler.utils.stack_gtfs.historic_stack`.

    Args:
        output_path: Directory to assemble the stitched feed(s) in.
        datasets: One or more raw NAP dataset dictionaries, as
            returned by `find_datasets`.
        start_date: Start of the date range to cover. Accepts
            "today", `date`/`datetime`, or `"%d-%m-%Y"`/`"%d%m%Y"`
            strings.
        end_date: End of the date range to cover.
        day_separation: Minimum number of days a publication is
            assumed to stay valid for, if service periods don't force
            it shorter.
        overwrite: If True, redo datasets whose stitched output
            already exists.
        aoi: Optional AOI passed through to
            `pyGTFSHandler.utils.stack_gtfs.historic_stack` to restrict
            stops.

    Returns:
        Paths to each dataset's stitched
        `{SOURCE_NAME}_{dataset_name}_{start}_{end}.zip`.
    """
    if not isinstance(datasets, list):
        datasets = [datasets]

    os.makedirs(output_path, exist_ok=True)
    start_date, end_date = input_date(start_date, end_date)
    datasets = copy(datasets)

    for dataset in datasets:
        dataset_id = dataset["conjuntoDatoId"]
        name = normalize_text(dataset["nombre"])
        response = self._get(f"/Fichero/historico/{dataset_id}")
        if response.status_code != 200:
            warnings.warn(f"No historic results for dataset {dataset_id} {name}")

        history = response.json()
        first_entry, first_entry_date = None, None
        historic_entries = []
        for entry in history:
            entry_date = datetime.strptime(entry["fecha"], "%Y-%m-%dT%H:%M:%S")
            if entry_date < start_date and (
                first_entry_date is None or first_entry_date < entry_date
            ):
                first_entry_date, first_entry = entry_date, entry
            if start_date <= entry_date <= end_date:
                historic_entries.append(entry)

        if first_entry is not None:
            historic_entries.append(first_entry)
        dataset["historic_entries"] = historic_entries

    all_paths = []
    for dataset in datasets:
        main_name = sanitize_filename(dataset["nombre"])
        zip_source_name = f"{self.SOURCE_NAME}_{main_name}"
        main_path = os.path.normpath(os.path.join(output_path, main_name))
        final_zip = os.path.join(
            output_path,
            f"{zip_source_name}_{start_date.strftime('%Y%m%d')}_{end_date.strftime('%Y%m%d')}.zip",
        )
        if not overwrite and os.path.isfile(final_zip):
            all_paths.append(final_zip)
            logger.info(f"'{final_zip}' already exists. Skipping.")
            continue

        entries = sorted(
            dataset["historic_entries"],
            key=lambda e: datetime.strptime(e["fecha"], "%Y-%m-%dT%H:%M:%S"),
        )
        if not entries:
            continue

        path_stack = self._download_and_stitch_history(
            entries, main_name, main_path, day_separation, end_date, overwrite
        )
        if not path_stack:
            continue

        if os.path.isfile(path_stack[-1] + ".zip"):
            os.remove(path_stack[-1] + ".zip")

        historic_stack(path_stack, main_path, aoi)
        zip_path = zip_stitched_feed(main_path, zip_source_name, start_date, end_date, output_path)
        logger.info(f"Finished stitching historic dataset '{main_name}'.")
        for f in path_stack:
            if os.path.isfile(f):
                os.remove(f)
            elif os.path.isdir(f):
                shutil.rmtree(f)
        shutil.rmtree(main_path)

        all_paths.append(zip_path)

    return all_paths

utils

Helpers specific to the Spanish NAP downloader (downloaders.spain.nap).

These functions are not reused by any other downloader (unlike downloaders.utils, which holds source-agnostic helpers), which is why they live under downloaders/spain/ instead: they deal with NAP-specific concerns -- parsing the flexible date formats accepted by NAP's find_files, and stitching together consecutive historic NAP GTFS publications (calendar.txt/calendar_dates.txt trimming) ahead of pyGTFSHandler.utils.stack_gtfs.historic_stack.

input_date

input_date(start_date: Optional[Union[str, date, datetime]], end_date: Optional[Union[str, date, datetime]]) -> Tuple[Optional[datetime], Optional[datetime]]

Normalize NAP's flexible start/end date inputs into datetimes.

Accepts None, the literal string "today", date/datetime objects, or strings in %d%m%Y or %d-%m-%Y format. If only one of start_date/end_date is given, the other is filled in with the same value.

Parameters:

Name	Type	Description	Default
start_date	Optional[Union[str, date, datetime]]	Start date, in any of the accepted forms, or None.	required
end_date	Optional[Union[str, date, datetime]]	End date, in any of the accepted forms, or None.	required

Returns:

Type	Description
Optional[datetime]	A tuple (start_date, end_date) of datetime objects, or
Optional[datetime]	(None, None) if both inputs were None.

Source code in pyGTFSHandler/downloaders/spain/utils.py

def input_date(
    start_date: Optional[Union[str, date, datetime]],
    end_date: Optional[Union[str, date, datetime]],
) -> Tuple[Optional[datetime], Optional[datetime]]:
    """Normalize NAP's flexible start/end date inputs into `datetime`s.

    Accepts `None`, the literal string `"today"`, `date`/`datetime`
    objects, or strings in `%d%m%Y` or `%d-%m-%Y` format. If only one of
    `start_date`/`end_date` is given, the other is filled in with the same
    value.

    Args:
        start_date: Start date, in any of the accepted forms, or `None`.
        end_date: End date, in any of the accepted forms, or `None`.

    Returns:
        A tuple `(start_date, end_date)` of `datetime` objects, or
        `(None, None)` if both inputs were `None`.
    """
    if end_date is None:
        end_date = start_date
    elif start_date is None:
        start_date = end_date

    if start_date is None and end_date is None:
        return None, None

    if start_date == "today":
        start_date = datetime.now().strftime("%d-%m-%Y")
    if end_date == "today":
        end_date = datetime.now().strftime("%d-%m-%Y")

    if isinstance(start_date, date) and not isinstance(start_date, datetime):
        start_date = datetime.combine(start_date, datetime.min.time())
    if isinstance(end_date, date) and not isinstance(end_date, datetime):
        end_date = datetime.combine(end_date, datetime.min.time())

    if isinstance(start_date, str):
        try:
            start_date = datetime.strptime(start_date, "%d%m%Y")
        except ValueError:
            start_date = datetime.strptime(start_date, "%d-%m-%Y")

    if isinstance(end_date, str):
        try:
            end_date = datetime.strptime(end_date, "%d%m%Y")
        except ValueError:
            end_date = datetime.strptime(end_date, "%d-%m-%Y")

    return start_date, end_date

resolve_publication_start_date

resolve_publication_start_date(calendar_path: Optional[str], calendar_dates_path: Optional[str], candidate_date: datetime) -> datetime

Determine a historic publication's true service start date.

NAP's /Fichero/historico only exposes a publication/upload date (fecha); that is not necessarily when the feed's own service actually starts (publications are sometimes uploaded ahead of or behind their own calendar.txt/calendar_dates.txt service period). _trim_calendars needs a start date that is actually inside the feed's service period, since it's used as the lower bound when trimming the previous publication's end_date in process_calendar/ process_calendar_dates.

If candidate_date (the publication date) falls inside the feed's own combined calendar.txt/calendar_dates.txt service date range, it is used as-is. Otherwise, it isn't a meaningful anchor for this feed, so the feed's own earliest service date is used instead.

Parameters:

Name	Type	Description	Default
calendar_path	Optional[str]	Path to the feed's calendar.txt, or None/a nonexistent path if the feed has none.	required
calendar_dates_path	Optional[str]	Path to the feed's calendar_dates.txt, or None/a nonexistent path if the feed has none.	required
candidate_date	datetime	The publication date to validate (NAP's fecha).	required

Returns:

Type	Description
datetime	candidate_date if it falls inside the feed's service date
datetime	range, otherwise the feed's earliest service date. Falls back to
datetime	candidate_date unchanged if no service dates could be read from
datetime	either file.

Source code in pyGTFSHandler/downloaders/spain/utils.py

def resolve_publication_start_date(
    calendar_path: Optional[str],
    calendar_dates_path: Optional[str],
    candidate_date: datetime,
) -> datetime:
    """Determine a historic publication's true service start date.

    NAP's `/Fichero/historico` only exposes a publication/upload date
    (`fecha`); that is not necessarily when the feed's own service
    actually starts (publications are sometimes uploaded ahead of or
    behind their own `calendar.txt`/`calendar_dates.txt` service period).
    `_trim_calendars` needs a start date that is actually inside the
    feed's service period, since it's used as the lower bound when
    trimming the previous publication's `end_date` in `process_calendar`/
    `process_calendar_dates`.

    If `candidate_date` (the publication date) falls inside the feed's own
    combined `calendar.txt`/`calendar_dates.txt` service date range, it is
    used as-is. Otherwise, it isn't a meaningful anchor for this feed, so
    the feed's own earliest service date is used instead.

    Args:
        calendar_path: Path to the feed's `calendar.txt`, or `None`/a
            nonexistent path if the feed has none.
        calendar_dates_path: Path to the feed's `calendar_dates.txt`, or
            `None`/a nonexistent path if the feed has none.
        candidate_date: The publication date to validate (NAP's `fecha`).

    Returns:
        `candidate_date` if it falls inside the feed's service date
        range, otherwise the feed's earliest service date. Falls back to
        `candidate_date` unchanged if no service dates could be read from
        either file.
    """
    min_date, max_date = None, None

    if calendar_path and os.path.isfile(calendar_path):
        sep = _sniff_separator(calendar_path)
        df = pd.read_csv(calendar_path, dtype=str, sep=sep)
        df = df.rename(columns={c: normalize_string(c) for c in df.columns})
        if len(df) and "start_date" in df.columns and "end_date" in df.columns:
            starts = pd.to_datetime(df["start_date"], format="%Y%m%d", errors="coerce")
            ends = pd.to_datetime(df["end_date"], format="%Y%m%d", errors="coerce")
            if starts.notna().any():
                min_date = starts.min() if min_date is None else min(min_date, starts.min())
            if ends.notna().any():
                max_date = ends.max() if max_date is None else max(max_date, ends.max())

    if calendar_dates_path and os.path.isfile(calendar_dates_path):
        sep = _sniff_separator(calendar_dates_path)
        df = pd.read_csv(calendar_dates_path, dtype=str, sep=sep)
        df = df.rename(columns={c: normalize_string(c) for c in df.columns})
        if len(df) and "date" in df.columns:
            dates = pd.to_datetime(df["date"], format="%Y%m%d", errors="coerce")
            if dates.notna().any():
                min_date = dates.min() if min_date is None else min(min_date, dates.min())
                max_date = dates.max() if max_date is None else max(max_date, dates.max())

    if min_date is None or max_date is None:
        return candidate_date

    if min_date <= candidate_date <= max_date:
        return candidate_date

    return min_date.to_pydatetime()

process_calendar

process_calendar(path: str, file_date: datetime, possible_dates: List[datetime], day_separation: int, calendar_path: str) -> Tuple[Optional[int], datetime]

Trim a historic calendar.txt so it stops before the next publication.

Used while stitching together successive NAP GTFS publications: caps end_date at whichever comes first among file_date + day_separation days, the file's own minimum end_date, or the next candidate publication date in possible_dates.

Parameters:

Name	Type	Description	Default
path	str	Path to the source calendar.txt.	required
file_date	datetime	Publication date of this GTFS file.	required
possible_dates	List[datetime]	Candidate dates of subsequent publications, in chronological order.	required
day_separation	int	Minimum number of days a publication is assumed to stay valid for, if service periods don't force it shorter.	required
calendar_path	str	Path to write the trimmed calendar.txt to.	required

Returns:

Type	Description
Optional[int]	A tuple (next_index, min_end_date): next_index is the index
datetime	into possible_dates of the publication that should follow this
Tuple[Optional[int], datetime]	one (or None if none qualifies), and min_end_date is the
Tuple[Optional[int], datetime]	computed cutoff date.

Source code in pyGTFSHandler/downloaders/spain/utils.py

def process_calendar(
    path: str,
    file_date: datetime,
    possible_dates: List[datetime],
    day_separation: int,
    calendar_path: str,
) -> Tuple[Optional[int], datetime]:
    """Trim a historic `calendar.txt` so it stops before the next publication.

    Used while stitching together successive NAP GTFS publications: caps
    `end_date` at whichever comes first among `file_date + day_separation`
    days, the file's own minimum `end_date`, or the next candidate
    publication date in `possible_dates`.

    Args:
        path: Path to the source `calendar.txt`.
        file_date: Publication date of this GTFS file.
        possible_dates: Candidate dates of subsequent publications, in
            chronological order.
        day_separation: Minimum number of days a publication is assumed to
            stay valid for, if service periods don't force it shorter.
        calendar_path: Path to write the trimmed `calendar.txt` to.

    Returns:
        A tuple `(next_index, min_end_date)`: `next_index` is the index
        into `possible_dates` of the publication that should follow this
        one (or `None` if none qualifies), and `min_end_date` is the
        computed cutoff date.
    """
    sep = _sniff_separator(path)
    df = pd.read_csv(path, dtype=str, sep=sep)
    df = df.rename(columns={c: normalize_string(c) for c in df.columns})
    df["start_date"] = df["start_date"].astype(int)
    df["end_date"] = df["end_date"].astype(int)
    date_int = int(file_date.strftime("%Y%m%d"))
    df["start_date"] = df["start_date"].apply(lambda x: min(x, date_int))

    if len(possible_dates) == 0:
        return None, file_date

    min_end_date = pd.to_datetime(df["end_date"].astype(str), format="%Y%m%d").min()
    min_end_date = min(file_date + timedelta(days=day_separation), min_end_date)
    min_end_date = min_end_date + timedelta(days=1)

    next_index = None
    for i in range(len(possible_dates)):
        if possible_dates[i] <= min_end_date:
            next_index = i
        else:
            break

    if next_index is not None:
        end_date = possible_dates[next_index]
        end_date_int = int((end_date - timedelta(days=1)).strftime("%Y%m%d"))
        if end_date > file_date:
            df["end_date"] = df["end_date"].apply(lambda x: min(x, end_date_int))
        else:
            df["end_date"] = df["end_date"].apply(lambda x: min(x, date_int))
        df.to_csv(calendar_path, index=False, sep=sep)
        return next_index, min_end_date

    df = df[df["start_date"] <= df["end_date"]]
    df.to_csv(calendar_path, index=False, sep=sep)
    return None, min_end_date

process_calendar_dates

process_calendar_dates(path: str, file_date: datetime, possible_dates: List[datetime], day_separation: int, calendar_dates_path: str, next_index: Optional[int] = None, has_calendar: bool = True) -> Tuple[Optional[int], Optional[datetime]]

Trim a historic calendar_dates.txt to match the stitched period.

Companion to process_calendar, used for GTFS publications that rely on (or supplement with) calendar_dates.txt exceptions rather than a plain weekly calendar.txt pattern.

Parameters:

Name	Type	Description	Default
path	str	Path to the source calendar_dates.txt.	required
file_date	datetime	Publication date of this GTFS file.	required
possible_dates	List[datetime]	Candidate dates of subsequent publications, in chronological order.	required
day_separation	int	Minimum number of days a publication is assumed to stay valid for, if service exceptions don't force it shorter.	required
calendar_dates_path	str	Path to write the trimmed calendar_dates.txt to.	required
next_index	Optional[int]	Index into possible_dates already selected by process_calendar, if this file also has a calendar.txt.	None
has_calendar	bool	Whether this publication also has a calendar.txt (only affects a diagnostic warning).	True

Returns:

Type	Description
Optional[int]	A tuple (next_index, cutoff_date), with the same meaning as
Optional[datetime]	process_calendar's return value.

Source code in pyGTFSHandler/downloaders/spain/utils.py

def process_calendar_dates(
    path: str,
    file_date: datetime,
    possible_dates: List[datetime],
    day_separation: int,
    calendar_dates_path: str,
    next_index: Optional[int] = None,
    has_calendar: bool = True,
) -> Tuple[Optional[int], Optional[datetime]]:
    """Trim a historic `calendar_dates.txt` to match the stitched period.

    Companion to `process_calendar`, used for GTFS publications that rely
    on (or supplement with) `calendar_dates.txt` exceptions rather than a
    plain weekly `calendar.txt` pattern.

    Args:
        path: Path to the source `calendar_dates.txt`.
        file_date: Publication date of this GTFS file.
        possible_dates: Candidate dates of subsequent publications, in
            chronological order.
        day_separation: Minimum number of days a publication is assumed to
            stay valid for, if service exceptions don't force it shorter.
        calendar_dates_path: Path to write the trimmed
            `calendar_dates.txt` to.
        next_index: Index into `possible_dates` already selected by
            `process_calendar`, if this file also has a `calendar.txt`.
        has_calendar: Whether this publication also has a `calendar.txt`
            (only affects a diagnostic warning).

    Returns:
        A tuple `(next_index, cutoff_date)`, with the same meaning as
        `process_calendar`'s return value.
    """
    sep = _sniff_separator(path)
    df = pd.read_csv(path, dtype=str, sep=sep)
    df = df.rename(columns={c: normalize_string(c) for c in df.columns})
    df["date"] = df["date"].astype(int)
    date_int = int(file_date.strftime("%Y%m%d"))
    df = df[df["date"] >= date_int]

    if next_index is not None:
        end_date = possible_dates[next_index]
        end_date_int = int((end_date - timedelta(days=1)).strftime("%Y%m%d"))
        df = df[df["date"] <= end_date_int]
        df.to_csv(calendar_dates_path, index=False, sep=sep)
        return next_index, end_date

    df["exception_type"] = df["exception_type"].astype(int)
    dates = pd.to_datetime(
        df.loc[df["exception_type"] == 1, "date"].astype(str), format="%Y%m%d"
    )
    if len(dates) == 0:
        if not has_calendar:
            warnings.warn(
                f"File {path.replace('calendar_dates.txt', '')} has no calendar.txt, "
                "and calendar_dates.txt has no exception_type 1 rows, so it has no "
                "service dates."
            )
        if next_index is not None and next_index < (len(possible_dates) - 1):
            return next_index + 1, None
        return None, None

    counts = dates.value_counts()
    max_count = counts.max()
    candidates = counts[counts == max_count].index
    min_end_date = candidates.max()
    min_end_date = min(file_date + timedelta(days=day_separation), min_end_date)
    min_end_date = min_end_date + timedelta(days=1)

    new_index = None
    for i in range(len(possible_dates)):
        if possible_dates[i] <= min_end_date:
            new_index = i
        else:
            break

    if new_index is not None:
        end_date = possible_dates[new_index]
        end_date_int = int((end_date - timedelta(days=1)).strftime("%Y%m%d"))
        if end_date > file_date:
            df = df[df["date"] <= end_date_int]
        else:
            df = df[df["date"] <= date_int]
        df.to_csv(calendar_dates_path, index=False, sep=sep)
        return new_index, min_end_date

    df.to_csv(calendar_dates_path, index=False, sep=sep)
    return None, min_end_date