Kundenverwaltung/update/migration-utf8mb4.php

<?php
/*
 * Migration einer Instanz-Datenbank auf das Schema update/schema.sql
 * (utf8mb4) mit Reparatur doppelt kodierter Texte.
 *
 * Hintergrund: Die Anwendung hat jahrelang UTF-8 aus dem Browser über eine
 * Verbindung geschrieben, die der Server als latin1 behandelt hat. Aus "ä"
 * wurde dabei "ä". Datensätze, die auf anderem Weg entstanden sind
 * (phpMyAdmin, neuere Server-Defaults), können dagegen korrekt sein. Deshalb
 * wird jeder Wert einzeln geprüft und nur repariert, wenn er eindeutig
 * doppelt kodiert ist.
 *
 * Aufruf im Installationsverzeichnis der Instanz (die Zugangsdaten kommen
 * aus deren Config):
 *
 *     php update/migration-utf8mb4.php pruefen
 *         Liest nur und gibt einen Bericht aus. Ändert nichts.
 *
 *     php update/migration-utf8mb4.php migrieren <zieldatenbank>
 *         Legt die Zieldatenbank mit update/schema.sql an und kopiert alle
 *         Daten hinein. Die Quelldatenbank bleibt unverändert. Die Ziel-
 *         datenbank darf noch nicht existieren oder muss leer sein.
 *
 * Für lokale Tests lässt sich die Config übergehen:
 *     --server=127.0.0.1:3307 --user=root --db=wagner
 * Das Passwort kommt dann aus der Umgebungsvariable MIGRATION_PW, damit es
 * nicht in der Prozessliste steht.
 */

if(PHP_SAPI != "cli")
	exit("Nur per Kommandozeile aufrufbar.\n");

require_once(__DIR__."/../incs/db.php");

function fehler($text)
{
	fwrite(STDERR, "FEHLER: ".$text."\n");
	exit(1);
}

/* ---------------------------------------------------------------------
 * Kodierung
 * ------------------------------------------------------------------ */

/*
 * Umkehrung dessen, was MariaDB beim Speichern gemacht hat: Unicode-Text
 * zurück in die Bytes von MySQL-"latin1" (das ist cp1252; die fünf dort
 * undefinierten Bytes bildet MySQL auf U+0081, U+008D, U+008F, U+0090 und
 * U+009D ab). Gibt null zurück, wenn ein Zeichen in latin1 nicht vorkommt -
 * dann kann der Wert nicht über eine latin1-Verbindung entstanden sein.
 */
function latin1Bytes($text)
{
	static $cp1252 = array(
		0x20AC => 0x80, 0x201A => 0x82, 0x0192 => 0x83, 0x201E => 0x84,
		0x2026 => 0x85, 0x2020 => 0x86, 0x2021 => 0x87, 0x02C6 => 0x88,
		0x2030 => 0x89, 0x0160 => 0x8A, 0x2039 => 0x8B, 0x0152 => 0x8C,
		0x017D => 0x8E, 0x2018 => 0x91, 0x2019 => 0x92, 0x201C => 0x93,
		0x201D => 0x94, 0x2022 => 0x95, 0x2013 => 0x96, 0x2014 => 0x97,
		0x02DC => 0x98, 0x2122 => 0x99, 0x0161 => 0x9A, 0x203A => 0x9B,
		0x0153 => 0x9C, 0x017E => 0x9E, 0x0178 => 0x9F,
		0x0081 => 0x81, 0x008D => 0x8D, 0x008F => 0x8F, 0x0090 => 0x90,
		0x009D => 0x9D,
	);
	$bytes = "";
	foreach(preg_split('//u', $text, -1, PREG_SPLIT_NO_EMPTY) as $zeichen)
	{
		$cp = codepoint($zeichen);
		if($cp < 0x80 || ($cp >= 0xA0 && $cp <= 0xFF))
			$bytes .= chr($cp);
		elseif(isset($cp1252[$cp]))
			$bytes .= chr($cp1252[$cp]);
		else
			return null;
	}
	return $bytes;
}

function codepoint($zeichen)
{
	$b = array_values(unpack("C*", $zeichen));
	switch(count($b))
	{
		case 1: return $b[0];
		case 2: return (($b[0] & 0x1F) << 6) | ($b[1] & 0x3F);
		case 3: return (($b[0] & 0x0F) << 12) | (($b[1] & 0x3F) << 6) | ($b[2] & 0x3F);
		default: return (($b[0] & 0x07) << 18) | (($b[1] & 0x3F) << 12) | (($b[2] & 0x3F) << 6) | ($b[3] & 0x3F);
	}
}

/*
 * Ordnet einen Textwert ein und liefert array(kategorie, neuer_wert).
 *
 *   ascii      nur ASCII, unverändert
 *   korrekt    enthält Nicht-ASCII, ist aber kein doppelt kodierter Text
 *   repariert  war doppelt kodiert
 *   mehrfach   war mehr als doppelt kodiert
 *   unklar     sieht teilweise doppelt kodiert aus, lässt sich aber nicht
 *              eindeutig zurückführen; bleibt unverändert
 */
function pruefeText($wert)
{
	if(!preg_match('/[^\x00-\x7F]/', $wert))
		return array("ascii", $wert);

	$runden = 0;
	$aktuell = $wert;
	while($runden < 5)
	{
		$bytes = latin1Bytes($aktuell);
		// Echte Umlaute in latin1-Bytes ("M\xFCller") sind kein gültiges
		// UTF-8. Nur wenn die Bytes gültiges UTF-8 mit Nicht-ASCII ergeben,
		// war der Text doppelt kodiert.
		if($bytes === null || !preg_match('//u', $bytes) || !preg_match('/[^\x00-\x7F]/', $bytes))
			break;
		$aktuell = $bytes;
		$runden++;
	}

	if($runden == 1)
		return array("repariert", $aktuell);
	if($runden > 1)
		return array("mehrfach", $aktuell);

	// Typische Reste doppelter Kodierung neben Zeichen, die nicht aus
	// latin1 stammen können: gemischter Wert, nicht automatisch lösbar.
	if(preg_match('/[\x{00C2}\x{00C3}][\x{0080}-\x{00BF}\x{0152}\x{0153}\x{0160}\x{0161}\x{0178}\x{017D}\x{017E}\x{0192}\x{02C6}\x{02DC}\x{2013}\x{2014}\x{2018}-\x{201E}\x{2020}-\x{2022}\x{2026}\x{2030}\x{2039}\x{203A}\x{20AC}\x{2122}]/u', $wert))
		return array("unklar", $wert);

	return array("korrekt", $wert);
}

/* ---------------------------------------------------------------------
 * Aufruf und Verbindung
 * ------------------------------------------------------------------ */

$modus = null;
$zielDb = null;
$optionen = array();
foreach(array_slice($argv, 1) as $arg)
{
	if(preg_match('/^--(server|user|db)=(.*)$/', $arg, $m))
		$optionen[$m[1]] = $m[2];
	elseif($modus === null)
		$modus = $arg;
	elseif($zielDb === null)
		$zielDb = $arg;
	else
		fehler("Unbekanntes Argument: ".$arg);
}

if($modus != "pruefen" && $modus != "migrieren")
	fehler("Aufruf: php update/migration-utf8mb4.php pruefen | migrieren <zieldatenbank>");
if($modus == "migrieren" && !preg_match('/^[A-Za-z0-9_]+$/', (string)$zielDb))
	fehler("Zieldatenbank fehlt oder enthält unzulässige Zeichen.");

if(isset($optionen["db"]))
{
	$server = isset($optionen["server"]) ? $optionen["server"] : "localhost";
	$user = isset($optionen["user"]) ? $optionen["user"] : "";
	$pw = (string)getenv("MIGRATION_PW");
	$quellDb = $optionen["db"];
}
else
{
	require_once(__DIR__."/../incs/constants.php");
	require_once(__DIR__."/../incs/settings.php");
	if(readsettings() != "valid")
		fehler("Konfiguration ".CONFIG_PATH.CONFIG_FILE." nicht lesbar. Im Installationsverzeichnis der Instanz aufrufen.");
	$server = $mysql_server;
	$user = $mysql_user;
	$pw = $mysql_pw;
	$quellDb = $mysql_db;
}

if($modus == "migrieren" && strcasecmp($zielDb, $quellDb) == 0)
	fehler("Quelle und Ziel sind dieselbe Datenbank.");

// Lesen immer als utf8mb4: Der Server wandelt dabei nur um, die Reparatur
// passiert ausschließlich hier im Skript.
$quelle = db_verbinden($server, $user, $pw, $quellDb) or fehler("Keine Verbindung zu ".$quellDb.": ".mysqli_connect_error());

/* Spalten einer Datenbank: tabelle(klein) => array(name, spalten) */
function tabellen($dbconn, $db)
{
	$tabellen = array();
	$zeilen = db_zeilen($dbconn, "SELECT table_name AS t, column_name AS c, data_type AS typ, is_nullable AS nullbar, column_default AS standard, character_set_name AS zeichensatz, extra
		FROM information_schema.columns WHERE table_schema = ? ORDER BY table_name, ordinal_position", array($db));
	foreach($zeilen as $z)
	{
		$schluessel = strtolower($z["t"]);
		if(!isset($tabellen[$schluessel]))
			$tabellen[$schluessel] = array("name" => $z["t"], "spalten" => array());
		$tabellen[$schluessel]["spalten"][$z["c"]] = $z;
	}
	return $tabellen;
}

$quellTabellen = tabellen($quelle, $quellDb);
if(!$quellTabellen)
	fehler("Datenbank ".$quellDb." enthält keine Tabellen.");

/* ---------------------------------------------------------------------
 * Zielschema anlegen
 * ------------------------------------------------------------------ */

$ziel = null;
if($modus == "migrieren")
{
	$vorhanden = db_wert($quelle, "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = ?", array($zielDb));
	if($vorhanden > 0)
		fehler("Zieldatenbank ".$zielDb." existiert und ist nicht leer.");
	mysqli_query($quelle, "CREATE DATABASE IF NOT EXISTS `".$zielDb."` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")
		or fehler("Zieldatenbank nicht anlegbar: ".mysqli_error($quelle));

	$ziel = db_verbinden($server, $user, $pw, $zielDb) or fehler("Keine Verbindung zu ".$zielDb.": ".mysqli_connect_error());
	$schema = file_get_contents(__DIR__."/schema.sql") or fehler("update/schema.sql nicht lesbar.");
	if(!mysqli_multi_query($ziel, $schema))
		fehler("Schema: ".mysqli_error($ziel));
	do {
		if($r = mysqli_store_result($ziel))
			mysqli_free_result($r);
	} while(mysqli_more_results($ziel) && mysqli_next_result($ziel));
	if(mysqli_errno($ziel))
		fehler("Schema: ".mysqli_error($ziel));

	// Beim Kopieren: ids 0 behalten, Fremdschlüssel erst hinterher prüfen.
	mysqli_query($ziel, "SET SESSION sql_mode = 'STRICT_ALL_TABLES,NO_AUTO_VALUE_ON_ZERO,NO_ENGINE_SUBSTITUTION'");
	mysqli_query($ziel, "SET FOREIGN_KEY_CHECKS = 0");
	$zielTabellen = tabellen($ziel, $zielDb);
}
else
{
	// Im Prüfmodus zählt das Schema aus der Datei nicht - geprüft wird,
	// was in der Quelle steht.
	$zielTabellen = $quellTabellen;
}

/* ---------------------------------------------------------------------
 * Tabellen durchgehen
 * ------------------------------------------------------------------ */

/* Wert für eine Spalte, die es in der Quelle nicht gibt */
function ersatzwert($spalte)
{
	if(stripos($spalte["extra"], "auto_increment") !== false)
		return null; // Spalte weglassen, die Datenbank vergibt die Werte
	if($spalte["standard"] !== null && strtoupper($spalte["standard"]) !== "NULL")
		return null; // Spalte weglassen, DEFAULT greift
	if($spalte["nullbar"] == "YES")
		return array(null);
	return array(leerwert($spalte));
}

/* Was der alte, nicht strikte Modus für einen fehlenden Wert eingesetzt hätte */
function leerwert($spalte)
{
	switch($spalte["typ"])
	{
		case "date": return "0000-00-00";
		case "datetime": case "timestamp": return "0000-00-00 00:00:00";
		case "time": return "00:00:00";
		case "char": case "varchar": case "text": case "tinytext": case "mediumtext": case "longtext": case "enum": case "set":
			return "";
		default: return 0;
	}
}

$bericht = array();
$hinweise = array();

foreach($quellTabellen as $schluessel => $q)
	if(!isset($zielTabellen[$schluessel]))
		$hinweise[] = "Tabelle ".$q["name"]." gibt es im neuen Schema nicht, sie wird nicht übernommen.";

foreach($zielTabellen as $schluessel => $z)
{
	if(!isset($quellTabellen[$schluessel]))
	{
		$hinweise[] = "Tabelle ".$z["name"]." fehlt in der Quelle und bleibt leer.";
		continue;
	}
	$q = $quellTabellen[$schluessel];

	$gemeinsam = array();
	$ergaenzt = array();
	foreach($z["spalten"] as $name => $spalte)
	{
		if(isset($q["spalten"][$name]))
			$gemeinsam[] = $name;
		elseif(($ersatz = ersatzwert($spalte)) !== null)
			$ergaenzt[$name] = $ersatz[0];
		elseif(stripos($spalte["extra"], "auto_increment") === false)
			$hinweise[] = "Spalte ".$z["name"].".".$name." fehlt in der Quelle, es gilt der Standardwert ".$spalte["standard"].".";
	}
	foreach(array_diff(array_keys($q["spalten"]), array_keys($z["spalten"])) as $name)
		$hinweise[] = "Spalte ".$q["name"].".".$name." gibt es im neuen Schema nicht, sie wird nicht übernommen.";
	foreach($ergaenzt as $name => $wert)
		$hinweise[] = "Spalte ".$z["name"].".".$name." fehlt in der Quelle und wird mit ".var_export($wert, true)." gefüllt.";

	// Im Ziel Pflicht, in der Quelle darf NULL stehen: dort den Leerwert
	// einsetzen, sonst lehnt der strikte Modus die Zeile ab.
	$nullErsatz = array();
	foreach($gemeinsam as $name)
		if($z["spalten"][$name]["nullbar"] == "NO" && $q["spalten"][$name]["nullbar"] == "YES")
			$nullErsatz[$name] = 0;

	$textSpalten = array();
	foreach($gemeinsam as $name)
		if($q["spalten"][$name]["zeichensatz"] !== null)
			$textSpalten[] = $name;

	$liste = implode(", ", array_map(function($n) { return "`".$n."`"; }, $gemeinsam));
	$res = mysqli_query($quelle, "SELECT ".$liste." FROM `".$q["name"]."`", MYSQLI_USE_RESULT)
		or fehler($q["name"].": ".mysqli_error($quelle));

	$stmt = null;
	if($ziel)
	{
		$spaltenZiel = array_merge($gemeinsam, array_keys($ergaenzt));
		$sql = "INSERT INTO `".$z["name"]."` (".implode(", ", array_map(function($n) { return "`".$n."`"; }, $spaltenZiel)).") VALUES (".implode(", ", array_fill(0, count($spaltenZiel), "?")).")";
		$stmt = mysqli_prepare($ziel, $sql) or fehler($z["name"].": ".mysqli_error($ziel));
		mysqli_begin_transaction($ziel);
	}

	$zeilen = 0;
	while($zeile = mysqli_fetch_assoc($res))
	{
		$zeilen++;
		foreach($textSpalten as $name)
		{
			if($zeile[$name] === null)
				continue;
			list($kategorie, $neu) = pruefeText($zeile[$name]);
			if(!isset($bericht[$q["name"]][$name]))
				$bericht[$q["name"]][$name] = array("ascii" => 0, "korrekt" => 0, "repariert" => 0, "mehrfach" => 0, "unklar" => 0, "unklar_ids" => array());
			$bericht[$q["name"]][$name][$kategorie]++;
			if($kategorie == "unklar" && isset($zeile["id"]) && count($bericht[$q["name"]][$name]["unklar_ids"]) < 20)
				$bericht[$q["name"]][$name]["unklar_ids"][] = $zeile["id"];
			$zeile[$name] = $neu;
		}
		foreach($nullErsatz as $name => $anzahl)
		{
			if($zeile[$name] === null)
			{
				$zeile[$name] = leerwert($z["spalten"][$name]);
				$nullErsatz[$name]++;
			}
		}
		if($stmt)
		{
			$werte = array_merge(array_values($zeile), array_values($ergaenzt));
			mysqli_stmt_execute($stmt, $werte)
				or fehler($z["name"].", Zeile ".$zeilen.(isset($zeile["id"]) ? " (id ".$zeile["id"].")" : "").": ".mysqli_stmt_error($stmt));
		}
	}
	mysqli_free_result($res);
	foreach($nullErsatz as $name => $anzahl)
		if($anzahl > 0)
			$hinweise[] = "Spalte ".$z["name"].".".$name.": ".$anzahl." Zeilen mit NULL, im neuen Schema Pflichtfeld, gefüllt mit ".var_export(leerwert($z["spalten"][$name]), true).".";

	if($stmt)
	{
		mysqli_commit($ziel);
		mysqli_stmt_close($stmt);
		$anzahl = db_wert($ziel, "SELECT COUNT(*) FROM `".$z["name"]."`");
		if($anzahl != $zeilen)
			fehler($z["name"].": ".$zeilen." Zeilen gelesen, aber ".$anzahl." im Ziel.");
	}
	$bericht[$q["name"]]["_zeilen"] = $zeilen;
}

/* ---------------------------------------------------------------------
 * Fremdschlüssel prüfen (nur Ziel)
 * ------------------------------------------------------------------ */

$verwaist = array();
if($ziel)
{
	mysqli_query($ziel, "SET FOREIGN_KEY_CHECKS = 1");
	$fks = db_zeilen($ziel, "SELECT table_name AS t, column_name AS c, referenced_table_name AS rt, referenced_column_name AS rc
		FROM information_schema.key_column_usage WHERE table_schema = ? AND referenced_table_name IS NOT NULL", array($zielDb));
	foreach($fks as $fk)
	{
		$anzahl = db_wert($ziel, "SELECT COUNT(*) FROM `".$fk["t"]."` k LEFT JOIN `".$fk["rt"]."` e ON e.`".$fk["rc"]."` = k.`".$fk["c"]."` WHERE k.`".$fk["c"]."` IS NOT NULL AND e.`".$fk["rc"]."` IS NULL");
		if($anzahl > 0)
			$verwaist[] = sprintf("%s.%s -> %s.%s: %d Zeilen ohne Gegenstück", $fk["t"], $fk["c"], $fk["rt"], $fk["rc"], $anzahl);
	}
}

/* ---------------------------------------------------------------------
 * Bericht
 * ------------------------------------------------------------------ */

echo "Migration utf8mb4 - ".($modus == "pruefen" ? "Prüfmodus (nichts geändert)" : "Migration nach ".$zielDb)."\n";
echo "Quelle: ".$quellDb."\n\n";

$summe = array("repariert" => 0, "mehrfach" => 0, "unklar" => 0, "korrekt" => 0);
printf("%-40s %8s %8s %9s %8s %7s\n", "Spalte", "Zeilen", "korrekt", "repariert", "mehrfach", "unklar");
foreach($bericht as $tabelle => $spalten)
{
	$zeilen = $spalten["_zeilen"];
	unset($spalten["_zeilen"]);
	foreach($spalten as $spalte => $w)
	{
		foreach($summe as $k => $v)
			$summe[$k] += $w[$k];
		if($w["korrekt"] + $w["repariert"] + $w["mehrfach"] + $w["unklar"] == 0)
			continue; // reine ASCII-Spalte
		printf("%-40s %8d %8d %9d %8d %7d\n", $tabelle.".".$spalte, $zeilen, $w["korrekt"], $w["repariert"], $w["mehrfach"], $w["unklar"]);
		if($w["unklar_ids"])
			echo "    unklar bei id: ".implode(", ", $w["unklar_ids"]).(count($w["unklar_ids"]) == 20 ? ", ..." : "")."\n";
	}
}
printf("\nSumme: %d korrekt, %d repariert, %d mehrfach kodiert, %d unklar\n", $summe["korrekt"], $summe["repariert"], $summe["mehrfach"], $summe["unklar"]);

if($hinweise)
	echo "\nHinweise:\n  - ".implode("\n  - ", $hinweise)."\n";
if($ziel)
	echo "\nFremdschlüssel: ".($verwaist ? "\n  - ".implode("\n  - ", $verwaist) : "alle Verweise gültig")."\n";