# coding: utf-8 import pymssql import csv import datetime from file2zip import * # from send_email import * import plugs.FileHelper as fileHelper from file2zip import FileToZip #類名 from send_email import SendMail #類名 class TestCSVInfo(fileHelper.BaseCheck, SendMail,FileToZip): report = '' def __init__(self, month): SendMail.__init__(self, month) def get_sel_excel(self): print(" 開始鏈接數據庫!") # 建立連接 conn = pymssql.connect( host='localhost', port=1433, user='test', password='test', database='Test', charset='utf8' ) dateFrom = '2020-04-01 0:00:00' dateTo = '2020-04-30 23:59:59' today = datetime.datetime.now().strftime('%Y-%m-%d') try: print(os.path.exists(month)) if os.path.exists(month) == False: os.makedirs(month) # 建立游標 cursor = conn.cursor() # <editor-fold desc="查詢語句"> titile = today + '_' + '商品' header = ['訂單號', '金額', '日期'] print("開始查詢表!") # 執行sql語句 _sql = self.getsql("test.sql") Sql = _sql.replace('v_from', str(dateFrom)).replace('v_to', str(dateTo)) print(titile + '——數據庫查詢中。。。') cursor.execute(Sql) # 獲取查詢到結果 res = cursor.fetchall() self.w_excel(res, header, titile) # </editor-fold> cursor.close() conn.close() #壓縮為zip self.addZip(month) # 發送郵件 self.SendMethod(month) except Exception as e: print("報錯{}".format(e)) cursor.close() conn.close() # 操作csv def w_excel(self, res, header, titile): with open(f'{month}/' + titile + '.csv', 'w', newline='', encoding='utf-8-sig') as csvfile: writer = csv.writer(csvfile) # 寫表頭 writer.writerow(header) # 寫入數據 for row in range(0, len(res)): writer.writerow(list(res[row])) print(titile + f"導出成功!,共有條數:{len(res)}") print(titile + "結束-------------------------------------------------------------------------") if __name__ == "__main__": _today = datetime.date.today() _first = _today.replace(day=1) last_month = _first - datetime.timedelta(days=1) month = last_month.strftime('%Y-%m') #獲取上個月月份 # month = datetime.datetime.now().strftime('%Y-%m') #獲取當前月份 TestCSVInfo(month).get_sel_excel()