python日期的处理

迷南。 2022-06-07 00:18 169阅读 0赞

最近写了一个python项目，因为要处理时间区间timezone问题，又要处理按年，按月，按天，按班次出报表。所以花了相当多的时间来处理python的时间问题，觉得有必要写篇文章记录下来,以便日后使用.

**1.dateutil.py**

import pytz
    from datetime import datetime,timedelta
    
    def second_interval(start, end):
        reverse = False
        if start > end:
            start, end = end, start
            reverse = True
        mydelta = (end.hour - start.hour)*60*60 + (end.minute - start.minute)*60 + end.second - start.second
        if reverse:
            mydelta =  - mydelta
        return mydelta
    
    def convert_datetime_timezone(dt, tz1, tz2):
        tz1 = pytz.timezone(tz1)
        tz2 = pytz.timezone(tz2)
        #dt = datetime.strptime(dt,"%Y-%m-%d %H:%M:%S")
        dt = tz1.localize(dt)
        dt = dt.astimezone(tz2)
        #dt = dt.strftime("%Y-%m-%d %H:%M:%S")
        return dt
    
    def datetime_localize(dt, tz1):
        tz1 = pytz.timezone(tz1)
        dt = tz1.localize(dt)
        #dt = dt.strftime("%Y-%m-%d %H:%M:%S")
        return dt
    
    
    def toUtcTime(localTime):
        utcTime=localTime.replace(tzinfo=(pytz.timezone("UTC")))
        return utcTime
                
    def withTimeZone(strTimezone, time_native):
        myTime=time_native.replace(tzinfo=(pytz.timezone(strTimezone)))
        return myTime
    
    
    def toLocalDate(strTimeZone, utcmoment_naive):
        utcmoment = utcmoment_naive.replace(tzinfo=pytz.utc)
        localDatetime = utcmoment.astimezone(pytz.timezone(strTimeZone))
        return localDatetime
    
    
    
    def checkWorkShift(startTime,endTime,shift,plantCalendar,myNow):
        local_startTime=convert_datetime_timezone(startTime,"UTC",shift["time_zone"])
        local_endTime=convert_datetime_timezone(endTime,"UTC",shift["time_zone"])
        for calendarItem in plantCalendar:
            shift_date=calendarItem["shift_date"]
            item_startTime=shift_date
            item_endTime=shift_date+(endTime-startTime)
            local_item_startTime=convert_datetime_timezone(item_startTime,"UTC",shift["time_zone"])
            local_item_endTime=convert_datetime_timezone(item_endTime,"UTC",shift["time_zone"])
            if(local_item_startTime==local_startTime and local_item_endTime==local_endTime):
                if(calendarItem["working"]):
                    return True
                else:
                    return False
        localDate=local_startTime.date()
        working=isWorking(localDate.weekday(),shift)
        if(working):
            myNow2=datetime.utcnow()
            left=local_startTime+(endTime-startTime)+(myNow2-myNow)+(myNow-endTime)
            right=convert_datetime_timezone(myNow2,"UTC",shift["time_zone"])
            if(left<=right):
                return True
        return False
    
    def isWorking(weekday,shift):
        if(weekday==0):
            if(shift["monday"]):
                return True
        if(weekday==1):
            if(shift["tuesday"]):
                return True
        if(weekday==2):
            if(shift["wednesday"]):
                return True
        if(weekday==3):
            if(shift["thursday"]):
                return True
        if(weekday==4):
            if(shift["friday"]):
                return True
        if(weekday==5):
            if(shift["saturday"]):
                return True
        if(weekday==6):
            if(shift["sunday"]):
                return True
    
    def isSheduleShift(myDateTime,shift,scheduleweekdays):
        oneDay = timedelta(days=1)
        localDateTime=toLocalDate(shift["time_zone"],myDateTime)
        localDate=localDateTime.date()
        localDateStart=datetime.combine(localDate, datetime.min.time())
        localDateEnd=localDateStart+oneDay
        startTime=toLocalDate(shift["time_zone"],datetime.combine(localDate,shift["from_time"]))
        endTime=toLocalDate(shift["time_zone"],datetime.combine(localDate, shift["to_time"]))
        localDateStart=toLocalDate(shift["time_zone"],localDateStart)
        localDateEnd=toLocalDate(shift["time_zone"],localDateEnd)
        
        if(startTime>=localDateStart and endTime<=localDateEnd):
            for wd in scheduleweekdays:
                    if(localDate.weekday()==wd):
                        return True
        return False
    
    def getSheduleWeekdays(shift):
        scheduleweekdays=[];
        if(shift["monday"]):
            scheduleweekdays.append(0)
        if(shift["tuesday"]):
            scheduleweekdays.append(1)
        if(shift["wednesday"]):
            scheduleweekdays.append(2)
        if(shift["thursday"]):
            scheduleweekdays.append(3)
        if(shift["friday"]):
            scheduleweekdays.append(4)
        if(shift["saturday"]):
            scheduleweekdays.append(5)
        if(shift["sunday"]):
            scheduleweekdays.append(6)
        return scheduleweekdays

**2.业务代码**

'''
    Created on Jun 28, 2017
    
    @author: Wilson.Ke
    '''
    from datetime import timedelta
    from oee_app.utils.dateutils import convert_datetime_timezone
    from oee_app.daos import lineoeeresultDao
    
    class Trend:
        
        def getTrend(self,viewBy,siteId,strtz,lineName,machineName,fromDate,toDate):
            datas=[]
            totalShiftTime=0
            if(viewBy==1):
                hours=self.getHours(fromDate,toDate)
                for hour in hours:
                    data={}
                    h1=hour["start_time"].replace(tzinfo=None)
                    h2=hour["end_time"].replace(tzinfo=None)
                    lineData=lineoeeresultDao.getLineOEE(siteId, lineName, machineName,h1,h2)
                    oee=0.0
                    schdowntime=0
                    for item in lineData:
                        schdowntime=schdowntime+float(item["schdowntime"])
                        oee=oee+float(item["oee"])
                    data["schdowntime"]=schdowntime    
                    data["oee"]=oee
                    data["name"]=str(hour["start_time"])+"~"+str(hour["end_time"])
                    data["shift_time"]=(h2-h1).total_seconds()
                    data["calendar_time"]=data["shift_time"]
                    data["nonschtime"]=0
                    datas.append(data)
            if(viewBy==2):
                days=self.getDays(fromDate,toDate)
                for day in days:
                    data={}
                    d1=day["start_time"].replace(tzinfo=None)
                    d2=day["end_time"].replace(tzinfo=None)
                    lineData=lineoeeresultDao.getLineOEE(siteId, lineName, machineName,d1,d2)
                    #data=self.getTrendEachDay(siteId, lineName, machineName,d1,d2)
                    oee=0.0;
                    totalShiftTime=0.0
                    totalNoneSchedule=0.0
                    schdowntime=0.0
                    totalCalendarTime=0.0
                    for item in lineData:
                        oee=oee+float(item["oee"])*float(item["shift_time"])
                        schdowntime=schdowntime+float(item["schdowntime"])
                        totalNoneSchedule=totalNoneSchedule+float(item["nonschtime"])
                        totalCalendarTime=totalCalendarTime+item["shift_time"]
                        if item["working"]==1:
                            totalShiftTime=totalShiftTime+item["shift_time"]
                    if(totalShiftTime!=0):
                        data["oee"]=oee/float(totalShiftTime);
                    data["calendar_time"]=totalCalendarTime
                    data["shift_time"]=totalShiftTime
                    data["schdowntime"]=schdowntime
                    day_start_time=convert_datetime_timezone(d1, "UTC", strtz)
                    day_end_time=convert_datetime_timezone(d2, "UTC", strtz)
                    data["name"]=day_start_time.strftime("%Y-%m-%d")+"~"+day_end_time.strftime("%Y-%m-%d")
                    data["nonschtime"]=totalNoneSchedule
                    datas.append(data)
            if(viewBy==3):
                weeks=self.getWeeks(fromDate,toDate)
                for week in weeks:
                    data={}
                    w1=week["start_time"].replace(tzinfo=None)
                    w2=week["end_time"].replace(tzinfo=None)
                    lineData=lineoeeresultDao.getLineOEE(siteId, lineName, machineName,w1,w2)
                    week_start_time=convert_datetime_timezone(w1, "UTC", strtz)
                    week_end_time=convert_datetime_timezone(w2, "UTC", strtz)
                    data["name"]=week_start_time.strftime("%Y-%m-%d")+"~"+week_end_time.strftime("%Y-%m-%d")
                    oee=0.0;
                    totalShiftTime=0.0
                    totalNoneSchedule=0.0
                    schdowntime=0.0
                    totalCalendarTime=0.0
                    for item in lineData:
                        oee=oee+item["oee"]*float(item["shift_time"])
                        schdowntime=schdowntime+float(item["schdowntime"])
                        totalNoneSchedule=totalNoneSchedule+float(item["nonschtime"])
                        totalCalendarTime=totalCalendarTime+item["shift_time"]
                        if item["working"]==1:
                            totalShiftTime=totalShiftTime+item["shift_time"]
                    if(totalShiftTime!=0):
                        data["oee"]=oee/totalShiftTime
                    data["calendar_time"]=totalCalendarTime
                    data["shift_time"]=totalShiftTime
                    data["schdowntime"]=schdowntime
                    data["nonschtime"]=totalNoneSchedule
                    datas.append(data) 
            if(viewBy==4):
                months=self.getMonths(strtz,fromDate,toDate)
                for month in months:
                    data={}
                    m1=month["start_time"].replace(tzinfo=None)
                    m2=month["end_time"].replace(tzinfo=None)
                    lineData=lineoeeresultDao.getLineOEE(siteId, lineName, machineName,m1,m2)
                    data["name"]=month["curr_month"]
                    oee=0.0;
                    totalShiftTime=0.0
                    totalNoneSchedule=0.0
                    totalCalendarTime=0.0
                    schdowntime=0.0
                    for item in lineData:
                        oee=float(oee)+float(item["oee"])*float(item["shift_time"])
                        schdowntime=schdowntime+float(item["schdowntime"])
                        totalNoneSchedule=totalNoneSchedule+float(item["nonschtime"])
                        totalCalendarTime=totalCalendarTime+item["shift_time"]
                        if item["working"]==1:
                            totalShiftTime=totalShiftTime+item["shift_time"]
                    if(totalShiftTime!=0):
                        data["oee"]=oee/totalShiftTime;
                    data["calendar_time"]=totalCalendarTime
                    data["shift_time"]=totalShiftTime
                    data["schdowntime"]=schdowntime
                    data["nonschtime"]=totalNoneSchedule
                    datas.append(data)
            return self.processData(datas,fromDate,toDate)
        
        def processData(self,datas,fromDate,toDate):
            items=[]
            for data in datas:
                item={}
    #             calendarTime=(toDate-fromDate).total_seconds()
                shiftTime=data["shift_time"]
                if shiftTime==0:
                    item["name"]=data["name"]
                    item["oee"]=0.0
                    item["teep"]=0.0
                else:
                    calendarTime=data["calendar_time"]
                    planedProductionTime=calendarTime-data["schdowntime"]
                    item["name"]=data["name"]
                    oee=data["oee"]
                    loading=(planedProductionTime-data["nonschtime"])/calendarTime
                    teep=float(loading)*float(oee)          
                    item["oee"]=round(oee*100,1)
                    item["teep"]=round(teep*100,1)
                items.append(item)
            return items
            
        
        def getTrendEachHour(self,siteId, lineName, machineName,startTime,endTime):
            data=lineoeeresultDao.getOEETrend(siteId, lineName, machineName, startTime, endTime)
            return data
        
        def getTrendEachDay(self,siteId, lineName, machineName,startTime,endTime):
            data=lineoeeresultDao.getOEETrend(siteId, lineName, machineName, startTime, endTime)
            return data
        
        def getTrendEachWeek(self,siteId, lineName, machineName,startTime,endTime):
            data=lineoeeresultDao.getOEETrend(siteId, lineName, machineName, startTime, endTime)
            return data
        def getTrendEachMonth(self,siteId, lineName, machineName,startTime,endTime):
            data=lineoeeresultDao.getOEETrend(siteId, lineName, machineName, startTime, endTime)
            return data
        
        def getHours(self,startTime,endTime):
            arrHour=[]
            oneHour = timedelta(hours=1)
            while startTime<endTime:
                currTime = startTime +oneHour
                myHour={}
                myHour["start_time"]=startTime
                myHour["end_time"]=currTime
                arrHour.append(myHour)
                startTime=currTime
            if(endTime>startTime):
                myHour={}
                myHour["start_time"]=startTime
                myHour["end_time"]=endTime
                arrHour.append(myHour)
            return arrHour
        
        def getDays(self,startTime,endTime):
            arrDay=[]
            oneDay = timedelta(days=1)
            while startTime<endTime:
                myDay={}
                myDay["start_time"]=startTime
                startTime = startTime +oneDay
                myDay["end_time"]=startTime
                arrDay.append(myDay)
            return arrDay
        
        def getWeeks(self,startTime,endTime):
            arrWeek=[]
            oneDay = timedelta(days=1)
            weekStart=startTime
            while startTime<endTime:
                if(startTime.weekday()==4):
                    myWeek={}
                    myWeek["start_time"]=weekStart
                    myWeek["end_time"]=startTime+oneDay
                    arrWeek.append(myWeek)
                    weekStart=myWeek["end_time"]
                startTime= startTime +oneDay
                
            if(endTime>weekStart):
                myWeek={}
                myWeek["start_time"]=weekStart
                myWeek["end_time"]=endTime
                arrWeek.append(myWeek)       
            return arrWeek
        
        def getMonths(self,strtz,startTime,endTime):
            oneDay = timedelta(days=1)
            arrMonth=[]
            monthStart=startTime
            while startTime<endTime:
                d1=monthStart.replace(tzinfo=None)
                localMonthStart=convert_datetime_timezone(d1, "UTC", strtz)
                d2=startTime.replace(tzinfo=None)
                localStart=convert_datetime_timezone(d2, "UTC", strtz)
                if(localMonthStart.month!=localStart.month):
                    myMonth={}
                    myMonth["curr_month"]=str(localMonthStart.year)+"-"+str(localMonthStart.month)
                    myMonth["start_time"]=monthStart
                    myMonth["end_time"]=startTime
                    arrMonth.append(myMonth)
                    monthStart=startTime
                startTime=startTime+oneDay
            d1=monthStart.replace(tzinfo=None)
            localMonthStart=convert_datetime_timezone(d1, "UTC", strtz)
            d2=startTime.replace(tzinfo=None)
            localStart=convert_datetime_timezone(d2, "UTC", strtz)
            if(endTime>=startTime and monthStart.month!=localStart.month):
                myMonth={}
                myMonth["curr_month"]=str(startTime.year)+"-"+str(startTime.month)
                myMonth["start_time"]=monthStart
                myMonth["end_time"]=endTime
                arrMonth.append(myMonth)
            return arrMonth