flightinfo_sever/Functions/utils.py

import datetime
import time
import json
import requests
from selenium.webdriver.support import expected_conditions as EC
from selenium import webdriver
from selenium.webdriver import ActionChains
from selenium.webdriver.common.by import By
from selenium.webdriver.support.wait import WebDriverWait
from selenium.webdriver.chrome.options import Options


class flight_list(object):

    def __init__(self):
        self.url = r"http://me.sichuanair.com/login.shtml"
        self.login_url = "https://login.sichuanair.com/idp/AuthnEngine?currentAuth=urn_oasis_names_tc_SAML_2.0_ac_classes_BAMUsernamePassword"
        self.flight_list_url = "https://me.sichuanair.com/api/v1/plugins/LM_FLIGHT_LIST"
        self.flight_list_third_url = "https://me.sichuanair.com/api/v1/plugins/LM_FLIGHT_THIRD_LIST"
        self.task_flight_list_url="https://me.sichuanair.com/api/v1/plugins/LM_TASK_ASSIGNMENT_LIST"
        self.LM_FJ_TASK_PG="https://me.sichuanair.com/api/v1/plugins/LM_FJ_TASK_PG"
        self.task_filght_third_list_url="https://cscsupplier.sichuanair.com/api/v1/plugins/LM_TASK_ASSIGNMENT_THIRD_LIST"
        self.LM_TSK_DINGDONG_url ="https://cscsupplier.sichuanair.com/api/v1/plugins/LM_TSK_DINGDONG"
        self.LM_TSK_SURE_PG_url="https://cscsupplier.sichuanair.com/api/v1/plugins/LM_TSK_SURE_PG"
        self.LM_TSK_EMP_PGLIST_url="https://cscsupplier.sichuanair.com/api/v1/plugins/LM_TSK_EMP_PGLIST"
        self.MM_GJJH_LIST="https://cscsupplier.sichuanair.com/api/v1/plugins/MM_GJJH_LIST"
        self.LM_TSK_HANDOVER_url="https://cscsupplier.sichuanair.com/api/v1/plugins/LM_TSK_HANDOVER"
        self.LM_FLIGHT_SEARCH_LIST='https://cscsupplier.sichuanair.com/api/v1/plugins/LM_FLIGHT_SEARCH_LIST'
        self.LM_TASK_ARCHIVE_LIST='https://cscsupplier.sichuanair.com/api/v1/plugins/LM_TASK_ARCHIVE_LIST'
        self.jobcard='https://cscsupplier.sichuanair.com'

        self.token = None
        #amro状态
        self.amro_status = 0
        self.flight_list_json = None
        self.flight_list_third_json = None


    def start(self, username, password):
        try:
            self.token = self.login(username, password)
        except Exception:
            pass

        return self.token

    def login(self, username, password):
        try:

            options = webdriver.EdgeOptions()
            options.use_chromium = True
            #options.add_experimental_option('excludeSwitches', ['enable-logging'])
            #options.add_argument("headless")
            #options.add_argument("disable-gpu")
            #options.add_argument('start-maximized')
            #options.add_argument('window-size=1920x1080')
            self.driver = webdriver.Edge(options=options)
            self.actions = ActionChains(self.driver)
        except:
            chrome_options = Options()
            chrome_options.add_argument("--headless")
            chrome_options.add_argument("window-size=1920x1080")
            chrome_options.add_argument("--start-maximized")
            chrome_options.add_argument('--disable-gpu')  # 如果不加这个选项，有时定位会出现问题
            self.driver = webdriver.Chrome(executable_path=r'D:\flightinfo\Google\Chrome\Application\chromedriver.exe',
                                           # executable_path=path
                                           options=chrome_options)

            self.actions = ActionChains(self.driver)


        self.driver.get(self.url)
        WebDriverWait(self.driver, 10).until(EC.presence_of_element_located((By.ID, 'singleSubmit')))
        loginbtn = self.driver.find_element('id', 'singleSubmit')
        time.sleep(0.5)
        self.actions.click(loginbtn)
        self.actions.perform()
        WebDriverWait(self.driver, 10).until(EC.presence_of_element_located((By.NAME, "j_username"))).send_keys(
            username)
        time.sleep(0.5)
        self.driver.find_element('name', "j_password").send_keys(password)
        #time.sleep(0.5)
        time.sleep(0.5)
        self.driver.find_element_by_xpath("//button[@type='button']").click()
        time.sleep(1)
        for cookie in self.driver.get_cookies():
            if "_amro_sk" in cookie.values():
                self.amro_status = 1
                self.driver.quit()
                return "_amro_sk=" +cookie["value"]
            else:
                pass
        self.driver.quit()
    def get_cookie(self):
        if self.token!=None:
            return self.token
        else:
            return None

    def checkCookieSts(self,cookie):
        url = "https://me.sichuanair.com/api/v1/plugins/PROCESS_CLAIM_TASK_PRO_LIST"
        data = {
            'user_id': '',
            'userId': '',
            'accountType': 'ARCHIVE',
            'page': '1',
            'rows': '11'
        }
        header = {"Accept": "application/json, text/javascript, */*; q=0.01",
                  "Cookie": cookie}
        result = requests.post(url, data, headers=header).json()

        #print(result['code'], result['msg'], result['data'])
        return result['code']


    def checkWorkjob(self,taskids,acno,taskType,actype,startDate,endDate,cookie):
        data = {"taskids": taskids,
                "acno": acno,
                "taskType": taskType,
                "actype": actype,
                "startDate": startDate,
                "endDate": endDate
                }
        header = {"Accept": "application/json, text/javascript, */*; q=0.01",
                  "Cookie": cookie}
        try:
            self.LM_FJ_TASK_PG_json = requests.post(url=self.LM_FJ_TASK_PG, data=data, headers=header).json()
        except Exception:
            self.LM_FJ_TASK_PG_json = None
        return  self.LM_FJ_TASK_PG_json

    def request_jobcard(self, cookie, filename):  # 需要自动逻辑白班夜班
        one_year_later = datetime.datetime.now() + datetime.timedelta(days=365)
        timestamp = int(one_year_later.timestamp()*1000)
        header = {
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
            "Cookie": cookie}
        try:
            task_jobcard_json = requests.get(url=f'https://cscsupplier.sichuanair.com{filename}?_cache_timestamp={timestamp}',headers=header).content
        except Exception:
            task_jobcard_json = None
        return task_jobcard_json

    def request_MM_GJJH_LIST(self,cookie):
        yesterday=(datetime.date.today() - datetime.timedelta(days=1)).strftime("%Y-%m-%d")
        today = datetime.date.today().strftime("%Y-%m-%d")
        data={"mfrpn":"",
            "zjcdats": yesterday,
            "zjcdate": today,
            "zwgh": "Y",
            "zlynam":"",
            "zbmtxt":"",
            "zkstxt":"",
            "zfdtxt":"",
            "zlgort": "TF20",
            "zghnam":"",
            "ghrbm":"",
            "ghrks":"",
            "ghrfd":"",
            "zghkcd":"",
            "sort": "ZBMTXT",
            "order": "asc"
            }
        header = {"Accept": "application/json, text/javascript, */*; q=0.01",
                  "Cookie": cookie}
        try:
            self.MM_GJJH_LIST_json = requests.post(url=self.MM_GJJH_LIST, data=data, headers=header).json()
        except Exception:
            self.MM_GJJH_LIST_json = None
        return self.MM_GJJH_LIST_json

    def request_task_filght_list(self, flightDate:str,cookie):   #需要自动逻辑白班夜班
        data = {"airportCode": "ZUTF",
                "ddate": flightDate,
                "notView":"",
                "baseCode": "TF01",
                "actype1":"(A319|A320|A321)",
                "aclocArea1": "()",
                "tasktype1": "()",
                "shift":"",
                "tasktype":"",
                "actype": "A319,A320,A321",
                "aclocArea":"",
                "isshiftex":"",
                "repush":"",
                "page": "1",
                "rows": "500"
                }
        header = {"Accept": "application/json, text/javascript, */*; q=0.01",
                  "Cookie": cookie}
        try:
            self.task_flight_list_json = requests.post(url=self.task_flight_list_url, data=data, headers=header).json()
        except Exception:
            self.task_flight_list_json = None
        #with open('./temp/task_flight_list_json/fl{}.json'.format(int(time.time())), 'w') as f:
            #json.dump(self.task_flight_list_json, f, indent=4)

        return self.task_flight_list_json


    def request_task_filght_third_list(self, flightDate:str, cookie):   #需要自动逻辑白班夜班
        data = {"airportCode": "ZUTF",
                "ddate": flightDate,
                "notView":"",
                "baseCode": "TF01",
                "isThird": "Y",
                "actype1":"",
                "aclocArea1": "()",
                "tasktype1": "()",
                "shift":"",
                "tasktype":"",
                "actype": "(A319|A320|A321|B737NG|B737MAX|C919)",
                "aclocArea":"",
                "isshiftex":"",
                "repush":"",
                "page": "1",
                "rows": "500"
                }
        header = {"Accept": "application/json, text/javascript, */*; q=0.01",
                  "Cookie": cookie}
        try:
            self.task_filght_third_list = requests.post(url=self.task_filght_third_list_url, data=data, headers=header).json()
        except Exception:
            self.task_filght_third_list = None
        #with open('./temp/task_filght_third_list/fl{}.json'.format(int(time.time())), 'w') as f:
        #    json.dump(self.task_filght_third_list, f, indent=4)
        return self.task_filght_third_list

    def request_LM_TASK_ARCHIVE_LIST(self, flightDate:str, cookie):   #需要自动逻辑白班夜班
        startdatePlStart=flightDate+" 00:00:00"
        flightDate2=(datetime.datetime.strptime(flightDate, "%Y%m%d")+datetime.timedelta(days=1)).strftime("%Y%m%d")
        startdatePlEnd=flightDate2+" 09:00:00"
        data = {"startdatePlStart": startdatePlStart,
                "startdatePlEnd": startdatePlEnd,
                "airportCode":'ZUTF',
                "taskTypeDefault": "LM",
                "actype1": '(A319|A320|A321)',
                "actype": "A319,A320,A321",
                "baseCode": "TF01",
                "page":"1",
                "rows": "500"
                }
        header = {"Accept": "application/json, text/javascript, */*; q=0.01","Cookie": cookie}
        try:
            self.task_LM_TASK_ARCHIVE_LIST = requests.post(url=self.LM_TASK_ARCHIVE_LIST, data=data, headers=header).json()
        except Exception:
            self.task_LM_TASK_ARCHIVE_LIST = None
        #with open('./temp/task_LM_FLIGHT_SEARCH_LIST/fl{}.json'.format(int(time.time())), 'w') as f:
        #    json.dump(self.task_LM_FLIGHT_SEARCH_LIST, f, indent=4)
        return self.task_LM_TASK_ARCHIVE_LIST

    def request_LM_FLIGHT_SEARCH_LIST(self, flightDate:str, cookie):   #需要自动逻辑白班夜班
        data = {"base4code": "ZUTF",
                "flightDate": flightDate,
                "flightDate1":flightDate,
                "jcType": "",
                "actype1": '(A319|A320|A321)',
                "dep_4code": "",
                "arr_4code": "",
                "acno":'',
                "actype": "A319,A320,A321",
                "flightNo":"",
                "notView":"",
                "page":"1",
                "onlyAf":"",
                "rows": "500"
                }
        header = {"Accept": "application/json, text/javascript, */*; q=0.01","Cookie": cookie}
        try:
            self.task_LM_FLIGHT_SEARCH_LIST = requests.post(url=self.LM_FLIGHT_SEARCH_LIST, data=data, headers=header).json()
        except Exception:
            self.task_LM_FLIGHT_SEARCH_LIST = None
        #with open('./temp/task_LM_FLIGHT_SEARCH_LIST/fl{}.json'.format(int(time.time())), 'w') as f:
        #    json.dump(self.task_LM_FLIGHT_SEARCH_LIST, f, indent=4)
        return self.task_LM_FLIGHT_SEARCH_LIST

    def request_filght_list(self, flightDate:str, cookie):
        data = {"base4code": "ZUTF",
                "flightDate": flightDate,#2023-06-02
                "tasktype": "",
                "notView": "",
                "jcType": "",
                "actype1": "(A319|A320|A321)",
                "actype": "A319,A320,A321",
                "acno": "",
                "dep_4code": "",
                "arr_4code": "",
                "flightNo": "",
                "repush": ""

                }
        header = {"Accept": "application/json, text/javascript, */*; q=0.01",
                  "Cookie": cookie}
        try:
            self.flight_list_json = requests.post(url=self.flight_list_url, data=data, headers=header).json()
        except Exception:
            self.flight_list_json = None
        #with open('./temp/filght_list/fl{}.json'.format(int(time.time())), 'w') as f:
        #    json.dump(self.flight_list_json, f, indent=4)
        return self.flight_list_json


    def request_LM_TSK_HANDOVER(self,dict:dict,cookie):
        data = {
                "taskids": dict["taskid"],
                "tasksts": dict["sts"],  # 必须,可能需要修改，不知道影响不
                "FunctionCode": "LM_TSK_HANDOVER"
                }
        header = {"Accept": "application/json, text/javascript, */*; q=0.01",
                  "Cookie": cookie}
        try:
            self.LM_TSK_HANDOVER = requests.post(url=self.LM_TSK_HANDOVER_url, data=data, headers=header).json()
            #print(self.LM_TSK_HANDOVER)
        except Exception:
            self.LM_TSK_HANDOVER = None
        return self.LM_TSK_HANDOVER

    def request_LM_TSK_DINGDONG(self,dict:dict,cookie):
        data = {"tasksts": "1",  # 必须
                "taskid": "{}".format(dict["taskid"]),
                "acno": dict["acno"],
                "actype": dict["actype"],
                "tasktype": dict["tasktype"],
                "tatd": dict["tatd"],
                "msgInfo": dict["msgInfo"],
                "bay": dict["bay"],
                "wxemp": dict["wxemp"],  # 必须dict["wxemp"]
                "fxemp": dict["fxemp"],
                "wx": dict["wx"],
                "fx": dict["fx"],
                "ecsj": dict["ecsj"],
                "ecsjEmp": "",
                "FunctionCode": "LM_TSK_DINGDONG"
                }
        header = {"Accept": "application/json, text/javascript, */*; q=0.01",
                  "Cookie": cookie}
        try:
            self.LM_TSK_DINGDONG = requests.post(url=self.LM_TSK_DINGDONG_url, data=data, headers=header).json()
        except Exception:
            self.LM_TSK_DINGDONG = None
        return self.LM_TSK_DINGDONG
    def request_flight_third_list(self, flightDate: str, cookie):
        data = {"base4code": "ZUTF",
                "flightDate": flightDate,
                "flightDateStart": flightDate,
                "tasktype": "",
                "notView": "",
                "isThird":"Y",
                "actype1": "(21N|319|320|321|32N|332|738|73G|A21N|A319|A319|A320|A321|A32N|A738|A73G|AZZZ|B737|C919)",
                "actype": "B737NG,B737MAX,A319,A320,A321,C919"
                #"actype": "A319%2CA320%2CA321%2CB737MAX%2CB737NG"
                }
        header = {"Accept": "application/json, text/javascript, */*; q=0.01",
                  "Cookie": cookie}
        try:
            self.flight_third_list = requests.post(url=self.flight_list_third_url, data=data, headers=header).json()
        except Exception:
            self.flight_third_list = None
        #with open('./temp/filght_list/fl{}.json'.format(int(time.time())), 'w') as f:
        #    json.dump(self.flight_third_list, f, indent=4)
        return self.flight_third_list

    def request_LM_TSK_SURE_PG(self,dict:dict,cookie):
        empNos = dict["empNos"]
        taskid = dict["taskid"]
        type = dict["type"]
        shiftDate = dict["shiftDate"]
        shift = dict["shift"]

        data = {"empNos": empNos,
                "taskid": taskid,
                "type": type,                 #ECSJ 二次送机  WX  FX
                "shiftDate": shiftDate,                   #夜班航前需要前一天
                "shift": shift,
                "FunctionCode": "LM_TSK_SURE_PG"
                }
        header = {"Accept": "application/json, text/javascript, */*; q=0.01",
                  "Cookie": cookie}
        try:
            #print(data)
            self.LM_TSK_SURE_PG = requests.post(url=self.LM_TSK_SURE_PG_url, data=data, headers=header).json()
            #print(self.LM_TSK_SURE_PG )
        except Exception:
            self.LM_TSK_SURE_PG = None
        return self.LM_TSK_SURE_PG

    def request_LM_TSK_EMP_PGLIST(self,taskid:str,shiftDate:str,shift:str,cookie):
        data = {"flightDate": shiftDate,
                "shift": shift,
                "type": "WX",
                "taskid":taskid,
                "wx": "wx",
                "baseCode": "TF01",       #EMP_NO，NAME，MAINLY_ROLE
                "deptNo": "",
                "fx": "fx",
                }
        header = {"Accept": "application/json, text/javascript, */*; q=0.01",
                  "Cookie": cookie}
        try:
            self.LM_TSK_EMP_PGLIST = requests.post(url=self.LM_TSK_EMP_PGLIST_url, data=data, headers=header).json()
        except Exception:
            self.LM_TSK_EMP_PGLIST = None
        #with open('./temp/LM_TSK_EMP_PGLIST/fl{}.json'.format(int(time.time())), 'w') as f:
        #    json.dump(self.LM_TSK_EMP_PGLIST, f, indent=4)
        return self.LM_TSK_EMP_PGLIST



if __name__ == '__main__':
    fl = flight_list()
    #fl.start()
    nowDay = datetime.datetime.now().replace(microsecond=0)
    nowDayStr = nowDay.strftime("%Y-%m-%d")