Maintainers
This module is maintained by the OCA.
-
+
+
+
OCA, or the Odoo Community Association, is a nonprofit organization whose mission is to support the collaborative development of Odoo features and promote its widespread use.
diff --git a/sale_forecast/wizards/__init__.py b/sale_forecast/wizards/__init__.py index 4021f4303409..a8f6960ea393 100644 --- a/sale_forecast/wizards/__init__.py +++ b/sale_forecast/wizards/__init__.py @@ -1,3 +1,4 @@ from . import sale_forecast_sheet from . import sale_forecast_sheet_line from . import sale_forecast_wizard +from . import wizard_sale_forecast_import diff --git a/sale_forecast/wizards/wizard_sale_forecast_import.py b/sale_forecast/wizards/wizard_sale_forecast_import.py new file mode 100644 index 000000000000..509d81df1ad8 --- /dev/null +++ b/sale_forecast/wizards/wizard_sale_forecast_import.py @@ -0,0 +1,183 @@ +# pylint: disable=no-member,protected-access,invalid-name,no-self-use +import base64 +import logging +from datetime import datetime + +from odoo import _, fields, models +from odoo.exceptions import UserError + +_logger = logging.getLogger(__name__) # pylint: disable=invalid-name + + +class WizardSaleForecastImport(models.TransientModel): + """Import sale forecast""" + + _name = "wizard.sale.forecast.import" + _description = "Import sale forecast records" + + file_import = fields.Binary("Import Forecast") + file_name = fields.Char("file name") + + def action_process_import(self): + """Actually process the uploaded file to import it.""" + self.ensure_one() + if not self.file_import: + raise UserError(_("Please attach a file containing product information.")) + ( + rows, + date_headers, + default_code_index, + date_index, + key_index, + ) = self._import_file() + aggregate_info = self._aggregate_info( + rows, date_headers, default_code_index, date_index, key_index + ) + self._process_import(aggregate_info) + + def _import_file(self): + def get_field_index(header_row, name): + """Get index of column in input file.""" + try: + index = header_row.index(name) + return index + except ValueError as error: + raise UserError( + _("Row header name %s is not found in file") % name + ) from error + + self.ensure_one() + lst = self._get_rows() + if not lst or not lst[0]: + raise UserError(_("Import file is empty or unreadable")) + rows = lst[1] + header_row = rows[1] + date_headers = header_row[4:] + product_headers = header_row[:4] + (product_category_index, product_index, default_code_index, key_index,) = ( + get_field_index(product_headers, name) + for name in [ + "Product Category", + "Product", + "Item Code (SKU)", + "Key", + ] + ) + date_index = [] + date_index += [get_field_index(header_row, name) for name in date_headers] + return rows, date_headers, default_code_index, date_index, key_index + + def _get_rows(self): + """Get rows from data_file.""" + self.ensure_one() + import_model = self.env["base_import.import"] + data_file = base64.b64decode(self.file_import) + importer = import_model.create({"file": data_file, "file_name": self.file_name}) + return importer._read_file({"quoting": '"', "separator": ","}) + + def _aggregate_info( + self, rows, date_headers, default_code_index, date_index, key_index + ): + aggregate_info = dict() + for row in rows[2:]: + if row[key_index].strip() == "Total": + continue + aggregate_info[row[default_code_index]] = [] + for date, index in zip(date_headers, date_index): + quantity = ( + float(row[index].replace(",", "").replace(".", "")) + if row[index] + else 0 + ) + if quantity <= 0: + continue + date_forecast = self._date_to_object(date.strip()) + if not date_forecast: + continue + aggregate_info[row[default_code_index]].append( + ( + row[key_index].strip(), + date_forecast, + quantity, + ) + ) + return aggregate_info + + def _date_to_object(self, date): + """No expired dates""" + date_object = datetime.strptime(date, "%b-%y") + if date_object.date() < fields.Date.today(): + return False + return date_object + + def _process_import(self, rows): + forecast_model = self.env["sale.forecast"] + location_model = self.env["stock.location"] + location_dict = { + "Sales": location_model.browse(25), + "CS consumption": location_model.browse(30), + "AS consumption": location_model.browse(18), + } + for default_code, location_date_quantity in rows.items(): + product = self.env["product.product"].search( + [("default_code", "=", default_code)] + ) + if not product: + _logger.warning( + "No product with default code %s exists.", + default_code, + ) + continue + if not location_date_quantity: + continue + for location, date, quantity in location_date_quantity: + if location not in location_dict.keys(): + _logger.warning( + "No location %s exists.", + location, + ) + continue + location_id = location_dict.get(location) + if not location_id: + _logger.warning( + "No location %s exists.", + location, + ) + continue + date_range_id = self._get_date_range_id(date) + if not date_range_id: + _logger.warning( + "No monthly date range exists for %s.", + date.strftime("%b-%y"), + ) + continue + vals = { + "product_id": product.id, + "location_id": location_id.id, + "product_uom_qty": quantity, + "date_range_id": date_range_id.id, + } + existing_forecast = forecast_model.search( + [ + ("product_id", "=", vals["product_id"]), + ("date_range_id", "=", vals["date_range_id"]), + ("location_id", "=", vals["location_id"]), + ] + ) + if existing_forecast: + _logger.warning( + "Forecast for product %s, location %s, date %s exists, updating...", + (product.name, location, date.strftime("%b-%y")), + ) + existing_forecast.write(vals) + continue + forecast_model.create(vals) + + def _get_date_range_id(self, date): + date_range_domain = [ + ("date_start", "<=", date), + ("date_end", ">", date), + ("type_name", "ilike", "Monthly"), + ("active", "=", True), + ] + return self.env["date.range"].search(date_range_domain) diff --git a/sale_forecast/wizards/wizard_sale_forecast_import.xml b/sale_forecast/wizards/wizard_sale_forecast_import.xml new file mode 100644 index 000000000000..711f0427754d --- /dev/null +++ b/sale_forecast/wizards/wizard_sale_forecast_import.xml @@ -0,0 +1,42 @@ + +