# (c) cavaliba.com - data - views_import.py """ Two independent tools: - texteditor: the RAW YAML/JSON Code Editor - prefills from an existing Schema or Instance ("Edit as YAML" links) or starts blank, auto-detects YAML vs JSON on paste. - import_tool: the Import Tool page - CSV file import and YAML/JSON file import sections. Each optionally runs entries through a Pipeline, then either just verifies parsing or imports via load_broker. Method inventory: texteditor, import_tool. Private helpers: _parse_rawdata, _import_csv, _import_file, _import_and_report, _report_csv_import. """ import json import yaml from django.conf import settings from django.contrib import messages from django.shortcuts import redirect, render from django.utils.translation import gettext as _ from app_data.aaa import start_view from app_data.data import Instance from app_data.filestore import tmp_local_filepath from app_data.forms import CSVImportForm, DataUploadForm from app_data.loader import load_broker, load_file_csv, load_file_json, load_file_yaml from app_data.pipeline import Pipeline from app_data.schema import Schema from app_home.log import DEBUG, ERROR, INFO, log # ------------------------------------------------------------------------- # RAW YAML/JSON CODE EDITOR # ------------------------------------------------------------------------- def texteditor(request, classname=None, id=None): """GET: prefill the editor with a schema's or instance's YAML, or start blank. POST: parse the textarea (auto-detecting YAML vs JSON), then either verify or import via load_broker (optionally through a Pipeline).""" context = start_view( request, app="data", view="texteditor", noauth="app_sirene:private", perm="p_data_import", noauthz="app_home:private", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] rawdata = "" # Get existing data or empty if request.method == "GET": # no id : => schema from classname # id : => instance # Schema if not id: schema = Schema.from_name(classname) try: rawdata = schema.to_yaml() except Exception: pass # Instance else: instance = Instance.from_id(id) # instance.print() if instance: rawdata = instance.to_yaml() elif request.method == "POST": rawdata = request.POST.get("rawdata", "") pipeline_name = request.POST.get("pipeline") or None pipeline = Pipeline.from_name(pipeline_name) if pipeline_name else None datalist, error = _parse_rawdata(rawdata) if error: messages.add_message(request, messages.ERROR, error) elif type(datalist) is not list: messages.add_message( request, messages.ERROR, _("Import failed - content is not a list") ) else: submit = request.POST.get("submit") if submit == "verify": messages.add_message(request, messages.SUCCESS, _("Check ok")) log(DEBUG, aaa=aaa, app="data", view="import", action="check", status="OK") elif submit == "import": _import_and_report(request, aaa, datalist, pipeline) return redirect("app_home:private") # edit context["rawdata"] = rawdata context["pipelines"] = Pipeline.list(is_enabled=True) return render(request, "app_data/texteditor.html", context) def _parse_rawdata(rawdata): """Parse RAW content as JSON or YAML, auto-detecting the format: if it looks like JSON (starts with "{" or "["), try json.loads first for a precise error message; otherwise, and as a fallback if that fails, use yaml.safe_load (which also accepts valid JSON). Returns (datalist, None) or (None, exception).""" stripped = rawdata.lstrip() if stripped[:1] in ("{", "["): try: return json.loads(rawdata), None except Exception: pass # fall through to YAML try: return yaml.safe_load(rawdata), None except Exception as e: return None, e # ------------------------------------------------------------------------- # IMPORT TOOL (CSV file + YAML/JSON file) # ------------------------------------------------------------------------- def import_tool(request): """GET: render the Import Tool page. POST: dispatch on mode=csv/file to the matching section handler.""" context = start_view( request, app="data", view="import_tool", noauth="app_sirene:private", perm="p_data_import", noauthz="app_home:private", ) if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if request.method == "POST": mode = request.POST.get("mode") if mode == "csv": redirect_to_home = _import_csv(request, aaa) elif mode == "file": redirect_to_home = _import_file(request, aaa) else: redirect_to_home = False if redirect_to_home: return redirect("app_home:private") context["upload_form"] = DataUploadForm() context["csv_form"] = CSVImportForm() context["pipelines"] = Pipeline.list(is_enabled=True) context["schemas"] = [s for s in Schema.listall() if s.has_create_permission(aaa)] return render(request, "app_data/import.html", context) # ------------------------------------------------------------------------- # CSV FILE IMPORT SECTION # ------------------------------------------------------------------------- def _import_csv(request, aaa): """Parse the uploaded CSV via load_file_csv (optional pipeline/schema/encoding/ separator), then either verify (report parse errors, no DB writes) or import (apply pipeline per row, load_broker per row, abort past CAVALIBA_CSV_IMPORT_MAX_FAILURE). Returns True if the caller should redirect home.""" form = CSVImportForm(request.POST, request.FILES) if not form.is_valid(): messages.add_message(request, messages.ERROR, _("Import failed - invalid form")) return False pipeline_name = request.POST.get("pipeline") or None pipeline = Pipeline.from_name(pipeline_name) if pipeline_name else None schema_name = request.POST.get("schema") or None encoding = form.cleaned_data.get("encoding") or "utf-8" csv_delimiter = form.cleaned_data.get("separator") or "," postfile = request.FILES["file"] filename = tmp_local_filepath() with open(filename, "wb+") as destination: for chunk in postfile.chunks(): destination.write(chunk) parse_errors = [] datalist = load_file_csv( filename=filename, pipeline_name=pipeline_name, schema_name=schema_name, encoding=encoding, csv_delimiter=csv_delimiter, errors=parse_errors, ) if datalist is None: datalist = [] submit = request.POST.get("submit") if submit == "verify": status = "OK" if not parse_errors else "KO" messages.add_message( request, messages.SUCCESS if not parse_errors else messages.WARNING, _("Check ok") + f" ({len(datalist)} rows, {len(parse_errors)} error(s))", ) for err in parse_errors[:20]: messages.add_message(request, messages.ERROR, f"row {err['row']}: {err['message']}") log(DEBUG, aaa=aaa, app="data", view="import", action="check", status=status) return False if submit == "import": count_ok = 0 count_ko = len(parse_errors) errors = list(parse_errors) aborted = False for datadict in datalist: if count_ko > settings.CAVALIBA_CSV_IMPORT_MAX_FAILURE: aborted = True break if pipeline: status = pipeline.apply(datadict) if status == "discard": continue result = load_broker([datadict], aaa=aaa) if result.get("errors"): count_ko += 1 errors.extend(result["errors"]) else: count_ok += 1 _report_csv_import(request, aaa, count_ok, count_ko, errors, aborted) return True return False def _report_csv_import(request, aaa, count_ok, count_ko, errors, aborted): """Report CSV import results: ok/ko counts, an early-abort note if the failure count crossed CAVALIBA_CSV_IMPORT_MAX_FAILURE, and the first 20 error messages.""" if aborted: messages.add_message( request, messages.ERROR, _("Import aborted - too many failures") + f" (ok={count_ok}, ko={count_ko}, max={settings.CAVALIBA_CSV_IMPORT_MAX_FAILURE})", ) elif count_ko: messages.add_message( request, messages.ERROR, _("Import partial or failed") + f" (ok={count_ok}, ko={count_ko})", ) else: messages.add_message(request, messages.SUCCESS, _("Import OK") + f" ({count_ok})") for err in errors[:20]: if isinstance(err, dict): messages.add_message(request, messages.ERROR, f"row {err['row']}: {err['message']}") else: messages.add_message(request, messages.ERROR, str(err)) log( INFO, aaa=aaa, app="data", view="import", action="import", status="KO" if (count_ko or aborted) else "OK", data=f"ok={count_ok} ko={count_ko} aborted={aborted}", ) # ------------------------------------------------------------------------- # YAML/JSON FILE IMPORT SECTION # ------------------------------------------------------------------------- def _import_file(request, aaa): """Parse the uploaded .yml/.yaml/.json file, then either verify or import (apply pipeline per row, load_broker per row). Returns True if the caller should redirect home.""" pipeline_name = request.POST.get("pipeline") or None pipeline = Pipeline.from_name(pipeline_name) if pipeline_name else None fileform = DataUploadForm(request.POST, request.FILES) if not fileform.is_valid(): messages.add_message(request, messages.ERROR, _("Import failed - invalid file")) return False postfile = request.FILES["file"] filename = tmp_local_filepath() with open(filename, "wb+") as destination: for chunk in postfile.chunks(): destination.write(chunk) if postfile.name.endswith(".yml") or postfile.name.endswith(".yaml"): datalist = load_file_yaml(filename=filename, pipeline_name=pipeline_name) elif postfile.name.endswith(".json"): datalist = load_file_json(filename=filename, pipeline_name=pipeline_name) else: messages.add_message(request, messages.ERROR, _("Import failed - unsupported file type")) return False if type(datalist) is not list: messages.add_message(request, messages.ERROR, _("Import failed - invalid file")) log( ERROR, aaa=aaa, app="data", view="import", action="file", status="KO", data=f"Content is not a list : {postfile.name}", ) return False log( INFO, aaa=aaa, app="data", view="import", action="file", status="OK", data=f"loaded: {postfile.name}", ) submit = request.POST.get("submit") if submit == "verify": messages.add_message(request, messages.SUCCESS, _("Check ok")) log(DEBUG, aaa=aaa, app="data", view="import", action="check", status="OK") return False if submit == "import": _import_and_report(request, aaa, datalist, pipeline) return True return False # ------------------------------------------------------------------------- # shared import loop (texteditor + file section) # ------------------------------------------------------------------------- def _import_and_report(request, aaa, datalist, pipeline): """Apply an optional Pipeline per entry then load via load_broker; report the ok/ko counts as a Django message. Shared by the RAW editor and the file import section.""" count_ok = 0 count_ko = 0 for datadict in datalist: if pipeline: status = pipeline.apply(datadict) if status == "discard": continue result = load_broker([datadict], aaa=aaa) if result.get("errors"): count_ko += result.get("count_ko", 0) else: count_ok += result.get("count_ok", 0) if count_ko: messages.add_message( request, messages.ERROR, _("Import partial or failed") + f" (ok={count_ok}, ko={count_ko})", ) log( DEBUG, aaa=aaa, app="data", view="import", action="import", status="KO", data=f"ok={count_ok} ko={count_ko}", ) else: messages.add_message(request, messages.SUCCESS, _("Import OK") + f" ({count_ok})") log( DEBUG, aaa=aaa, app="data", view="import", action="import", status="OK", data=f"imported: {count_ok} objects", )