gitrusprusandCursor
4eb4c145b6
Users can register, log in, and manage profile/password in a personal account page; server data is scoped by owner across contacts, maps, tags, and import. Co-authored-by: Cursor <cursoragent@cursor.com>
234 lines
8.7 KiB
Python
234 lines
8.7 KiB
Python
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, owner=None):
|
|
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
|
|
lookup = {'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(),
|
|
}
|
|
if owner is not None:
|
|
lookup['owner'] = owner
|
|
defaults['owner'] = owner
|
|
Contact.objects.get_or_create(**lookup, defaults=defaults)
|
|
created += 1
|
|
return {
|
|
'total': len(rows),
|
|
'created': created,
|
|
'skipped': skipped,
|
|
'errors': errors,
|
|
}
|