"""Read-only 1C OData v3 export. No credentials or rows are logged.""" import argparse import base64 import csv import getpass import io import json import os import re import ssl import sys import urllib.error import urllib.parse import urllib.request MAX_BYTES = 10 * 1024 * 1024 MAX_ROWS = 20000 class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise ValueError('Redirect refused; configure the final HTTPS publication address.') def identifier(value): if not re.fullmatch(r'[^\W\d]\w*', value, re.UNICODE): raise ValueError('Invalid object or field name.') return value def validate_base(url): p = urllib.parse.urlsplit(url) if p.scheme != 'https' or not p.hostname or p.username or p.password or p.query or p.fragment: raise ValueError('Use an HTTPS publication URL without credentials, query or fragment.') if not p.path.rstrip('/').endswith('/odata/standard.odata'): raise ValueError('Expected the /odata/standard.odata publication address.') return url.rstrip('/') + '/' class Client: def __init__(self, base, username, password): self.base = validate_base(base) if ':' in username or not username or not password: raise ValueError('Provide a read-only 1C account.') self.auth = 'Basic ' + base64.b64encode((username + ':' + password).encode()).decode() # Do not inherit environment proxy settings for internal business credentials. self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect(), urllib.request.HTTPSHandler(context=ssl.create_default_context())) def get(self, path='', params=None): url = self.base + urllib.parse.quote(path, safe='') if params: url += '?' + urllib.parse.urlencode(params) req = urllib.request.Request(url, headers={'Authorization': self.auth, 'Accept': 'application/json'}, method='GET') with self.opener.open(req, timeout=30) as response: raw = response.read(MAX_BYTES + 1) if len(raw) > MAX_BYTES: raise ValueError('Response too large; select fewer fields.') return json.loads(raw) def objects(self): body = self.get(params={'$format': 'json'}) values = body.get('value') if not isinstance(values, list): raise ValueError('Unsupported service document. Confirm OData JSON support.') return [identifier(x['name']) for x in values if isinstance(x, dict) and isinstance(x.get('name'), str)] def export(self, entity, columns, key): entity, key = identifier(entity), identifier(key) columns = [identifier(x) for x in columns] if not columns or len(columns) > 100 or len(set(columns)) != len(columns) or key not in columns: raise ValueError('Select 1–100 unique fields including the unique key.') rows, seen, expected = [], set(), None while True: body = self.get(entity, {'$format': 'json', '$select': ','.join(columns), '$orderby': key, '$top': '500', '$skip': str(len(rows)), '$inlinecount': 'allpages'}) values = body.get('value') count = body.get('odata.count') if not isinstance(values, list) or count is None: raise ValueError('Missing rows or total count; incomplete exports are refused.') count = int(count) if count < 0 or count > MAX_ROWS: raise ValueError('Export exceeds 20,000 rows; use a scoped 1C publication.') if expected is None: expected = count if expected != count: raise ValueError('Source changed during export; retry during a stable period.') for record in values: if not isinstance(record, dict) or any(c not in record for c in columns): raise ValueError('Source schema changed or selected field is missing.') uid = record[key] if not isinstance(uid, (str, int)) or isinstance(uid, bool) or uid == '' or uid in seen: raise ValueError('Missing or duplicate unique key; choose a unique scalar key.') seen.add(uid) row = [record[c] for c in columns] if any(isinstance(v, (dict, list)) for v in row): raise ValueError('Nested fields require a configuration-specific adapter.') rows.append(row) if len(rows) == expected: return rows if not values or len(rows) > expected or len(rows) > MAX_ROWS: raise ValueError('Incomplete or inconsistent source response.') def csv_bytes(columns, rows): def safe(v): if isinstance(v, str) and v.lstrip().startswith(('=', '+', '-', '@')): return "'" + v return v stream = io.StringIO(newline='') writer = csv.writer(stream) writer.writerow(columns) writer.writerows([[safe(v) for v in row] for row in rows]) return ('\ufeff' + stream.getvalue()).encode('utf-8') def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--url', required=True) parser.add_argument('--list', action='store_true') parser.add_argument('--entity') parser.add_argument('--fields', help='Comma separated scalar field names') parser.add_argument('--key', help='Unique scalar key included in fields') parser.add_argument('--output') args = parser.parse_args() username = os.environ.get('BEXUVRA_1C_USER') or input('Read-only 1C user: ') password = os.environ.get('BEXUVRA_1C_PASSWORD') or getpass.getpass('1C password: ') client = Client(args.url, username, password) if args.list: print('\n'.join(client.objects())) return if not all([args.entity, args.fields, args.key, args.output]): parser.error('Export requires --entity, --fields, --key and --output.') columns = args.fields.split(',') rows = client.export(args.entity, columns, args.key) data = csv_bytes(columns, rows) if len(data) > MAX_BYTES: raise ValueError('CSV exceeds the platform file limit.') # Exclusive creation avoids overwriting existing customer data. fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, 'wb') as output: output.write(data) print('Export complete: %d rows. Upload the CSV and confirm column mapping in Bexuvra.' % len(rows)) if __name__ == '__main__': try: main() except (ValueError, OSError, urllib.error.URLError) as error: # Server errors can contain account information; do not echo them. print('Export failed. Check HTTPS, read-only permissions, selected fields and source consistency. No partial report was accepted.', file=sys.stderr) sys.exit(1)