import csv import io import json from contacts.models import Contact def monica_contact_fields(contact_data): phone = '' email = '' notes_parts = [] for block in contact_data or []: if block.get('type') == 'contact_field': for val in block.get('values') or []: props = val.get('properties') or {} value = str(props.get('data') or '').strip() if not value: continue if '@' in value and '.' in value: email = email or value else: phone = phone or value elif block.get('type') == 'note': for val in block.get('values') or []: body = str((val.get('properties') or {}).get('body') or '').strip() if body: notes_parts.append(body) return phone, email, '\n'.join(notes_parts) def normalize_rows(data): if isinstance(data, list): return data if isinstance(data, dict): account = data.get('account') if isinstance(account, dict): account_data = account.get('data') if isinstance(account_data, list): rows = [] for block in account_data: if block.get('type') != 'contact': continue for c in block.get('values') or []: props = c.get('properties') or {} first = str(props.get('first_name') or '').strip() last = str(props.get('last_name') or '').strip() middle = str(props.get('middle_name') or '').strip() name = ' '.join(filter(None, [first, middle, last])) or ' '.join( filter(None, [first, last]) ) if not name: continue phone, email, notes = monica_contact_fields(c.get('data')) rows.append({ 'name': name, 'email': email, 'phone': phone, 'organization': '', 'position': '', 'notes': notes, }) if rows: return rows if 'contacts' in data: rows = [] for c in data['contacts']: first = str(c.get('first_name') or '').strip() last = str(c.get('last_name') or '').strip() name = ' '.join(filter(None, [first, last])) phone = '' for ph in c.get('phone_numbers') or []: phone = str(ph.get('number') or ph.get('content') or '') if phone: break email = '' for em in c.get('emails') or []: email = str(em.get('email') or em.get('content') or '') if email: break org = '' position = '' for comp in c.get('companies') or []: org = str(comp.get('name') or comp.get('company_name') or '') position = str(comp.get('job') or comp.get('position') or comp.get('title') or '') if org: break if not org: org = str(c.get('company') or c.get('company_name') or '').strip() position = str(c.get('job') or c.get('position') or '').strip() rows.append({ 'name': name, 'email': email, 'phone': phone, 'organization': org, 'position': position, 'notes': str(c.get('information') or c.get('description') or c.get('notes') or '').strip(), }) return rows for key in ('results', 'data', 'items', 'people', 'persons'): if key in data and isinstance(data[key], list): return data[key] return [] def unfold_vcard_lines(text): lines = text.replace('\r\n', '\n').replace('\r', '\n').split('\n') unfolded = [] for line in lines: if line.startswith((' ', '\t')) and unfolded: unfolded[-1] += line[1:] else: unfolded.append(line) return unfolded def unescape_vcard_value(value): return ( str(value or '') .replace('\\n', '\n') .replace('\\N', '\n') .replace('\\,', ',') .replace('\\;', ';') .replace('\\\\', '\\') .strip() ) def name_from_vcard_n(value): parts = unescape_vcard_value(value).split(';') family = (parts[0] if len(parts) > 0 else '').strip() given = (parts[1] if len(parts) > 1 else '').strip() additional = (parts[2] if len(parts) > 2 else '').strip() return ' '.join(filter(None, [given, additional, family])).strip() def parse_vcf_rows(text): props = None rows = [] for line in unfold_vcard_lines(text): trimmed = line.strip() if not trimmed: continue upper = trimmed.upper() if upper == 'BEGIN:VCARD': props = {} continue if upper == 'END:VCARD': if props: name = ( (props.get('FN') or [''])[0] or name_from_vcard_n((props.get('N') or [''])[0]) ).strip() org_raw = (props.get('ORG') or [''])[0] org = org_raw.split(';')[0].strip() if org_raw else '' email = (props.get('EMAIL') or [''])[0].replace('mailto:', '').strip() phone = (props.get('TEL') or [''])[0].replace('tel:', '').strip() rows.append({ 'name': name, 'email': email, 'phone': phone, 'organization': org, 'position': (props.get('TITLE') or [''])[0].strip(), 'notes': '\n'.join(props.get('NOTE') or []).strip(), }) props = None continue if props is None: continue if ':' not in trimmed: continue raw_key, value = trimmed.split(':', 1) key = raw_key.split(';')[0].upper() decoded = unescape_vcard_value(value.replace('mailto:', '').replace('tel:', '')) props.setdefault(key, []).append(decoded) return rows def parse_upload_file(file): filename = file.name.lower() if filename.endswith('.csv'): text = file.read().decode('utf-8-sig') reader = csv.DictReader(io.StringIO(text)) return list(reader), None if filename.endswith('.json'): raw = json.loads(file.read().decode('utf-8')) rows = normalize_rows(raw) if not rows: return None, 'Не удалось распознать формат JSON. Ожидается массив контактов или экспорт Monica CRM.' return rows, None if filename.endswith('.vcf') or filename.endswith('.vcard'): text = file.read().decode('utf-8-sig') rows = parse_vcf_rows(text) if not rows: return None, 'В файле vCard не найдено контактов.' return rows, None return None, 'Поддерживаются только CSV, JSON и vCard (.vcf) файлы.' def import_contacts_from_rows(rows): created = 0 skipped = 0 errors = [] for i, row in enumerate(rows): name = str( row.get('name') or row.get('Name') or row.get('ФИО') or ' '.join(filter(None, [ str(row.get('first_name') or '').strip(), str(row.get('last_name') or '').strip(), ])) ).strip() if not name: errors.append(f'Строка {i + 1}: отсутствует поле "name"') skipped += 1 continue Contact.objects.get_or_create( name=name, defaults={ 'email': str(row.get('email') or '').strip(), 'phone': str(row.get('phone') or '').strip(), 'organization': str(row.get('organization') or '').strip(), 'position': str(row.get('position') or '').strip(), 'notes': str(row.get('notes') or '').strip(), }, ) created += 1 return { 'total': len(rows), 'created': created, 'skipped': skipped, 'errors': errors, }