Skip to content
Extraits de code Groupes Projets

[UPD] get all import functions in this module

Fusionnées Rémi - Le Filament a demandé de fusionner 16.0-upd-remi vers 16.0
3 files
+ 208
132
Comparer les modifications
  • Côte à côte
  • En ligne

Fichiers

+ 132
98
@@ -2,6 +2,7 @@
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html)
from datetime import datetime
from os.path import splitext
from dateutil.relativedelta import relativedelta
from odoo import _, models
@@ -40,119 +41,152 @@ class AccOperation(models.Model):
# ------------------------------------------------------
# Actions
# ------------------------------------------------------
def import_check_docs(self, files):
# ------------------------------------------------------
# Business methods
# ------------------------------------------------------
def _import_check_docs(self, model):
raise_error = False
error_msg = ""
mess = ""
if not model:
raise ValidationError(_("Le type de courbes à importer n'est pas défini"))
self.ensure_one()
files = self.env["ir.attachment"].search(
[("res_model", "=", "acc.operation"), ("res_id", "=", self.id)]
)
if not files:
raise ValidationError(
raise UserError(
_(
"Aucun fichier à intégrer. Pour joindre des fichiers, "
"veuillez cliquer sur l’icône trombone se trouvant dans "
"le bloc Chatter à droite"
"le bloc Chatter à droite (ou en-dessous en fonction de "
"la résolution de votre écran)."
)
)
else:
files.mapped("name")
for file in files:
file_suffix = splitext(file.name)[1]
file_name = splitext(file.name)[0]
if file_suffix != ".csv":
raise ValidationError(
_(
"Le fichier %s ne semble pas être au bon format. "
"Format CSV attendu."
)
% file.name
)
data_filename = file_name.split("_")
for file in files:
file_suffix = splitext(file.name)[1]
file_name = splitext(file.name)[0]
if len(data_filename) != 5:
raise ValidationError(
_(
"Le fichier %s ne semble pas correspondre "
"au fichier attendu. Ex de nom de fichier correct: "
"'23555861012080_09022023_08032023_Conso_CDC'."
)
% file.name
if file_suffix != ".csv":
raise_error = True
error_msg += (
_(
"Le fichier %s ne semble pas être au bon format. "
"Format CSV attendu.\n"
)
% file.name
)
date_begin_str = data_filename[1]
date_end_str = data_filename[2]
# Contrôles sur le type de données CDC
if data_filename[3] not in ["Prod", "Conso", "Autoconso", "Surplus"]:
raise UserError(
_(
"Le fichier %s ne correspond à aucun type "
"de courbes de charge. il doit contenir Prod, Conso, "
"Autoconso ou Surplus."
)
% file.name
)
data_filename = file_name.split("_")
computed_data_type = data_filename[3].lower()
# Contrôle sur le type de données
file_suffix = data_filename[4]
if file_suffix != "CDC":
raise UserError(
_(
"Le fichier %s n'a pas le bon type de format, "
"il doit se terminer par 'CDC'."
)
% file.name
if len(data_filename) != 5:
raise_error = True
error_msg += (
_(
"Le fichier %s ne semble pas correspondre "
"au fichier attendu. Ex de nom de fichier correct: "
"'23555861012080_09022023_08032023_Conso_CDC'.\n"
)
# Contrôle PRM
id_prm = data_filename[0]
# Vérification existance PRM
counter_id = self.env["acc.counter"].search([("name", "=", id_prm)])
if not counter_id:
raise ValidationError(
_("Fichier %(file)s : L'ID PRM %(prm)s ne semble pas exister.")
% {"file": file.name, "prm": id_prm}
% file.name
)
continue
date_begin_str = data_filename[1]
date_end_str = data_filename[2]
# Contrôles sur le type de données CDC
if data_filename[3] not in ["Prod", "Conso", "Autoconso", "Surplus"]:
raise_error = True
error_msg += (
_(
"Le fichier %s ne correspond à aucun type "
"de courbes de charge. il doit contenir Prod, Conso, "
"Autoconso ou Surplus."
)
% file.name
)
date_format = "%d%m%Y"
date_begin_obj = datetime.strptime(date_begin_str, date_format)
date_end_obj = datetime.strptime(
date_end_str, date_format
) + relativedelta(days=1)
# Vérification si des données existent déjà
# pour cet intervalle de date
start_datetime = local_to_utc(date_begin_obj, "Europe/Paris")
end_datetime = local_to_utc(date_end_obj, "Europe/Paris")
domain_all = [
("date_slot", ">=", start_datetime),
("date_slot", "<", end_datetime),
("acc_operation_id", "=", self.id),
("comp_data_type", "=", computed_data_type),
("acc_counter_id", "=", counter_id.id),
]
rec_ids = self.env["acc.enedis.cdc"].search(domain_all)
if rec_ids:
mess += (
"<p>Fichier "
+ file.name
+ ": des données existent déjà pour la période du "
+ str(date_begin_obj)
+ " au "
+ str(date_end_obj)
+ ". Etes-vous sûr de vouloir écraser "
"les données existantes?</p>"
computed_data_type = data_filename[3].lower()
if computed_data_type == "conso":
computed_data_type = "cons"
elif computed_data_type == "autoconso":
computed_data_type = "autocons"
# Contrôle sur le type de données
file_suffix = data_filename[4]
if file_suffix != "CDC":
raise_error = True
error_msg += (
_(
"Le fichier %s n'a pas le bon type de format, "
"il doit se terminer par 'CDC'.\n"
)
else:
mess += (
"<p>Fichier "
+ file.name
+ ": <strong>Prêt à importer</strong></p>"
)
return mess
% file.name
)
# ------------------------------------------------------
# Business methods
# ------------------------------------------------------
# Contrôle PRM
id_prm = data_filename[0]
# Vérification existence PRM
counter_id = self.env["acc.counter"].search([("name", "=", id_prm)])
if not counter_id:
raise_error = True
error_msg += _(
"Fichier %(file)s : L'ID PRM %(prm)s ne semble pas exister.\n"
) % {"file": file.name, "prm": id_prm}
date_format = "%d%m%Y"
date_begin_obj = datetime.strptime(date_begin_str, date_format)
date_end_obj = datetime.strptime(date_end_str, date_format) + relativedelta(
days=1
)
# Vérification si des données existent déjà
# pour cet intervalle de date
start_datetime = local_to_utc(date_begin_obj, "Europe/Paris")
end_datetime = local_to_utc(date_end_obj, "Europe/Paris")
domain_all = [
("date_slot", ">=", start_datetime),
("date_slot", "<", end_datetime),
("acc_operation_id", "=", self.id),
("comp_data_type", "=", computed_data_type),
("acc_counter_id", "=", counter_id.id),
]
rec_ids = self.env[model].search(domain_all)
if rec_ids:
mess += (
"<p>Fichier "
+ file.name
+ ": des données existent déjà pour la période du "
+ str(date_begin_obj)
+ " au "
+ str(date_end_obj)
+ ". Etes-vous sûr de vouloir écraser "
"les données existantes ?</p>"
)
else:
mess += (
"<p>Fichier " + file.name + ": <strong>Prêt à importer</strong></p>"
)
if raise_error:
raise ValidationError(error_msg)
if mess:
wizard = self.env["acc.operation.import.wizard"].create(
{
"operation_id": self.id,
"attachment_ids": files,
"model": model,
"message": mess,
}
)
return {
"name": "Confirmation d'import",
"type": "ir.actions.act_window",
"view_mode": "form",
"res_model": "acc.operation.import.wizard",
"res_id": wizard.id,
"target": "new",
}
Chargement en cours