#!/usr/bin/env python3 """Export CRM conversations created in [since, until) to JSONL (Python 3.10+). Example: python3 export_history.py --since 2026-06-01 --until 2026-09-01 Required environment: CRM_BASE_URL, CRM_ACCOUNT_ID, CRM_API_TOKEN. This is a paginated read, not a transactional snapshot. Reconcile live accounts. """ import argparse from datetime import date, datetime, time, timedelta, timezone import json import os from pathlib import Path import random import time as clock from urllib.error import HTTPError, URLError from urllib.parse import urlencode, urlsplit from urllib.request import Request, urlopen, HTTPRedirectHandler, build_opener class NoRedirect(HTTPRedirectHandler): # Never forward a token to a redirect destination. def redirect_request(self, req, fp, code, msg, headers, newurl): return None class Client: def __init__(self): base = os.environ['CRM_BASE_URL'].rstrip('/') parsed = urlsplit(base) if parsed.scheme != 'https' or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path: raise ValueError('CRM_BASE_URL must be an HTTPS origin, without credentials or path') account = os.environ['CRM_ACCOUNT_ID'] if not account.isdigit(): raise ValueError('CRM_ACCOUNT_ID must be numeric') self.base = f'{base}/api/v1/accounts/{account}' self.headers = {'api_access_token': os.environ['CRM_API_TOKEN'], 'Accept': 'application/json'} self.opener = build_opener(NoRedirect()) self.last_request = 0.0 def read(self, path, *, params=None, body=None): # POST is used only for the read-only conversations/filter endpoint. url = self.base + path + ('?' + urlencode(params) if params else '') data = json.dumps(body).encode() if body is not None else None headers = dict(self.headers) if data is not None: headers['Content-Type'] = 'application/json' for attempt in range(5): clock.sleep(max(0, 1.2 - (clock.monotonic() - self.last_request))) self.last_request = clock.monotonic() try: with self.opener.open(Request(url, data=data, headers=headers), timeout=30) as response: return json.load(response) except HTTPError as error: retry = error.code in (429, 500, 502, 503, 504) error.close() if not retry or attempt == 4: raise RuntimeError(f'HTTP {error.code} reading {path}; export incomplete') from None except (URLError, TimeoutError): if attempt == 4: raise RuntimeError(f'Network error reading {path}; export incomplete') from None clock.sleep(min(2 ** (attempt + 1), 30) + random.uniform(0, 1)) raise RuntimeError('Read attempts exhausted') def messages(client, conversation_id): collected = {} before = None while True: params = {'history_scan': 'true'} if before is not None: params['before'] = before page = client.read(f'/conversations/{conversation_id}/messages', params=params)['payload'] if not page: break oldest = min(message['id'] for message in page) if before is not None and oldest >= before: raise RuntimeError('Message cursor did not advance; export incomplete') for message in page: collected[message['id']] = message before = oldest return sorted(collected.values(), key=lambda message: (message['created_at'], message['id'])) def export(client, since, until, destination): # Built-in date filters cast to date and use strict >/< comparisons. # Widen by a day at either end, then select the exact UTC interval locally. filters = {'payload': [ {'attribute_key': 'created_at', 'filter_operator': 'is_greater_than', 'values': [(since - timedelta(days=1)).isoformat()], 'query_operator': 'AND'}, {'attribute_key': 'created_at', 'filter_operator': 'is_less_than', 'values': [(until + timedelta(days=1)).isoformat()], 'query_operator': None}, ]} start = datetime.combine(since, time.min, timezone.utc).timestamp() end = datetime.combine(until, time.min, timezone.utc).timestamp() partial = destination.with_name(destination.name + '.partial') # Exclusive creation protects previous files, including an incomplete run. if destination.exists(): raise FileExistsError(f'{destination} already exists; choose another output') seen = set() scanned = set() count = 0 page_number = 1 # Set permissions in the creation syscall, before any other process can open it. with open(partial, 'x', encoding='utf-8', opener=lambda path, flags: os.open(path, flags, 0o600)) as output: while True: page = client.read('/conversations/filter', params={'page': page_number}, body=filters)['payload'] if not page: break page_ids = {conversation['id'] for conversation in page} if page_ids <= scanned: raise RuntimeError('Conversation pages repeated; export incomplete') scanned.update(page_ids) for conversation in page: cid = conversation['id'] if cid in seen or not start <= conversation['created_at'] < end: continue seen.add(cid) record = {'conversation': conversation, 'messages': messages(client, cid)} output.write(json.dumps(record, ensure_ascii=False) + '\n') output.flush() count += 1 page_number += 1 # Link fails if another process created destination during the run. os.link(partial, destination) partial.unlink() return count def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--since', type=date.fromisoformat, required=True, help='Inclusive UTC date (YYYY-MM-DD)') parser.add_argument('--until', type=date.fromisoformat, required=True, help='Exclusive UTC date (YYYY-MM-DD)') parser.add_argument('--output', type=Path, default=Path('crm-history.jsonl')) args = parser.parse_args() if args.since >= args.until: parser.error('--since must precede --until') try: count = export(Client(), args.since, args.until, args.output) except (KeyError, ValueError, OSError, RuntimeError) as error: parser.exit(1, f'Export stopped: {error}\nAny .partial file is incomplete.\n') print(f'Exported {count} conversations to {args.output}. Reconcile if the account changed during the scan.') if __name__ == '__main__': main()