osmapi.dom

  1"""
  2DOM parsing for the OpenStreetMap API.
  3"""
  4
  5from datetime import datetime
  6import xml.dom.minidom
  7import xml.parsers.expat
  8import logging
  9from typing import Any, Union, Optional
 10from xml.dom.minidom import Element
 11
 12from . import errors
 13from . import xmlbuilder
 14
 15logger = logging.getLogger(__name__)
 16
 17
 18def OsmResponseToDom(
 19    response: bytes, tag: str, single: bool = False, allow_empty: bool = False
 20) -> Union[Element, list[Element]]:
 21    """
 22    Returns the (sub-) DOM parsed from an OSM response
 23    """
 24    try:
 25        dom = xml.dom.minidom.parseString(response)
 26        osm_dom = dom.getElementsByTagName("osm")[0]
 27        all_data = osm_dom.getElementsByTagName(tag)
 28        first_element = all_data[0]
 29    except IndexError as e:
 30        if allow_empty:
 31            return []
 32        raise errors.XmlResponseInvalidError(
 33            f"The XML response from the OSM API is invalid: {e!r}"
 34        )
 35    except xml.parsers.expat.ExpatError as e:
 36        raise errors.XmlResponseInvalidError(
 37            f"The XML response from the OSM API is invalid: {e!r}"
 38        )
 39
 40    if single:
 41        return first_element
 42    return list(all_data)
 43
 44
 45def dom_parse_node(dom_element: Element) -> dict[str, Any]:
 46    """
 47    Returns NodeData for the node.
 48    """
 49    result = _dom_get_attributes(dom_element)
 50    result["tag"] = _dom_get_tag(dom_element)
 51    return result
 52
 53
 54def dom_parse_way(dom_element: Element) -> dict[str, Any]:
 55    """
 56    Returns WayData for the way.
 57    """
 58    result = _dom_get_attributes(dom_element)
 59    result["tag"] = _dom_get_tag(dom_element)
 60    result["nd"] = _dom_get_nd(dom_element)
 61    return result
 62
 63
 64def dom_parse_relation(dom_element: Element) -> dict[str, Any]:
 65    """
 66    Returns RelationData for the relation.
 67    """
 68    result = _dom_get_attributes(dom_element)
 69    result["tag"] = _dom_get_tag(dom_element)
 70    result["member"] = _dom_get_member(dom_element)
 71    return result
 72
 73
 74def dom_parse_changeset(
 75    dom_element: Element, include_discussion: bool = False
 76) -> dict[str, Any]:
 77    """
 78    Returns ChangesetData for the changeset.
 79    """
 80    result = _dom_get_attributes(dom_element)
 81    result["tag"] = _dom_get_tag(dom_element)
 82    if include_discussion:
 83        result["discussion"] = _dom_get_discussion(dom_element)
 84
 85    return result
 86
 87
 88def dom_parse_note(dom_element: Element) -> dict[str, Any]:
 89    """
 90    Returns NoteData for the note.
 91    """
 92    result = _dom_get_attributes(dom_element)
 93    result["id"] = xmlbuilder._get_xml_value(dom_element, "id")
 94    result["status"] = xmlbuilder._get_xml_value(dom_element, "status")
 95
 96    result["date_created"] = _parse_date(
 97        xmlbuilder._get_xml_value(dom_element, "date_created")
 98    )
 99    result["date_closed"] = _parse_date(
100        xmlbuilder._get_xml_value(dom_element, "date_closed")
101    )
102    result["comments"] = _dom_get_comments(dom_element)
103
104    return result
105
106
107def _dom_get_attributes(dom_element: Element) -> dict[str, Any]:
108    """
109    Returns a formated dictionnary of attributes of a dom_element.
110    """
111
112    def is_true(v: str) -> bool:
113        return v == "true"
114
115    attribute_mapping: dict[str, Any] = {
116        "uid": int,
117        "changeset": int,
118        "version": int,
119        "id": int,
120        "lat": float,
121        "lon": float,
122        "open": is_true,
123        "visible": is_true,
124        "ref": int,
125        "comments_count": int,
126        "timestamp": _parse_date,
127        "created_at": _parse_date,
128        "closed_at": _parse_date,
129        "date": _parse_date,
130    }
131    result: dict[str, Any] = {}
132    for k, v in dom_element.attributes.items():
133        try:
134            result[k] = attribute_mapping[k](v)
135        except KeyError:
136            result[k] = v
137    return result
138
139
140def _dom_get_tag(dom_element: Element) -> dict[str, str]:
141    """
142    Returns the dictionnary of tags of a dom_element.
143    """
144    result: dict[str, str] = {}
145    for t in dom_element.getElementsByTagName("tag"):
146        k = t.attributes["k"].value
147        v = t.attributes["v"].value
148        result[k] = v
149    return result
150
151
152def _dom_get_nd(dom_element: Element) -> list[int]:
153    """
154    Returns the list of nodes of a dom_element.
155    """
156    result: list[int] = []
157    for t in dom_element.getElementsByTagName("nd"):
158        result.append(int(int(t.attributes["ref"].value)))
159    return result
160
161
162def _dom_get_discussion(dom_element: Element) -> list[dict[str, Any]]:
163    """
164    Returns the dictionnary of comments of a dom_element.
165    """
166    result: list[dict[str, Any]] = []
167    try:
168        discussion = dom_element.getElementsByTagName("discussion")[0]
169        for t in discussion.getElementsByTagName("comment"):
170            comment = _dom_get_attributes(t)
171            comment["text"] = xmlbuilder._get_xml_value(t, "text")
172            result.append(comment)
173    except IndexError:
174        pass
175    return result
176
177
178def _dom_get_comments(dom_element: Element) -> list[dict[str, Any]]:
179    """
180    Returns the list of comments of a dom_element.
181    """
182    result: list[dict[str, Any]] = []
183    for t in dom_element.getElementsByTagName("comment"):
184        comment: dict[str, Any] = {}
185        comment["date"] = _parse_date(xmlbuilder._get_xml_value(t, "date"))
186        comment["action"] = xmlbuilder._get_xml_value(t, "action")
187        comment["text"] = xmlbuilder._get_xml_value(t, "text")
188        comment["html"] = xmlbuilder._get_xml_value(t, "html")
189        comment["uid"] = xmlbuilder._get_xml_value(t, "uid")
190        comment["user"] = xmlbuilder._get_xml_value(t, "user")
191        result.append(comment)
192    return result
193
194
195def _dom_get_member(dom_element: Element) -> list[dict[str, Any]]:
196    """
197    Returns a list of relation members.
198    """
199    result: list[dict[str, Any]] = []
200    for m in dom_element.getElementsByTagName("member"):
201        result.append(_dom_get_attributes(m))
202    return result
203
204
205def _parse_date(date_string: Optional[str]) -> Union[datetime, str, None]:
206    date_formats = ["%Y-%m-%d %H:%M:%S UTC", "%Y-%m-%dT%H:%M:%SZ"]
207    for date_format in date_formats:
208        try:
209            result = datetime.strptime(date_string, date_format)  # type: ignore[arg-type]  # noqa: E501
210            return result
211        except (ValueError, TypeError):
212            logger.debug(f"{date_string} does not match {date_format}")
213
214    return date_string