#!/usr/bin/env python3 """Turn a CRM property history export into the normalized transitions CSV. Usage: python3 normalize_history.py long --in FILE.csv --out transitions.csv --object leads \ --map record=LeadId,field=Field,value=NewValue,time=CreatedDate,source=CreatedById \ [--only-fields Status,Owner] [--field-map Status=lead_status,Owner=owner] \ [--keep-values "Owner=^(005|00G)"] python3 normalize_history.py wide --in export.csv --list-sources sources-to-label.csv \ --object contacts --field lifecyclestage --id-column "Record ID" --group value,time,source python3 normalize_history.py wide --in export.csv --out transitions.csv --object contacts \ --field lifecyclestage --id-column "Record ID" --group value,time,source \ [--first-group-column NAME] python3 normalize_history.py jsonl --in history.jsonl --out transitions.csv --object contacts \ --map record=KEY,field=KEY,value=KEY,time=KEY,source=KEY.SUBKEY python3 normalize_history.py self-test Modes: long one change per row (Salesforce history objects, warehouse history tables) wide one row per record with repeating groups in the order given by --group ("skip" ignores a column inside a group); groups start at --first-group-column or at the first column after --id-column jsonl one JSON object per line; --map accepts dot paths such as source.type Common options: --source-map FILE CSV with columns raw,label: replaces raw sources (user emails, integration user IDs, app names) with TYPE:detail labels --list-sources FILE writes every distinct raw source with its count to FILE (columns raw,label,count; label filled from --source-map when known) for the user to label; prints counts only, never values. With --list-sources, --out is optional --value-map FILE CSV with columns raw,label: replaces display labels with the internal values the audit config uses --keep-values FIELD=REGEX (repeatable) keeps a FIELD row only when its raw value matches REGEX, for example Owner=^(005|00G) keeps the ID form of a Salesforce owner change and drops the duplicate name form --assume-tz ZONE zone for timestamps without an offset (default UTC); an IANA name or an offset such as +02:00 --cap N report records with N or more rows for one field --append add rows to an existing output file instead of replacing it old=COLUMN in --map (long, jsonl) also writes each field's starting value: the old value of its oldest change, dated at the record's created row (a row whose field is --created-field, default "created"); history objects that start at the first change need this for T04 and T08 Timestamps: ISO 8601 with Z or an offset (+00:00 or +0000), "YYYY-MM-DD HH:MM[:SS]" read in --assume-tz, and epoch milliseconds. Output is ISO 8601 UTC with Z. Output columns: record_id,object,field,value,changed_at,source, sorted by object, record_id, field and changed_at. Changes with the same timestamp keep the order of the change: file order in long and jsonl mode, oldest group first in wide mode (groups run newest to oldest, left to right). Only these columns are written. A source or value that looks like an email address and is not in --source-map or --value-map becomes UNMAPPED_USER:, so no address reaches the output file. Standard library only. Reads and writes local files, never calls the network. Exit 0 ok, 2 bad input, including a run that finds no usable row. """ import argparse import contextlib import csv import hashlib import io import json import os import re import sys import tempfile from collections import Counter from datetime import datetime, timedelta, timezone try: from zoneinfo import ZoneInfo except ImportError: # Python older than 3.9 ZoneInfo = None OUT_COLUMNS = ["record_id", "object", "field", "value", "changed_at", "source"] EMAIL_RE = re.compile(r"[^@\s]+@[^@\s]+\.[A-Za-z]{2,}") PII_HEADER_RE = re.compile(r"e-?mail|phone|mobile|first ?name|last ?name|full ?name|^name$", re.I) ISO_RE = re.compile( r"^(\d{4})-(\d{2})-(\d{2})[T ](\d{2}):(\d{2})(?::(\d{2})(?:[.,](\d{1,9}))?)?" r"\s*(Z|[+-]\d{2}:?\d{2})?$", re.I) DATE_RE = re.compile(r"^(\d{4})-(\d{2})-(\d{2})$") EPOCH_MS_RE = re.compile(r"^\d{12,14}$") class InputError(Exception): pass def resolve_tz(name): if not name or name.upper() == "UTC": return timezone.utc m = re.match(r"^([+-])(\d{2}):?(\d{2})$", name) if m: sign = 1 if m.group(1) == "+" else -1 try: return timezone(sign * timedelta(hours=int(m.group(2)), minutes=int(m.group(3)))) except ValueError: raise InputError("offset %r is out of range" % name) if ZoneInfo is None: raise InputError("--assume-tz %s needs Python 3.9 or later; pass an offset such as +02:00" % name) try: return ZoneInfo(name) except Exception: raise InputError("unknown time zone %r" % name) def parse_time(text, assume_tz=timezone.utc, stats=None): """Aware UTC datetime for the text, or None when it is empty or unparseable.""" s = str(text if text is not None else "").strip() if not s: return None if EPOCH_MS_RE.match(s): return datetime.fromtimestamp(int(s) / 1000.0, tz=timezone.utc) d = DATE_RE.match(s) if d: if stats is not None: stats["date only (order inside a day is lost)"] += 1 s = s + "T00:00:00" m = ISO_RE.match(s) if not m: return None year, month, day, hour, minute = (int(m.group(i)) for i in range(1, 6)) second = int(m.group(6) or 0) micro = int((m.group(7) or "0")[:6].ljust(6, "0")) zone = m.group(8) try: if zone is None: tz = assume_tz elif zone.upper() == "Z": tz = timezone.utc else: tz = resolve_tz(zone if ":" in zone else zone[:3] + ":" + zone[3:]) return datetime(year, month, day, hour, minute, second, micro, tzinfo=tz).astimezone(timezone.utc) except (ValueError, InputError): return None def format_time(dt): ms = dt.microsecond // 1000 if ms: return dt.strftime("%Y-%m-%dT%H:%M:%S.") + "%03dZ" % ms return dt.strftime("%Y-%m-%dT%H:%M:%SZ") def parse_pairs(text, what): """'a=b,c=d' -> {'a': 'b', 'c': 'd'}.""" out = {} for part in (text or "").split(","): if not part.strip(): continue if "=" not in part: raise InputError("%s entry %r needs the form key=value" % (what, part)) key, value = part.split("=", 1) out[key.strip()] = value.strip() return out def load_map(path, what): if not path: return {} try: with open(path, newline="", encoding="utf-8-sig") as fh: reader = csv.DictReader(fh) if not reader.fieldnames or "raw" not in reader.fieldnames or "label" not in reader.fieldnames: raise InputError("%s %s needs the columns raw,label" % (what, path)) return {(r["raw"] or "").strip(): (r["label"] or "").strip() for r in reader if (r["raw"] or "").strip()} except OSError as exc: raise InputError("cannot read %s: %s" % (path, exc)) def warn_pii(header): flagged = [h for h in header if h and PII_HEADER_RE.search(h.strip())] if flagged: print("warning: the input has columns that may hold personal data (%s). Only the mapped " "columns are written, but delete the input file once you are done." % ", ".join(flagged)) def rows_long(args, tz, stats): mapping = parse_pairs(args.map, "--map") for key in ("record", "value", "time"): if key not in mapping: raise InputError("--map needs %s=COLUMN" % key) if "field" not in mapping and not args.field: raise InputError("long mode needs field=COLUMN in --map or a fixed --field") only = set(x.strip() for x in args.only_fields.split(",")) if args.only_fields else None field_map = parse_pairs(args.field_map, "--field-map") with open(args.inp, newline="", encoding="utf-8-sig") as fh: reader = csv.DictReader(fh) header = reader.fieldnames or [] warn_pii(header) missing = [c for c in mapping.values() if c not in header] if missing: raise InputError("columns not in the header: %s (header: %s)" % (", ".join(missing), ", ".join(header))) for row in reader: stats["rows read"] += 1 yield (row.get(mapping["record"]), row.get(mapping["field"]) if "field" in mapping else args.field, row.get(mapping["value"]), row.get(mapping["time"]), row.get(mapping.get("source", ""), ""), only, field_map, row.get(mapping["old"]) if "old" in mapping else None) def rows_jsonl(args, tz, stats): mapping = parse_pairs(args.map, "--map") for key in ("record", "value", "time"): if key not in mapping: raise InputError("--map needs %s=KEY" % key) if "field" not in mapping and not args.field: raise InputError("jsonl mode needs field=KEY in --map or a fixed --field") only = set(x.strip() for x in args.only_fields.split(",")) if args.only_fields else None field_map = parse_pairs(args.field_map, "--field-map") def dig(obj, path): for part in path.split("."): if not isinstance(obj, dict) or part not in obj: return None obj = obj[part] return obj with open(args.inp, encoding="utf-8-sig") as fh: for line_no, line in enumerate(fh, start=1): if not line.strip(): continue try: obj = json.loads(line) except ValueError: raise InputError("line %d is not valid JSON" % line_no) if line_no == 1: warn_pii(list(obj.keys()) if isinstance(obj, dict) else []) stats["rows read"] += 1 yield (dig(obj, mapping["record"]), dig(obj, mapping["field"]) if "field" in mapping else args.field, dig(obj, mapping["value"]), dig(obj, mapping["time"]), dig(obj, mapping["source"]) if "source" in mapping else "", only, field_map, dig(obj, mapping["old"]) if "old" in mapping else None) def rows_wide(args, tz, stats): groups = [g.strip().lower() for g in args.group.split(",") if g.strip()] if "value" not in groups or "time" not in groups or set(groups) - {"value", "time", "source", "skip"}: raise InputError("--group lists value, time and optionally source and skip, in column order") if not args.field: raise InputError("wide mode needs --field (the property the export is for)") with open(args.inp, newline="", encoding="utf-8-sig") as fh: reader = csv.reader(fh) header = next(reader, None) if not header: raise InputError("%s is empty" % args.inp) warn_pii(header) if args.id_column not in header: raise InputError('id column "%s" not in header (first columns: %s)' % (args.id_column, ", ".join(header[:8]))) id_idx = header.index(args.id_column) if args.first_group_column: if args.first_group_column not in header: raise InputError('column "%s" not in header' % args.first_group_column) start = header.index(args.first_group_column) else: start = id_idx + 1 for row in reader: if not any(cell.strip() for cell in row): continue stats["records read"] += 1 record = row[id_idx] if id_idx < len(row) else "" for g0 in range(start, len(row), len(groups)): cells = (row[g0:g0 + len(groups)] + [""] * len(groups))[:len(groups)] group = dict(zip(groups, cells)) stats["rows read"] += 1 yield (record, args.field, group.get("value", ""), group.get("time", ""), group.get("source", ""), None, {}, None) def parse_keep(items): """['Owner=^(005|00G)'] -> {'Owner': compiled regex}; the field is the raw field name.""" keep = {} for item in items or []: if "=" not in item: raise InputError("--keep-values %r needs the form FIELD=REGEX" % item) field, pattern = item.split("=", 1) try: keep[field.strip()] = re.compile(pattern) except re.error as exc: raise InputError("--keep-values %r: %s" % (item, exc)) return keep def normalize(args, out_rows=None): """Rows are (object, record, field, value, changed_at, source, tiebreak, is_start).""" tz = resolve_tz(args.assume_tz) source_map = load_map(args.source_map, "--source-map") source_map_lower = {k.lower(): v for k, v in source_map.items()} value_map = load_map(args.value_map, "--value-map") keep = parse_keep(getattr(args, "keep_values", None)) stats, sources_masked, values_masked, raw_sources = Counter(), set(), set(), Counter() reader = {"long": rows_long, "wide": rows_wide, "jsonl": rows_jsonl}[args.mode] rows, created, oldest = [], {}, {} created_field = (getattr(args, "created_field", None) or "created").lower() def masked(raw, seen): label = "UNMAPPED_USER:" + hashlib.sha256(raw.lower().encode("utf-8")).hexdigest()[:8] seen.add(label) return label def label_for(raw): raw = str(raw if raw is not None else "").strip() label = source_map.get(raw, source_map_lower.get(raw.lower())) if label is None: label = masked(raw, sources_masked) if EMAIL_RE.search(raw) else raw return label def value_for(raw): """--value-map first; an email-like value it does not map (an owner email) is masked like a source.""" if raw in value_map: return value_map[raw] return masked(raw, values_masked) if EMAIL_RE.search(raw) else raw try: for order, item in enumerate(reader(args, tz, stats)): record, field, value, when_text, source, only, field_map, old = item record = str(record if record is not None else "").strip() field = str(field if field is not None else "").strip() value = str(value if value is not None else "").strip() when_text = str(when_text if when_text is not None else "").strip() if field.lower() == created_field and record: when = parse_time(when_text, tz, stats) if when is not None and (record not in created or when < created[record][0]): created[record] = (when, source) raw_sources[str(source if source is not None else "").strip()] += 1 stats["skipped: created rows (used for starting values)"] += 1 continue if not value and not when_text: stats["skipped: no value and no time"] += 1 continue if not record: stats["skipped: no record id"] += 1 continue if only is not None and field not in only: stats["skipped: field not selected"] += 1 continue if field in keep and not keep[field].search(value): stats["skipped: value filtered by --keep-values"] += 1 continue field = field_map.get(field, field) when = parse_time(when_text, tz, stats) if when is None: stats["skipped: unparseable time"] += 1 continue raw_sources[str(source if source is not None else "").strip()] += 1 # Same-timestamp changes keep the order of the change. Wide exports run newest to # oldest, left to right, so the right-hand (older) group sorts first there. tiebreak = -order if args.mode == "wide" else order rows.append((args.object, record, field, value_for(value), when, label_for(source), tiebreak, False)) if old is not None and ((record, field) not in oldest or when < oldest[(record, field)][0]): oldest[(record, field)] = (when, str(old).strip()) except OSError as exc: raise InputError("cannot read %s: %s" % (args.inp, exc)) for (record, field), (first_change, old_value) in sorted(oldest.items()): if not old_value: continue if record in created and created[record][0] <= first_change: when, source = created[record] rows.append((args.object, record, field, value_for(old_value), when, label_for(source), float("-inf"), True)) stats["starting values written"] += 1 else: stats["skipped: starting value with no created row"] += 1 rows.sort(key=lambda r: (r[0], r[1], r[2], r[4], r[6])) if out_rows is not None: out_rows.extend(rows) elif args.out: exists = args.append and os.path.exists(args.out) and os.path.getsize(args.out) > 0 with open(args.out, "a" if exists else "w", newline="", encoding="utf-8") as fh: writer = csv.writer(fh, lineterminator="\n") if not exists: writer.writerow(OUT_COLUMNS) for obj, record, field, value, when, label, _, _ in rows: writer.writerow([record, obj, field, value, format_time(when), label]) if getattr(args, "list_sources", None): with open(args.list_sources, "w", newline="", encoding="utf-8") as fh: writer = csv.writer(fh, lineterminator="\n") writer.writerow(["raw", "label", "count"]) for raw, n in sorted(raw_sources.items(), key=lambda kv: (-kv[1], kv[0])): writer.writerow([raw, source_map.get(raw, source_map_lower.get(raw.lower(), "")), n]) per_record = Counter((r[1], r[2]) for r in rows if not r[7]) # revisions only, not starting values say = (lambda *a: None) if getattr(args, "quiet", False) else print read = ("%d records, %d value groups read" % (stats["records read"], stats["rows read"]) if args.mode == "wide" else "%d rows read" % stats["rows read"]) say("%s: %s, %d rows %s for %d records and %d fields" % ( args.mode, read, len(rows), "written" if args.out or out_rows is not None else "found", len(set(r[1] for r in rows)), len(set(r[2] for r in rows)))) for reason in sorted(k for k in stats if k.startswith(("skipped", "date only", "starting"))): say(" %s: %d" % (reason, stats[reason])) if sources_masked: say(" %d email-like sources were replaced with UNMAPPED_USER labels; add them to --source-map " "as role labels (for example CRM_UI:rep_a)" % len(sources_masked)) if values_masked: say(" %d email-like values were replaced with UNMAPPED_USER labels; add them to --value-map " "as role labels (for example rep_a)" % len(values_masked)) if getattr(args, "list_sources", None): unlabelled = sum(1 for raw in raw_sources if raw not in source_map and raw.lower() not in source_map_lower) say(" %d distinct sources written to %s, %d without a label; fill the label column with TYPE:detail " "labels and pass the file as --source-map" % (len(raw_sources), os.path.basename(args.list_sources), unlabelled)) if args.cap: at_cap = sorted(k for k, n in per_record.items() if n >= args.cap) say(" records at or above the cap of %d rows: %d%s" % ( args.cap, len(at_cap), (" (for example %s)" % ", ".join(r for r, _ in at_cap[:5])) if at_cap else "")) return rows def self_test(): checks = [] def check(name, condition): checks.append((name, bool(condition))) utc = timezone.utc check("Z suffix", parse_time("2026-07-24T14:00:05Z") == datetime(2026, 7, 24, 14, 0, 5, tzinfo=utc)) check("+0000 offset with milliseconds", parse_time("2026-08-12T14:00:00.250+0000") == datetime(2026, 8, 12, 14, 0, 0, 250000, tzinfo=utc)) check("naive time in an assumed zone", parse_time("2026-07-01 10:00", resolve_tz("America/New_York")) == datetime(2026, 7, 1, 14, 0, tzinfo=utc)) check("epoch milliseconds", parse_time("1782914400000") == datetime(2026, 7, 1, 14, 0, tzinfo=utc)) check("garbage is None", parse_time("next Tuesday") is None) with tempfile.TemporaryDirectory() as tmp: wide = os.path.join(tmp, "wide.csv") with open(wide, "w", newline="") as fh: csv.writer(fh).writerows([ ["Record ID", "v1", "t1", "s1", "v2", "t2", "s2", "v3", "t3", "s3"], ["c-1", "lead", "2026-07-24 14:00:05", "rep.a@example.com", "", "2026-07-24 14:00:00", "wf", "marketingqualifiedlead", "2026-07-10 14:00:00", "wf"], ["c-2", "lead", "2026-07-02 14:00:00", "form", "", "", "", "", "", ""]]) args = argparse.Namespace(mode="wide", inp=wide, out=None, object="contacts", field="lifecyclestage", id_column="Record ID", group="value,time,source", first_group_column=None, source_map=None, value_map=None, assume_tz="UTC", cap=3, append=False, map=None, only_fields=None, field_map=None, quiet=True) got = normalize(args, []) check("wide: empty groups skipped, clear kept", [(r[1], r[3]) for r in got] == [("c-1", "marketingqualifiedlead"), ("c-1", ""), ("c-1", "lead"), ("c-2", "lead")]) check("wide: email source masked", got[2][5].startswith("UNMAPPED_USER:") and "@" not in got[2][5]) same = os.path.join(tmp, "same.csv") with open(same, "w", newline="") as fh: csv.writer(fh).writerows([ ["Record ID", "v1", "t1", "s1", "v2", "t2", "s2", "v3", "t3", "s3"], ["c-3", "lead", "2026-07-24 14:00", "wf", "", "2026-07-24 14:00", "wf", "marketingqualifiedlead", "2026-07-10 14:00", "wf"]]) args.inp, args.source_map, args.list_sources = same, None, os.path.join(tmp, "sources.csv") got = normalize(args, []) check("wide: a clear and a set in the same minute stay in change order", [r[3] for r in got] == ["marketingqualifiedlead", "", "lead"]) with open(args.list_sources, newline="") as fh: listed = list(csv.DictReader(fh)) check("--list-sources writes each raw source with its count and an empty label", listed == [{"raw": "wf", "label": "", "count": "3"}]) longf = os.path.join(tmp, "long.csv") with open(longf, "w", newline="") as fh: csv.writer(fh).writerows([ ["LeadId", "Field", "OldValue", "NewValue", "CreatedDate", "CreatedById"], ["L-1", "created", "", "", "2026-08-12T13:55:00.000+0000", "user-int"], ["L-1", "Owner", "user-a", "user-b", "2026-08-12T15:00:03.000+0000", "user-int"], ["L-1", "Status", "Open", "Working", "2026-08-12T14:00:00.000+0000", "user-a"], ["L-1", "Industry", "", "Retail", "2026-08-12T15:00:01.000+0000", "user-int"]]) smap = os.path.join(tmp, "map.csv") with open(smap, "w", newline="") as fh: csv.writer(fh).writerows([["raw", "label"], ["user-int", "INTEGRATION:enrichment"], ["user-a", "CRM_UI:rep_a"]]) args = argparse.Namespace(mode="long", inp=longf, out=None, object="leads", field=None, map="record=LeadId,field=Field,value=NewValue,time=CreatedDate,source=CreatedById," "old=OldValue", only_fields="Status,Owner", field_map="Status=lead_status,Owner=owner", source_map=smap, value_map=None, assume_tz="UTC", cap=None, append=False, quiet=True) got = normalize(args, []) check("long: fields selected, renamed, sorted, starting values dated at creation", [(r[2], r[3], r[5], format_time(r[4])) for r in got] == [("lead_status", "Open", "INTEGRATION:enrichment", "2026-08-12T13:55:00Z"), ("lead_status", "Working", "CRM_UI:rep_a", "2026-08-12T14:00:00Z"), ("owner", "user-a", "INTEGRATION:enrichment", "2026-08-12T13:55:00Z"), ("owner", "user-b", "INTEGRATION:enrichment", "2026-08-12T15:00:03Z")]) with open(longf, "a", newline="") as fh: csv.writer(fh).writerow(["L-1", "Owner", "Rep A", "Rep B", "2026-08-12T15:00:03.000+0000", "user-int"]) args.keep_values = ["Owner=^user-"] got = normalize(args, []) check("--keep-values drops the name form of a duplicated owner change", [r[3] for r in got if r[2] == "owner"] == ["user-a", "user-b"]) jl = os.path.join(tmp, "h.jsonl") with open(jl, "w") as fh: fh.write(json.dumps({"id": "c-9", "property": "lifecyclestage", "value": "lead", "timestamp": 1782914400000, "source": {"type": "FORM"}}) + "\n") args = argparse.Namespace(mode="jsonl", inp=jl, out=None, object="contacts", field=None, map="record=id,field=property,value=value,time=timestamp,source=source.type", only_fields=None, field_map=None, source_map=None, value_map=None, assume_tz="UTC", cap=None, append=False, quiet=True) got = normalize(args, []) check("jsonl: dot path and epoch time", got and got[0][5] == "FORM" and format_time(got[0][4]) == "2026-07-01T14:00:00Z") with open(jl, "w") as fh: fh.write(json.dumps({"id": "c-9", "property": "hubspot_owner_id", "value": "Rep.C@example.com", "timestamp": 1782914400000, "source": {"type": "CRM_UI"}}) + "\n") got = normalize(args, []) check("an email-like value not in --value-map is masked like a source", got and got[0][3] == "UNMAPPED_USER:" + hashlib.sha256(b"rep.c@example.com").hexdigest()[:8]) check("an out-of-range offset is unparseable, not a crash", parse_time("2026-07-01T10:00:00+25:00") is None) with tempfile.TemporaryDirectory() as tmp: wrong = os.path.join(tmp, "wrong.csv") with open(wrong, "w", newline="") as fh: csv.writer(fh).writerows([["Record ID", "t1", "v1", "s1"], ["c-1", "2026-07-01 10:00", "lead", "form"]]) with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()): code = main(["wide", "--in", wrong, "--out", os.path.join(tmp, "o.csv"), "--object", "contacts", "--field", "lifecyclestage", "--group", "value,time,source"]) check("a wrong --group that leaves no usable row exits 2", code == 2) failed = [name for name, ok in checks if not ok] for name in failed: print("FAILED: %s" % name) print("self-test %s (%d checks)" % ("passed" if not failed else "failed", len(checks))) return 1 if failed else 0 def main(argv): if argv[:1] == ["self-test"]: return self_test() parser = argparse.ArgumentParser(description="Normalize a CRM property history export.") parser.add_argument("mode", choices=["long", "wide", "jsonl"]) parser.add_argument("--in", dest="inp", required=True) parser.add_argument("--out", help="transitions CSV to write (optional with --list-sources)") parser.add_argument("--object", required=True, help="object name written to every row, e.g. contacts") parser.add_argument("--field", help="fixed field name (wide mode, or when the file has no field column)") parser.add_argument("--map", help="record=COL,field=COL,value=COL,time=COL,source=COL (long, jsonl)") parser.add_argument("--only-fields", help="comma list of raw field names to keep (long, jsonl)") parser.add_argument("--field-map", help="Raw=normalized,... renames fields (long, jsonl)") parser.add_argument("--id-column", default="Record ID", help="record ID column (wide)") parser.add_argument("--group", default="value,time,source", help="column order inside one group (wide)") parser.add_argument("--first-group-column", help="header of the first column of the first group (wide)") parser.add_argument("--source-map") parser.add_argument("--list-sources", help="write distinct raw sources (raw,label,count) to this CSV") parser.add_argument("--value-map") parser.add_argument("--keep-values", action="append", help="FIELD=REGEX: keep FIELD rows whose raw value matches") parser.add_argument("--assume-tz", default="UTC") parser.add_argument("--cap", type=int) parser.add_argument("--append", action="store_true") parser.add_argument("--created-field", default="created", help="field name of creation rows (long, jsonl)") args = parser.parse_args(argv) try: if not args.out and not args.list_sources: raise InputError("pass --out FILE, --list-sources FILE, or both") if not normalize(args): raise InputError("no usable rows in %s: check --group or --map against the header" % args.inp) except InputError as exc: print("error: %s" % exc, file=sys.stderr) return 2 except OSError as exc: print("error: %s" % exc, file=sys.stderr) return 2 return 0 if __name__ == "__main__": try: code = main(sys.argv[1:]) except Exception as exc: # malformed input that slipped past the checks above print("error: could not process the input (%s: %s)" % (type(exc).__name__, exc), file=sys.stderr) code = 2 sys.exit(code)