api_requests.py

import os
import configparser
import requests
import logging
from dataclasses import dataclass
from datetime import datetime, timedelta
from functools import lru_cache
from pathlib import Path

# Logging setup
logging.basicConfig(level=logging.DEBUG, format='[%(levelname)s] %(asctime)s - %(message)s')

# Read config
CONFIG_SECTION = 'ttlock_admin_panel'
CONFIG_ENV_VAR = 'TTLOCK_CONFIG_PATH'


class TTLockConfigError(RuntimeError):
    """Raised when the TTLock configuration file is missing or malformed."""


@dataclass(frozen=True)
class TTLockConfig:
    client_id: str
    client_secret: str
    redirect_uri: str


def _resolve_config_path() -> Path:
    """Resolve the configuration path from the environment or package defaults."""
    env_path = os.environ.get(CONFIG_ENV_VAR)
    if env_path:
        candidate = Path(env_path).expanduser()
    else:
        candidate = Path(__file__).resolve().parent / 'config.ini'

    if not candidate.is_file():
        raise TTLockConfigError(
            "TTLock configuration file not found. Looked for "
            f"'{candidate}'. Set the {CONFIG_ENV_VAR} environment variable to the correct path."
        )

    return candidate


@lru_cache(maxsize=1)
def get_ttlock_config() -> TTLockConfig:
    """Load and validate the TTLock configuration file."""
    config_path = _resolve_config_path()
    parser = configparser.ConfigParser()

    logging.debug(f"Reading config from {config_path}")
    read_files = parser.read(config_path)
    if not read_files:
        raise TTLockConfigError(
            f"Failed to read TTLock configuration at '{config_path}'."
        )

    if CONFIG_SECTION not in parser:
        raise TTLockConfigError(
            f"Missing '{CONFIG_SECTION}' section in TTLock configuration file at {config_path}."
        )

    section = parser[CONFIG_SECTION]
    missing_keys = [key for key in ('client_id', 'client_secret', 'redirect_uri') if not section.get(key)]
    if missing_keys:
        raise TTLockConfigError(
            "Missing TTLock configuration keys: " + ", ".join(missing_keys)
        )

    return TTLockConfig(
        client_id=section['client_id'],
        client_secret=section['client_secret'],
        redirect_uri=section['redirect_uri'],
    )

# API constants
BASE_URL = "https://api.ttlock.com"
header = {'Content-Type': 'application/x-www-form-urlencoded'}


def get_token(email, password):
    config = get_ttlock_config()
    payload = {
        'grant_type': 'password',
        'client_id': config.client_id,
        'client_secret': config.client_secret,
        'redirect_uri': config.redirect_uri,
        'username': email,
        'password': password
    }
    url = f"{BASE_URL}/oauth2/token"
    logging.debug(f"[get_token] Sending payload: {payload}")
    response = requests.post(url, headers=header, data=payload)
    logging.debug(f"[get_token] Response: {response.status_code} {response.text}")
    return response


def refresh_tocken(refresh_token):
    config = get_ttlock_config()
    payload = {
        'grant_type': 'refresh_token',
        'client_id': config.client_id,
        'client_secret': config.client_secret,
        'redirect_uri': config.redirect_uri,
        'refresh_token': refresh_token
    }
    url = f"{BASE_URL}/oauth2/token"
    logging.debug("[refresh_token] Refreshing token...")
    response = requests.post(url, headers=header, data=payload)
    logging.debug(f"[refresh_token] Response: {response.status_code} {response.text}")
    return response


def lock_list(accessToken, pageNo):
    config = get_ttlock_config()
    payload = {
        'clientId': config.client_id,
        'accessToken': accessToken,
        'pageNo': pageNo,
        'pageSize': 50,
        'date': int(datetime.now().timestamp() * 1000)
    }
    url = f"{BASE_URL}/v3/lock/list"
    logging.debug(f"[lock_list] Payload: {payload}")
    response = requests.post(url, headers=header, data=payload)
    logging.debug(f"[lock_list] Response: {response.status_code} {response.text}")
    return response


def unlock_records(accessToken, lockId, pageNo):
    config = get_ttlock_config()
    now = datetime.now()
    payload = {
        'clientId': config.client_id,
        'accessToken': accessToken,
        'lockId': lockId,
        'startDate': int((now - timedelta(days=7)).timestamp() * 1000),
        'endDate': int(now.timestamp() * 1000),
        'pageNo': pageNo,
        'pageSize': 100,
        'date': int(now.timestamp() * 1000)
    }
    url = f"{BASE_URL}/v3/lockRecord/list"
    logging.debug(f"[unlock_records] Payload: {payload}")
    response = requests.post(url, headers=header, data=payload)
    logging.debug(f"[unlock_records] Response: {response.status_code} {response.text}")
    return response


def unlock_records_one_day(accessToken, lockId, pageNo):
    config = get_ttlock_config()
    now = datetime.now()
    payload = {
        'clientId': config.client_id,
        'accessToken': accessToken,
        'lockId': lockId,
        'startDate': int((now - timedelta(days=1)).timestamp() * 1000),
        'endDate': int(now.timestamp() * 1000),
        'pageNo': pageNo,
        'pageSize': 100,
        'date': int(now.timestamp() * 1000)
    }
    url = f"{BASE_URL}/v3/lockRecord/list"
    logging.debug(f"[unlock_records_one_day] Payload: {payload}")
    response = requests.post(url, headers=header, data=payload)
    logging.debug(f"[unlock_records_one_day] Response: {response.status_code} {response.text}")
    return response


def list_passwords(accessToken, lockId, pageNo):
    config = get_ttlock_config()
    payload = {
        'clientId': config.client_id,
        'accessToken': accessToken,
        'lockId': lockId,
        'pageNo': pageNo,
        'pageSize': 50,
        'date': int(datetime.now().timestamp() * 1000)
    }
    url = f"{BASE_URL}/v3/lock/listKeyboardPwd"
    logging.debug(f"[list_passwords] Payload: {payload}")
    response = requests.post(url, headers=header, data=payload)
    logging.debug(f"[list_passwords] Response: {response.status_code} {response.text}")
    return response


def get_all_unlock_records(accessToken):
    logging.debug("[get_all_unlock_records] Fetching all locks...")
    response = lock_list(accessToken, 1)
    try:
        locks = response.json().get("list", [])
        lock_ids = [lock['lockId'] for lock in locks]
    except Exception as e:
        logging.error(f"[get_all_unlock_records] Error parsing lock list: {e}")
        return []

    all_unlocks = []
    for lock_id in lock_ids:
        record_response = unlock_records_one_day(accessToken, lock_id, 1)
        try:
            records = record_response.json().get("list", [])
            all_unlocks.extend(records)
        except Exception as e:
            logging.warning(f"[get_all_unlock_records] Skipped lockId {lock_id} due to error: {e}")
    return all_unlocks


def create_password(accessToken, lockId, keyboardPwd, keyboardPwdName, startDate, endDate):
    payload = {
        'clientId': client_id,
        'accessToken': accessToken,
        'lockId': lockId,
        'keyboardPwd': keyboardPwd,
        'keyboardPwdName': keyboardPwdName,
        'startDate': startDate,
        'endDate': endDate,
        'addType': 2,
        'date': int(datetime.now().timestamp() * 1000)
    }
    url = f"{BASE_URL}/v3/keyboardPwd/add"
    logging.debug(f"[create_password] Payload: {payload}")
    response = requests.post(url, headers=header, data=payload)
    logging.debug(f"[create_password] Response: {response.status_code} {response.text}")
    return response