DRYing parse_atom_georss

03_berlin.py

-import xml.etree.ElementTree as ET
-from datetime import datetime
-
-from citygml_download import TMP_DIR, Bundesland, download_all_files, download_file
-from zipfile_download import ZipFile
+from citygml_download import Bundesland, download_all_files
+from georss_download import parse_atom_georss
 
 BERLIN = Bundesland(
     "Berlin",
@@ -12,33 +9,14 @@ BERLIN = Bundesland(
 )
 
 if __name__ == "__main__":
-    atom_path = TMP_DIR / f"{BERLIN}.atom"
-    download_file(BERLIN.source, TMP_DIR / atom_path)
-    tree = ET.parse(atom_path)
-    root = tree.getroot()
-    zip_files = []
-    ns = {"atom": "http://www.w3.org/2005/Atom"}
+    zip_files = parse_atom_georss(BERLIN)
+    import rich
 
-    updated_node = root.find("atom:updated", ns)
-    if updated_node is None:
-        raise ValueError("No date has been found")
-    updated_date = updated_node.text
-    if updated_date is None:
-        raise ValueError("No date has been found")
-    source_date = datetime.strptime(updated_date, "%Y-%m-%dT%H:%M:%SZ")
-    for file_elem in root.findall("atom:entry/atom:link", ns):
-        href = file_elem.attrib["href"]
-        if href.endswith(".zip"):
-            zip_files.append(
-                ZipFile(
-                    url=href,
-                    bundesland=BERLIN,
-                    source_date=source_date,
-                )
-            )
+    rich.print(zip_files)
+    exit()
 
     download_all_files(zip_files)
     # NOTE: Extracted files are called xml. Rename them to gml:
-    for xml_path in BERLIN.download_folder.glob('*.xml'):
-        gml_path = xml_path.with_suffix('.gml')
+    for xml_path in BERLIN.download_folder.glob("*.xml"):
+        gml_path = xml_path.with_suffix(".gml")
         xml_path.rename(gml_path)

14_sachsen_anhalt.py

@@ -2,8 +2,8 @@
 # NOTE: Multiple sources are available. Large ZIP might be hard to update
 # ATOM : https://geodatenportal.sachsen-anhalt.de/arcgisinspire/rest/directories/web/INSPIRE_ALKIS/ALKIS_LOD2_BU/datasetlod2.xml
 
-from citygml_download import TMP_DIR, Bundesland, download_all_files, download_file
-from zipfile_download import ZipFile
+from citygml_download import Bundesland, download_all_files
+from georss_download import parse_atom_georss
 
 SACHSEN_ANHALT = Bundesland(
     "Sachsen-Anhalt",
@@ -12,3 +12,10 @@ SACHSEN_ANHALT = Bundesland(
     license="dl-de/by-2-0",
     kontakt="Landesamt für Vermessung und Geoinformation Sachsen-Anhalt <service.lvermgeo@sachsen-anhalt.de>",
 )
+
+
+if __name__ == "__main__":
+    zip_files = parse_atom_georss(SACHSEN_ANHALT)
+    import rich
+    rich.print(zip_files)
+    # download_all_files(zip_files)

georss_download.py

+import xml.etree.ElementTree as ET
+from datetime import datetime
+
+from citygml_download import TMP_DIR, Bundesland, download_file
+from zipfile_download import ZipFile
+
+
+def parse_atom_georss(bundesland: Bundesland) -> list[ZipFile]:
+    atom_path = TMP_DIR / f"{bundesland}.atom"
+    download_file(bundesland.source, atom_path)
+
+    tree = ET.parse(atom_path)
+    root = tree.getroot()
+    zip_files = []
+
+    ns = {"atom": "http://www.w3.org/2005/Atom"}
+
+    updated_node = root.find("atom:updated", ns)
+    if updated_node is None:
+        raise ValueError("No date has been found")
+    updated_date = updated_node.text
+    if updated_date is None:
+        raise ValueError("No date has been found")
+    source_date = datetime.strptime(updated_date, "%Y-%m-%dT%H:%M:%SZ")
+    for file_elem in root.findall("atom:entry/atom:link", ns):
+        href = file_elem.attrib["href"]
+        if href.endswith(".zip"):
+            zip_files.append(
+                ZipFile(
+                    url=href,
+                    bundesland=bundesland,
+                    source_date=source_date,
+                )
+            )
+    return zip_files