#!/usr/bin/env python3 """ import.py — Kennzeichen-Daten aus Git-Repo + data.db nach PocketBase importieren Voraussetzungen: pip install requests Verwendung: # Erst PocketBase Admin-Account anlegen unter /admin python3 import.py \ --pb-url http://localhost:8090 \ --pb-email admin@example.com \ --pb-password deinpasswort \ --repo-path /pfad/zum/plates-repo \ --db-path /pfad/zur/data.db """ import argparse import csv import os import sqlite3 import sys import requests # ── Länder-Mapping: Ordnername im Repo → Tabellenname in data.db ────────────── COUNTRY_MAP = { "germany": "deutschland", "austria": "austria", "switzerland": "schweiz", "luxembourg": "luxemburg", "poland": "polen", "norway": "norwegen", "uk": "uk", "italy": "italien", "france": "frankreich", "greece": "greece", "slovakia": "slowakei", "croatia": "kroatien", "ukraine": "ukraine", "serbia": "serbien", "russia": "russland", "belarus": "belarus", "montenegro": "montenegro", "northmacedonia": "nmk", "ireland": "irland", "bulgaria": "bulgarien", "romania": "romania", "moldova": "moldawien", "czechia": "tschechien", "turkey": "turkey", "kosovo": "kosovo", "slovenia": "slowenien", } # ── PocketBase Collection-Schemas ───────────────────────────────────────────── COLLECTIONS = { "kennzeichen": { "name": "kennzeichen", "type": "base", "schema": [ {"name": "code", "type": "text", "required": True}, {"name": "name", "type": "text", "required": True}, {"name": "derivation", "type": "text", "required": False}, {"name": "region", "type": "text", "required": False}, {"name": "note", "type": "text", "required": False}, {"name": "active", "type": "bool", "required": True}, {"name": "country", "type": "text", "required": True}, # Aus data.db angereichert: {"name": "lat", "type": "number", "required": False}, {"name": "lon", "type": "number", "required": False}, {"name": "population", "type": "number", "required": False}, {"name": "points", "type": "number", "required": False}, {"name": "alt_code", "type": "text", "required": False}, ], }, "diplomatenkennzeichen": { "name": "diplomatenkennzeichen", "type": "base", "schema": [ {"name": "code", "type": "text", "required": True}, {"name": "country", "type": "text", "required": True}, {"name": "organisation", "type": "text", "required": False}, {"name": "type", "type": "text", "required": True}, {"name": "city", "type": "text", "required": False}, {"name": "note", "type": "text", "required": False}, ], }, } class PocketBaseClient: def __init__(self, base_url: str): self.base_url = base_url.rstrip("/") self.token = None def authenticate(self, email: str, password: str): r = requests.post( f"{self.base_url}/api/admins/auth-with-password", json={"identity": email, "password": password}, ) r.raise_for_status() self.token = r.json()["token"] print(f"✓ Eingeloggt als {email}") def _headers(self): return {"Authorization": self.token, "Content-Type": "application/json"} def collection_exists(self, name: str) -> bool: r = requests.get( f"{self.base_url}/api/collections/{name}", headers=self._headers(), ) return r.status_code == 200 def create_collection(self, schema: dict): name = schema["name"] if self.collection_exists(name): print(f" Collection '{name}' existiert bereits, überspringe.") return r = requests.post( f"{self.base_url}/api/collections", headers=self._headers(), json=schema, ) r.raise_for_status() print(f" ✓ Collection '{name}' erstellt") def upsert_record(self, collection: str, data: dict): """Insert oder Update anhand des 'code'-Feldes.""" # Prüfe ob Eintrag schon existiert r = requests.get( f"{self.base_url}/api/collections/{collection}/records", headers=self._headers(), params={"filter": f"code='{data['code']}' && country='{data.get('country','')}'", "perPage": 1}, ) r.raise_for_status() items = r.json().get("items", []) if items: record_id = items[0]["id"] r2 = requests.patch( f"{self.base_url}/api/collections/{collection}/records/{record_id}", headers=self._headers(), json=data, ) r2.raise_for_status() else: r2 = requests.post( f"{self.base_url}/api/collections/{collection}/records", headers=self._headers(), json=data, ) r2.raise_for_status() def load_db_geodata(db_path: str) -> dict: """ Lädt GPS-Koordinaten, Einwohnerzahl und Punkte aus data.db. Gibt dict zurück: { (land, code.upper()): {lat, lon, population, points, alt_code} } """ conn = sqlite3.connect(db_path) cur = conn.cursor() geo = {} cur.execute("SELECT name FROM sqlite_master WHERE type='table'") tables = [r[0] for r in cur.fetchall() if r[0] != "laender"] for table in tables: try: cur.execute(f"SELECT kennzeichen, map1, map2, anzahl, punkte, alt FROM {table}") for row in cur.fetchall(): kz, map1, map2, anzahl, punkte, alt = row if kz: geo[(table, kz.upper())] = { "lat": map1, "lon": map2, "population": anzahl, "points": punkte, "alt_code": alt, } except sqlite3.OperationalError: pass conn.close() print(f"✓ {len(geo)} Geo-Einträge aus data.db geladen") return geo def import_kennzeichen(pb: PocketBaseClient, repo_path: str, geo: dict): total = 0 for folder, db_table in COUNTRY_MAP.items(): csv_path = os.path.join(repo_path, folder, "kennzeichen.csv") if not os.path.exists(csv_path): continue count = 0 with open(csv_path, newline="", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: code = row.get("code", "").strip() if not code: continue record = { "code": code, "name": row.get("name", "").strip(), "derivation": row.get("derivation", "").strip(), "region": row.get("region", "").strip(), "note": row.get("note", "").strip(), "active": row.get("active", "true").strip().lower() == "true", "country": folder, } # Geo-Daten aus data.db anreichern geo_key = (db_table, code.upper()) if geo_key in geo: record.update(geo[geo_key]) pb.upsert_record("kennzeichen", record) count += 1 print(f" ✓ {folder}: {count} Kennzeichen importiert") total += count print(f"\n✓ Gesamt: {total} Kennzeichen in PocketBase") def import_diplomatenkennzeichen(pb: PocketBaseClient, repo_path: str): total = 0 for folder in os.listdir(repo_path): csv_path = os.path.join(repo_path, folder, "diplomatenkennzeichen.csv") if not os.path.exists(csv_path): continue count = 0 with open(csv_path, newline="", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: code = row.get("code", "").strip() if not code: continue record = { "code": code, "country": row.get("country", "").strip(), "organisation": row.get("organisation", "").strip(), "type": row.get("type", "embassy").strip(), "city": row.get("city", "").strip(), "note": row.get("note", "").strip(), "base_country": folder, } pb.upsert_record("diplomatenkennzeichen", record) count += 1 print(f" ✓ {folder}: {count} Diplomatenkennzeichen importiert") total += count print(f"\n✓ Gesamt: {total} Diplomatenkennzeichen in PocketBase") def main(): parser = argparse.ArgumentParser(description="Kennzeichen → PocketBase Import") parser.add_argument("--pb-url", default="http://localhost:8090") parser.add_argument("--pb-email", required=True) parser.add_argument("--pb-password", required=True) parser.add_argument("--repo-path", required=True, help="Pfad zum plates-Repo") parser.add_argument("--db-path", required=True, help="Pfad zur data.db") parser.add_argument("--skip-schema", action="store_true", help="Collections nicht neu erstellen") args = parser.parse_args() pb = PocketBaseClient(args.pb_url) pb.authenticate(args.pb_email, args.pb_password) if not args.skip_schema: print("\n→ Collections anlegen...") for schema in COLLECTIONS.values(): pb.create_collection(schema) print("\n→ Geo-Daten laden...") geo = load_db_geodata(args.db_path) print("\n→ Kennzeichen importieren...") import_kennzeichen(pb, args.repo_path, geo) print("\n→ Diplomatenkennzeichen importieren...") import_diplomatenkennzeichen(pb, args.repo_path) print("\n✓ Import abgeschlossen!") if __name__ == "__main__": main()