Script codé brutalement pour passer la base LDAP de ISO à UTF-8

diff --git a/reencodeldap.py b/reencodeldap.py
new file mode 100755 (executable)
index 0000000..7944818
--- /dev/null
+++ b/reencodeldap.py
@@ -0,0 +1,189 @@
+#!/usr/bin/env python
+# -*- encoding: utf-8 -*-
+
+# Pour essayer de passer la base LDAP de ISO à UTF-8
+
+# Je vais essayer de faire ça plus joliment que le premier, 
+#  cette fois-ci avec un binding
+
+import ldap, ldap.modlist
+import sys
+import json
+import copy
+
+TEST_BASE = True
+INTERACTIVE = False
+
+if "-i" in sys.argv or "--interactive" in sys.argv:
+    INTERACTIVE = True
+
+if "--no-interactive" in sys.argv:
+    INTERACTIVE = False
+
+SAVE_MANUAL = False
+if INTERACTIVE:
+    USER_ENCODING = sys.stdin.encoding
+
+# On corrige certaines choses à la main
+corrections_manuelles_file = "corrections.json"
+
+# idée pour voir si on décode n'importe comment
+usual_table = u""" 
+abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ[]\\/^_`{}()|~0123456789:;<=>?@.-,+*'\"&%$!àáâäæèéêëìíîïòóôöùúûüýÿœŷỳçÀÁÂÄÆÈÉÊËÌÍÎÏÒÓÔÖÙÚÛÜÝŸŒŶỲǵ£§°\t#"""
+
+# NB : u"\xc3" c'est à mais apparemment il y arrive pas…
+known_problems_table=[u"¨", u"©", u"Ã", u" ", u"¢", u"´", u'\xc3', u"¯",
+                      u'\x81', u'\x82', u'\x80', u'\x89', u'\x83', u'\x99',
+                      u'\x9c', u'\x9d', u'\xac',
+                      u"¡", u"®", u"¼", u"¦",
+                      u'\xad', u'\xb6', u"«", u"»", u"ª", u"º", u"¿", u"¡"]
+
+def con_ldap(uri="ldap://127.0.0.1", test=TEST_BASE):
+    """Ouvre une connexion à la base LDAP"""
+    con = ldap.initialize(uri)
+    if test:
+        mdp = "75bdb64f32"
+    else:
+        sys.path.append("/etc/crans/secrets/")
+        import secrets
+        mdp = secrets.ldap_password
+    con.bind_s(who="cn=admin,dc=crans,dc=org", cred=mdp)
+    return con
+
+def load_base(con):
+    """dumpe la base"""
+    b=con.search_s(base="ou=data,dc=crans,dc=org", scope=ldap.SCOPE_SUBTREE)
+    return b
+
+def load_manual():
+    """Charge la liste des corrections manuelles"""
+    with open(corrections_manuelles_file) as f:
+        manual = json.load(f)
+    return manual
+
+def save_manual(manual):
+    """Enregistre la liste des correction manuelles"""
+    with open(corrections_manuelles_file, "w") as f:
+        json.dump(manual, f)
+
+def is_strange(s):
+    """Retourne True si la chaîne s (unicode) contient des caractères qu'on n'est pas habitué à voir"""
+    assert isinstance(s, unicode)
+    strange_chars = (set(s).symmetric_difference(set(usual_table))).intersection(set(s))
+    st = [i for i in strange_chars if not "\\u" in i.encode("raw_unicode_escape")]
+    return st != []
+
+#def tryapf4242(s):
+#    """Prend une chaîne (unicode) qui était encodée en APF4242 et la renvoie décodée un cran de plus.
+#       Si ce n'est pas possible, renvoie la chaîne elle-même"""
+#    assert isinstance(s, unicode)
+#    try:
+#       s42 = s.encode("raw_unicode_escape").decode("UTF-8")
+#        print u"*** APF4242 trouvé ***\n" + s
+#        return s42
+#    except UnicodeError:
+#        return s
+
+class AbandonReplace(Exception):
+    pass
+
+def get_replacement(champ, s):
+    """Affiche une chaîne qu'on n'a pas su corriger automatiquement 
+       et demande à l'utilisateur de préciser par quoi il faut la remplacer"""
+    print "%s : %r" % (champ, s)
+    repl = raw_input("Remplacer par : ")
+    if repl == "":
+        print "Abandon"
+        raise AbandonReplace
+    repl = repl.decode(USER_ENCODING)
+    return repl
+    
+def reencode(champ, s, manual={}):
+    """Prend un str et essaye de le renvoyer correctement encodé"""
+    assert isinstance(s, str)
+    sunicode = s.decode("UTF-8")
+    # On vérifie ensuite qu'on a bien fait
+    try:
+        # Le cas le plus fréquent dans la base LDAP :
+        sunicode = sunicode.encode("ISO8859-1").decode("UTF-8")
+    except UnicodeError:
+        # Si ça n'a pas marché :
+        sunicode = sunicode.encode("UTF-8").decode("UTF-8")
+    dunno, strange = None, False
+    if is_strange(sunicode):
+#        sunicode = tryapf4242(sunicode)
+        # on essaye de le remplacer par une valeur du dictionnaire de remplacements manuels
+        if sunicode in manual:
+            sunicode = manual.get(sunicode, sunicode)
+        elif INTERACTIVE:
+            try:
+                replacement = get_replacement(champ, sunicode)
+                manual.update({sunicode: replacement})
+                sunicode = replacement
+            except AbandonReplace:
+                pass
+        if is_strange(sunicode):
+            strange=True
+    if strange:
+            dunno = sunicode
+    reencodes = sunicode.encode("UTF-8")
+    return reencodes, dunno, manual
+
+def perform(bbase, manual={}):
+    """Renvoie :
+        - une nouvelle base réencodée
+        - une liste de valeurs où on n'est pas satisfait du résultat obtenu
+        - un dico des valeurs qui ont été entrées à la main"""
+    base = copy.deepcopy(bbase)
+    dunnos = []
+    for iobj in range(len(base)):
+        for k in base[iobj][1].keys():
+            for iattr in range(len(base[iobj][1][k])):
+                attr = base[iobj][1][k][iattr]
+                attr, dunno, manual = reencode(k, attr, manual)
+                base[iobj][1][k][iattr] = attr
+                if dunno:
+                    dunnos.append([[iobj, k, iattr], dunno])
+    return base, dunnos, manual
+
+def compute_modlists(avant, apres):
+    """Renvoie une liste de (dn,modlist) pour pouvoir effectuer les modification"""
+    modlists = []
+    for iobj in range(len(avant)):
+        dn1, dn2 = avant[iobj][0], apres[iobj][0]
+        try:
+            assert dn1==dn2
+        except AssertionError:
+            print "%s != %s" % (dn1, dn2)
+            assert False
+        modlist = ldap.modlist.modifyModlist(avant[iobj][1], apres[iobj][1])
+        modlists.append((dn1, modlist))
+    return modlists
+
+def modify(con, modlists):
+    """Effectue les modifications sur la base LDAP"""
+    for dn, modlist in modlists:
+        con.modify_s(dn, modlist)
+
+def do():
+    """Fait tout"""
+    con = con_ldap()
+    base = load_base(con)
+    manual = load_manual()
+    base_new, dunnos, manual = perform(base, manual)
+    if SAVE_MANUAL:
+        save_manual(manual)
+    elif INTERACTIVE:
+        ans = ""
+        while ans == "":
+            ans = raw_input("Se souvenir des modifs manuelles ?")
+        if ans.lower() in ["y","yes","o","oui"]:
+            save_manual(manual)
+    #import pprint
+    #pprint.pprint(dunnos)
+    modlists = [m for m in compute_modlists(base, base_new) if m[1] != []]
+    return modlists
+    
+
+if __name__ == "__main__":
+    modlists = do()