380 lines
14 KiB
Python
380 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
|
|
from __future__ import print_function
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
SOURCE_REF_PATTERN = re.compile(r"^[0-9a-f]{64}$")
|
|
VIN_PATTERN = re.compile(r"^[A-Z0-9]{6,64}$")
|
|
MAX_ROWS = 500
|
|
|
|
|
|
class ImportError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class Client:
|
|
def __init__(self, base_url, token, timeout):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.token = token
|
|
self.timeout = timeout
|
|
|
|
def data(self, method, path, payload=None):
|
|
body = None
|
|
headers = {"Authorization": "Bearer " + self.token}
|
|
if payload is not None:
|
|
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
request = urllib.request.Request(
|
|
self.base_url + path,
|
|
data=body,
|
|
headers=headers,
|
|
method=method
|
|
)
|
|
try:
|
|
response = urllib.request.urlopen(request, timeout=self.timeout)
|
|
status = response.getcode()
|
|
raw = response.read()
|
|
except urllib.error.HTTPError as error:
|
|
status = error.code
|
|
raw = error.read()
|
|
if status < 200 or status >= 300:
|
|
code = ""
|
|
message = "HTTP {0}".format(status)
|
|
try:
|
|
envelope = json.loads(raw.decode("utf-8"))
|
|
code = envelope.get("code") or envelope.get("errorCode") or ""
|
|
message = envelope.get("message") or envelope.get("error") or message
|
|
except (ValueError, UnicodeDecodeError):
|
|
pass
|
|
detail = "{0}: {1}".format(code, message) if code else message
|
|
raise ImportError("{0} {1} failed: {2}".format(method, path, detail))
|
|
try:
|
|
envelope = json.loads(raw.decode("utf-8"))
|
|
except (ValueError, UnicodeDecodeError):
|
|
raise ImportError("{0} {1} returned invalid JSON".format(method, path))
|
|
if not isinstance(envelope, dict) or "data" not in envelope:
|
|
raise ImportError("{0} {1} returned no data envelope".format(method, path))
|
|
return envelope["data"]
|
|
|
|
|
|
def clean(value):
|
|
return (value or "").strip()
|
|
|
|
|
|
def read_rows(path):
|
|
rows = []
|
|
seen = set()
|
|
try:
|
|
handle = open(path, "r", encoding="utf-8-sig", newline="")
|
|
except (IOError, OSError) as error:
|
|
raise ImportError("cannot open CSV: {0}".format(error))
|
|
with handle:
|
|
reader = csv.DictReader(handle)
|
|
required = {"vin", "source_ref", "provider_name", "evidence"}
|
|
fields = set(reader.fieldnames or [])
|
|
missing = sorted(required - fields)
|
|
if missing:
|
|
raise ImportError("CSV missing columns: " + ",".join(missing))
|
|
for line_number, raw in enumerate(reader, 2):
|
|
vin = clean(raw.get("vin")).upper()
|
|
source_ref = clean(raw.get("source_ref")).lower()
|
|
provider_name = clean(raw.get("provider_name"))
|
|
evidence = clean(raw.get("evidence"))
|
|
expected_protocol = clean(raw.get("expected_protocol")).upper()
|
|
expected_current_provider = clean(raw.get("expected_current_provider"))
|
|
if not any((vin, source_ref, provider_name, evidence, expected_protocol, expected_current_provider)):
|
|
continue
|
|
if not VIN_PATTERN.match(vin):
|
|
raise ImportError("line {0}: invalid VIN".format(line_number))
|
|
if not SOURCE_REF_PATTERN.match(source_ref):
|
|
raise ImportError("line {0}: source_ref must be a 64-character lowercase SHA-256 value".format(line_number))
|
|
if not provider_name:
|
|
raise ImportError("line {0}: provider_name is required".format(line_number))
|
|
if len(provider_name) > 128:
|
|
raise ImportError("line {0}: provider_name exceeds 128 characters".format(line_number))
|
|
if not evidence:
|
|
raise ImportError("line {0}: evidence is required".format(line_number))
|
|
if len(evidence) > 255:
|
|
raise ImportError("line {0}: evidence exceeds 255 characters".format(line_number))
|
|
key = (vin, source_ref)
|
|
if key in seen:
|
|
raise ImportError("line {0}: duplicate VIN/source_ref".format(line_number))
|
|
seen.add(key)
|
|
rows.append({
|
|
"line": line_number,
|
|
"vin": vin,
|
|
"sourceRef": source_ref,
|
|
"providerName": provider_name,
|
|
"evidence": evidence,
|
|
"expectedProtocol": expected_protocol,
|
|
"expectedCurrentProvider": expected_current_provider
|
|
})
|
|
if len(rows) > MAX_ROWS:
|
|
raise ImportError("CSV exceeds the {0}-row safety limit".format(MAX_ROWS))
|
|
if not rows:
|
|
raise ImportError("CSV contains no provider rows")
|
|
return rows
|
|
|
|
|
|
def diagnostic_path(vin):
|
|
return "/api/v2/operations/vehicles/{0}/sources".format(urllib.parse.quote(vin, safe=""))
|
|
|
|
|
|
def update_path(vin, source_ref):
|
|
return "/api/v2/operations/vehicles/{0}/sources/{1}".format(
|
|
urllib.parse.quote(vin, safe=""),
|
|
urllib.parse.quote(source_ref, safe="")
|
|
)
|
|
|
|
|
|
def sources(diagnostic):
|
|
evidence = diagnostic.get("evidence") if isinstance(diagnostic, dict) else None
|
|
result = evidence.get("locationSources") if isinstance(evidence, dict) else None
|
|
return result if isinstance(result, list) else []
|
|
|
|
|
|
def find_source(diagnostic, row):
|
|
for source in sources(diagnostic):
|
|
if clean(source.get("sourceRef")).lower() == row["sourceRef"]:
|
|
return source
|
|
raise ImportError(
|
|
"line {0}: source_ref no longer exists for VIN {1}".format(row["line"], row["vin"])
|
|
)
|
|
|
|
|
|
def policy_version(diagnostic, row):
|
|
policy = diagnostic.get("policy") if isinstance(diagnostic, dict) else None
|
|
version = policy.get("version") if isinstance(policy, dict) else None
|
|
if not isinstance(version, int) or version < 1:
|
|
raise ImportError(
|
|
"line {0}: invalid source policy version for VIN {1}".format(row["line"], row["vin"])
|
|
)
|
|
return version
|
|
|
|
|
|
def effective_provider(source):
|
|
return clean(source.get("providerOverride")) or clean(source.get("sourceLabel"))
|
|
|
|
|
|
def validate_source(row, source):
|
|
source_protocol = clean(source.get("protocol")).upper()
|
|
if row["expectedProtocol"] and source_protocol != row["expectedProtocol"]:
|
|
raise ImportError(
|
|
"line {0}: protocol changed from {1} to {2}".format(
|
|
row["line"], row["expectedProtocol"], source_protocol or "-"
|
|
)
|
|
)
|
|
current_provider = effective_provider(source)
|
|
if row["expectedCurrentProvider"] and current_provider != row["expectedCurrentProvider"]:
|
|
raise ImportError(
|
|
"line {0}: current provider changed from {1} to {2}".format(
|
|
row["line"], row["expectedCurrentProvider"], current_provider or "-"
|
|
)
|
|
)
|
|
priority = source.get("priority")
|
|
if not isinstance(priority, int) or priority < 1 or priority > 1000:
|
|
raise ImportError("line {0}: current priority is invalid".format(row["line"]))
|
|
if not isinstance(source.get("enabled"), bool):
|
|
raise ImportError("line {0}: current enabled state is invalid".format(row["line"]))
|
|
|
|
|
|
def preflight(client, rows):
|
|
diagnostics = {}
|
|
changes = 0
|
|
for row in rows:
|
|
vin = row["vin"]
|
|
if vin not in diagnostics:
|
|
diagnostics[vin] = client.data("GET", diagnostic_path(vin))
|
|
policy_version(diagnostics[vin], row)
|
|
source = find_source(diagnostics[vin], row)
|
|
validate_source(row, source)
|
|
if clean(source.get("providerOverride")) != row["providerName"]:
|
|
changes += 1
|
|
return diagnostics, changes
|
|
|
|
|
|
def export_missing(client, path):
|
|
vehicles = []
|
|
offset = 0
|
|
total = None
|
|
while total is None or offset < total:
|
|
page = client.data("POST", "/api/v2/access/vehicles", payload={
|
|
"connectionState": "master_data",
|
|
"limit": 200,
|
|
"offset": offset
|
|
})
|
|
if not isinstance(page, dict) or not isinstance(page.get("items"), list):
|
|
raise ImportError("access vehicle query returned an invalid page")
|
|
if total is None:
|
|
total = page.get("total")
|
|
if not isinstance(total, int) or total < 0:
|
|
raise ImportError("access vehicle query returned an invalid total")
|
|
vehicles.extend(page["items"])
|
|
if not page["items"]:
|
|
break
|
|
offset += len(page["items"])
|
|
rows = []
|
|
for vehicle in vehicles:
|
|
issues = vehicle.get("masterDataIssues") or []
|
|
if "JT808 接入方未维护" not in issues:
|
|
continue
|
|
vin = clean(vehicle.get("vin")).upper()
|
|
if not VIN_PATTERN.match(vin):
|
|
raise ImportError("access vehicle query returned an invalid VIN")
|
|
diagnostic = client.data("GET", diagnostic_path(vin))
|
|
for source in sources(diagnostic):
|
|
if clean(source.get("protocol")).upper() != "JT808":
|
|
continue
|
|
if clean(source.get("providerOverride")):
|
|
continue
|
|
current_provider = effective_provider(source)
|
|
source_ref = clean(source.get("sourceRef")).lower()
|
|
if not SOURCE_REF_PATTERN.match(source_ref):
|
|
raise ImportError("diagnostic returned an invalid source_ref for VIN " + vin)
|
|
rows.append({
|
|
"vin": vin,
|
|
"plate": clean(vehicle.get("plate")),
|
|
"source_ref": source_ref,
|
|
"terminal_label": clean(source.get("terminalLabel")),
|
|
"expected_protocol": "JT808",
|
|
"expected_current_provider": current_provider,
|
|
"provider_name": "",
|
|
"evidence": ""
|
|
})
|
|
if len(rows) > MAX_ROWS:
|
|
raise ImportError("missing provider export exceeds the {0}-row safety limit".format(MAX_ROWS))
|
|
if not rows:
|
|
raise ImportError("no JT808 provider gaps were found")
|
|
fieldnames = [
|
|
"vin", "plate", "source_ref", "terminal_label",
|
|
"expected_protocol", "expected_current_provider", "provider_name", "evidence"
|
|
]
|
|
try:
|
|
handle = open(path, "x", encoding="utf-8-sig", newline="")
|
|
except FileExistsError:
|
|
raise ImportError("export path already exists: " + path)
|
|
except (IOError, OSError) as error:
|
|
raise ImportError("cannot create export CSV: {0}".format(error))
|
|
with handle:
|
|
writer = csv.DictWriter(handle, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
return len(rows), len(set(row["vin"] for row in rows))
|
|
|
|
|
|
def apply_rows(client, rows, diagnostics):
|
|
updated = 0
|
|
unchanged = 0
|
|
for row in rows:
|
|
diagnostic = diagnostics[row["vin"]]
|
|
source = find_source(diagnostic, row)
|
|
validate_source(row, source)
|
|
if clean(source.get("providerOverride")) == row["providerName"]:
|
|
unchanged += 1
|
|
continue
|
|
payload = {
|
|
"version": policy_version(diagnostic, row),
|
|
"providerName": row["providerName"],
|
|
"providerEvidence": row["evidence"],
|
|
"enabled": source["enabled"],
|
|
"priority": source["priority"],
|
|
"remark": clean(source.get("policyRemark"))
|
|
}
|
|
diagnostic = client.data(
|
|
"PUT",
|
|
update_path(row["vin"], row["sourceRef"]),
|
|
payload=payload
|
|
)
|
|
diagnostics[row["vin"]] = diagnostic
|
|
saved = find_source(diagnostic, row)
|
|
if clean(saved.get("providerOverride")) != row["providerName"]:
|
|
raise ImportError(
|
|
"line {0}: server did not persist provider for VIN {1}".format(row["line"], row["vin"])
|
|
)
|
|
updated += 1
|
|
print(
|
|
"source_provider_row=updated vin={0} source_ref={1} provider={2}".format(
|
|
row["vin"], row["sourceRef"][:12], row["providerName"]
|
|
)
|
|
)
|
|
return updated, unchanged
|
|
|
|
|
|
def parser():
|
|
value = argparse.ArgumentParser(
|
|
description="Preflight and apply an authoritative source-provider CSV through the audited admin API."
|
|
)
|
|
mode = value.add_mutually_exclusive_group(required=True)
|
|
mode.add_argument("--csv", help="UTF-8 CSV with vin,source_ref,provider_name,evidence")
|
|
mode.add_argument(
|
|
"--export-missing",
|
|
metavar="PATH",
|
|
help="Write a fillable CSV for current JT808 provider gaps; refuses to overwrite an existing file."
|
|
)
|
|
value.add_argument("--base-url", default="http://127.0.0.1:20300")
|
|
value.add_argument("--token-env", default="SOURCE_PROVIDER_ADMIN_TOKEN")
|
|
value.add_argument("--timeout", type=float, default=20.0)
|
|
value.add_argument(
|
|
"--apply",
|
|
action="store_true",
|
|
help="Apply changes after a full preflight. Without this flag the command is read-only."
|
|
)
|
|
return value
|
|
|
|
|
|
def main(argv=None):
|
|
args = parser().parse_args(argv)
|
|
token = os.environ.get(args.token_env, "")
|
|
if not token:
|
|
print("source_provider_import=failed reason=token_environment_missing", file=sys.stderr)
|
|
return 2
|
|
try:
|
|
client = Client(args.base_url, token, args.timeout)
|
|
if args.export_missing:
|
|
if args.apply:
|
|
raise ImportError("--apply cannot be used with --export-missing")
|
|
row_count, vehicle_count = export_missing(client, args.export_missing)
|
|
print(
|
|
"source_provider_export=ok rows={0} vehicles={1} path={2}".format(
|
|
row_count, vehicle_count, args.export_missing
|
|
)
|
|
)
|
|
return 0
|
|
rows = read_rows(args.csv)
|
|
diagnostics, changes = preflight(client, rows)
|
|
unchanged = len(rows) - changes
|
|
if not args.apply:
|
|
print(
|
|
"source_provider_import=ready rows={0} vehicles={1} changes={2} unchanged={3} mode=dry-run".format(
|
|
len(rows), len(diagnostics), changes, unchanged
|
|
)
|
|
)
|
|
return 0
|
|
updated, unchanged = apply_rows(client, rows, diagnostics)
|
|
except ImportError as error:
|
|
print(
|
|
"source_provider_import=failed reason={0}".format(str(error).replace("\n", " ")),
|
|
file=sys.stderr
|
|
)
|
|
return 1
|
|
print(
|
|
"source_provider_import=ok rows={0} vehicles={1} updated={2} unchanged={3}".format(
|
|
len(rows), len(diagnostics), updated, unchanged
|
|
)
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|