warlock/senaro_dealer_app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b2a846f7a0e45829b68ceb068e4254da8a5e47a3
senaro_dealer_app/scripts/import_service_points.py
T
warlock b2a846f7a0 sync
2026-10-10 08:25:32 +05:30

407 lines
13 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Import ServicePoints.csv through the same API the app uses to add a point.
Login (staff, same as the app):
POST {base}/auth/staff-login
{"phone": "...", "password": "..."}
Create (same body as ServicePointFormScreen):
POST {base}/service-points
{"name", "district", "address", "phone", "notes"}
Authorization: Bearer <token>
The CSV has no header. Columns are town, name, address, phone.
`district` must be one of the names in lib/labels.dart (sriLankaDistricts).
The spreadsheet town is stored in `notes` so the finer location is kept.
Examples:
python3 scripts/import_service_points.py --dry-run
python3 scripts/import_service_points.py --phone 07XXXXXXXX --password '***'
SENARO_TOKEN=eyJ... python3 scripts/import_service_points.py
"""
from __future__ import annotations
import argparse
import csv
import getpass
import json
import os
import sys
import urllib.error
import urllib.request
from collections import Counter
from pathlib import Path
DEFAULT_BASE_URL = "https://snrodealer.playpoolstudios.com/api"
ROOT = Path(__file__).resolve().parent.parent
# Official district names, matching lib/labels.dart.
DISTRICTS = (
"Ampara",
"Anuradhapura",
"Badulla",
"Batticaloa",
"Colombo",
"Galle",
"Gampaha",
"Hambantota",
"Jaffna",
"Kalutara",
"Kandy",
"Kegalle",
"Kilinochchi",
"Kurunegala",
"Mannar",
"Matale",
"Matara",
"Moneragala",
"Mullaitivu",
"Nuwara Eliya",
"Polonnaruwa",
"Puttalam",
"Ratnapura",
"Trincomalee",
"Vavuniya",
)
# Spreadsheet town (normalized) -> official district.
TOWN_TO_DISTRICT = {
# Ampara
"akkarapattu": "Ampara",
"ampara": "Ampara",
"dehiattakandiya": "Ampara",
"hingurana": "Ampara",
"kalmunai": "Ampara",
"samanthurai": "Ampara",
# Anuradhapura
"anuradhapura": "Anuradhapura",
"eppawala": "Anuradhapura",
"galenbidunuwewa": "Anuradhapura",
"horowpothana": "Anuradhapura",
"kahatagasdigiliya": "Anuradhapura",
"kebitigollewa": "Anuradhapura",
"kekirawa": "Anuradhapura",
"madawachchiya": "Anuradhapura",
"parakramapura": "Anuradhapura",
"thambuththegama": "Anuradhapura",
# Badulla
"badulla": "Badulla",
"bandarawela": "Badulla",
"mahiyanganaya": "Badulla",
"welimada": "Badulla",
# Batticaloa
"batticoloa": "Batticaloa",
"kokkadicholai": "Batticaloa",
# Colombo
"avissawella": "Colombo",
"hanwella": "Colombo",
"kotikawatta": "Colombo",
"kottawa": "Colombo",
"malabe": "Colombo",
"moratuwa": "Colombo",
"panchikawatte": "Colombo",
"piliyandala": "Colombo",
"polgasowita": "Colombo",
# Galle
"ambalangoda": "Galle",
"baddegama": "Galle",
"batapola": "Galle",
"elpitiya": "Galle",
"galle": "Galle",
"pitigala": "Galle",
"udugama": "Galle",
# Gampaha
"biyagama": "Gampaha",
"divulapitiya": "Gampaha",
"gampaha": "Gampaha",
"ganemulla": "Gampaha",
"kadawatha": "Gampaha",
"kandana": "Gampaha",
"kiridiwela": "Gampaha",
"minuwangoda": "Gampaha",
"negombo": "Gampaha",
"wattala": "Gampaha",
"weliweriya": "Gampaha",
"yakkala": "Gampaha",
# Hambantota
"agunukolapalassa": "Hambantota",
"ambalanthota": "Hambantota",
"middeniya": "Hambantota",
"thissamaharama": "Hambantota",
"walasmulla": "Hambantota",
# Jaffna
"chavakachchri": "Jaffna",
"jaffna": "Jaffna",
# Kalutara
"aluthgama": "Kalutara",
"bandaragama": "Kalutara",
"horana": "Kalutara",
"ingiriya": "Kalutara",
"kaluthara": "Kalutara",
"maggona": "Kalutara",
"mathugama": "Kalutara",
"panadura": "Kalutara",
# Kandy
"digana": "Kandy",
"gampola": "Kandy",
"kandy": "Kandy",
"nawalapitiya": "Kandy",
"teldeniya": "Kandy",
# Kegalle
"dehiowita": "Kegalle",
"kegalle": "Kegalle",
"kithulgala": "Kegalle",
"warakapola": "Kegalle",
# Kilinochchi
"kilinochchi": "Kilinochchi",
# Kurunegala
"bingiriya": "Kurunegala",
"galgamuwa": "Kurunegala",
"kuliyapitiya": "Kurunegala",
"kurunegala": "Kurunegala",
"mahawa": "Kurunegala",
"mawathagama": "Kurunegala",
"narammala": "Kurunegala",
"nikawaratiya": "Kurunegala",
"polgahawela": "Kurunegala",
"polpitigama": "Kurunegala",
"wariyapola": "Kurunegala",
# Mannar
"mannar": "Mannar",
# Matale
"dambulla": "Matale",
"matale": "Matale",
# Matara
"akuressa": "Matara",
"kamburupitiya": "Matara",
"matara": "Matara",
"pitabeddara": "Matara",
"urubokka": "Matara",
# Moneragala
"butthala": "Moneragala",
"monaragala": "Moneragala",
"wellawaya": "Moneragala",
# Mullaitivu
"puthukkudiyiruppu": "Mullaitivu",
# Nuwara Eliya
"hatton": "Nuwara Eliya",
"nuwara eliya": "Nuwara Eliya",
"thalawakele": "Nuwara Eliya",
# Polonnaruwa
"aralaganwila": "Polonnaruwa",
"manampitiya": "Polonnaruwa",
"medirigiriya": "Polonnaruwa",
"polonnaruwa": "Polonnaruwa",
"sevanapitiya": "Polonnaruwa",
# Puttalam
"anamaduwa": "Puttalam",
"galmuruwa": "Puttalam",
"halawatha (chillaw)": "Puttalam",
"nathandiya": "Puttalam",
"norochchole": "Puttalam",
# Ratnapura
"balangoda": "Ratnapura",
"eheliyagoda": "Ratnapura",
"embilipitiya": "Ratnapura",
"kalawana": "Ratnapura",
"opanayaka": "Ratnapura",
"pelmadulla": "Ratnapura",
"rathnapura": "Ratnapura",
# Trincomalee
"kantale": "Trincomalee",
"kinniya": "Trincomalee",
"trincomalee": "Trincomalee",
# Vavuniya
"vauniya": "Vavuniya",
}
class ApiError(Exception):
def __init__(self, status: int, message: str):
super().__init__(f"HTTP {status}: {message}")
self.status = status
self.message = message
def normalize(value: str) -> str:
return " ".join(value.strip().casefold().split())
def request_json(method: str, url: str, body: dict | None = None, token: str | None = None) -> dict:
data = None if body is None else json.dumps(body).encode()
headers = {"Accept": "application/json", "Content-Type": "application/json"}
if token:
headers["Authorization"] = f"Bearer {token}"
request = urllib.request.Request(url, data=data, headers=headers, method=method)
try:
with urllib.request.urlopen(request, timeout=30) as response:
raw = response.read().decode()
return json.loads(raw) if raw else {}
except urllib.error.HTTPError as error:
raw = error.read().decode()
message = raw or str(error.reason)
try:
payload = json.loads(raw)
except json.JSONDecodeError:
payload = None
if isinstance(payload, dict) and isinstance(payload.get("message"), str):
message = payload["message"]
raise ApiError(error.code, message) from error
def load_rows(path: Path) -> list[dict]:
rows: list[dict] = []
unmapped: list[tuple[int, str]] = []
with path.open(newline="", encoding="utf-8-sig") as handle:
for line_no, record in enumerate(csv.reader(handle), start=1):
if not record or all(not cell.strip() for cell in record):
continue
if len(record) != 4:
raise SystemExit(f"{path}:{line_no}: expected 4 columns, got {len(record)}")
town, name, address, phone = (cell.strip() for cell in record)
district = TOWN_TO_DISTRICT.get(normalize(town))
if district not in DISTRICTS:
unmapped.append((line_no, town))
continue
if not name:
raise SystemExit(f"{path}:{line_no}: missing service point name")
rows.append(
{
"line": line_no,
"town": town,
"body": {
"name": name,
"district": district,
"address": address,
"phone": phone,
"notes": town,
},
}
)
if unmapped:
print("Unmapped towns (add them to TOWN_TO_DISTRICT before importing):", file=sys.stderr)
for line_no, town in unmapped:
print(f" line {line_no}: {town}", file=sys.stderr)
raise SystemExit(2)
return rows
def point_key(name: str, district: str, phone: str) -> tuple[str, str, str]:
return (normalize(name), district, normalize(phone))
def existing_keys(base_url: str, token: str) -> set[tuple[str, str, str]]:
payload = request_json("GET", f"{base_url}/service-points", token=token)
items = payload.get("data", payload)
if not isinstance(items, list):
raise SystemExit("Unexpected /service-points response; expected a data list")
keys: set[tuple[str, str, str]] = set()
for item in items:
if not isinstance(item, dict):
continue
keys.add(
point_key(
str(item.get("name") or ""),
str(item.get("district") or ""),
str(item.get("phone") or ""),
)
)
return keys
def login(base_url: str, phone: str, password: str) -> str:
payload = request_json(
"POST",
f"{base_url}/auth/staff-login",
{"phone": phone, "password": password},
)
token = payload.get("token")
if not isinstance(token, str) or not token:
raise SystemExit("Staff login succeeded but no token was returned")
user = payload.get("user") if isinstance(payload.get("user"), dict) else {}
name = user.get("name") or user.get("phone") or phone
print(f"Signed in as {name}")
return token
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("--csv", type=Path, default=ROOT / "ServicePoints.csv", help="CSV path")
parser.add_argument("--base-url", default=os.environ.get("SENARO_API_BASE_URL", DEFAULT_BASE_URL))
parser.add_argument("--phone", default=os.environ.get("SENARO_PHONE"), help="Staff phone")
parser.add_argument("--password", default=os.environ.get("SENARO_PASSWORD"), help="Staff password")
parser.add_argument("--token", default=os.environ.get("SENARO_TOKEN"), help="Existing bearer token")
parser.add_argument("--dry-run", action="store_true", help="Print payloads and do not call the API")
parser.add_argument(
"--no-skip-existing",
action="store_true",
help="Create a point even when the same name, district, and phone already exist",
)
return parser.parse_args()
def main() -> None:
args = parse_args()
base_url = args.base_url.rstrip("/")
rows = load_rows(args.csv)
counts = Counter(row["body"]["district"] for row in rows)
print(f"{len(rows)} service points in {args.csv}")
for district in DISTRICTS:
if counts[district]:
print(f" {district}: {counts[district]}")
if args.dry_run:
print()
for row in rows:
body = row["body"]
print(f"{body['district']}\t{row['town']}\t{body['name']}\t{body['phone']}")
print("\nDry run only. Nothing was sent.")
return
token = args.token
if not token:
phone = args.phone or input("Staff phone: ").strip()
password = args.password or getpass.getpass("Staff password: ")
if not phone or not password:
raise SystemExit("Staff phone and password are required")
token = login(base_url, phone, password)
known: set[tuple[str, str, str]] = set()
if not args.no_skip_existing:
known = existing_keys(base_url, token)
print(f"{len(known)} service points already on the server")
created = 0
skipped = 0
failed: list[str] = []
for row in rows:
body = row["body"]
key = point_key(body["name"], body["district"], body["phone"])
label = f"line {row['line']}: {body['name']} ({body['district']})"
if key in known:
skipped += 1
print(f"skip {label}")
continue
try:
request_json("POST", f"{base_url}/service-points", body, token=token)
except ApiError as error:
failed.append(f"{label}: {error.message}")
print(f"fail {label}: {error.message}", file=sys.stderr)
continue
except urllib.error.URLError as error:
failed.append(f"{label}: {error.reason}")
print(f"fail {label}: {error.reason}", file=sys.stderr)
continue
known.add(key)
created += 1
print(f"added {label}")
print(f"\nAdded {created}, skipped {skipped}, failed {len(failed)}")
if failed:
raise SystemExit(1)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink