需要放置二个个配置文件在脚本根目录:
1. config.ini 数据库基础配置
[mysql] ip = 172.16.0.232 id = root pw = Company database = employeeinfo[mssql] ip = 172.16.0.80 id = mes pw = hr@mes database = emhr[oracle] ip = 192.168.1.243:1521/TOPPROD id = MISTEST pw = mistest database = mydb
2. config.ini 最后一次修改的时间
[insert] last_insert = 2018-11-11
3.停用成功后发送邮件
#!/usr/bin/python3
# coding = utf-8
import sys
import smtplib
from import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
from email.mime.application import MIMEApplication
from datetime import datetime
from jinja2 import Template,DictLoaderfrom disableuser import *def mail(path):now = w()) # 獵取當前時間now = now[:16]mail_host = 'mail.Company'mail_user = 'robot'mail_pass = 'whoareyou2009'sender = 'robot@Company' #寄件人# receivers =['lifajia@Company','mis-erp@Company']receivers = ['b-at@Company','b_at_sec@Company','lifajia@Company'] #收件人列表 b-at-group@Company(二廠自動化小組群組) b_at_sec(二廠自動化小組文員)
message = MIMEMultipart() #创建一个带附件的实例message.attach(MIMEText('庫存明細更新時間 : '+now,'plain','utf-8')) #郵件正文message['From'] = Header('System','utf-8') #郵件發件人Frommessage['To'] = Header('自動化小組','utf-8') #收件人subject = '庫存明細' #郵件標題message['Subject'] = Header(subject,'utf-8')# xlsxpart = MIMEApplication(open('庫存明細.xls', 'rb').read())try:xlsxpart = MIMEApplication(open(path, 'rb').read())xlsxpart.add_header('Content-Disposition', 'attachment', filename=path)message.attach(xlsxpart)except Exception as err_msg:print("打開表格失敗%s"%err_msg)log_txt('打開表格失敗,可能沒有表格生成%s'%err_it()try:smtpObj = smtplib.SMTP()t(mail_host, 25) # 25 为 SMTP 端口号
smtpObj.login(mail_user, mail_pass)smtpObj.sendmail(sender, receivers, message.as_string())smtpObj.quit()print("邮件发送成功")log_txt('邮件发送成功')except Exception as e:print("Error: 邮件發送失敗 %s"%e)log_txt('Error: 邮件發送失敗 %s'%e) #寫日誌def send_mail(ad,tiptop,begin,alluser=None): # 參數說明:ad已完成用戶 tiptop已完成用戶 alluser遍歷全部用戶 begin開始日期now = day().date() # 獵取當前時間mail_host = 'mail.Company'mail_user = 'robot'mail_pass = 'whoareyou2009'sender = 'robot@Company' # 寄件人# receivers = ['it@Company'] #收件人列表 receivers = ['lifajia@Company'] #收件人列表 context_ad = dict()context_tiptop = dict()all_number = 0if alluser is not None:for i in ad:context_ad[i] = alluser[i]print(context_ad)for x in tiptop:if x[0:2] != "T0":context_tiptop[x] = alluser[x]else:context_tiptop[x] = alluser[x[2:]]all_number = alluser.__len__()mail_msg = """<html><body><h3>管理者您好,於{{ begin }}日至{{ now }}日期間共離職{{ all_number }}人,根據離職數據自動停用的賬號名單如下:</h3><h4>TIPTOP停用名單</h4>{% if tiptop %}<table border="1" bordercolor="#a0c6e5" style="border-collapse:collapse;"><tr><th>工號</th><th>姓名</th><th>性別</th><th>部門</th><th>崗位</th><th>級別</th></tr>{% for key,value in tiptop.items() %}<tr><td>{{ key }} </td><td>{{ value[0] }}</td><td>{{ value[1] }}</td><td>{{ value[2] }}</td><td>{{ value[3] }}</td><td>{{ value[4] }}</td></tr>{% endfor %}</table>{% else %}<ul><li>無TIPTOP賬號停用</li> </ul>{% endif %}<h4>AD域停用名單</h4>{% if ad %}<table border="1" bordercolor="#a0c6e5" style="border-collapse:collapse;"><tr><th>工號</th><th>姓名</th><th>性別</th><th>部門</th><th>崗位</th><th>級別</th></tr>{% for key,value in ad.items() %}<tr><td>{{ key }} </td><td>{{ value[0] }}</td><td>{{ value[1] }}</td><td>{{ value[2] }}</td><td>{{ value[3] }}</td><td>{{ value[4] }}</td></tr>{% endfor %}</table>{% else %}<ul><li>無AD域賬號停用</li> </ul>{% endif %}<br><br><br><br><br><br><br><br><hr><p>說明:因正處測試階段,AD域賬號處理過程中可能會出現錯誤停用賬號的情況,如出現,請登錄域控手動啟用該賬號即可.</p><p>此作業為每周日晚23:30執行一次,將統計周一至周日的所有離職數據進行處理.</p><p>此郵件為系統發送,請勿回覆.</p></body></html>"""#jinja2生成HTMLtemplate = Template(mail_msg)# der(lists=a))html = der(ad=context_ad,tiptop=context_tiptop,all_number=all_number,now=now,begin=begin) #jinja2生成動態HTML# print(html)
message = MIMEMultipart() # 创建一个带附件的实例message.attach(MIMEText(html, 'html', 'utf-8')) # 郵件正文message['From'] = Header('System', 'utf-8') # 郵件發件人Frommessage['To'] = Header('MIS', 'utf-8') # 收件人subject = '離職人員賬號停用名單' # 郵件標題message['Subject'] = Header(subject, 'utf-8')try:smtpObj = smtplib.SMTP()t(mail_host, 25) # 25 为 SMTP 端口号
smtpObj.login(mail_user, mail_pass)smtpObj.sendmail(sender, receivers, message.as_string())smtpObj.quit()print("邮件发送成功")log_txt("邮件发送成功")except Exception as e:print("Error: 邮件發送失敗,%s"%e)log_txt("Error: 邮件發送失敗,%s"%e) # 寫日誌
if __name__ == '__main__':send_mail()
4.停用程序
#!/usr/bin/python3
# coding = utf-8
import os
import sys
import pymysql
from ldap3 import *
from datetime import datetime
import time
import configparser
import cx_Oracle
from sendmail import *def log_txt(msg):now = w()) # 獵取當前時間now = now[:16]try:log_txt = open(&#','a',encoding='utf-8')except:ists(&#',encoding='utf-8')log_txt = open(&#','a',encoding='utf-8')log_txt.write('n%s %s'%(now,msg))log_txt.close()def openconf(database):config = configparser.ConfigParser()msg_conf = {}ad('config.ini') == []:print('沒有找到配置文件!')log_txt('沒有找到配置文件!')time.sleep(it()try:msg_conf['ip'] = config[database]['ip']msg_conf['id'] = config[database]['id']msg_conf['pw'] = config[database]['pw']msg_conf['database'] = config[database]['database'] # 数据库名except:print('配置文件讀取錯誤,程序十秒后將退出!')log_txt('配置文件讀取錯誤,程序十秒后將退出!')# os.system('pause')time.sleep(it()global date_configdate_config = configparser.ConfigParser()date_conf ={}if ad('data.ini') == []:print('沒有找到上次修改日期配置文件!')log_txt('沒有找到上次修改日期配置文件!')time.sleep(it()try:date_conf['last_insert'] = date_config["insert"]['last_insert']except:print('沒有找到上次修改日期配置文件!')log_txt('沒有找到上次修改日期配置文件!')# os.system('pause')time.sleep(it()return msg_conf, date_confdef saveconf(): #新增,更新后寫入日期,用作下次更新條件# if option == 'in':date_config.set('insert','last_insert',str(today))with open('data.ini','w') as configfile:date_config.write(configfile)def openmysql():global todayglobal begin_insertdbauth,date_conf = openconf('mysql') #獲取認證信息和日期配置文件today = day().date() #获取今日日期begin_insert = date_conf['last_insert'] #上次同步日期# today = "2018-08-08" #測試用日期# db = t(dbauth['ip'],dbauth['id'],dbauth['pw'],dbauth['database'],charset ='utf8',as_dict = True) #輸出字典try:# db = t('172.16.0.232', 'root', 'Company', 'employeeinfo', charset='utf8')db = t(dbauth['ip'],dbauth['id'],dbauth['pw'],dbauth['database'],charset='utf8')cursor = db.cursor()# cursor1 = db.cursor()except Exception as e :log_txt('連接MySQL錯誤' + str(e))print(e)print("連接Mysql錯誤,無法連接服務器")time.sleep(it()try:#獲取當天新增加的所有離職信息# ute ("""UPDATE leaveapplication,employee SET leaveapplication.EmpName=employee.EmpName WHERE leaveapplication.EmpNo = employee.EmpNo""")# dbmit()# ute ("""SELECT * FROM employee WHERE LeaveDate BETWEEN '{} 00:00:00' AND '{} 23:59:59' AND DutyName NOT LIKE '%作業員%'""".format(begin_insert,today)) #讀取配置文件ute ("""SELECT * FROM employee WHERE LeaveDate BETWEEN '{} 00:00:00' AND '{} 23:59:59' AND Incumbency = '0'""".format(begin_insert,today)) #讀取配置文件data = cursor.fetchall()except Exception as e:log_txt('執行MySQL語句錯誤' + str(e))print(e)print('執行MySQL語句錯誤')cursor.close()db.close()time.sleep(it()cursor.close()db.close()select_dict = {}if data.__len__() != 0:for cache in data: #用戶處理轉換字典# try:# ute("""SELECT EmpNo,EmpName FROM employee WHERE EmpName = '{}'""".format(cache[1])) == 1:select_dict[cache[0]] = (cache[1],cache[2],cache[129],cache[11],cache[12]) #0工號 1姓名 2性別 129部門名稱 11崗位 12 員工級別# else:# print(cache[1]+'發現同名,請手動處理!')# log_txt(cache[1]+'發現同名,請手動處理!')#出現多個同名人員,郵件通知資訊進行手動處理.# except Exception as e:# log_txt('for執行MySQL語句錯誤' + str(e))# print(e)if select_dict != {}:return select_dictreturn Nonedef LDAP(EmpName_Dict): #停用AD域賬號#LDAP主程序try:server = Server("ldaps://Company:636", use_ssl=True)conn = Connection(server, user="Company\user_registration", password="whoareyou2009", authentication=NTLM, auto_bind=True)except:print('連接AD域控失敗!程序將在十秒后退出!')log_txt('連接AD域控失敗,程序退出!')time.sleep(it()dn_look_path = 'ou=Company,dc=Company,dc=com,dc=cn'print('---------------AD域賬號停用作業----------------')log_txt('---------------AD域賬號停用作業----------------')succeed_AD = []for EmpNo in EmpName_Dict.keys():user_Duty = EmpName_Dict[EmpNo][3]if '作業員' in user_Duty: #如果是作業員則跳過continueaduser = EmpName_Dict[EmpNo][0] #根據鍵 找到值if conn.search(dn_look_path, '(&(objectClass=User)(description={}))'.format(EmpNo), attributes=['cn','sAMAccountName','Description','Mail']): #根據工號查找(描述)# change = {'userAccountControl': [('MODIFY_REPLACE', [66048])], 'Mail': [('MODIFY_REPLACE', ['123@qq'])]} #嘗試將用戶啟用并添加郵箱user_msg = ies[0]try:# print("嘗試停用!!!!測試!!!")changes = {'userAccountControl': [('MODIFY_REPLACE', [66050])],'Mail':[('MODIFY_REPLACE', [])]} #嘗試將用戶停用并刪除郵箱dify(_dn, changes=changes)succeed_AD.append(EmpNo) #停用成功列表print(aduser + ' ' + EmpNo +'--------------->AD域賬號停用成功!')log_txt(aduser + ' ' + EmpNo +'--------------->AD域賬號停用成功!')except:print(aduser + ' ' + EmpNo +' AD域找到用戶但停用失敗!')log_txt(aduser + ' ' + EmpNo +' AD域找到用戶但停用失敗!')else:print(aduser + ' ' + EmpNo +' AD域中沒有找到該用戶!')log_txt(aduser + ' ' + EmpNo +' AD域中沒有找到該用戶!')conn.unbind()conn.closedsaveconf()return succeed_ADdef TIPTOP(EmpNo_Dict): #停用TIPTOP賬號# XXXX = "select name from V$DATABASE" #查询当前连接的数据库i = 0while True:try:db_ora = t('MISTEST', 'mistest@2', "192.168.1.243:1521/TOPPROD", encoding='UTF-8') # 登录数据库,尝试三次连接数据库.cursor_ora = db_ora.cursor()# print(">>>>>连接TIPTOP数据库成功")log_txt('连接TIPTOP数据库成功!')breakexcept:i += 1# sys.stdout.write('r连接数据库失败，正在尝试第{}次'.format(i))# sys.stdout.flush()time.sleep(1.5)if i != 3:continueelse:print("t 尝试三次连接数据库失败,请检查网络连接!")log_txt('连接oracle数据库失败,请检查网络连接!')time.sleep(it()SEL_SQL = """SELECT COUNT(*) FROM DS.ZX_FILE WHERE ZX01=:empno""" #查詢SQL = """UPDATE DS.ZX_FILE SET ZXACTI=:acti WHERE ZX01=:empno""" #根據工號設置ZXACTI 用戶狀態 Y 啟用 N停用 让用户不能登录TIPTOPSQL_USER_MSG_ACTI = """UPDATE SINXON.GEN_FILE SET GENACTI=:acti WHERE GEN01=:empno""" #aooi040 员工资料冻结(无法开单)print('---------------TIPTOP賬號停用作業----------------')log_txt('---------------TIPTOP賬號停用作業----------------')succeed_TIPTOP = []for EmpNo in EmpNo_Dict.keys():try:ute(SEL_SQL,empno='0'+EmpNo)sel_num = cursor_ora.fetchall()ute(SEL_SQL,empno='T'+'0'+EmpNo)sel_num_T = cursor_ora.fetchall()if sel_num[0][0] == 1 : #先檢測用戶是否存在(因為直接更改值的話,即便沒有這個用戶,也不會報錯),所以需要檢測用戶是否存在.ute(SQL,acti='N',empno='0'+EmpNo)ute(SQL_USER_MSG_ACTI,acti='N',empno='0'+EmpNo)db_oramit()log_txt(EmpNo + ' ' + EmpNo_Dict[EmpNo][0] + '--------------->TIPTOP賬號已成功停用!')succeed_TIPTOP.append(EmpNo) # 成功停用名單else:log_txt(EmpNo + ' ' + EmpNo_Dict[EmpNo][0] + ' TIPOP中沒有找到該用戶!')print(EmpNo + ' ' + EmpNo_Dict[EmpNo][0] + ' TIPOP中沒有找到該用戶!')if sel_num_T[0][0] == 1 :ute(SQL,acti='N',empno='T'+'0'+EmpNo)db_oramit()log_txt('T'+'0'+EmpNo + ' ' + EmpNo_Dict[EmpNo][0] + '--------------->TIPTOP賬號已成功停用!')succeed_TIPTOP.append('T'+'0'+EmpNo) # 成功停用名單else:log_txt('T'+'0'+EmpNo +' '+EmpNo_Dict[EmpNo][0]+' TIPOP中沒有找到該用戶!')print('T'+'0'+EmpNo +' '+EmpNo_Dict[EmpNo][0]+' TIPOP中沒有找到該用戶!')except Exception as e:print(e)log_txt(e)cursor_ora.close()db_ora.close()saveconf()return succeed_TIPTOPif __name__ == '__main__':leave_person = openmysql()if leave_person != None: # like {'19262': '韓團結', '39289': '王納', '30106': '成朋飛', '31700': '宋光星'}succeed_ad = LDAP(leave_person)succeed_tiptop = TIPTOP(leave_person)send_mail(succeed_ad,succeed_tiptop,begin_insert,leave_person)print('運行成功,10秒后將自動退出程序.')time.sleep(it()else:print('沒有需要處理的用戶,10秒后將自動退出程序.')log_txt('沒有需要處理的用戶,將自動退出程序.')send_mail(ad=[],tiptop=[],begin=begin_insert)saveconf()time.sleep(it()
转载于:.html
本文发布于:2024-02-05 02:41:39，感谢您对本站的认可！
本文链接：https://www.4u4v.net/it/170722123262274.html
版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系，我们将在24小时内删除。
| 留言与评论（共有 0 条评论） |
