reencodeldap.py

   1 #!/usr/bin/env python
   2 # -*- encoding: utf-8 -*-
   3
   4 # Pour essayer de passer la base LDAP de ISO à UTF-8
   5
   6 # Je vais essayer de faire ça plus joliment que le premier,
   7 #  cette fois-ci avec un binding
   8
   9 import ldap, ldap.modlist
  10 import sys
  11 import json
  12 import copy
  13
  14 TEST_BASE = True
  15 INTERACTIVE = False
  16
  17 if "-i" in sys.argv or "--interactive" in sys.argv:
  18     INTERACTIVE = True
  19
  20 if "--no-interactive" in sys.argv:
  21     INTERACTIVE = False
  22
  23 SAVE_MANUAL = False
  24 if INTERACTIVE:
  25     USER_ENCODING = sys.stdin.encoding
  26
  27 # On corrige certaines choses à la main
  28 corrections_manuelles_file = "corrections.json"
  29
  30 # idée pour voir si on décode n'importe comment
  31 usual_table = u"""
  32 abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ[]\\/^_`{}()|~0123456789:;<=>?@.-,+*'\"&%$!àáâäæèéêëìíîïòóôöùúûüýÿœŷỳçÀÁÂÄÆÈÉÊËÌÍÎÏÒÓÔÖÙÚÛÜÝŸŒŶỲÇµ£§°\t#"""
  33
  34 # NB : u"\xc3" c'est Ã mais apparemment il y arrive pas…
  35 known_problems_table=[u"¨", u"©", u"Ã", u" ", u"¢", u"´", u'\xc3', u"¯",
  36                       u'\x81', u'\x82', u'\x80', u'\x89', u'\x83', u'\x99',
  37                       u'\x9c', u'\x9d', u'\xac',
  38                       u"¡", u"®", u"¼", u"¦",
  39                       u'\xad', u'\xb6', u"«", u"»", u"ª", u"º", u"¿", u"¡"]
  40
  41 def con_ldap(uri="ldap://127.0.0.1", test=TEST_BASE):
  42     """Ouvre une connexion à la base LDAP"""
  43     con = ldap.initialize(uri)
  44     if test:
  45         mdp = "75bdb64f32"
  46     else:
  47         sys.path.append("/etc/crans/secrets/")
  48         import secrets
  49         mdp = secrets.ldap_password
  50     con.bind_s(who="cn=admin,dc=crans,dc=org", cred=mdp)
  51     return con
  52
  53 def load_base(con):
  54     """dumpe la base"""
  55     b=con.search_s(base="ou=data,dc=crans,dc=org", scope=ldap.SCOPE_SUBTREE)
  56     return b
  57
  58 def load_manual():
  59     """Charge la liste des corrections manuelles"""
  60     with open(corrections_manuelles_file) as f:
  61         manual = json.load(f)
  62     return manual
  63
  64 def save_manual(manual):
  65     """Enregistre la liste des correction manuelles"""
  66     with open(corrections_manuelles_file, "w") as f:
  67         json.dump(manual, f)
  68
  69 def is_strange(s):
  70     """Retourne True si la chaîne s (unicode) contient des caractères qu'on n'est pas habitué à voir"""
  71     assert isinstance(s, unicode)
  72     strange_chars = (set(s).symmetric_difference(set(usual_table))).intersection(set(s))
  73     st = [i for i in strange_chars if not "\\u" in i.encode("raw_unicode_escape")]
  74     return st != []
  75
  76 #def tryapf4242(s):
  77 #    """Prend une chaîne (unicode) qui était encodée en APF4242 et la renvoie décodée un cran de plus.
  78 #       Si ce n'est pas possible, renvoie la chaîne elle-même"""
  79 #    assert isinstance(s, unicode)
  80 #    try:
  81 #       s42 = s.encode("raw_unicode_escape").decode("UTF-8")
  82 #        print u"*** APF4242 trouvé ***\n" + s
  83 #        return s42
  84 #    except UnicodeError:
  85 #        return s
  86
  87 class AbandonReplace(Exception):
  88     pass
  89
  90 def get_replacement(champ, s):
  91     """Affiche une chaîne qu'on n'a pas su corriger automatiquement
  92        et demande à l'utilisateur de préciser par quoi il faut la remplacer"""
  93     print "%s : %r" % (champ, s)
  94     repl = raw_input("Remplacer par : ")
  95     if repl == "":
  96         print "Abandon"
  97         raise AbandonReplace
  98     repl = repl.decode(USER_ENCODING)
  99     return repl
 100
 101 def reencode(champ, s, manual={}):
 102     """Prend un str et essaye de le renvoyer correctement encodé"""
 103     assert isinstance(s, str)
 104     sunicode = s.decode("UTF-8")
 105     # On vérifie ensuite qu'on a bien fait
 106     try:
 107         # Le cas le plus fréquent dans la base LDAP :
 108         sunicode = sunicode.encode("ISO8859-1").decode("UTF-8")
 109     except UnicodeError:
 110         # Si ça n'a pas marché :
 111         sunicode = sunicode.encode("UTF-8").decode("UTF-8")
 112     dunno, strange = None, False
 113     if is_strange(sunicode):
 114 #        sunicode = tryapf4242(sunicode)
 115         # on essaye de le remplacer par une valeur du dictionnaire de remplacements manuels
 116         if sunicode in manual:
 117             sunicode = manual.get(sunicode, sunicode)
 118         elif INTERACTIVE:
 119             try:
 120                 replacement = get_replacement(champ, sunicode)
 121                 manual.update({sunicode: replacement})
 122                 sunicode = replacement
 123             except AbandonReplace:
 124                 pass
 125         if is_strange(sunicode):
 126             strange=True
 127     if strange:
 128             dunno = sunicode
 129     reencodes = sunicode.encode("UTF-8")
 130     return reencodes, dunno, manual
 131
 132 def perform(bbase, manual={}):
 133     """Renvoie :
 134         - une nouvelle base réencodée
 135         - une liste de valeurs où on n'est pas satisfait du résultat obtenu
 136         - un dico des valeurs qui ont été entrées à la main"""
 137     base = copy.deepcopy(bbase)
 138     dunnos = []
 139     for iobj in range(len(base)):
 140         for k in base[iobj][1].keys():
 141             for iattr in range(len(base[iobj][1][k])):
 142                 attr = base[iobj][1][k][iattr]
 143                 attr, dunno, manual = reencode(k, attr, manual)
 144                 base[iobj][1][k][iattr] = attr
 145                 if dunno:
 146                     dunnos.append([[iobj, k, iattr], dunno])
 147     return base, dunnos, manual
 148
 149 def compute_modlists(avant, apres):
 150     """Renvoie une liste de (dn,modlist) pour pouvoir effectuer les modification"""
 151     modlists = []
 152     for iobj in range(len(avant)):
 153         dn1, dn2 = avant[iobj][0], apres[iobj][0]
 154         try:
 155             assert dn1==dn2
 156         except AssertionError:
 157             print "%s != %s" % (dn1, dn2)
 158             assert False
 159         modlist = ldap.modlist.modifyModlist(avant[iobj][1], apres[iobj][1])
 160         modlists.append((dn1, modlist))
 161     return modlists
 162
 163 def modify(con, modlists):
 164     """Effectue les modifications sur la base LDAP"""
 165     for dn, modlist in modlists:
 166         con.modify_s(dn, modlist)
 167
 168 def do():
 169     """Fait tout"""
 170     con = con_ldap()
 171     base = load_base(con)
 172     manual = load_manual()
 173     base_new, dunnos, manual = perform(base, manual)
 174     if SAVE_MANUAL:
 175         save_manual(manual)
 176     elif INTERACTIVE:
 177         ans = ""
 178         while ans == "":
 179             ans = raw_input("Se souvenir des modifs manuelles ?")
 180         if ans.lower() in ["y","yes","o","oui"]:
 181             save_manual(manual)
 182     #import pprint
 183     #pprint.pprint(dunnos)
 184     modlists = [m for m in compute_modlists(base, base_new) if m[1] != []]
 185     return modlists
 186
 187
 188 if __name__ == "__main__":
 189     modlists = do()