// Importiert echte Betriebe als nicht bestätigte Einträge. Zwei Quellen, zwei Schichten (siehe schema.ts):
//   --osm  Datei aus scripts/osm/extrahieren.py (OpenStreetMap, ODbL)  -> Tabelle osm_betriebe
//   --web  betriebe-websites.json (Angaben von der Website des Betriebs, je Feld Beleg und Abrufdatum) -> web_betriebe
// Je Betrieb entsteht ein Eintrag in `betriebe`, der auf die Schichten verweist. Die Schichten werden nicht vermischt.
// Wiederholbar: Ein zweiter Lauf aktualisiert über osm_typ/osm_id und die Kennung der Website-Datei statt zu verdoppeln.
// Entfernte Einträge (status = entfernt) kommen nicht zurück. Die Daten selbst liegen nicht im Repository.
//
//   node scripts/betriebe-importieren.ts --osm daten/osm-betriebe.json --web daten/betriebe-websites.json [--trockenlauf]
//
// --trockenlauf  macht alles in einer Transaktion und rollt am Ende zurück (gleiche Zahlen wie der echte Lauf)
// --aufraeumen-erzwingen  erlaubt, dass eine Datei mehr als die Hälfte der bisherigen Zeilen ihrer Schicht entfallen lässt
// DATABASE_URL aus der Umgebung, sonst die lokale Entwicklungsdatenbank.
import { createHash, randomUUID } from "node:crypto";
import { readFileSync } from "node:fs";
import postgres from "postgres";
import {
  ARTEN,
  anzeigeBauen,
  ortAbleiten,
  webDateiPruefen,
  type OsmSchicht,
  type WebFeldWert,
  type WebSchicht,
} from "../src/lib/betriebeRegeln.ts";
import { stallSlug } from "../src/lib/text.ts";

const argv = process.argv.slice(2);
const wert = (name: string) => {
  const i = argv.indexOf(name);
  return i >= 0 ? argv[i + 1] : undefined;
};
const osmDatei = wert("--osm");
const webDatei = wert("--web");
const trocken = argv.includes("--trockenlauf");
const erzwingen = argv.includes("--aufraeumen-erzwingen");
if (!osmDatei && !webDatei) {
  console.error("Aufruf: node scripts/betriebe-importieren.ts [--osm datei.json] [--web datei.json] [--trockenlauf]");
  process.exit(2);
}

const url = process.env.DATABASE_URL ?? "postgres://stallscout:entwicklung@127.0.0.1:54329/stallscout";
const sql = postgres(url, { max: 1, onnotice: () => {} });

const TAGS_ERLAUBT = new Set([
  "name", "leisure", "sport", "amenity", "animal", "animal_boarding", "animal_breeding", "building", "club",
  "addr:street", "addr:housenumber", "addr:postcode", "addr:city", "addr:place", "addr:suburb", "addr:village",
  "website", "contact:website", "url",
]);

type OsmZeile = {
  osm_typ: string; osm_id: number; name: string; art: string; lat: number; lon: number; strasse: string | null;
  plz: string | null; ort: string | null; website: string | null; tags: Record<string, string>; zusammengefasst: string[];
};

/** Datumsspalten liefert postgres.js als Date */
const tag = (d: unknown) => (d instanceof Date ? d.toISOString().slice(0, 10) : String(d).slice(0, 10));

/** jsonb ordnet Schlüssel um, deshalb für Vergleiche sortiert ausgeben */
const kanon = (v: unknown): string =>
  JSON.stringify(v, (_k, x) => (x && typeof x === "object" && !Array.isArray(x) ? Object.fromEntries(Object.entries(x).sort(([a], [b]) => a.localeCompare(b))) : x));

function jsonLesen(pfad: string): unknown {
  return JSON.parse(readFileSync(pfad, "utf8"));
}

class Abbruch extends Error {}

const zaehler = {
  osm: { datei: 0, abgelehnt: 0, neu: 0, aktualisiert: 0, unveraendert: 0, entfernt_uebersprungen: 0, entfallen: 0 },
  web: { datei: 0, abgelehnt: 0, neu: 0, aktualisiert: 0, unveraendert: 0, entfernt_uebersprungen: 0, entfallen: 0, verknuepft: 0, getrennt_gelassen: 0 },
  betriebe: { neu: 0, geloescht: 0, ort_aktualisiert: 0 },
};
const jeArt: Record<string, number> = {};
const hinweise: string[] = [];

type Betrieb = {
  id: string; slug: string; status: string; osm_typ: string | null; osm_id: string | null; web_schluessel: string | null;
  code_hash: string | null; kreis_ags: string | null; ort_slug: string | null;
};

async function lauf(tx: postgres.TransactionSql) {
  const betriebe = [...(await tx<Betrieb[]>`SELECT id, slug, status::text, osm_typ, osm_id::text, web_schluessel, code_hash, kreis_ags, ort_slug FROM betriebe`)];
  const nachOsm = new Map(betriebe.filter((b) => b.osm_typ).map((b) => [`${b.osm_typ}/${b.osm_id}`, b]));
  const nachWeb = new Map(betriebe.filter((b) => b.web_schluessel).map((b) => [b.web_schluessel!, b]));

  // Stand der Schichten im Speicher, damit Anzeige und Ortszuordnung ohne weitere Abfragen gehen
  const osmSchicht = new Map<string, OsmSchicht>();
  const webSchicht = new Map<string, WebSchicht>();
  for (const z of await tx<Record<string, unknown>[]>`SELECT * FROM osm_betriebe`) {
    osmSchicht.set(`${z.osm_typ}/${z.osm_id}`, {
      osmTyp: z.osm_typ as string, osmId: Number(z.osm_id), osmStand: tag(z.osm_stand), art: z.art as string,
      name: z.name as string, lat: z.lat as number, lon: z.lon as number, strasse: z.strasse as string | null, plz: z.plz as string | null,
      ort: z.ort as string | null, website: z.website as string | null,
    });
  }
  for (const z of await tx<Record<string, unknown>[]>`SELECT schluessel, felder FROM web_betriebe`) {
    webSchicht.set(z.schluessel as string, { schluessel: z.schluessel as string, felder: z.felder as Record<string, WebFeldWert> });
  }

  /** Ort und Kreis neu ableiten und, wenn nötig, am Eintrag nachziehen */
  const ortNachziehen = async (b: Betrieb) => {
    const o = b.osm_typ ? (osmSchicht.get(`${b.osm_typ}/${b.osm_id}`) ?? null) : null;
    const w = b.web_schluessel ? (webSchicht.get(b.web_schluessel) ?? null) : null;
    const t = ortAbleiten(o, w);
    const kreis = t?.ort.kreis ?? null;
    const ort = t?.ort.slug ?? null;
    if (kreis !== b.kreis_ags || ort !== b.ort_slug) {
      await tx`UPDATE betriebe SET kreis_ags = ${kreis}, ort_slug = ${ort}, aktualisiert_am = now() WHERE id = ${b.id}`;
      b.kreis_ags = kreis;
      b.ort_slug = ort;
      zaehler.betriebe.ort_aktualisiert++;
    }
  };

  const neuerBetrieb = async (osmKey: string | null, webKey: string | null): Promise<Betrieb> => {
    const id = randomUUID();
    const o = osmKey ? (osmSchicht.get(osmKey) ?? null) : null;
    const w = webKey ? (webSchicht.get(webKey) ?? null) : null;
    const anzeige = anzeigeBauen(o, w)!;
    const t = ortAbleiten(o, w);
    const slug = stallSlug(anzeige.name.wert, t?.ort.name ?? anzeige.ort?.wert ?? "", id);
    const [typ, oid] = osmKey ? osmKey.split("/") : [null, null];
    const b: Betrieb = {
      id, slug, status: "unbestaetigt", osm_typ: typ, osm_id: oid, web_schluessel: webKey, code_hash: null,
      kreis_ags: t?.ort.kreis ?? null, ort_slug: t?.ort.slug ?? null,
    };
    await tx`INSERT INTO betriebe (id, slug, osm_typ, osm_id, web_schluessel, kreis_ags, ort_slug)
             VALUES (${id}, ${slug}, ${typ}, ${oid}, ${webKey}, ${b.kreis_ags}, ${b.ort_slug})`;
    zaehler.betriebe.neu++;
    return b;
  };

  // ------------------------------------------------------------------------------------------------ OSM-Schicht
  if (osmDatei) {
    const d = jsonLesen(osmDatei) as { meta: { osm_stand: string; lauf: string; lizenz?: string }; betriebe: OsmZeile[] };
    if (!d?.meta?.osm_stand || !d.meta.lauf || !Array.isArray(d.betriebe)) throw new Abbruch("OSM-Datei hat nicht das Format von scripts/osm/extrahieren.py");
    const gesehen = new Set<string>();
    zaehler.osm.datei = d.betriebe.length;
    for (const r of d.betriebe) {
      const key = `${r.osm_typ}/${r.osm_id}`;
      if (!r.name || !(r.art in ARTEN) || !Number.isFinite(r.lat) || !Number.isFinite(r.lon) || !["node", "way", "relation"].includes(r.osm_typ) || gesehen.has(key)) {
        zaehler.osm.abgelehnt++;
        continue;
      }
      gesehen.add(key);
      const tags = Object.fromEntries(Object.entries(r.tags ?? {}).filter(([k]) => TAGS_ERLAUBT.has(k)));
      const b = nachOsm.get(key);
      if (b?.status === "entfernt") {
        zaehler.osm.entfernt_uebersprungen++;
        continue;
      }
      const neuZeile = {
        stand: d.meta.osm_stand, tags, art: r.art, name: r.name, lat: r.lat, lon: r.lon, strasse: r.strasse, plz: r.plz, ort: r.ort,
        website: r.website, zus: r.zusammengefasst ?? [],
      };
      const [alt] = await tx<Record<string, unknown>[]>`SELECT * FROM osm_betriebe WHERE osm_typ = ${r.osm_typ} AND osm_id = ${r.osm_id}`;
      const gleich =
        alt &&
        tag(alt.osm_stand) === neuZeile.stand && alt.art === r.art && alt.name === r.name && alt.lat === r.lat && alt.lon === r.lon &&
        alt.strasse === r.strasse && alt.plz === r.plz && alt.ort === r.ort && alt.website === r.website &&
        kanon(alt.tags) === kanon(tags) && JSON.stringify(alt.zusammengefasst) === JSON.stringify(neuZeile.zus);
      if (gleich) zaehler.osm.unveraendert++;
      else {
        await tx`INSERT INTO osm_betriebe (osm_typ, osm_id, osm_stand, tags, art, name, lat, lon, strasse, plz, ort, website, zusammengefasst, import_lauf)
                 VALUES (${r.osm_typ}, ${r.osm_id}, ${neuZeile.stand}, ${tx.json(tags)}, ${r.art}, ${r.name}, ${r.lat}, ${r.lon}, ${r.strasse}, ${r.plz},
                         ${r.ort}, ${r.website}, ${tx.array(neuZeile.zus)}, ${d.meta.lauf})
                 ON CONFLICT (osm_typ, osm_id) DO UPDATE SET osm_stand = excluded.osm_stand, tags = excluded.tags, art = excluded.art,
                   name = excluded.name, lat = excluded.lat, lon = excluded.lon, strasse = excluded.strasse, plz = excluded.plz, ort = excluded.ort,
                   website = excluded.website, zusammengefasst = excluded.zusammengefasst, import_lauf = excluded.import_lauf, importiert_am = now()`;
        if (alt) zaehler.osm.aktualisiert++;
        else zaehler.osm.neu++;
      }
      osmSchicht.set(key, {
        osmTyp: r.osm_typ, osmId: r.osm_id, osmStand: d.meta.osm_stand, art: r.art, name: r.name, lat: r.lat, lon: r.lon, strasse: r.strasse,
        plz: r.plz, ort: r.ort, website: r.website,
      });
      jeArt["osm:" + r.art] = (jeArt["osm:" + r.art] ?? 0) + 1;
      if (b) await ortNachziehen(b);
      else {
        const nb = await neuerBetrieb(key, null);
        nachOsm.set(key, nb);
      }
    }
    // Zeilen, die OSM nicht mehr liefert (oder die nicht mehr passen)
    const vorhanden = [...osmSchicht.keys()].filter((k) => !gesehen.has(k));
    if (vorhanden.length > 0 && gesehen.size < (vorhanden.length + gesehen.size) / 2 && !erzwingen) {
      throw new Abbruch(`Die OSM-Datei lässt ${vorhanden.length} von ${vorhanden.length + gesehen.size} Zeilen entfallen. Mit --aufraeumen-erzwingen trotzdem.`);
    }
    for (const key of vorhanden) {
      const b = nachOsm.get(key);
      if (b && b.status !== "unbestaetigt") continue; // übernommene Betriebe behalten ihre Herkunft
      const [typ, id] = key.split("/");
      await tx`DELETE FROM osm_betriebe WHERE osm_typ = ${typ} AND osm_id = ${Number(id)}`;
      osmSchicht.delete(key);
      zaehler.osm.entfallen++;
      if (b) {
        if (b.web_schluessel) {
          await tx`UPDATE betriebe SET osm_typ = NULL, osm_id = NULL, aktualisiert_am = now() WHERE id = ${b.id}`;
          b.osm_typ = null;
          b.osm_id = null;
          await ortNachziehen(b);
        } else {
          await tx`DELETE FROM betriebe WHERE id = ${b.id}`;
          zaehler.betriebe.geloescht++;
        }
        nachOsm.delete(key);
      }
    }
  }

  // ------------------------------------------------------------------------------------------------ Website-Schicht
  if (webDatei) {
    const inhalt = readFileSync(webDatei, "utf8");
    const lauf = `web-${new Date().toISOString().slice(0, 10)}-${createHash("sha256").update(inhalt).digest("hex").slice(0, 12)}`;
    const pruefung = webDateiPruefen(JSON.parse(inhalt));
    zaehler.web.datei = pruefung.saetze.length + pruefung.abgelehnt.length;
    zaehler.web.abgelehnt = pruefung.abgelehnt.length;
    const grundZahlen: Record<string, number> = {};
    for (const a of pruefung.abgelehnt) grundZahlen["abgelehnt: " + a.grund.replace(/„.*“/, "„…“")] = (grundZahlen["abgelehnt: " + a.grund.replace(/„.*“/, "„…“")] ?? 0) + 1;
    for (const v of pruefung.verworfeneFelder) grundZahlen[`Feld ${v.feld} verworfen: ${v.grund}`] = (grundZahlen[`Feld ${v.feld} verworfen: ${v.grund}`] ?? 0) + 1;
    if (pruefung.nichtGespeichert) grundZahlen["Felder bewusst nicht gespeichert (Telefon, E-Mail, Betreiber)"] = pruefung.nichtGespeichert;
    for (const [g, n] of Object.entries(grundZahlen)) hinweise.push(`  ${n} × ${g}`);

    const gesehen = new Set<string>();
    for (const s of pruefung.saetze) {
      gesehen.add(s.schluessel);
      const b0 = nachWeb.get(s.schluessel);
      if (b0?.status === "entfernt") {
        zaehler.web.entfernt_uebersprungen++;
        continue;
      }
      // Hinweis auf ein OSM-Objekt, das entfernt wurde: Betrieb wollte raus, also auch hier nichts anlegen
      const osmKey = s.osm ? `${s.osm.typ}/${s.osm.id}` : null;
      const osmBetrieb = osmKey ? nachOsm.get(osmKey) : undefined;
      if (osmBetrieb?.status === "entfernt") {
        zaehler.web.entfernt_uebersprungen++;
        continue;
      }
      const felder = s.felder as Record<string, WebFeldWert>;
      const [alt] = await tx<{ felder: unknown; osm_typ: string | null; osm_id: string | null }[]>`SELECT felder, osm_typ, osm_id::text FROM web_betriebe WHERE schluessel = ${s.schluessel}`;
      const gleich = alt && kanon(alt.felder) === kanon(felder) && alt.osm_typ === (s.osm?.typ ?? null) && alt.osm_id === (s.osm ? String(s.osm.id) : null);
      if (gleich) zaehler.web.unveraendert++;
      else {
        await tx`INSERT INTO web_betriebe (schluessel, osm_typ, osm_id, felder, import_lauf)
                 VALUES (${s.schluessel}, ${s.osm?.typ ?? null}, ${s.osm?.id ?? null}, ${tx.json(felder)}, ${lauf})
                 ON CONFLICT (schluessel) DO UPDATE SET osm_typ = excluded.osm_typ, osm_id = excluded.osm_id, felder = excluded.felder,
                   import_lauf = excluded.import_lauf, importiert_am = now()`;
        if (alt) zaehler.web.aktualisiert++;
        else zaehler.web.neu++;
      }
      webSchicht.set(s.schluessel, { schluessel: s.schluessel, felder });
      const artWert = felder.art?.wert;
      jeArt["website:" + (artWert && artWert in ARTEN ? artWert : "ohne Art")] = (jeArt["website:" + (artWert && artWert in ARTEN ? artWert : "ohne Art")] ?? 0) + 1;

      // Eintrag zuordnen: schon bekannt, sonst zum OSM-Betrieb, sonst neu
      let b = b0;
      if (!b && osmBetrieb && !osmBetrieb.web_schluessel) {
        await tx`UPDATE betriebe SET web_schluessel = ${s.schluessel}, aktualisiert_am = now() WHERE id = ${osmBetrieb.id}`;
        osmBetrieb.web_schluessel = s.schluessel;
        nachWeb.set(s.schluessel, osmBetrieb);
        b = osmBetrieb;
        zaehler.web.verknuepft++;
      } else if (b && !b.osm_typ && osmBetrieb && !osmBetrieb.web_schluessel && osmBetrieb.id !== b.id) {
        // Bisher eigener Eintrag, jetzt als derselbe Betrieb wie ein OSM-Eintrag erkannt: zusammenführen, wenn kein Brief unterwegs ist
        if (b.code_hash || osmBetrieb.code_hash || b.status !== "unbestaetigt") {
          zaehler.web.getrennt_gelassen++;
          hinweise.push(`  ${s.schluessel}: gehört laut Datei zu ${osmKey}, beide haben aber schon einen Code oder sind übernommen, bleiben getrennt`);
        } else {
          await tx`DELETE FROM betriebe WHERE id = ${b.id}`;
          await tx`UPDATE betriebe SET web_schluessel = ${s.schluessel}, aktualisiert_am = now() WHERE id = ${osmBetrieb.id}`;
          osmBetrieb.web_schluessel = s.schluessel;
          nachWeb.set(s.schluessel, osmBetrieb);
          b = osmBetrieb;
          zaehler.betriebe.geloescht++;
          zaehler.web.verknuepft++;
        }
      }
      if (!b) {
        b = await neuerBetrieb(null, s.schluessel);
        nachWeb.set(s.schluessel, b);
      } else await ortNachziehen(b);
    }
    // Zeilen der Website-Schicht, die in der Datei nicht mehr stehen
    const vorhanden = [...webSchicht.keys()].filter((k) => !gesehen.has(k));
    if (vorhanden.length > 0 && gesehen.size < (vorhanden.length + gesehen.size) / 2 && !erzwingen) {
      throw new Abbruch(`Die Website-Datei lässt ${vorhanden.length} von ${vorhanden.length + gesehen.size} Zeilen entfallen. Mit --aufraeumen-erzwingen trotzdem.`);
    }
    for (const key of vorhanden) {
      const b = nachWeb.get(key);
      if (b && b.status !== "unbestaetigt") continue;
      await tx`DELETE FROM web_betriebe WHERE schluessel = ${key}`;
      webSchicht.delete(key);
      zaehler.web.entfallen++;
      if (b) {
        if (b.osm_typ) {
          await tx`UPDATE betriebe SET web_schluessel = NULL, aktualisiert_am = now() WHERE id = ${b.id}`;
          b.web_schluessel = null;
          await ortNachziehen(b);
        } else {
          await tx`DELETE FROM betriebe WHERE id = ${b.id}`;
          zaehler.betriebe.geloescht++;
        }
        nachWeb.delete(key);
      }
    }
  }
}

try {
  await sql.begin(async (tx) => {
    await lauf(tx);
    if (trocken) throw new Abbruch("__trockenlauf__");
  });
} catch (e) {
  if (!(e instanceof Abbruch) || e.message !== "__trockenlauf__") {
    console.error(e instanceof Abbruch ? `Abgebrochen: ${e.message}` : e);
    await sql.end();
    process.exit(1);
  }
}

console.log(trocken ? "TROCKENLAUF (nichts gespeichert)" : "Import fertig");
if (osmDatei) console.log(`OSM-Schicht (${osmDatei}):`, zaehler.osm);
if (webDatei) console.log(`Website-Schicht (${webDatei}):`, zaehler.web);
console.log("Einträge:", zaehler.betriebe);
console.log("Je Schicht und Art:", Object.fromEntries(Object.entries(jeArt).sort()));
if (hinweise.length) console.log("Hinweise:\n" + hinweise.join("\n"));
const [ges] = await sql<{ n: number; osm: number; web: number; beide: number }[]>`
  SELECT count(*)::int AS n, count(*) FILTER (WHERE osm_typ IS NOT NULL)::int AS osm, count(*) FILTER (WHERE web_schluessel IS NOT NULL)::int AS web,
         count(*) FILTER (WHERE osm_typ IS NOT NULL AND web_schluessel IS NOT NULL)::int AS beide FROM betriebe WHERE status <> 'entfernt'`;
console.log(`In der Datenbank${trocken ? " (Stand vor dem Trockenlauf)" : ""}: ${ges.n} Einträge, davon ${ges.osm} mit OSM, ${ges.web} mit Website, ${ges.beide} mit beidem`);
await sql.end();