5418fc48c5
Switches column references from the old title-case logistics CSV (IMEI, Device Name, License Plate No., Telephone, Fuel/100km, ...) to the snake_case Mitieng export shipped on 2026-04-27 (imei, device_name, vehicle_number, driver_phone, fuel_100km, ...). Without this, the bulk device-update API tool fails with KeyError: 'IMEI' on the new CSV. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
359 lines
15 KiB
Python
359 lines
15 KiB
Python
#!/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("vehicle_number")),
|
|
"vehicle_model": clean(row.get("vehicle_models")),
|
|
"driver_name": clean(row.get("driver_name")),
|
|
"driver_phone": clean(row.get("driver_phone")),
|
|
"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()
|