#!/usr/bin/env python3 """ Tracksolid Pro - Bulk Vehicle Information Updater Updates vehicle details via the jimi.open.device.update API endpoint using data from the Fireside logistics CSV. Signing approach taken directly from tspostman.py (confirmed working): - POST as x-www-form-urlencoded (NOT JSON) - All parameter values cast to strings before signing - expires_in passed as string '7200', not integer Usage: python tracksolid_update.py [--dry-run] [--csv path/to/file.csv] [--limit N] Environment variables (or edit CONFIG below): TS_USER_ID - Your Tracksolid account username TS_USER_PWD_MD5 - MD5 hash of your password (lowercase) TS_APP_KEY - Your appKey from JIMI TS_APP_SECRET - Your appSecret from JIMI TS_API_URL - API base URL (defaults to EU node) TS_CSV_PATH - Path to the logistics CSV """ import hashlib import time import json import logging import argparse import os import sys from datetime import datetime, timezone from pathlib import Path import requests import pandas as pd # ────────────────────────────────────────────────────────────────────────────── # CONFIGURATION — edit here or set environment variables # ────────────────────────────────────────────────────────────────────────────── CONFIG = { "user_id": os.getenv("TS_USER_ID", "Fireside Communications"), "user_pwd_md5": os.getenv("TS_USER_PWD_MD5", "81a1b005efd3596073e38efd8a2fd3fd"), "app_key": os.getenv("TS_APP_KEY", "8FB345B8693CCD00BB70D528C0D4019E"), "app_secret": os.getenv("TS_APP_SECRET", "3177c89993b446c6aced0d7c56375d2c"), # EU node confirmed for this account "api_url": os.getenv("TS_API_URL", "https://eu-open.tracksolidpro.com/route/rest"), "expires_in": "7200", # string, not int — matches tspostman.py "request_delay": 0.5, # seconds between API calls } CSV_PATH = os.getenv("TS_CSV_PATH", "20260414_FS__Logistics_-_final_fixed.csv") # ────────────────────────────────────────────────────────────────────────────── # LOGGING # ────────────────────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)-8s %(message)s", handlers=[ logging.StreamHandler(sys.stdout), logging.FileHandler("tracksolid_update.log", encoding="utf-8"), ], ) log = logging.getLogger(__name__) # ────────────────────────────────────────────────────────────────────────────── # SIGNING UTILITIES (ported directly from tspostman.py) # ────────────────────────────────────────────────────────────────────────────── def utc_timestamp() -> str: """UTC time formatted as yyyy-MM-dd HH:mm:ss.""" return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") def build_sign(params: dict, app_secret: str) -> str: """ Tracksolid signing algorithm (matches tspostman.py exactly): 1. Sort all param keys alphabetically, exclude 'sign' and empty values. 2. Concatenate key+value pairs with no separators. 3. Wrap with appSecret on both sides. 4. MD5 -> UPPERCASE 32-char string. All values must already be strings before calling this. """ sorted_keys = sorted( k for k in params if k != "sign" and params[k] is not None and str(params[k]).strip() != "" ) param_string = "".join(f"{k}{params[k]}" for k in sorted_keys) raw_string = f"{app_secret}{param_string}{app_secret}" return hashlib.md5(raw_string.encode("utf-8")).hexdigest().upper() # ────────────────────────────────────────────────────────────────────────────── # TRACKSOLID CLIENT # ────────────────────────────────────────────────────────────────────────────── class TracksolidClient: def __init__(self, cfg: dict): self.cfg = cfg self._token: str | None = None self._token_expires_at: float = 0.0 self.session = requests.Session() # No Content-Type set here — requests sets it automatically to # application/x-www-form-urlencoded when data= is used (matching tspostman.py) def _post(self, params: dict) -> dict: """ Sign and POST params using x-www-form-urlencoded encoding. Confirmed working approach from tspostman.py: - Cast all values to strings - Use data= (form-encoded), NOT json= """ str_params = { k: str(v) for k, v in params.items() if v is not None and str(v).strip() != "" } str_params["sign"] = build_sign(str_params, self.cfg["app_secret"]) log.debug("POST %s params=%s", self.cfg["api_url"], str_params) resp = self.session.post( self.cfg["api_url"], data=str_params, # form-encoded, NOT json= timeout=30, ) resp.raise_for_status() data = resp.json() log.debug("Response: %s", json.dumps(data)) return data def _common_params(self, method: str) -> dict: return { "method": method, "timestamp": utc_timestamp(), "app_key": self.cfg["app_key"], "sign_method": "md5", "v": "1.0", "format": "json", } def get_token(self) -> str: """Return a valid access token, fetching a new one only when needed.""" if self._token and time.time() < self._token_expires_at - 60: log.debug("Reusing cached token.") return self._token log.info("Obtaining new access token ...") params = self._common_params("jimi.oauth.token.get") params.update({ "user_id": self.cfg["user_id"], "user_pwd_md5": self.cfg["user_pwd_md5"], "expires_in": self.cfg["expires_in"], # already a string per tspostman.py }) data = self._post(params) if data.get("code") != 0: raise RuntimeError( f"Auth failed — code={data.get('code')} message={data.get('message')}" ) self._token = data["result"]["accessToken"] self._token_expires_at = time.time() + int(data["result"]["expiresIn"]) log.info("Token acquired. Valid for %s seconds.", data["result"]["expiresIn"]) return self._token def update_vehicle(self, imei: str, fields: dict, dry_run: bool = False) -> dict: """ Call jimi.open.device.update for one IMEI. fields: dict using local key names (see row_to_fields below). Returns the API response dict. """ if dry_run: log.info("[DRY-RUN] IMEI %s -> %s", imei, fields) return {"code": 0, "message": "dry-run"} token = self.get_token() params = self._common_params("jimi.open.device.update") params["access_token"] = token params["imei"] = str(imei) # Map local field names -> Tracksolid API field names api_field_map = { "license_plate": "vehicle_number", "vehicle_name": "vehicle_name", "vehicle_icon": "vehicle_icon", "driver_name": "driver_name", "driver_phone": "driver_phone", "vehicle_model": "vehicle_models", "vin": "carFrame", "engine_number": "engineNumber", "device_name": "device_name", "fuel_per_100": "oilWear", "sim": "sim", "remarks": "remarks", } for local_key, api_key in api_field_map.items(): val = fields.get(local_key) if val: params[api_key] = str(val) return self._post(params) # ────────────────────────────────────────────────────────────────────────────── # CSV PARSING # ────────────────────────────────────────────────────────────────────────────── def clean(val) -> str | None: """Return None for NaN/empty values, else a stripped string.""" if pd.isna(val) or str(val).strip() in ("", "nan", "NaN"): return None s = str(val).strip() if s.endswith(".0") and s[:-2].lstrip("+-").isdigit(): s = s[:-2] return s def row_to_fields(row: pd.Series) -> dict: """Map one CSV row to local field names used by update_vehicle().""" return { "device_name": clean(row.get("Device Name")), "vehicle_name": clean(row.get("Vehicle Name")), "vehicle_icon": clean(row.get("Vehicle Icon")), "license_plate": clean(row.get("License Plate No.")), "vehicle_model": clean(row.get("Vehicle Model")), "driver_name": clean(row.get("Driver Name")), "driver_phone": clean(row.get("Telephone")), "sim": clean(row.get("SIM")), "vin": clean(row.get("VIN")), "engine_number": clean(row.get("Engine Number")), "fuel_per_100": clean(row.get("Fuel/100km")), "remarks": clean(row.get("Remarks")), } # ────────────────────────────────────────────────────────────────────────────── # MAIN # ────────────────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="Bulk-update Tracksolid vehicle info from CSV." ) parser.add_argument( "--dry-run", action="store_true", help="Show what would be sent without making update API calls." ) parser.add_argument( "--csv", default=CSV_PATH, help=f"Path to the logistics CSV (default: {CSV_PATH})" ) parser.add_argument( "--limit", type=int, default=None, help="Only process the first N rows (useful for testing)." ) args = parser.parse_args() # ── Load CSV ─────────────────────────────────────────────────────────────── csv_file = Path(args.csv) if not csv_file.exists(): log.error("CSV file not found: %s", csv_file) sys.exit(1) df = pd.read_csv(csv_file, dtype={"IMEI": str}) log.info("Loaded %d rows from %s", len(df), csv_file) if args.limit: df = df.head(args.limit) log.info("Limiting to first %d rows.", args.limit) df = df[df["IMEI"].notna() & (df["IMEI"].str.strip() != "")] log.info("%d rows have a valid IMEI.", len(df)) if df.empty: log.warning("No rows to process. Exiting.") sys.exit(0) # ── Initialise client & verify auth ─────────────────────────────────────── client = TracksolidClient(CONFIG) if not args.dry_run: try: client.get_token() except Exception as exc: log.error("Authentication failed: %s", exc) log.error("Check TS_USER_ID, TS_USER_PWD_MD5, TS_APP_KEY, TS_APP_SECRET.") sys.exit(1) # ── Process rows ─────────────────────────────────────────────────────────── results = [] success = 0 failed = 0 skipped = 0 for idx, row in df.iterrows(): imei = str(row["IMEI"]).strip() fields = row_to_fields(row) # Skip rows where every updatable field is empty if not any(fields.values()): log.warning("Row %d (IMEI %s): no updatable fields, skipping.", idx + 1, imei) skipped += 1 continue log.info( "Row %d/%d — IMEI: %s | Plate: %s | Driver: %s", idx + 1, len(df), imei, fields.get("license_plate") or "—", fields.get("driver_name") or "—", ) try: resp = client.update_vehicle(imei, fields, dry_run=args.dry_run) if resp.get("code") == 0: log.info(" OK") success += 1 else: log.warning( " API error — code=%s message=%s", resp.get("code"), resp.get("message") ) failed += 1 results.append({ "imei": imei, **fields, "api_code": resp.get("code"), "api_message": resp.get("message"), }) except Exception as exc: log.error(" Exception for IMEI %s: %s", imei, exc) failed += 1 results.append({ "imei": imei, **fields, "api_code": "EXCEPTION", "api_message": str(exc), }) if not args.dry_run: time.sleep(CONFIG["request_delay"]) # ── Summary & audit CSV ─────────────────────────────────────────────────── log.info("-" * 60) log.info( "Done. OK: %d Failed: %d Skipped: %d | Total: %d", success, failed, skipped, success + failed + skipped ) out_path = "tracksolid_update_results.csv" pd.DataFrame(results).to_csv(out_path, index=False) log.info("Audit results written to %s", out_path) if __name__ == "__main__": main()