changesets.rs (8202B)
1 use color_eyre::eyre::Result; 2 use quick_xml::{ 3 events::{BytesStart, Event}, 4 name::QName, 5 Reader, 6 }; 7 use std::{ 8 borrow::Cow, 9 collections::{BTreeSet, HashMap}, 10 convert::Infallible, 11 fs::File, 12 io::{BufReader, Write}, 13 }; 14 use tracing::{debug, error, info, warn}; 15 use zstd::stream::Decoder; 16 17 #[derive(Debug, Clone, PartialEq)] 18 pub struct Changeset { 19 pub id: u64, 20 pub created_at: String, 21 pub closed_at: Option<String>, 22 pub open: bool, 23 pub user: String, 24 pub uid: u64, 25 pub min_lat: Option<f64>, 26 pub max_lat: Option<f64>, 27 pub min_lon: Option<f64>, 28 pub max_lon: Option<f64>, 29 pub tags: HashMap<String, String>, 30 } 31 32 impl Changeset { 33 fn new_from_element( 34 reader: &mut Reader<BufReader<Decoder<'_, BufReader<File>>>>, 35 element: &BytesStart, 36 changeset_list: &[u64], 37 ) -> Result<Option<Self>> { 38 let changeset_attributes: HashMap<String, String> = element 39 .attributes() 40 .filter_map(|attr_result| attr_result.ok()) 41 .map(|attr| { 42 let key = reader 43 .decoder() 44 .decode(attr.key.local_name().as_ref()) 45 .or_else(|err| { 46 debug!( 47 "unable to read key in DefaultSettings attribute {:?}, utf8 error {:?}", 48 &attr, err 49 ); 50 Ok::<Cow<'_, str>, Infallible>(std::borrow::Cow::from("")) 51 }) 52 .unwrap() 53 .to_string(); 54 let value = attr 55 .decode_and_unescape_value(reader) 56 .or_else(|err| { 57 debug!( 58 "unable to read key in DefaultSettings attribute {:?}, utf8 error {:?}", 59 &attr, err 60 ); 61 Ok::<Cow<'_, str>, Infallible>(std::borrow::Cow::from("")) 62 }) 63 .unwrap() 64 .to_string(); 65 (key, value) 66 }) 67 .collect(); 68 69 //debug!("changeset_attributes: {:?}", changeset_attributes); 70 71 let id = changeset_attributes.get("id").unwrap().parse().unwrap(); 72 if !changeset_list.contains(&id) { 73 return Ok(None); 74 } 75 76 let mut changeset = Changeset { 77 id: changeset_attributes.get("id").unwrap().parse().unwrap(), 78 created_at: changeset_attributes.get("created_at").unwrap().to_string(), 79 closed_at: changeset_attributes.get("closed_at").map(|s| s.to_string()), 80 open: changeset_attributes.get("open").unwrap().parse().unwrap(), 81 user: changeset_attributes 82 .get("user") 83 .map(|s| s.to_string()) 84 .unwrap_or_else(|| "Unknown".to_string()), 85 uid: changeset_attributes 86 .get("uid") 87 .unwrap_or(&"0".to_string()) 88 .parse() 89 .unwrap(), 90 min_lat: changeset_attributes 91 .get("min_lat") 92 .map(|s| s.parse().unwrap()), 93 max_lat: changeset_attributes 94 .get("max_lat") 95 .map(|s| s.parse().unwrap()), 96 min_lon: changeset_attributes 97 .get("min_lon") 98 .map(|s| s.parse().unwrap()), 99 max_lon: changeset_attributes 100 .get("max_lon") 101 .map(|s| s.parse().unwrap()), 102 tags: HashMap::new(), 103 }; 104 105 let mut new_buf = Vec::new(); 106 107 loop { 108 let event = reader.read_event_into(&mut new_buf)?; 109 110 match event { 111 Event::End(ref e) => { 112 if e.name() == element.name() { 113 break; 114 } 115 } 116 Event::Start(ref e) => { 117 let name = e.name(); 118 if name == QName(b"tag") { 119 let mut key = Cow::Borrowed(""); 120 let mut value = Cow::Borrowed(""); 121 122 for attr_result in element.attributes() { 123 let a = attr_result?; 124 match a.key.as_ref() { 125 b"k" => key = a.decode_and_unescape_value(reader)?, 126 b"v" => value = a.decode_and_unescape_value(reader)?, 127 _ => (), 128 } 129 } 130 131 changeset.tags.insert(key.to_string(), value.to_string()); 132 } else { 133 warn!("Unexpected tag: {:?}", name); 134 //reader.read_to_end_into(e.name(), &mut new_buf); 135 } 136 } 137 _ => { 138 if let Event::Text(ref text) = event { 139 if text.borrow().starts_with(b"\n") { 140 continue; 141 } 142 } else if let Event::End(ref e) = event { 143 if e.name() == QName(b"tag") { 144 continue; 145 } 146 } 147 warn!("Unexpected event in changeset: {:?}", event); 148 // Write the data to file for debugging 149 150 let mut file = std::fs::File::create("debug.xml")?; 151 file.write_all(&new_buf)?; 152 file.sync_all()?; 153 } 154 } 155 new_buf = Vec::new(); 156 } 157 158 Ok(Some(changeset)) 159 } 160 } 161 162 pub fn uncompress_changeset_file<'a>( 163 file: File, 164 ) -> Reader<BufReader<Decoder<'a, BufReader<File>>>> { 165 // Decompress the changeset file 166 info!("Decompressing changeset file"); 167 let reader: BufReader<Decoder<BufReader<File>>> = BufReader::new(Decoder::new(file).unwrap()); 168 Reader::from_reader(reader) 169 } 170 171 pub fn parse_changeset( 172 changeset_data: &mut Reader<BufReader<Decoder<'_, BufReader<File>>>>, 173 changeset_list: &[u64], 174 ) -> Result<Vec<Changeset>> { 175 // == Handling empty elements == 176 // To simply our processing code 177 // we want the same events for empty elements, like: 178 // <DefaultSettings Language="es" Greeting="HELLO"/> 179 // <Text/> 180 changeset_data.expand_empty_elements(true); 181 182 let mut changesets = Vec::new(); 183 let changeset_hashset = changeset_list.iter().cloned().collect::<BTreeSet<u64>>(); 184 let mut buf = Vec::new(); 185 186 // Parse the changeset file 187 info!("Parsing changeset file"); 188 loop { 189 // If we already have all of them then break 190 // We compare the ids even if its a little more expensive 191 let ids_parsed = changesets 192 .iter() 193 .map(|c: &Changeset| c.id) 194 .collect::<BTreeSet<u64>>(); 195 if changeset_hashset.is_subset(&ids_parsed) { 196 break; 197 } 198 199 let event = changeset_data.read_event_into(&mut buf)?; 200 match event { 201 Event::Start(element) => { 202 if let b"changeset" = element.name().as_ref() { 203 // TODO: What do we do in case of an error? 204 let changeset = 205 Changeset::new_from_element(changeset_data, &element, changeset_list); 206 207 match changeset { 208 Ok(Some(changeset)) => { 209 changesets.push(changeset); 210 } 211 Err(err) => { 212 error!( 213 "unable to read changeset element {:?}, utf8 error {:?}", 214 &element, err 215 ); 216 } 217 _ => {} 218 } 219 } 220 //changeset_data.read_to_end_into(element.name(), &mut buf); 221 } 222 Event::Eof => break, // exits the loop when reaching end of file 223 _ => (), // There are `Event` types not considered here 224 } 225 buf = Vec::new(); 226 } 227 Ok(changesets) 228 }