You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

249 lines
9.4 KiB

  1. # Copyright 2014-2016 Akretion (Alexis de Lattre
  2. # <alexis.delattre@akretion.com>)
  3. # Copyright 2014 Lorenzo Battistini <lorenzo.battistini@agilebg.com>
  4. # Copyright 2017 ForgeFlow, S.L. <.com>
  5. # Copyright 2018 Aitor Bouzas <aitor.bouzas@adaptivecity.com>
  6. # Copyright 2016-2020 Tecnativa - Pedro M. Baeza
  7. # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
  8. import csv
  9. import io
  10. import logging
  11. import os
  12. import tempfile
  13. import zipfile
  14. import requests
  15. from odoo import _, api, fields, models
  16. from odoo.exceptions import UserError
  17. logger = logging.getLogger(__name__)
  18. class CityZipGeonamesImport(models.TransientModel):
  19. _name = "city.zip.geonames.import"
  20. _description = "Import City Zips from Geonames"
  21. country_ids = fields.Many2many("res.country", string="Countries")
  22. letter_case = fields.Selection(
  23. [("unchanged", "Unchanged"), ("title", "Title Case"), ("upper", "Upper Case")],
  24. string="Letter Case",
  25. default="unchanged",
  26. help="Converts retreived city and state names to Title Case "
  27. "(upper case on each first letter of a word) or Upper Case "
  28. "(all letters upper case).",
  29. )
  30. @api.model
  31. def transform_city_name(self, city, country):
  32. """Override it for transforming city name (if needed)
  33. :param city: Original city name
  34. :param country: Country record
  35. :return: Transformed city name
  36. """
  37. res = city
  38. if self.letter_case == "title":
  39. res = city.title()
  40. elif self.letter_case == "upper":
  41. res = city.upper()
  42. return res
  43. @api.model
  44. def _domain_search_city_zip(self, row, city_id=False):
  45. domain = [("name", "=", row[1])]
  46. if city_id:
  47. domain += [("city_id", "=", city_id)]
  48. return domain
  49. @api.model
  50. def select_state(self, row, country):
  51. code = row[country.geonames_state_code_column or 4]
  52. return self.env["res.country.state"].search(
  53. [("country_id", "=", country.id), ("code", "=", code)], limit=1
  54. )
  55. @api.model
  56. def select_city(self, row, country, state_id):
  57. # This has to be done by SQL for performance reasons avoiding
  58. # left join with ir_translation on the translatable field "name"
  59. self.env.cr.execute(
  60. "SELECT id, name FROM res_city "
  61. "WHERE name = %s AND country_id = %s AND state_id = %s LIMIT 1",
  62. (self.transform_city_name(row[2], country), country.id, state_id),
  63. )
  64. row_city = self.env.cr.fetchone()
  65. return (row_city[0], row_city[1]) if row_city else (False, False)
  66. @api.model
  67. def select_zip(self, row, country, state_id):
  68. city_id, _ = self.select_city(row, country, state_id)
  69. return self.env["res.city.zip"].search(
  70. self._domain_search_city_zip(row, city_id)
  71. )
  72. @api.model
  73. def prepare_state(self, row, country):
  74. return {
  75. "name": row[country.geonames_state_name_column or 3],
  76. "code": row[country.geonames_state_code_column or 4],
  77. "country_id": country.id,
  78. }
  79. @api.model
  80. def prepare_city(self, row, country, state_id):
  81. vals = {
  82. "name": self.transform_city_name(row[2], country),
  83. "state_id": state_id,
  84. "country_id": country.id,
  85. }
  86. return vals
  87. @api.model
  88. def prepare_zip(self, row, city_id):
  89. vals = {"name": row[1], "city_id": city_id}
  90. return vals
  91. @api.model
  92. def get_and_parse_csv(self, country):
  93. country_code = country.code
  94. config_url = self.env["ir.config_parameter"].get_param(
  95. "geonames.url", default="http://download.geonames.org/export/zip/%s.zip"
  96. )
  97. url = config_url % country_code
  98. logger.info("Starting to download %s" % url)
  99. res_request = requests.get(url)
  100. if res_request.status_code != requests.codes.ok:
  101. raise UserError(
  102. _("Got an error %d when trying to download the file %s.")
  103. % (res_request.status_code, url)
  104. )
  105. f_geonames = zipfile.ZipFile(io.BytesIO(res_request.content))
  106. tempdir = tempfile.mkdtemp(prefix="odoo")
  107. f_geonames.extract("%s.txt" % country_code, tempdir)
  108. data_file = open(
  109. os.path.join(tempdir, "%s.txt" % country_code), "r", encoding="utf-8"
  110. )
  111. data_file.seek(0)
  112. reader = csv.reader(data_file, delimiter=" ")
  113. parsed_csv = [row for i, row in enumerate(reader)]
  114. data_file.close()
  115. logger.info("The geonames zipfile has been decompressed")
  116. return parsed_csv
  117. def _create_states(self, parsed_csv, search_states, max_import, country):
  118. # States
  119. state_vals_list = []
  120. state_dict = {}
  121. for i, row in enumerate(parsed_csv):
  122. if max_import and i == max_import:
  123. break
  124. state = self.select_state(row, country) if search_states else False
  125. if not state:
  126. state_vals = self.prepare_state(row, country)
  127. if state_vals not in state_vals_list:
  128. state_vals_list.append(state_vals)
  129. else:
  130. state_dict[state.code] = state.id
  131. created_states = self.env["res.country.state"].create(state_vals_list)
  132. for i, vals in enumerate(state_vals_list):
  133. state_dict[vals["code"]] = created_states[i].id
  134. return state_dict
  135. def _create_cities(
  136. self, parsed_csv, search_cities, max_import, state_dict, country
  137. ):
  138. # Cities
  139. city_vals_list = []
  140. city_dict = {}
  141. for i, row in enumerate(parsed_csv):
  142. if max_import and i == max_import:
  143. break
  144. state_id = state_dict[row[country.geonames_state_code_column or 4]]
  145. city_id, city_name = (
  146. self.select_city(row, country, state_id)
  147. if search_cities
  148. else (False, False)
  149. )
  150. if not city_id:
  151. city_vals = self.prepare_city(row, country, state_id)
  152. if city_vals not in city_vals_list:
  153. city_vals_list.append(city_vals)
  154. else:
  155. city_dict[(city_name, state_id)] = city_id
  156. ctx = dict(self.env.context)
  157. ctx.pop("lang", None) # make sure no translation is added
  158. created_cities = self.env["res.city"].with_context(ctx).create(city_vals_list)
  159. for i, vals in enumerate(city_vals_list):
  160. city_dict[(vals["name"], vals["state_id"])] = created_cities[i].id
  161. return city_dict
  162. def run_import(self):
  163. for country in self.country_ids:
  164. parsed_csv = self.get_and_parse_csv(country)
  165. self._process_csv(parsed_csv, country)
  166. return True
  167. def _process_csv(self, parsed_csv, country):
  168. state_model = self.env["res.country.state"]
  169. zip_model = self.env["res.city.zip"]
  170. res_city_model = self.env["res.city"]
  171. # Store current record list
  172. old_zips = set(zip_model.search([("city_id.country_id", "=", country.id)]).ids)
  173. search_zips = len(old_zips) > 0
  174. old_cities = set(res_city_model.search([("country_id", "=", country.id)]).ids)
  175. search_cities = len(old_cities) > 0
  176. current_states = state_model.search([("country_id", "=", country.id)])
  177. search_states = len(current_states) > 0
  178. max_import = self.env.context.get("max_import", 0)
  179. logger.info("Starting to create the cities and/or city zip entries")
  180. # Pre-create states and cities
  181. state_dict = self._create_states(parsed_csv, search_states, max_import, country)
  182. city_dict = self._create_cities(
  183. parsed_csv, search_cities, max_import, state_dict, country
  184. )
  185. # Zips
  186. zip_vals_list = []
  187. for i, row in enumerate(parsed_csv):
  188. if max_import and i == max_import:
  189. break
  190. # Don't search if there aren't any records
  191. zip_code = False
  192. state_id = state_dict[row[country.geonames_state_code_column or 4]]
  193. if search_zips:
  194. zip_code = self.select_zip(row, country, state_id)
  195. if not zip_code:
  196. city_id = city_dict[
  197. (self.transform_city_name(row[2], country), state_id)
  198. ]
  199. zip_vals = self.prepare_zip(row, city_id)
  200. if zip_vals not in zip_vals_list:
  201. zip_vals_list.append(zip_vals)
  202. else:
  203. old_zips.remove(zip_code.id)
  204. self.env["res.city.zip"].create(zip_vals_list)
  205. if not max_import:
  206. if old_zips:
  207. logger.info("removing city zip entries")
  208. self.env["res.city.zip"].browse(list(old_zips)).unlink()
  209. logger.info(
  210. "%d city zip entries deleted for country %s"
  211. % (len(old_zips), country.name)
  212. )
  213. old_cities -= set(city_dict.values())
  214. if old_cities:
  215. logger.info("removing city entries")
  216. self.env["res.city"].browse(list(old_cities)).unlink()
  217. logger.info(
  218. "%d res.city entries deleted for country %s"
  219. % (len(old_cities), country.name)
  220. )
  221. logger.info(
  222. "The wizard to create cities and/or city zip entries from "
  223. "geonames has been successfully completed."
  224. )
  225. return True