首页 发现 帮助
注册 登录
GlobalCube
/
gctools
3
0
派生 0
文件 工单管理 0 合并请求 0 Wiki
浏览代码

Black Formatter

gc-server3 5 月之前
父节点
40f2a2d170
当前提交
dff662e348
共有 8 个文件被更改，包括 91 次插入 和 89 次删除
分列视图 显示文件统计
  1. 14 8
      misc/apache_ldap.py
  2. 2 2
      misc/csv_cleanup.py
  3. 34 31
      misc/dispatch.py
  4. 12 11
      misc/file_rename.py
  5. 7 5
      misc/headers.py
  6. 1 1
      misc/logger.py
  7. 5 15
      misc/mail.py
  8. 16 16
      misc/zip_backup.py

+ 14 - 8
misc/apache_ldap.py
查看文件 

@@ -1,22 +1,28 @@
 
-from ldap3 import Server, Connection, ALL, ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES
 
+from ldap3 import ALL, ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES, Connection, Server
 
+
 
 # from ldap3.core.exceptions import LDAPCursorError
 
 
 
 def connect_ldap3():
 
-    server = Server('localhost:10389', get_info=ALL, use_ssl=False)
 
+    server = Server("localhost:10389", get_info=ALL, use_ssl=False)
 
     # conn = Connection(server, 'uid=admin,ou=system', 'gc01gapsC$', auto_bind='TLS_AFTER_BIND')  # 'uid=admin,ou=system'
 
     # status = conn.search('ou=cognos,dc=ibm,dc=com', '(objectclass=person)', 'SUBTREE')
 
     # print(conn.entries)
 
-    conn = Connection(server, user='uid=Global1,ou=cognos,dc=ibm,dc=com', password='Cognos#11')
 
+    conn = Connection(server, user="uid=Global1,ou=cognos,dc=ibm,dc=com", password="Cognos#11")
 
     if conn.bind():
 
-        conn.search('ou=cognos,dc=ibm,dc=com', '(objectclass=person)', 'SUBTREE', attributes=[ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES])
 
-        format_string = '{:15} {:25} {:19} {:25} {}'
 
+        conn.search(
 
+            "ou=cognos,dc=ibm,dc=com",
 
+            "(objectclass=person)",
 
+            "SUBTREE",
 
+            attributes=[ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES],
 
+        )
 
+        format_string = "{:15} {:25} {:19} {:25} {}"
 
         for e in conn.entries:
 
             desc = ""
 
             email = ""
 
-            if 'description' in e:
 
+            if "description" in e:
 
                 desc = e.description
 
-            if 'mail' in e:
 
+            if "mail" in e:
 
                 email = e.mail
 
             # print(e.entry_to_json())
 
             print(format_string.format(str(e.uid), str(e.givenName), str(e.createTimestamp)[:19], str(email), desc))
 
@@ -24,7 +30,7 @@ def connect_ldap3():
 
         print(conn.result)
 
 
 
-if __name__ == '__main__':
 
+if __name__ == "__main__":
 
     # connect_pyldap()
 
     connect_ldap3()

+ 2 - 2
misc/csv_cleanup.py
查看文件 

@@ -1,8 +1,8 @@
 
-import plac
 
 import re
 
-from pathlib import Path
 
 from datetime import datetime
 
+from pathlib import Path
 
 
+import plac
 
 
 MIN_AGE = datetime.now().timestamp() - 12 * 60 * 60

+ 34 - 31
misc/dispatch.py
查看文件 

@@ -1,13 +1,14 @@
 
 import csv
 
-from pathlib import Path
 
-from mail import mail
 
+from datetime import datetime
 
 from itertools import groupby
 
-import config
 
+from pathlib import Path
 
+
 
 import jinja2
 
-from datetime import datetime
 
+from mail import mail
 
 
+import config
 
 
-cfg = config.Config('GAPS3.ini')
 
+cfg = config.Config("GAPS3.ini")
 
 
 html = """
 
     <!DOCTYPE html>
 
@@ -36,51 +37,53 @@ html = """
 
 def dispatch(xml_filename, recipient=None, report=None, cube=None):
 
     publish_subdir = config.joinpath(cfg.cognos7.publish_dir, xml_filename[:-4])
 
 
-    with open(cfg.xml_dir + '/info/versand.csv', 'r') as frh:
 
-        csv_reader = csv.DictReader(frh, delimiter=';')
 
+    with open(cfg.xml_dir + "/info/versand.csv", "r") as frh:
 
+        csv_reader = csv.DictReader(frh, delimiter=";")
 
         dispatch_list = [x for x in csv_reader]
 
 
     def dispatch_filter(x):
 
-        return all([
 
-            x['Datei'].lower() == xml_filename,
 
-            x['Versand'] == 'J',
 
-            (recipient is None or x['Empfaenger'] == recipient),
 
-            (report is None or x['Report'].lower() == report.lower()),
 
-            (cube is None or x['Cube'].lower() == cube.lower())
 
-        ])
 
+        return all(
 
+            [
 
+                x["Datei"].lower() == xml_filename,
 
+                x["Versand"] == "J",
 
+                (recipient is None or x["Empfaenger"] == recipient),
 
+                (report is None or x["Report"].lower() == report.lower()),
 
+                (cube is None or x["Cube"].lower() == cube.lower()),
 
+            ]
 
+        )
 
 
     dispatch_list = list(filter(dispatch_filter, dispatch_list))
 
-    dispatch_group = groupby(dispatch_list, lambda x: x['Report'] + '_' + x['PDF-Schicht'])
 
+    dispatch_group = groupby(dispatch_list, lambda x: x["Report"] + "_" + x["PDF-Schicht"])
 
     report_mails = {}
 
     for k, v in dispatch_group:
 
-        report_mails[k] = '<br/>'.join([x['Empfaenger'] for x in v])
 
+        report_mails[k] = "<br/>".join([x["Empfaenger"] for x in v])
 
 
     template = jinja2.Template(html)
 
     mail_batch = []
 
 
-    for group_name, v in groupby(dispatch_list, lambda x: x['Empfaenger']):
 
+    for group_name, v in groupby(dispatch_list, lambda x: x["Empfaenger"]):
 
         group_filenames = []
 
         d_group = list(v)
 
         for row in d_group:
 
-            ext = '.xls' if row['XLS'] == 'J' else '.pdf'
 
+            ext = ".xls" if row["XLS"] == "J" else ".pdf"
 
             filename = f"{publish_subdir}/{row['Report']}_{row['PDF-Schicht']}{ext}"
 
 
-            row['Empfaenger'] = report_mails[row['Report'] + '_' + row['PDF-Schicht']]
 
-            row['Dateiname'] = []
 
-            row['Stand'] = 'nicht verf&uuml;gbar'
 
-            row['Schicht'] = row['PDF-Schicht'] if row['PDF-Schicht'] != '0' else 'komplett'
 
+            row["Empfaenger"] = report_mails[row["Report"] + "_" + row["PDF-Schicht"]]
 
+            row["Dateiname"] = []
 
+            row["Stand"] = "nicht verf&uuml;gbar"
 
+            row["Schicht"] = row["PDF-Schicht"] if row["PDF-Schicht"] != "0" else "komplett"
 
             if Path(filename).exists():
 
                 mtime = datetime.fromtimestamp(Path(filename).stat().st_mtime)
 
-                row['Stand'] = mtime.strftime('%d.%m.%Y, %H:%M')
 
+                row["Stand"] = mtime.strftime("%d.%m.%Y, %H:%M")
 
                 if (datetime.now() - mtime).total_seconds() <= 60 * 60 * 12:
 
-                    row['Dateiname'].append((row['Name'] + ext, filename))
 
-                    group_filenames.append((row['Name'] + ext, filename))
 
+                    row["Dateiname"].append((row["Name"] + ext, filename))
 
+                    group_filenames.append((row["Name"] + ext, filename))
 
 
             if cfg.versand_separat:
 
-                mail_batch.append((group_name, row['Name'], template.render(reports=[row]), row['Dateiname']))
 
+                mail_batch.append((group_name, row["Name"], template.render(reports=[row]), row["Dateiname"]))
 
 
         if not cfg.versand_separat:
 
-            mail_batch.append((group_name, 'GAPS-Versand', template.render(reports=d_group), group_filenames))
 
+            mail_batch.append((group_name, "GAPS-Versand", template.render(reports=d_group), group_filenames))
 
     return mail_batch
 
 
 
@@ -92,8 +95,8 @@ def send_mail(mail_batch):
 
 
 def main(xml_filename, recipient=None, report=None, cube=None):
 
     xml_filename = xml_filename.lower()
 
-    if xml_filename[-4:] != '.xml':
 
-        xml_filename = xml_filename + '.xml'
 
+    if xml_filename[-4:] != ".xml":
 
+        xml_filename = xml_filename + ".xml"
 
 
     mail_batch = dispatch(xml_filename, recipient, report, cube)
 
     for m in mail_batch:
 
@@ -101,5 +104,5 @@ def main(xml_filename, recipient=None, report=None, cube=None):
 
     send_mail([mail_batch[0]])
 
 
 
-if __name__ == '__main__':
 
-    main('GAPS_Vers_Tag', 'robert.bedner@gmail.com')
 
+if __name__ == "__main__":
 
+    main("GAPS_Vers_Tag", "robert.bedner@gmail.com")

+ 12 - 11
misc/file_rename.py
查看文件 

@@ -1,26 +1,27 @@
 
 import csv
 
-from pathlib import Path
 
 from datetime import datetime
 
+from pathlib import Path
 
 
 base_dir = "C:/Projekte/Python"
 
 # timestamp = date.today().strftime('%Y%m%d')
 
 
+
 
 def main():
 
-    csv_reader = csv.reader(open(base_dir + '/gctools/DSG_Uebersetzung.csv', 'r', encoding='latin-1'), delimiter=';')
 
-    translation = dict([(row[0], row[1]) for row in csv_reader if row[1] != ''])
 
-    target = Path(base_dir + '/export_dsg')
 
-    for filename in Path(base_dir + '/export').glob('*'):
 
-        for (old, new) in translation.items():
 
-            if '_' + old in filename.name:
 
+    csv_reader = csv.reader(open(base_dir + "/gctools/DSG_Uebersetzung.csv", "r", encoding="latin-1"), delimiter=";")
 
+    translation = dict([(row[0], row[1]) for row in csv_reader if row[1] != ""])
 
+    target = Path(base_dir + "/export_dsg")
 
+    for filename in Path(base_dir + "/export").glob("*"):
 
+        for old, new in translation.items():
 
+            if "_" + old in filename.name:
 
                 new_name = filename.name.replace(old, new)
 
-                new_timestamp = datetime.fromtimestamp(filename.stat().st_mtime).strftime('%Y%m%d')
 
+                new_timestamp = datetime.fromtimestamp(filename.stat().st_mtime).strftime("%Y%m%d")
 
 
-                new_name = new_name[:-4] + '_' + new_timestamp + new_name[-4:]
 
+                new_name = new_name[:-4] + "_" + new_timestamp + new_name[-4:]
 
                 new_filename = target.joinpath(new_name)
 
-                print(filename.name + ' => ' + new_filename.name)
 
+                print(filename.name + " => " + new_filename.name)
 
                 filename.rename(new_filename)
 
                 break
 
 
 
-if __name__ == '__main__':
 
+if __name__ == "__main__":
 
     main()

+ 7 - 5
misc/headers.py
查看文件 

@@ -4,13 +4,15 @@ from pathlib import Path
 
 
 def main(base_dir):
 
     path = Path(base_dir)
 
-    target = path.joinpath('Kopfzeilen')
 
+    target = path.joinpath("Kopfzeilen")
 
     os.makedirs(target, exist_ok=True)
 
 
-    for filename in Path(base_dir).glob('*.csv'):
 
-        with open(filename, 'r', encoding='latin-1') as frh, open(target.joinpath(filename.name), 'w', encoding='latin-1') as fwh:
 
+    for filename in Path(base_dir).glob("*.csv"):
 
+        with open(filename, "r", encoding="latin-1") as frh, open(
 
+            target.joinpath(filename.name), "w", encoding="latin-1"
 
+        ) as fwh:
 
             fwh.write(frh.readline())
 
 
 
-if __name__ == '__main__':
 
-    main('C:/GlobalCube/System/OPTIMA/Export')
 
+if __name__ == "__main__":
 
+    main("C:/GlobalCube/System/OPTIMA/Export")

+ 1 - 1
misc/logger.py
查看文件 

@@ -1,4 +1,4 @@
 
 from datetime import datetime
 
 
 with open("test.log", "a") as f:
 
-    f.write('Running at ' + datetime.now().isoformat() + '\n')
 
+    f.write("Running at " + datetime.now().isoformat() + "\n")

+ 5 - 15
misc/mail.py
查看文件 

@@ -1,11 +1,12 @@
 
-from dataclasses import dataclass
 
-from pathlib import Path
 
 import smtplib
 
 import ssl
 
-from email.mime.text import MIMEText
 
-from email.mime.multipart import MIMEMultipart
 
+from dataclasses import dataclass
 
 from email.mime.application import MIMEApplication
 
+from email.mime.multipart import MIMEMultipart
 
+from email.mime.text import MIMEText
 
 from email.utils import make_msgid
 
+from pathlib import Path
 
+
 
 import config
 
 
 
@@ -17,17 +18,6 @@ class Message:
 
     attachment: list
 
 
 
-class SMTP_365:
 
-    def __init__(self, host="", port=0, local_hostname=None, timeout=30, source_address=None):
 
-        pass
 
-
 
-    def login(self, user, password, *, initial_response_ok=True):
 
-        pass
 
-
 
-    def sendmail(self, from_addr, to_addrs, msg, mail_options=(), rcpt_options=()):
 
-        pass
 
-
 
-
 
 class mail:
 
     connected = False

+ 16 - 16
misc/zip_backup.py
查看文件 

@@ -1,32 +1,32 @@
 
-import zipfile
 
 import glob
 
+import zipfile
 
 from os import path
 
 
 
 class zip_backup:
 
-    root_dir = ''
 
+    root_dir = ""
 
 
     ignore = []
 
     backup_list = []
 
 
     def ignore_list(self):
 
-        gitignore = self.root_dir + '\\.gitignore'
 
+        gitignore = self.root_dir + "\\.gitignore"
 
         if not path.exists(gitignore):
 
             pass
 
-        with open(gitignore, 'r') as f:
 
+        with open(gitignore, "r") as f:
 
             for line in f.readlines():
 
-                line = line.strip().replace('/', '\\').lower()
 
-                if line[:1] == '*':
 
-                    if line[-1] == '*':
 
+                line = line.strip().replace("/", "\\").lower()
 
+                if line[:1] == "*":
 
+                    if line[-1] == "*":
 
                         line = line[1:-1]
 
                     else:
 
-                        line = line[1:] + '\\'
 
+                        line = line[1:] + "\\"
 
                 else:
 
-                    line = '\\' + line + '\\'
 
+                    line = "\\" + line + "\\"
 
                 self.ignore.append(line)
 
 
     def ignored(self, filename):
 
-        rel_filename = '\\' + filename.replace(self.root_dir, '').lower() + '\\'
 
+        rel_filename = "\\" + filename.replace(self.root_dir, "").lower() + "\\"
 
 
         for e in self.ignore:
 
             if e in rel_filename:
 
@@ -34,24 +34,24 @@ class zip_backup:
 
         return False
 
 
     def check_dir(self, current_dir):
 
-        if self.root_dir == '':
 
+        if self.root_dir == "":
 
             self.root_dir = current_dir
 
             self.ignore_list()
 
 
-        for entry in glob.glob(current_dir + '\\*'):
 
+        for entry in glob.glob(current_dir + "\\*"):
 
             if path.isdir(entry):
 
                 self.check_dir(entry)
 
             elif not self.ignored(entry):
 
                 self.backup_list.append(entry)
 
 
     def zip_to_file(self, zip_file):
 
-        with zipfile.ZipFile(zip_file, 'w', compression=zipfile.ZIP_DEFLATED, compresslevel=9) as zip:
 
+        with zipfile.ZipFile(zip_file, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as zip:
 
             for e in self.backup_list:
 
                 zip.write(e)
 
 
 
-if __name__ == '__main__':
 
+if __name__ == "__main__":
 
     z_backup = zip_backup()
 
-    z_backup.check_dir('C:\\GAPS_Autosys')
 
+    z_backup.check_dir("C:\\GAPS_Autosys")
 
     # print(backup_list[:10])
 
-    z_backup.zip_to_file('C:\\GAPS_Autosys\\Test.zip')
 
+    z_backup.zip_to_file("C:\\GAPS_Autosys\\Test.zip")