# (c) cavaliba.com - data - exporter.py """ Export engine, plus the export-folder viewer helpers. All non-HTTP export logic lives here so that the Exporter UI (app_data.views_exporter), the async task (app_data.tasks.task_export) and - later - the API/CLI share one implementation: request parsing (spec_from_post), sizing (export_count), the selectable column list (export_field_catalog / export_column_catalog), and the writers (export_write / export_render), which stream one instance at a time so a very large export never builds the whole dataset in memory. Method inventory: spec_from_post, export_queryset, export_count, export_field_catalog, export_column_catalog, export_filename, export_content_type, export_write, export_render, list_export_folder, send_file, get_file_content. Private helpers: _post_columns, _static_injected_names, _explicit_columns, _iter_instances, _write_csv, _write_yaml, _write_json, _instance_dict, _enumerate_subfields, _expand_enumerate_columns, _drop_orphan_enumerate. """ import csv import hashlib import io import json import os import re from dataclasses import dataclass from datetime import datetime import yaml from django.conf import settings from django.http import FileResponse from django.utils.translation import gettext as _ from app_data.data import Instance from app_data.dataview import DataView from app_data.schema import Schema from app_data.search import get_instance_from_advanced_query from app_data.yamlutil import MyYamlDumper from app_home.configuration import get_configuration EXPORT_FORMATS = ("csv", "yaml", "json") # builtin instance attributes usable as a column name, in column-selector order EXPORT_BUILTIN_FIELDS = ( "_schema", "keyname", "displayname", "is_enabled", "last_sync", "last_update", ) # columns the column selector always keeps checked: the schema column is what makes # an export re-importable, keyname is what identifies the row EXPORT_MANDATORY_FIELDS = ("_schema", "keyname") # progress is reported (and abort checked) every this many scanned instances BATCH_SIZE = 100 # how many per-instance error messages are kept in the task output / modal MAX_ERROR_KEPT = 20 _SCHEMA_RE = re.compile(r"^[a-zA-Z0-9()_\/\.\s]+$") _QUERY_RE = re.compile(r'^[a-zA-Z0-9()_\/\.\s:\'"*><=-]*$') _COLUMNS_RE = re.compile(r"^[!]?[a-zA-Z0-9_\-\.,\*\s]*$") # -------------------------------------------------------- # ExportSpec - one fully validated export request # -------------------------------------------------------- @dataclass class ExportSpec: """One validated export request, independent of HTTP. `columns` carries three modes: "*" (every schema column), "!a,b" (every column except those - the field_export_match() grammar), or "a,b,c" (exactly those columns, in that order). The first two are "filter mode" and are passed straight through as the `fields` argument of Instance.get_csv_columns()/ get_dict_for_export(); the third is "explicit mode" and emits the named columns verbatim. The Exporter UI always sends an explicit list (its column selector enumerates what it checked); filter mode is kept for API/CLI callers. The schema column (CSV `_schema` / YAML-JSON `classname`) is always emitted - it is what makes an export re-importable, so it is not an option. A YAML/JSON export always carries each exported enumerate field's own subfields ("myenum__widget"), right after it - and only for enumerate fields that are themselves exported, since a subfield without its parent carries no context. They are never emitted in CSV: they come from each row's own selected value, and one fixed header cannot depend on it (see Instance.get_csv_columns()). Naming one explicitly is still honored everywhere, CSV included - the caller then owns the consequences. """ schema: str query: str = "" columns: str = "*" format: str = "csv" refs: list | None = None rev: int | None = None delimiter: str = "," encoding: str = "utf-8" action: str | None = None def to_dict(self): """Plain-dict form, for DataTask.params (Celery JSON serializer safe).""" return { "schema": self.schema, "query": self.query, "columns": self.columns, "format": self.format, "refs": self.refs, "rev": self.rev, "delimiter": self.delimiter, "encoding": self.encoding, "action": self.action, } @classmethod def from_dict(cls, data): """Rebuild a spec from to_dict() output, as stored in DataTask.params. Unknown keys are dropped rather than raising: a DataTask queued by a previous release carries that release's option set (e.g. the retired with_schema_column / with_injected / with_enumerate_subfields), and the worker must still be able to run it after an upgrade. """ known = cls.__dataclass_fields__ return cls(**{k: v for k, v in (data or {}).items() if k in known}) def _post_columns(post): """Read the column list out of a form POST, as a single ExportSpec.columns string. The Exporter's column selector posts one `columns` input per checked column, so the browser sends the name repeated, in the selector's own order - that order is the export's column order. They are joined here into the same comma-separated string an API/CLI caller would pass as a single `columns` value (which also keeps "*" and "!a,b" working for those callers). """ getlist = getattr(post, "getlist", None) values = getlist("columns") if getlist else None if values and len(values) > 1: names = [v.strip() for v in values if v and v.strip()] return ",".join(names) if names else "*" return (post.get("columns") or "").strip() or "*" def spec_from_post(post): """Validate an Exporter form POST into an ExportSpec. Returns (spec, None) or (None, error_message). Permission checks are the caller's job (the schema's own read permission, see views_exporter). Parameters: post (QueryDict): request.POST from the Exporter form Returns: tuple: (ExportSpec, None) on success, (None, str) on a rejected input """ schema_name = (post.get("schema") or "").strip() if not schema_name or not _SCHEMA_RE.match(schema_name): return None, _("Invalid schema") fmt = (post.get("format") or "").strip() if fmt not in EXPORT_FORMATS: return None, _("Invalid format") query = (post.get("query") or "").strip() if not _QUERY_RE.match(query): return None, _("Invalid search query") columns = _post_columns(post) if not _COLUMNS_RE.match(columns): return None, _("Invalid column list") raw_refs = (post.get("refs") or "").strip() refs = [r.strip() for r in raw_refs.split(",") if r.strip()] if raw_refs else None rev = None raw_rev = (post.get("rev") or "").strip() if raw_rev: try: rev = int(raw_rev) except (TypeError, ValueError): return None, _("Revisions must be a number") if rev < 1: return None, _("Revisions must be a positive number") delimiter = (post.get("delimiter") or "").strip() or get_configuration(keyname="CSV_DELIMITER") if fmt == "csv" and len(delimiter) != 1: return None, _("CSV delimiter must be a single character") encoding = (post.get("encoding") or "").strip() or "utf-8" if encoding not in settings.SUPPORTED_ENCODING: return None, _("Unknown CSV encoding: ") + encoding action = (post.get("action") or "").strip() or None if action and action not in settings.ACTION_LIST: return None, _("Invalid action") spec = ExportSpec( schema=schema_name, query=query, columns=columns, format=fmt, refs=refs, rev=rev, delimiter=delimiter, encoding=encoding, action=action, ) return spec, None # -------------------------------------------------------- # sizing / queryset # -------------------------------------------------------- def export_queryset(spec): """ DataInstance queryset for an export, ordered by id. The explicit order_by matters: the writers walk the queryset with .iterator(), and an unordered keyset walk can repeat or skip rows across the DB's internal batches. All filtering stays in the DB (Q objects), see search.py. Parameters: spec (ExportSpec): the export request Returns: QuerySet: DataInstance rows, unsliced, ordered by id """ qs = get_instance_from_advanced_query(classname=spec.schema, query=spec.query) return qs.order_by("id") def export_count(spec): """ Number of instances an export would scan, without loading any of them. This is the figure compared against EXPORT_INTERACTIVE_MAX_SIZE to pick the sync or the async path. It counts rows matched by the query, before each instance's own read permission is checked, so the number of rows actually written can be lower (reported separately as count_ok). Parameters: spec (ExportSpec): the export request Returns: int: matched row count """ return export_queryset(spec).count() def _static_injected_names(schema): """Injected "__" names derivable from the schema alone. Mirrors Instance.inject_schema_subfields(): every "schema" field declares its target and the subfields to inject in dataformat_ext ("TargetClass sub1 sub2"), and a subfield that is itself a schema/external field on the target is skipped (it cannot be injected). Derived statically rather than read off a sample row because the sample row's own reference may be empty - the subfields then exist on no field of that instance and would be missing from the column selector, which is the only place a user can ask for them. Parameters: schema (Schema): the exported schema Returns: list: injected field names, in schema field order """ reply = [] for fieldname in sorted( schema.fields, key=lambda fn: (schema.fields[fn].get("order") or 0, fn) ): fielddef = schema.fields[fieldname] if fielddef.get("dataformat") != "schema": continue parts = (fielddef.get("dataformat_ext") or "").split() if len(parts) < 2: continue target = Schema.from_name(parts[0]) if not target: continue for subfield in parts[1:]: targetdef = target.fields.get(subfield) if not targetdef: continue if targetdef.get("dataformat") in ("schema", "external"): continue reply.append(f"{fieldname}__{subfield}") return reply def export_field_catalog(schema_name, spec=None): """ Field names usable as export columns, grouped, as the source for the UI's column selector (see export_column_catalog()). Declared schema fields come from the Schema. Injected schema subfields ("myuser__email") are derived from the schema too (_static_injected_names()) and unioned with whatever one sample row carries - the sample row is what also surfaces resolved externals, which exist only on a loaded Instance. That union makes the catalog a best-effort hint, never a contract: an external's resolution depends on the row. An export must never fail on a name the catalog did not list, nor on a listed name a given row does not have. Enumerate subfields are not bucketed at all: they are derived from each row's own selected value, so they can never be offered as a fixed column - a YAML/JSON export picks them up from their parent instead (see _expand_enumerate_columns()). Parameters: schema_name (str): classname spec (ExportSpec): optional, to pick the sample row from the same query Returns: dict: {"builtin": [...], "schema": [...], "injected": [...], "external": [...]} """ catalog = { "builtin": list(EXPORT_BUILTIN_FIELDS), "schema": [], "injected": [], "external": [], } schema = Schema.from_name(schema_name) if not schema: return catalog catalog["schema"] = sorted( (fn for fn in schema.fields if not fn.startswith("_")), key=lambda fn: (schema.fields[fn].get("order") or 0, fn), ) catalog["injected"] = _static_injected_names(schema) sample_spec = spec or ExportSpec(schema=schema_name) iobj = export_queryset(sample_spec).first() if not iobj: return catalog instance = Instance(iobj=iobj, expand=True) if not instance: return catalog bucket_by_type = {"schema": "injected", "external": "external"} for fieldname in instance.ordered_fields(): ifield = instance.fields[fieldname] if not ifield.is_injected: continue bucket = bucket_by_type.get(ifield.injected_type) if bucket and fieldname not in catalog[bucket]: catalog[bucket].append(fieldname) return catalog def export_column_catalog(schema_name, spec=None): """ Selectable export columns for the Exporter's column selector, as one flat ordered list - the order is also the export's column order. `_schema` and `keyname` come first and are flagged mandatory: the schema column is what makes an export re-importable and keyname identifies the row, so neither is optional. Then the remaining builtin attributes, the declared schema fields, and the injected subfields (schema references and resolved externals). Enumerate subfields are deliberately absent (export_field_catalog() does not even collect them): they are derived from each row's own selected value, so the list genuinely varies per row and cannot be offered as a fixed column. A YAML/JSON export carries them automatically for every enumerate column selected here. Parameters: schema_name (str): classname spec (ExportSpec): optional, to pick the sample row from the same query Returns: list: dicts with keys name, label, group, mandatory """ if not schema_name: return [] catalog = export_field_catalog(schema_name, spec=spec) schema = Schema.from_name(schema_name) reply = [] seen = set() def add(name, group, label=None, mandatory=False): """Append one selectable column, skipping a name already listed.""" if not name or name in seen: return seen.add(name) reply.append( { "name": name, "label": label or name, "group": group, "mandatory": mandatory, } ) for name in EXPORT_BUILTIN_FIELDS: add(name, "builtin", mandatory=name in EXPORT_MANDATORY_FIELDS) for name in catalog["schema"]: fielddef = schema.fields.get(name, {}) if schema else {} add(name, "schema", label=fielddef.get("displayname") or name) for name in catalog["injected"] + catalog["external"]: add(name, "injected") return reply def export_filename(spec): """Download filename for an export: "-.".""" stamp = datetime.now().strftime("%Y%m%d-%H%M%S") return f"{spec.schema}-{stamp}.{spec.format}" def export_content_type(spec): """HTTP content type for an export, charset-tagged for CSV.""" if spec.format == "csv": return f"text/csv; charset={spec.encoding}" if spec.format == "yaml": return "text/yaml" return "application/json" # -------------------------------------------------------- # writers # -------------------------------------------------------- def _explicit_columns(spec): """Return the typed column list for explicit mode, or None in filter mode ("*" / "!a,b" - both handled by field_export_match() downstream).""" spec_text = (spec.columns or "*").strip() if not spec_text or spec_text == "*" or spec_text.startswith("!"): return None return DataView.parse_dv_fields(spec_text) def _iter_instances(spec, aaa, emit, on_batch=None): """Shared per-instance loop for all three formats. Walks the export queryset with .iterator(), skips instances the caller cannot read, and hands each readable Instance to `emit`. A failure on one instance is counted and recorded instead of aborting the whole export - until failures exceed settings.CAVALIBA_MAX_EXPORT_ERROR, which stops the run the same way an import batch gives up. `on_batch(count_done, total, count_ok, count_ko)` returning "abort" also stops it (that is how a DataTask abort is honored). Returns the stats dict the task stores as its output. """ qs = export_queryset(spec) total = qs.count() max_error = settings.CAVALIBA_MAX_EXPORT_ERROR stats = { "total": total, "count_done": 0, "count_ok": 0, "count_ko": 0, "errors": [], "aborted": False, } for iobj in qs.iterator(): stats["count_done"] += 1 try: instance = Instance(iobj=iobj, expand=True) if instance and instance.has_read_permission(aaa=aaa): emit(instance) stats["count_ok"] += 1 except Exception as exc: stats["count_ko"] += 1 if len(stats["errors"]) < MAX_ERROR_KEPT: keyname = getattr(iobj, "keyname", "?") stats["errors"].append(f"{keyname}: {exc}") if stats["count_ko"] > max_error: stats["aborted"] = True break if on_batch and stats["count_done"] % BATCH_SIZE == 0: reply = on_batch(stats["count_done"], total, stats["count_ok"], stats["count_ko"]) if reply == "abort": stats["aborted"] = True break return stats def _enumerate_subfields(instance, parent): """Names of `parent`'s own enumerate subfields on this instance, in field order. Read off the loaded instance rather than derived from the schema: an enumerate's subfields come from the value the row actually selected (see Instance.inject_enumerate_subfields()), so they differ from row to row. """ if parent not in instance.fields or instance.fields[parent].dataformat != "enumerate": return [] prefix = parent + "__" return [ fieldname for fieldname in instance.ordered_fields() if fieldname.startswith(prefix) and instance.fields[fieldname].is_injected and instance.fields[fieldname].injected_type == "enumerate" ] def _expand_enumerate_columns(instance, columns): """Insert each selected enumerate column's subfields right after it. Subfields follow their parent, with no option of their own: a subfield is exported when the enumerate field it comes from is, so checking `myenum` is what brings `myenum__widget` along. A subfield the caller also named explicitly stays where it was named (the dict keeps the first insertion). """ reply = [] for name in columns: reply.append(name) reply += _enumerate_subfields(instance, name) return reply def _drop_orphan_enumerate(instance, data): """Remove enumerate subfields whose own enumerate field is not in the export. get_dict_for_export() walks ordered_fields(), which already contains every injected field of an expanded instance, so a `columns` filter excluding "myenum" would otherwise still leave its "myenum__widget" behind - a subfield with no parent to read it against. Same rule explicit mode applies by construction (see _expand_enumerate_columns()). """ for fieldname, ifield in instance.fields.items(): if not (ifield.is_injected and ifield.injected_type == "enumerate"): continue if fieldname.split("__", 1)[0] not in data: data.pop(fieldname, None) return data def _instance_dict(spec, instance, explicit, dv): """One YAML/JSON document for an instance, in explicit or filter mode. In explicit mode the CSV-only column names are mapped to their YAML/JSON keys - `_schema` is the `classname` key and `_action` is handled by the spec - rather than resolved as fields, which would emit an empty `_schema` key. """ if explicit is not None: columns = _expand_enumerate_columns( instance, [n for n in explicit if n not in ("_schema", "_action")] ) if "keyname" not in columns: columns.insert(0, "keyname") if "classname" not in columns: columns.insert(0, "classname") data = {name: dv.cell(instance, name, "value") for name in columns} if spec.refs: refs_dict = instance.get_export_refs(refs=spec.refs, selected_fields=columns) if refs_dict: data["_refs"] = refs_dict if spec.rev: revlist = instance.get_export_revisions(rev=spec.rev) if revlist: data["_revision"] = revlist if spec.action: data["_action"] = spec.action return data data = instance.get_dict_for_export( refs=spec.refs, rev=spec.rev, fields=spec.columns, action=spec.action ) return _drop_orphan_enumerate(instance, data) def _write_csv(spec, aaa, out, on_batch=None): """Write a CSV export to `out`, one row at a time. The header is resolved from the first exported instance (filter mode needs a loaded instance to know its injected field names), then reused for every row so the file keeps one fixed header. An export that matched nothing still gets a header, built from an empty instance of the schema. """ writer = csv.writer(out, delimiter=spec.delimiter) explicit = _explicit_columns(spec) dv = DataView() if explicit is not None else None state = {"columns": None} def columns_for(instance): """Resolve the one header row, from the explicit list or from the instance.""" if explicit is not None: # _action first, then _schema: the cell a human edits before # re-importing is the leftmost one in a spreadsheet columns = ["_action"] if spec.action else [] columns.append("_schema") columns += [n for n in explicit if n not in ("_schema", "_action")] return columns # injected=False: in filter mode the header is resolved from the first # exported instance, and an injected subfield only exists on a row whose # reference is set - a row without it would emit one cell less and shift # every column after it. Naming an injected column explicitly is supported # (and is what the Exporter UI does), since each cell is then resolved by # name and an absent one is an empty cell. return instance.get_csv_columns( fields=spec.columns, injected=False, action=spec.action, ) def emit(instance): """Write the header on the first instance, then this instance's row.""" if state["columns"] is None: state["columns"] = columns_for(instance) writer.writerow(state["columns"]) columns = state["columns"] if explicit is None: writer.writerow(instance.get_csv_line(columns, action=spec.action)) return row = [] for name in columns: if name == "_schema": row.append(instance.classname) elif name == "_action": row.append(spec.action or "") else: row.append(dv.cell(instance, name, "csv")) writer.writerow(row) stats = _iter_instances(spec, aaa, emit, on_batch=on_batch) # nothing matched (or nothing was readable): still emit the header, resolved # from an empty instance of the schema, so the download is a valid CSV rather # than a zero-byte file. Skipped for an aborted run - that file is discarded, # and the schema lookup itself may be what failed. if state["columns"] is None and not stats["aborted"]: try: writer.writerow(columns_for(Instance(classname=spec.schema))) except Exception as exc: stats["errors"].append(f"header: {exc}") return stats def _write_yaml(spec, aaa, out, on_batch=None): """Write a YAML export to `out`, one document at a time. Each instance is dumped as a one-element list ("- key: value"), so the concatenation is a single valid top-level sequence - the same shape the old single yaml.dump(datalist) produced, without holding every instance in memory. """ explicit = _explicit_columns(spec) dv = DataView() if explicit is not None else None state = {"count": 0} def emit(instance): """Append one YAML document for this instance.""" data = _instance_dict(spec, instance, explicit, dv) out.write(yaml.dump([data], allow_unicode=True, Dumper=MyYamlDumper, sort_keys=False)) state["count"] += 1 stats = _iter_instances(spec, aaa, emit, on_batch=on_batch) if not state["count"]: out.write("[]\n") return stats def _write_json(spec, aaa, out, on_batch=None): """Write a JSON export to `out`, one document at a time, as a top-level array (documents indented by 4 to match the old json.dumps(datalist, indent=4)).""" explicit = _explicit_columns(spec) dv = DataView() if explicit is not None else None state = {"count": 0} def emit(instance): """Append one JSON object for this instance.""" data = _instance_dict(spec, instance, explicit, dv) chunk = json.dumps(data, indent=4, ensure_ascii=False) chunk = "\n".join(" " + line for line in chunk.splitlines()) out.write("[\n" if not state["count"] else ",\n") out.write(chunk) state["count"] += 1 stats = _iter_instances(spec, aaa, emit, on_batch=on_batch) out.write("\n]\n" if state["count"] else "[]\n") return stats _WRITER_BY_FORMAT = {"csv": _write_csv, "yaml": _write_yaml, "json": _write_json} def export_write(spec, aaa, out, on_batch=None): """ Write an export into a text stream, one instance at a time. This is the single entry point shared by the synchronous view and the async task; nothing above it needs to know about csv/yaml/json. Parameters: spec (ExportSpec): validated export request aaa (dict): auth context; each instance's read permission is checked out (file-like): text stream, opened with spec.encoding for a file target on_batch (callable): progress/abort callback, see _iter_instances() Returns: dict: stats - total, count_done, count_ok, count_ko, errors, aborted """ return _WRITER_BY_FORMAT[spec.format](spec, aaa, out, on_batch=on_batch) def export_render(spec, aaa, on_batch=None): """ Render a small export in memory, for the synchronous download path. Parameters: spec (ExportSpec): validated export request aaa (dict): auth context on_batch (callable): optional progress/abort callback Returns: tuple: (raw bytes, filename, content type, stats dict), or (None, None, None, error_message) if the data cannot be encoded in the requested encoding """ buffer = io.StringIO() stats = export_write(spec, aaa, buffer, on_batch=on_batch) try: raw = buffer.getvalue().encode(spec.encoding) except LookupError: return None, None, None, _("Unknown CSV encoding: ") + spec.encoding except UnicodeEncodeError: return None, None, None, _("Some data cannot be encoded as ") + spec.encoding return raw, export_filename(spec), export_content_type(spec), stats # -------------------------------------------------------- # list_export_folder - List files in export folder # -------------------------------------------------------- # 3.24 def list_export_folder(): """ List all files in the export folder, sorted by filename. Returns: list: List of dicts with keys: filename, size (bytes), creation_date (ISO format) """ export_folder = settings.CAVALIBA_EXPORT_FOLDER files_list = [] if not os.path.exists(export_folder): return files_list for filename in os.listdir(export_folder): # Ignore dot files if filename.startswith("."): continue filepath = os.path.join(str(export_folder), filename) # Only include files, not directories if os.path.isfile(filepath): stat_info = os.stat(filepath) creation_date = datetime.fromtimestamp(stat_info.st_ctime).strftime( "%Y-%m-%d %H:%M:%S" ) # Calculate md5sum of filename filename_md5 = hashlib.md5(filename.encode()).hexdigest() files_list.append( { "filename": filename, "size": stat_info.st_size, "creation_date": creation_date, "filename_md5": filename_md5, } ) # Sort by filename files_list.sort(key=lambda x: x["filename"]) return files_list def _export_folder_filepath(filename): """Resolve a name to a path inside the export folder, or None. Callers match a filename_md5 against list_export_folder() first, so a crafted name should never reach here - this stays as defense in depth: the name is reduced to its basename and the resolved path must still sit directly inside the export folder. """ export_folder = os.path.realpath(str(settings.CAVALIBA_EXPORT_FOLDER)) basename = os.path.basename(filename or "") if not basename or basename.startswith("."): return None filepath = os.path.realpath(os.path.join(export_folder, basename)) if os.path.dirname(filepath) != export_folder: return None if not os.path.isfile(filepath): return None return filepath # -------------------------------------------------------- # send_file - Send file as response # -------------------------------------------------------- def send_file(filename): """ Send a file from the export folder as a downloadable response. Parameters: filename (str): Name of the file to send Returns: FileResponse: Response object for downloading the file, or None if file not found """ filepath = _export_folder_filepath(filename) if not filepath: return None try: response = FileResponse(open(filepath, "rb")) response["Content-Disposition"] = f'attachment; filename="{os.path.basename(filepath)}"' return response except Exception: return None # -------------------------------------------------------- # get_file_content - Get raw file content # -------------------------------------------------------- def get_file_content(filename): """ Read and return the raw content of a file from the export folder. Parameters: filename (str): Name of the file to read Returns: str: Raw content of the file, or None if file not found or error reading """ filepath = _export_folder_filepath(filename) if not filepath: return None try: with open(filepath, encoding="utf-8") as f: content = f.read() if content: return content return "" except Exception: return None