import datetime import dateutil import logging import json import re import requests def init(CONFIG, logger): return Mod(CONFIG, logger) class Mod: def __init__(self, CONFIG, logging): self.URL = "https://www.e-license.jp/el25/pc/" self.CONFIG = CONFIG self.logging = logging self.targets = [] self.calender = None self.ss = self.authenticate() def __del__(self): res = self.ss.get( url = self.URL+"logout.action", params = { "b.schoolCd": self.CONFIG["school"], "senisakiCd": 4, }) res.encoding = "shift_jis" self.calender = Calender(res.text) self.logging.info("logout succeeded") def cycle(self): if len(self.targets) == 0: return res = self.ss.post( url = self.URL+"p04a.action", data = { "b.processCd": "A", "b.kamokuCd" : self.calender.params["kamokuCd"], "b.page" : 1, "b.schoolCd" : self.CONFIG["school"], }) res.encoding = "shift_jis" self.calender = Calender(res.text) now = datetime.datetime.utcnow() + datetime.timedelta(hours=9) today = now.strftime("%Y%m%d") for target in self.targets: for slot in target[1]: begins = now.combine(now.date(), TIMETABLE[slot]) border = begins - datetime.timedelta(hours=2) if target[0] != today or now < border: if self.calender.checkAvailable(target[0], slot): self.make_reserve(target[0], slot) target[2]() break self.targets = [] def watch(self, date, slots, cb): self.logging.debug("watch request accepted, {} {}".format(date, slots)) self.targets.append((date, slots, cb)) def make_reserve(self, date, slot): res = self.ss.post( url = self.URL+"p03a.action", data = { "b.schoolCd" : self.CONFIG["school"], "b.processCd" : "V", "b.kamokuCd" : self.calender.params["kamokuCd"], "b.instructorTypeCd" : self.calender.params["instructorTypeCd"], "b.dateInformationType": date, "b.infoPeriodNumber" : slot, "b.carModelCd" : self.calender.params["carModelCd"], "b.instructorCd" : self.calender.params["instructorCd"], "b.page" : 1, "b.groupCd" : self.calender.group, }) res.encoding = "shift_jis" self.calender = Calender(res.text) self.logging.info("reservation made, {}:{}".format(date, slots)) def authenticate(self): ss = requests.Session() try: res = ss.post( url = self.URL+"p01a.action", data = { "b.studentId" : self.CONFIG["username"], "b.password" : self.CONFIG["password"], "b.schoolCd" : self.CONFIG["school"], "b.processCd" : "", "b.kamokuCd" : "", "method:doLogin" : "ログイン".encode("shift_jis"), }) res.encoding = "shift_jis" self.calender = Calender(res.text) self.logging.info("authenticated") return ss except Exception as e: self.logging.error("authentication failure ({})".format(e)) return None class Calender: def __init__(self, text): m = re.search(RE_ERROR, text) if m is None: m = re.search(RE_ALERT, text) if m is not None: raise Exception(m[1]) self.avails = re.findall(RE_AVAIL_SLOT, text) self.params = {} for name in RE_PARAMS: m = re.search(RE_PARAMS[name], text) if m is not None: self.params[name] = m[1] def checkAvailable(self, date, slot): for avail in self.avails: if avail[0] == date and avail[1] == str(slot): return True return False TIMETABLE = { 1: datetime.time( 9,20), 2: datetime.time(10,20), 3: datetime.time(11,20), 4: datetime.time(12,20), 5: datetime.time(14,10), 6: datetime.time(15,10), 7: datetime.time(16,10), 8: datetime.time(17,10), 9: datetime.time(18,10), 10: datetime.time(19,10), } RE_ALERT = re.compile(r"window\.alert\('(.*)'\)") RE_ERROR = re.compile(r"<.*? class=\"errorTitle\">メッセージ<\/.*?>\s*<.*? class=\"errorDisp\">(.*?)<\/.*?>", re.MULTILINE) RE_AVAIL_SLOT = re.compile(r"^\s*$") RE_PARAMS = { "kamokuCd" : None, "instructorTypeCd": None, "carModelCd" : None, "instructorCd" : None, "groupCd" : None, } for name in RE_PARAMS: RE_PARAMS[name] = re.compile( r"