SKEEN-Proyecto/migracion/detectar_duplicados.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Detecta duplicados de pacientes por telefono + similitud de nombre.
Solo lee; genera /root/migracion/reporte_duplicados.csv para revision."""

import csv
import re
import unicodedata
import difflib
import xmlrpc.client

c = xmlrpc.client.ServerProxy('http://localhost:8069/xmlrpc/2/common')
uid = c.authenticate('skeen_odoo', 'admin', 'skeen_admin_2026', {})
m = xmlrpc.client.ServerProxy('http://localhost:8069/xmlrpc/2/object')


def kw(model, method, args, kwar=None):
    return m.execute_kw('skeen_odoo', uid, 'skeen_admin_2026', model, method, args, kwar or {})


def norm_name(s):
    s = unicodedata.normalize('NFKD', s or '')
    s = ''.join(ch for ch in s if not unicodedata.combining(ch))
    s = re.sub(r'[^a-z0-9 ]', ' ', s.lower())
    return ' '.join(sorted(s.split()))  # orden independiente


def norm_phone(p):
    return re.sub(r'\D', '', p or '')


FIELDS = ['id', 'name', 'phone', 'email', 'birth_date', 'legacy_id',
          'total_visits', 'last_visit', 'create_date', 'active']

patients = []
offset = 0
while True:
    batch = kw('res.partner', 'search_read',
               [[['is_patient', '=', True]]],
               {'fields': FIELDS, 'limit': 1000, 'offset': offset, 'order': 'id'})
    if not batch:
        break
    patients.extend(batch)
    offset += len(batch)
print(f'Pacientes leidos: {len(patients)}')

# Agrupar por telefono
by_phone = {}
for p in patients:
    ph = norm_phone(p['phone'])
    if ph:
        by_phone.setdefault(ph, []).append(p)

reporte = []
grupos_dup = 0
grupos_familia = 0
for phone, group in sorted(by_phone.items()):
    if len(group) < 2:
        continue
    names = [norm_name(p['name']) for p in group]
    # similitud maxima dentro del grupo
    max_ratio = 0.0
    for i in range(len(names)):
        for j in range(i + 1, len(names)):
            max_ratio = max(max_ratio, difflib.SequenceMatcher(None, names[i], names[j]).ratio())
    tipo = 'duplicado_probable' if max_ratio >= 0.80 else 'telefono_compartido'
    if tipo == 'duplicado_probable':
        grupos_dup += 1
    else:
        grupos_familia += 1
    # candidato a conservar: mas visitas, luego mas antiguo
    canon = sorted(group, key=lambda p: (-(p['total_visits'] or 0), p['create_date']))[0]
    for p in group:
        reporte.append({
            'telefono': phone,
            'tipo': tipo,
            'similitud_nombre': f'{max_ratio:.2f}',
            'conservar': 'SI' if p['id'] == canon['id'] else '',
            'id': p['id'],
            'nombre': p['name'],
            'email': p['email'] or '',
            'fecha_nac': p['birth_date'] or '',
            'visitas': p['total_visits'] or 0,
            'ultima_visita': p['last_visit'] or '',
            'legacy_id': p['legacy_id'] or '',
            'creado': p['create_date'],
        })

# Nombres identicos con telefonos distintos (posible duplicado extra)
by_name = {}
for p in patients:
    n = norm_name(p['name'])
    if len(n) >= 8:
        by_name.setdefault(n, []).append(p)
extra = 0
for name, group in by_name.items():
    phones = {norm_phone(p['phone']) for p in group}
    if len(group) > 1 and len(phones) > 1:
        extra += 1
        canon = sorted(group, key=lambda p: (-(p['total_visits'] or 0), p['create_date']))[0]
        for p in group:
            reporte.append({
                'telefono': norm_phone(p['phone']),
                'tipo': 'mismo_nombre_distinto_telefono',
                'similitud_nombre': '1.00',
                'conservar': 'SI' if p['id'] == canon['id'] else '',
                'id': p['id'],
                'nombre': p['name'],
                'email': p['email'] or '',
                'fecha_nac': p['birth_date'] or '',
                'visitas': p['total_visits'] or 0,
                'ultima_visita': p['last_visit'] or '',
                'legacy_id': p['legacy_id'] or '',
                'creado': p['create_date'],
            })

out = '/root/migracion/reporte_duplicados.csv'
with open(out, 'w', encoding='utf-8', newline='') as f:
    w = csv.DictWriter(f, fieldnames=['telefono', 'tipo', 'similitud_nombre', 'conservar',
                                      'id', 'nombre', 'email', 'fecha_nac', 'visitas',
                                      'ultima_visita', 'legacy_id', 'creado'])
    w.writeheader()
    w.writerows(reporte)

dup_rows = [r for r in reporte if r['tipo'] == 'duplicado_probable']
print(f'\nGrupos por telefono con duplicado probable: {grupos_dup} ({len(dup_rows)} registros)')
print(f'Grupos telefono compartido (familia): {grupos_familia}')
print(f'Grupos mismo nombre, distinto telefono: {extra}')
print(f'\nReporte: {out} ({len(reporte)} filas)')
print('\nMuestra de duplicados probables:')
shown = set()
for r in dup_rows[:20]:
    key = r['telefono']
    if key not in shown:
        print(f"  tel {key}:")
        shown.add(key)
    print(f"    [{'CONSERVAR' if r['conservar'] else 'fusionar '}] id={r['id']} {r['nombre']} (visitas={r['visitas']})")