Skip to content

Instantly share code, notes, and snippets.

@mark99i

mark99i/ezviz_api.py

Last active March 5, 2023 02:56
Show Gist options
  • Save mark99i/00dbff3ceb2616dfc6be521a007ffd05 to your computer and use it in GitHub Desktop.
Save mark99i/00dbff3ceb2616dfc6be521a007ffd05 to your computer and use it in GitHub Desktop.
Download ZIP
EZVIZ Cloud cameras (RU endpoints)
Raw
ezviz_api.py
import base64
import json
import time
from hashlib import md5
from urllib.parse import urlencode
from requests import get, post, put, Response
import jwt
class BaseRequest:
success: bool
code: int
message: str
class LoginResult(BaseRequest):
session_id: str
session_rf: str
area_id: int
need_refresh: bool
def tostr(self):
return json.dumps({
"session_id": self.session_id,
"session_rf": self.session_rf,
"areaid": self.area_id
}, indent=4)
@staticmethod
def loadstr(data: str) -> 'LoginResult':
lr = LoginResult()
try:
data = json.loads(data)
lr.session_rf = data['session_rf']
lr.session_id = data['session_id']
lr.area_id = data['areaid']
nt = int(time.time())
dec = jwt.decode(lr.session_id, verify=False, options={'verify_signature': False})
exp = dec.get('exp')
iat = dec.get('iat')
halftime = round((exp - iat) / 2)
lr.need_refresh = iat + halftime < nt
dec = jwt.decode(lr.session_rf, verify=False, options={'verify_signature': False})
exp = dec.get('exp')
lr.success = exp > nt
except Exception as e:
lr.success = False
lr.message = f"EzvizApi:LoginResult:{e}"
return lr
class RecordsResult(BaseRequest):
records: list[dict[str, str]]
class CamerasResult(BaseRequest):
class Page:
offset: int
limit: int
has_next: bool
class Camera:
class Connection:
ip: str
mask: str
gw: str
signal: int
ssid: str
wan_ip: str
cmd_port: int
stream_port: int
name: str
serial: str
device_code: str
model: str
mac: str
need_upgrade: bool
owner_username: str
enc_enabled: bool
enc_password: str
connection: Connection
class CameraList(list[Camera]):
def find_by(self, name: str = None, serial: str = None):
if serial is not None:
for i in self:
if i.serial == serial: return i
if name is not None:
for i in self:
if i.name == name: return i
return None
page: Page
cameras: CameraList
class Constants:
LOGIN_ENDPOINT = 'https://api.ezvizru.com/v3/users/login/v5'
LOGIN_RF_ENDPOINT = 'https://api.ezvizru.com/v3/apigateway/login'
RECORDS_ENDPOINT = 'https://apiirus.ezvizru.com/v3/streaming/records'
CAMERAS_ENDPOINT = 'https://apiirus.ezvizru.com/v3/userdevices/v1/resources/pagelist'
FC = "896be422a6df398453e3dd4a6894721c"
UA = "okhttp/3.12.1"
@staticmethod
def get_auth_headers() -> dict:
return {
"featurecode": Constants.FC,
"clienttype": "3",
"osversion": "12",
"clientversion": "5.9.8.0215",
"nettype": "LTE",
"customno": "1000001",
"ssid": "<unknown ssid>",
"clientno": "google",
"appid": "ys7",
"language": "ru_RU",
"lang": "ru",
"sessionid": "",
"content-type": "application/x-www-form-urlencoded",
"user-agent": Constants.UA,
}
@staticmethod
def get_req_headers(lr: LoginResult) -> dict:
return {
"featurecode": Constants.FC,
"appid": "ys7",
"sessionid": lr.session_id,
"areaid": str(lr.area_id),
"user-agent": Constants.UA,
}
def login(account: str, password: str, device_name: str = "M2012K11AG") -> LoginResult:
login_pl = {
"account": account,
"password": md5(password.encode("utf-8")).hexdigest(),
"featureCode": Constants.FC,
"msgType": "0",
"bizType": "",
"cuName": base64.b64encode(device_name.encode('utf-8')),
"smsCode": "",
}
res = post(Constants.LOGIN_ENDPOINT, data=login_pl, headers=Constants.get_auth_headers())
res = json.loads(res.content)
lr = LoginResult()
lr.code = res['meta']['code']
lr.message = res['meta']['message']
lr.success = lr.code == 200
if lr.success:
lr.session_id = res['loginSession']['sessionId']
lr.session_rf = res['loginSession']['rfSessionId']
lr.area_id = res['loginArea']['areaId']
lr.need_refresh = False
return lr
def get_camera_record_list(lr: LoginResult, serial: str, date: str, tfrom: str, tto: str, max_records: int = 100, channel_no: int = 1) -> RecordsResult:
dstart = f"{date}T{tfrom}"
dend = f"{date}T{tto}"
qs = {
"deviceSerial": serial,
"channelNo": channel_no,
"channelSerial": serial,
"startTime": dstart,
"stopTime": dend,
"size": max_records
}
res: Response = get(f'{Constants.RECORDS_ENDPOINT}?{urlencode(qs)}', headers=Constants.get_req_headers(lr))
res: dict = json.loads(res.content)
rr = RecordsResult()
rr.code = res['meta']['code']
rr.message = res['meta']['message']
rr.success = rr.code == 200
if rr.success:
rr.records = res['records']
return rr
def get_cameras(lr: LoginResult) -> CamerasResult:
qs = {
"groupId": -1,
"limit": 100,
"offset": 0,
"filter": "WIFI,UPGRADE,CONNECTION,STATUS"
}
res: Response = get(f'{Constants.CAMERAS_ENDPOINT}?{urlencode(qs)}', headers=Constants.get_req_headers(lr))
res: dict = json.loads(res.content)
cr = CamerasResult()
cr.code = res['meta']['code']
cr.message = res['meta']['message']
cr.success = cr.code == 200
if not cr.success: return cr
cr.page = CamerasResult.Page()
cr.page.limit = res['page']['offset']
cr.page.limit = res['page']['limit']
cr.page.has_next = res['page']['hasNext']
cr.cameras = CamerasResult.CameraList()
for i in res['deviceInfos']:
cam = CamerasResult.Camera()
cam.name = i['name']
cam.device_code = i['deviceType']
cam.model = i['deviceCategory'] + " " + i['deviceSubCategory']
cam.serial = i['deviceSerial']
cam.mac = i['mac']
cam.owner_username = i['userName']
cam.need_upgrade = res['UPGRADE'][cam.serial]['isNeedUpgrade'] > 0
cam.enc_enabled = res['STATUS'][cam.serial]['isEncrypt'] > 0
cam.enc_password = res['STATUS'][cam.serial]['encryptPwd']
cam.connection = CamerasResult.Camera.Connection()
cam.connection.ip = res['WIFI'][cam.serial]['address']
cam.connection.gw = res['WIFI'][cam.serial]['gateway']
cam.connection.mask = res['WIFI'][cam.serial]['mask']
cam.connection.signal = res['WIFI'][cam.serial]['signal']
cam.connection.ssid = res['WIFI'][cam.serial]['ssid']
cam.connection.wan_ip = res['CONNECTION'][cam.serial]['wanIp']
cam.connection.cmd_port = res['CONNECTION'][cam.serial]['localCmdPort']
cam.connection.stream_port = res['CONNECTION'][cam.serial]['localStreamPort']
cr.cameras.append(cam)
return cr
def refresh_session(lr: LoginResult) -> bool:
rf_pl = {
"refreshSessionId": lr.session_rf,
"featureCode": Constants.FC
}
res = put(Constants.LOGIN_RF_ENDPOINT, data=rf_pl, headers=Constants.get_auth_headers())
res = json.loads(res.content)
if res['meta']['code'] != 200:
lr.message = res['meta']['message']
return False
lr.session_rf = res['sessionInfo']['refreshSessionId']
lr.session_id = res['sessionInfo']['sessionId']
lr.need_refresh = False
return True
if __name__ == '__main__':
#
# lr.json:
# { "session_id": "...", "session_rf": "...", "areaid": 000}
#
test_lr: LoginResult = LoginResult.loadstr(open('lr.json', 'r').read())
if not test_lr.success:
print("need relogin (session_rf expired)")
else:
if test_lr.need_refresh:
print("need session refresh (session_id can be expired)")
if refresh_session(test_lr):
with open('lr.json', 'w') as f:
f.write(test_lr.tostr())
else:
raise Exception(f"Refresh session error: {test_lr.message}")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment