# coding=utf-8 from typing import TYPE_CHECKING, Optional import json import logging import datetime import hashlib from urlparse import urljoin import requests logger = logging.getLogger(__name__) class TianAnException(Exception): def __init__(self, errCode, errMsg, client=None, request=None, response=None): super(TianAnException, self).__init__(errMsg) self.errCode = errCode self.errMsg = errMsg self.client = client self.request = request self.response = response def __str__(self): if self.client: _repr = '{klass}(client: {client}, errCode: {errCode}, errMsg: {errMsg})'.format( klass=self.__class__.__name__, client=repr(self.client), errCode=self.errCode, errMsg=self.errMsg) else: _repr = '{klass}(errCode: {errCode}, errMsg: {errMsg})'.format( klass=self.__class__.__name__, errCode=self.errCode, errMsg=self.errMsg) return _repr def __repr__(self): return str(self) class Constant(object): COMPANY_CODE = 1009 APP_KEY = "k1009yc83einjr9hpkc21i6hzecqw2n7ou" class TianAn(object): URL_BASE = "http://47.96.41.38:8089/api/" def __init__(self, companyCode, appKey, timeout=None, autoRetry=False): self._companyCode = companyCode self._appKey = appKey self._timeout = timeout self._autoRetry = autoRetry def _request(self, method, path, **kwargs): logger.info("[TianAn request] method = {}, path = {}, kwargs = {}".format(method, path, kwargs)) url = urljoin(self.URL_BASE, path) headers = { "Content-Type": "application/x-www-form-urlencoded", } kwargs.setdefault("params", {}) kwargs.setdefault("timeout", self._timeout) processor = kwargs.pop("processor", None) with requests.sessions.Session() as _session: res = _session.request( method=method, url=url, headers=headers, **kwargs ) try: res.raise_for_status() except requests.RequestException as rre: logger.info("[{} send request] error! status code = {}, error = {}".format(self.__class__.__name__, res.status_code, rre)) raise TianAnException( errCode='HTTP{}'.format(res.status_code), errMsg=rre.message, client=self, request=rre.request, response=rre.response ) return self._handle_result( res, method, url, processor, **kwargs ) def _handle_result(self, res, method, url, processor, **kwargs): result = json.loads(res.content.decode('utf-8', 'ignore'), strict=False) logger.info("[TianAn _handle_result] method = {}, url = {}, result = {}".format(method, url, result)) if processor: return processor(self, result) if "code" in result and result["code"] == "0": return result # 重试机制待调试 if self._autoRetry: return self._request(method, res.request.path_url, processor=processor, **kwargs) raise TianAnException( errCode=result.get("code"), errMsg=result.get("msg"), client=self ) def _post(self, **kwargs): return self._request("post", **kwargs) def _get(self, **kwargs): return self._request("get", **kwargs) def _get_token(self, qydm=None, key=None, timestamp=None): """ :param qydm: 企业代码 固定 :param key: 固定 :param timestamp: 时间戳 各位为 yyyy-MM-dd hh24:mm:ss """ qydm = qydm or self._companyCode key = key or self._appKey timestamp = timestamp or datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") logger.debug("[TianAn get_token] qydm = {}, key = {}, timestamp = {}".format(qydm, key, timestamp)) s = "{qydm}{key}{timestamp}".format( qydm=qydm, key=key, timestamp=timestamp ) token = hashlib.md5(s.encode("utf-8")).hexdigest().lower() logger.debug("[TianAn get_token] token = {}".format(token)) return token def submit_charge_insurance(self, orderNo, logicalCode, startTime, province, city, area, address, finishTime=None): qydm, nowtime = self._companyCode, datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") token = self._get_token(qydm, timestamp=nowtime) data = { "qydm": qydm, "token": token, "ddh": orderNo, "sbh": logicalCode, "kssj": startTime, "jssj": finishTime or startTime, "sheng": province, "shi": city, "xian": area, "xxdz": address, "nowtime": nowtime } return self._post(path="adddevdata", data=data) def submit_site_insurance(self, logicalCode, partNums, addTime, address, devType=1, remarks=None): """ :param logicalCode: 设备编号 :param partNums: 插座个数 :param addTime: 安装时间 :param address: 安装地址 :param devType: 设备类型 暂时固定为1 :param remarks: 备注 选填 """ qydm, nowtime = self._companyCode, datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") token = self._get_token(qydm, timestamp=nowtime) data = { "qydm": qydm, "token": token, "nowtime": nowtime, "czlx": "add", "sbbh": logicalCode, "czgs": partNums, "azsj": addTime, "azwz": address, "sblx": devType, "beizhu": remarks or "" } return self._post(path="sbgl", data=data) def edit_site_insurance(self, logicalCode, partNums, addTime, address, devType=1, remarks=None): qydm, nowtime = self._companyCode, datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") token = self._get_token(qydm, timestamp=nowtime) data = { "qydm": qydm, "token": token, "nowtime": nowtime, "czlx": "edit", "sbbh": logicalCode, "czgs": partNums, "azsj": addTime, "azwz": address, "sblx": devType, "beizhu": remarks or "" } return self._post(path="sbgl", data=data) def del_site_insurance(self, logicalCode): qydm, nowtime = self._companyCode, datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") token = self._get_token(qydm, timestamp=nowtime) data = { "qydm": qydm, "token": token, "nowtime": nowtime, "czlx": "del", "sbbh": logicalCode, } return self._post(path="sbgl", data=data) def query_site_insurance(self, startTime, endTime): # type: (Optional[datetime.datetime, str], Optional[datetime.datetime, str]) -> list """ :param startTime: 订单提交的起始日期 :param endTime: 订单提交的结束日期 放回的rows [ { 'aymd': '20230403', 保单生成日期 'bxdh': 'testbx66', 保单号 'bxdhsbxxlist': [ # 对应的设备列表 {'sbbh': 'xxx'}, {'sbbh': 'xxxx'} ], 'bxjssj': '2024-04-03 14:45:47', # 保险开始时间 'bxkssj': '2023-04-03 14:45:45', # 保险结束时间 'bxqx': '1', # 期限 单位年 'dysbnum': 2, # 对应的设备数量 'qydm': '1003', # 企业代码 无用 'sbxxbxdhbid': '2023040314461490434913954', 主键 遇到问题时候查询（售后） 'scsjendymd': '20230331', 该保险单号对应的上传设备开始日期 'scsjstartymd': '20230325' 该保险单号对应的上传设备结束日期 } ] """ qydm, nowtime = self._companyCode, datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") token = self._get_token(qydm, timestamp=nowtime) if isinstance(startTime, (unicode, str)): try: startTime = datetime.datetime.strptime(startTime, "%Y-%m-%d") except ValueError: raise TianAnException(errCode="", errMsg=u"请传入正确的时间格式:%Y-%m-%d") startTime = startTime.strftime("%Y%m%d") if isinstance(endTime, (unicode, str)): try: endTime = datetime.datetime.strptime(endTime, "%Y-%m-%d") except ValueError: raise TianAnException(errCode="", errMsg=u"请传入正确的时间格式:%Y-%m-%d") endTime = endTime.strftime("%Y%m%d") data = { "qydm": qydm, "token": token, "nowtime": nowtime, "startymd": startTime, "endymd": endTime, } result = self._post(path="getSbglBxdh", data=data) if "rows" not in result: raise TianAnException(errCode="", errMsg=u"未获取到数据") return result["rows"] def get_client(): return TianAn(Constant.COMPANY_CODE, Constant.APP_KEY)