companies: Adress-Klassifizierung persistieren (Roadmap 1a.2, E1 = 3)

Die Klassifizierung aus dem Trockenlauf wird jetzt nach company_email_groups
geschrieben — eine Zeile je Adresse, denn ob zusammengeführt werden darf, ist
eine Eigenschaft der Adresse und nicht der Firma. Die Zusammenführung (1a.3)
und die laufende Zuordnung neuer API-Meldungen (1a.4) lesen dort, statt den
Trockenlauf jedes Mal neu zu rechnen.

Die Logik liegt jetzt in CompanyAddressClassifier und wird von Bericht und
Persistierung gleichermaßen benutzt. Beide müssen dieselbe Antwort geben —
sonst zeigt der Bericht etwas anderes, als die Zusammenführung später tut.

manual_classification übersteuert die berechnete Klasse und überlebt jede
Neuberechnung. Nötig für Grenzfälle, die nur ein Mensch entscheiden kann:
presse@pr4you.de ist eine echte Agentur, sieht in den Daten aber aus wie eine
Wegwerfadresse. effectiveClassification() ist die einzige Stelle, die
Abnehmer fragen dürfen.

Eigene Tabelle statt Spalten auf companies: Die Tabelle trägt einen
FULLTEXT-Index, jedes ADD COLUMN dort ist ein Rebuild. Die Zuordnung Firma →
Gruppe ist ohnehin nur die normalisierte Adresse, und die eine Stelle, die
festlegt, was "dieselbe Adresse" heißt, ist jetzt Company::emailKey().

E1 ist entschieden: Schwellwert 3, nach drei Live-Läufen. Bei 2 wird jede
zweite Schreibweise zur "Agentur", bei 4 wandern echte Mehrkunden-Adressen in
die Handarbeit. Konfiguriert in config/newsroom.php; Bericht und
Klassifizierung lesen den Wert von dort.

Die Ähnlichkeitsprüfung ist quadratisch und jetzt bei 25 Repräsentanten
gedeckelt — ab da steht die Klasse fest, und yahoo@yahoo.com mit über 4000
Namen hätte sonst neun Millionen Levenshtein-Aufrufe gekostet. Laufzeit über
den ganzen Bestand: 25 s → 3 s, Ergebnis unverändert.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

app/Services/Company/CompanyAddressClassifier.php

@@ -0,0 +1,251 @@
+<?php
+
+namespace App\Services\Company;
+
+use App\Enums\CompanyAddressClass;
+use App\Models\Company;
+use App\Models\PressRelease;
+use App\Scopes\PortalScope;
+
+/**
+ * Klassifiziert Firmen-E-Mail-Adressen für die Newsroom-Konsolidierung.
+ *
+ * Eine Logik, zwei Abnehmer: der Trockenlauf-Bericht (1a.1) und die
+ * Persistierung in `company_email_groups` (1a.2). Beide müssen dieselbe
+ * Antwort geben — sonst zeigt der Bericht etwas anderes, als die
+ * Zusammenführung später tut.
+ *
+ * Schlüssel ist die FIRMEN-E-Mail, nicht die Pressekontaktadresse. Der
+ * Pressekontakt ist der Firma untergeordnet und wechselt; die Firmenidentität
+ * darf nicht an einer Person hängen.
+ *
+ * Fünf Klassen statt der drei Signale aus dem Regelwerk. Zwei Ergänzungen
+ * kommen aus den Daten: die Namensvariante (Rechtsform, Tippfehler, Zusatz —
+ * dieselbe Firma) und die Adresse ohne jede Veröffentlichung. Letztere ist der
+ * Grund, warum der Trockenlauf vor jeder Zusammenführung stehen muss: Hinter
+ * `yahoo@yahoo.com` stehen 4376 Firmen und keine einzige Meldung. Würde über
+ * sie zusammengeführt, verschmölzen Tausende Unbeteiligte zu einem Newsroom.
+ */
+class CompanyAddressClassifier
+{
+    /**
+     * Ab so vielen unterschiedlichen Firmennamen auf einer Adresse kommt sie
+     * für die Nie-veröffentlicht-Prüfung überhaupt in Frage. Bewusst
+     * großzügig: Eine echte Agentur mit 50 Kunden soll noch als Agentur gelten.
+     */
+    private const JunkThreshold = 25;
+
+    /**
+     * Die Namensanzahl allein trennt Müll NICHT von einer großen Agentur —
+     * `pr@deutsche-stadtauskunft.ag` trägt 133 Namen und hat trotzdem 134
+     * Firmen mit veröffentlichten Meldungen. Das trennscharfe Signal ist die
+     * Veröffentlichung: Hinter einer Platzhalter- oder Wegwerfadresse ist nie
+     * etwas erschienen, hinter einer Agentur laufend.
+     */
+    private const JunkPublishedRatio = 0.05;
+
+    /**
+     * Rechtsformen und Zusätze, die für die Namensgleichheit nichts bedeuten.
+     * „Hoppe Unternehmensberatung" und „Hoppe Unternehmensberatung GmbH" sind
+     * dieselbe Firma.
+     *
+     * Als einzelne Wörter geführt und vom Ende her wiederholt entfernt, bis
+     * keines mehr passt. So fallen „GmbH & Co. KG", „Ltd. & Co. KG" und
+     * „AG & Co. KGaA" gleichermaßen weg, ohne dass jede Kombination einzeln
+     * aufgeführt sein muss — die Satzzeichen sind zu dem Zeitpunkt ohnehin
+     * schon fort.
+     *
+     * @var list<string>
+     */
+    private const LegalFormTokens = [
+        'gmbh', 'mbh', 'ug', 'haftungsbeschraenkt', 'ohg', 'kgaa', 'kg', 'ag', 'se', 'co', 'cie', 'und',
+        'ek', 'ev', 'gbr', 'partg', 'ltd', 'llc', 'inc', 'bv', 'nv', 'sa', 'srl', 'plc', 'mvz',
+    ];
+
+    /**
+     * Zwei normalisierte Namen gelten als derselbe, wenn sie sich nur um
+     * wenige Zeichen unterscheiden (Tippfehler: „Berursverband" /
+     * „Berufsverband", „S Research" / „JS Research") oder denselben Anfang
+     * teilen (Zusätze: „Praxis Prof Dr Dhom Kollegen" / „Praxis Prof Dr Dhom
+     * Partner"). Unter EINER Firmen-E-Mail ist ein gemeinsamer Anfang aus
+     * drei Wörtern ein sehr starkes Zeichen für dieselbe Einheit.
+     */
+    private const SimilarMaxDistance = 2;
+
+    private const SimilarDistanceRatio = 0.15;
+
+    private const SharedPrefixWords = 3;
+
+    /**
+     * Alle Adressen mit mehr als einer Firma, klassifiziert.
+     *
+     * @param  int  $collectionThreshold  ab wie vielen abweichenden Namen eine Adresse als Sammeladresse gilt (E1)
+     * @return list<array{email: string, klasse: CompanyAddressClass, firmen: int, namen: int, namen_normalisiert: int, beispiele: list<string>, firmen_mit_pm: int}>
+     */
+    public function classifyAll(int $collectionThreshold): array
+    {
+        // Einmal alle Firmen mit mindestens einer veröffentlichten Meldung,
+        // statt je Firma eine korrelierte Unterabfrage: Das läuft über den
+        // gesamten Bestand, und das ist eine Abfrage gegen hunderttausend.
+        $withPublication = array_flip(
+            PressRelease::query()
+                ->withoutGlobalScope(PortalScope::class)
+                ->where('status', 'published')
+                ->whereNotNull('company_id')
+                ->distinct()
+                ->pluck('company_id')
+                ->all()
+        );
+
+        $rows = Company::query()
+            ->withoutGlobalScope(PortalScope::class)
+            ->whereNotNull('email')
+            ->where('email', '!=', '')
+            ->orderBy('email')
+            ->orderBy('id')
+            ->get(['id', 'email', 'name']);
+
+        $groups = [];
+
+        foreach ($rows as $row) {
+            $key = Company::emailKey($row->email);
+
+            if ($key === null) {
+                continue;
+            }
+
+            $groups[$key] ??= ['email' => $key, 'firmen' => 0, 'beispiele' => [], 'firmen_mit_pm' => 0, 'normalisiert' => []];
+            $groups[$key]['firmen']++;
+            $groups[$key]['firmen_mit_pm'] += isset($withPublication[$row->id]) ? 1 : 0;
+
+            $name = trim((string) $row->name);
+
+            if (! in_array($name, $groups[$key]['beispiele'], true)) {
+                $groups[$key]['beispiele'][] = $name;
+            }
+
+            $groups[$key]['normalisiert'][$this->normalise($name)] = true;
+        }
+
+        $result = [];
+
+        foreach ($groups as $group) {
+            if ($group['firmen'] < 2) {
+                continue;
+            }
+
+            $group['namen'] = count($group['beispiele']);
+            $group['namen_normalisiert'] = $this->countDistinct(array_keys($group['normalisiert']));
+            unset($group['normalisiert']);
+            $group['klasse'] = $this->classify($group, $collectionThreshold);
+
+            $result[] = $group;
+        }
+
+        return $result;
+    }
+
+    /**
+     * @param  array{firmen: int, namen: int, namen_normalisiert: int, firmen_mit_pm: int}  $group
+     */
+    private function classify(array $group, int $collectionThreshold): CompanyAddressClass
+    {
+        if ($group['namen'] >= self::JunkThreshold
+            && $group['firmen_mit_pm'] / max(1, $group['firmen']) < self::JunkPublishedRatio) {
+            return CompanyAddressClass::NeverPublished;
+        }
+
+        if ($group['namen'] === 1) {
+            return CompanyAddressClass::Duplicate;
+        }
+
+        // Normalisiert derselbe Name: Rechtsform, Schreibweise oder Zusatz
+        // weicht ab, die Firma ist dieselbe. Genau der Fall „gleicher ODER
+        // ÄHNLICHER Firmenname" aus dem Regelwerk.
+        if ($group['namen_normalisiert'] === 1) {
+            return CompanyAddressClass::NameVariant;
+        }
+
+        return $group['namen_normalisiert'] >= $collectionThreshold
+            ? CompanyAddressClass::CollectionAddress
+            : CompanyAddressClass::Proposal;
+    }
+
+    /**
+     * Firmenname auf seinen Kern reduziert: Kleinschreibung, Umlaute
+     * aufgelöst, Rechtsform und Satzzeichen entfernt.
+     */
+    public function normalise(string $name): string
+    {
+        $value = mb_strtolower($name);
+        $value = strtr($value, ['ä' => 'ae', 'ö' => 'oe', 'ü' => 'ue', 'ß' => 'ss']);
+        $value = (string) preg_replace('/[^a-z0-9]+/', ' ', $value);
+        $value = trim((string) preg_replace('/\s+/', ' ', $value));
+
+        $words = explode(' ', $value);
+
+        // Vom Ende her, solange das letzte Wort eine Rechtsform ist — aber nie
+        // bis auf nichts: „AG" allein ist ein Firmenname, kein Zusatz.
+        while (count($words) > 1 && in_array(end($words), self::LegalFormTokens, true)) {
+            array_pop($words);
+        }
+
+        return implode(' ', $words);
+    }
+
+    /**
+     * Zahl der wirklich verschiedenen Namen in einer Liste normalisierter
+     * Namen. Ähnliche werden gierig demselben Cluster zugeschlagen.
+     *
+     * Gedeckelt bei JunkThreshold: Ab so vielen verschiedenen Namen steht die
+     * Klasse fest (Sammeladresse oder nie veröffentlicht), und der Vergleich
+     * ist quadratisch — bei `yahoo@yahoo.com` mit über 4000 Namen wären das
+     * neun Millionen Levenshtein-Aufrufe für ein Ergebnis, das ohnehin
+     * „viele" heißt. Der zurückgegebene Wert ist dann eine Untergrenze.
+     *
+     * @param  list<string>  $names
+     */
+    private function countDistinct(array $names): int
+    {
+        $representatives = [];
+
+        foreach ($names as $name) {
+            if (count($representatives) >= self::JunkThreshold) {
+                break;
+            }
+
+            foreach ($representatives as $representative) {
+                if ($this->similar($name, $representative)) {
+                    continue 2;
+                }
+            }
+
+            $representatives[] = $name;
+        }
+
+        return count($representatives);
+    }
+
+    private function similar(string $a, string $b): bool
+    {
+        if ($a === $b) {
+            return true;
+        }
+
+        $length = max(mb_strlen($a), mb_strlen($b));
+        $allowed = max(self::SimilarMaxDistance, (int) ceil($length * self::SimilarDistanceRatio));
+
+        // levenshtein() arbeitet auf Bytes und ist auf 255 begrenzt — die
+        // Namen sind zu diesem Zeitpunkt ASCII und kurz.
+        if (levenshtein(substr($a, 0, 255), substr($b, 0, 255)) <= $allowed) {
+            return true;
+        }
+
+        $wordsA = explode(' ', $a);
+        $wordsB = explode(' ', $b);
+
+        return count($wordsA) >= self::SharedPrefixWords
+            && count($wordsB) >= self::SharedPrefixWords
+            && array_slice($wordsA, 0, self::SharedPrefixWords) === array_slice($wordsB, 0, self::SharedPrefixWords);
+    }
+}