#!/usr/bin/env python3
"""Method for the slim barcode table used by unsupport.

This file is the alteration method under the Open Database License.
The table itself is not published. The copy unsupport uses was built
with the rules in this file on 8 October 2026.

Sources, read in this order: Open Food Facts, Open Beauty Facts,
Open Pet Food Facts, Open Products Facts. The inputs are the CSV
dumps named food, beauty, pet, and products.

Kept for the served table: barcode, product name, brand, and which
list the row came from. brand_owner, labels, Nutri-Score, ingredients,
stores, and pack photos are not copied.

Dropped: a row with no barcode, a barcode whose length is not 8, 12,
13, or 14, a failed GS1 check digit, and a row with neither a product
name nor a brand. Placeholders such as "unknown" count as empty.
If product_name is empty, product_name_de, then product_name_en, then
product_name_fr is used when that column exists. Those dumps only
contained product_name.

One barcode stays one row across all four lists. The October 2026
table kept the row with the higher score (2 for a brand, 1 for a
name) and left an equal score on the first row.

This rebuild fills empty fields and keeps a product name that is
already stored. A later row that only has a brand does not erase that
name. The language is not chosen: the dumps used here only provide
product_name, with product_name_de, product_name_en, then
product_name_fr as fallbacks when product_name is empty.
"""

from __future__ import annotations

import gzip
import sqlite3
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent
RAW = ROOT / "raw"
DB_PATH = ROOT / "barcode-slice.sqlite"

SOURCES = (
    ("food", RAW / "food.csv.gz"),
    ("beauty", RAW / "beauty.csv.gz"),
    ("pet", RAW / "pet.csv.gz"),
    ("products", RAW / "products.csv.gz"),
)

EMPTY = {
    "",
    "-",
    "?",
    ".",
    "n/a",
    "na",
    "none",
    "null",
    "unknown",
    "unbekannt",
    "inconnu",
    "inconnue",
    "non renseigne",
    "non renseigné",
}


def clean(value: str | None) -> str | None:
    if value is None:
        return None
    text = " ".join(value.replace("\x00", " ").split())
    if text.casefold() in EMPTY:
        return None
    return text


def check_digit_ok(code: str) -> bool | None:
    if not code.isdigit() or len(code) not in (8, 12, 13, 14):
        return None
    digits = [int(char) for char in code]
    total = 0
    for index, digit in enumerate(reversed(digits[:-1])):
        total += digit * (3 if index % 2 == 0 else 1)
    expected = (10 - (total % 10)) % 10
    return expected == digits[-1]


def completeness(name: str | None, brand: str | None) -> int:
    return (2 if brand else 0) + (1 if name else 0)


def open_rows(path: Path):
    with gzip.open(path, "rt", encoding="utf-8", errors="replace", newline="") as handle:
        header = handle.readline().rstrip("\n").split("\t")
        index = {name: position for position, name in enumerate(header)}
        for line in handle:
            yield header, index, line.rstrip("\n").split("\t")


def cell(row: list[str], index: dict[str, int], name: str) -> str:
    position = index.get(name)
    if position is None or position >= len(row):
        return ""
    return row[position]


def main() -> int:
    out = Path(sys.argv[1]) if len(sys.argv) > 1 else DB_PATH
    raw = Path(sys.argv[2]) if len(sys.argv) > 2 else RAW
    sources = tuple((name, raw / f"{name}.csv.gz") for name, _path in SOURCES)
    if out.exists():
        out.unlink()
    out.parent.mkdir(parents=True, exist_ok=True)
    db = sqlite3.connect(out)
    db.execute("PRAGMA journal_mode=OFF")
    db.execute("PRAGMA synchronous=OFF")
    db.execute("PRAGMA temp_store=MEMORY")
    db.execute(
        """
        CREATE TABLE products (
          code TEXT NOT NULL,
          name TEXT,
          brand TEXT,
          source TEXT NOT NULL,
          brands_n INTEGER NOT NULL,
          de INTEGER NOT NULL,
          PRIMARY KEY (code)
        )
        """
    )
    stats: dict[str, dict[str, int]] = {}
    conflicts: list[tuple[str, str, str]] = []

    for source, path in sources:
        if not path.exists():
            print(f"missing {path}", file=sys.stderr)
            return 1
        counter = {
            "rows": 0,
            "kept": 0,
            "replaced": 0,
            "no_code": 0,
            "bad_check": 0,
            "odd_code": 0,
            "neither": 0,
            "name": 0,
            "brand": 0,
            "both": 0,
            "name_only": 0,
            "brand_only": 0,
            "multi_brand": 0,
            "de": 0,
            "de_brand": 0,
            "de_name": 0,
            "de_both": 0,
            "name_de_field": 0,
            "prefix_02": 0,
        }
        print(f"reading {source}", flush=True)
        batch: list[tuple] = []
        seen_header = False
        for _header, index, row in open_rows(path):
            if not seen_header:
                seen_header = True
                print(
                    "  columns",
                    "product_name" in index,
                    "brands" in index,
                    "product_name_de" in index,
                    "countries_tags" in index,
                    flush=True,
                )
            counter["rows"] += 1
            code = cell(row, index, "code").strip()
            if not code:
                counter["no_code"] += 1
                continue
            digit_ok = check_digit_ok(code)
            if digit_ok is None:
                counter["odd_code"] += 1
                continue
            if not digit_ok:
                counter["bad_check"] += 1
                continue
            if code.startswith("02"):
                counter["prefix_02"] += 1
            name = clean(cell(row, index, "product_name"))
            brand = clean(cell(row, index, "brands"))
            if not name:
                for alt in ("product_name_de", "product_name_en", "product_name_fr"):
                    name = clean(cell(row, index, alt))
                    if name:
                        break
            de_name = clean(cell(row, index, "product_name_de"))
            if de_name:
                counter["name_de_field"] += 1
            countries = cell(row, index, "countries_tags").casefold()
            in_de = "en:germany" in countries or "deutschland" in countries
            if in_de:
                counter["de"] += 1
            if name:
                counter["name"] += 1
            if brand:
                counter["brand"] += 1
            if name and brand:
                counter["both"] += 1
            elif name:
                counter["name_only"] += 1
            elif brand:
                counter["brand_only"] += 1
            else:
                counter["neither"] += 1
                continue
            if in_de and brand:
                counter["de_brand"] += 1
            if in_de and name:
                counter["de_name"] += 1
            if in_de and name and brand:
                counter["de_both"] += 1
            brands_n = len([part for part in brand.split(",") if clean(part)]) if brand else 0
            if brands_n > 1:
                counter["multi_brand"] += 1
            if len(row) < max(index.get("code", 0), index.get("brands", 0), index.get("product_name", 0)) + 1:
                counter["short_row"] = counter.get("short_row", 0) + 1
                continue
            batch.append((code, name, brand, source, brands_n, 1 if in_de else 0))
            if len(batch) >= 20000:
                flush(db, batch, counter)
                batch.clear()
            if counter["rows"] % 500000 == 0:
                print(f"  {source} {counter['rows']}", flush=True)
        if batch:
            flush(db, batch, counter)
        db.commit()
        stats[source] = counter
        print(f"done {source} {counter}", flush=True)

    db.commit()
    print("indexing", flush=True)
    db.execute("CREATE INDEX products_source ON products(source)")
    db.commit()
    db.execute("VACUUM")
    write_summary(db, stats, conflicts, out)
    db.close()
    print(f"sqlite {out} bytes {out.stat().st_size}", flush=True)
    return 0


def flush(db, batch, counter) -> None:
    before = db.total_changes
    db.executemany(
        """
        INSERT INTO products (code, name, brand, source, brands_n, de)
        VALUES (?, ?, ?, ?, ?, ?)
        ON CONFLICT(code) DO UPDATE SET
          name = COALESCE(products.name, excluded.name),
          brand = COALESCE(products.brand, excluded.brand),
          brands_n = CASE
            WHEN products.brand IS NOT NULL THEN products.brands_n
            ELSE excluded.brands_n
          END,
          de = CASE WHEN products.de = 1 OR excluded.de = 1 THEN 1 ELSE 0 END
        WHERE products.name IS NULL OR products.brand IS NULL
        """,
        batch,
    )
    changed = db.total_changes - before
    counter["kept"] += changed
    counter["unchanged_dup"] = counter.get("unchanged_dup", 0) + (len(batch) - changed)


def write_summary(db, stats, conflicts, sqlite_path: Path) -> None:
    out = sqlite_path.parent / "report.txt"
    lines = []
    total = db.execute("SELECT COUNT(*) FROM products").fetchone()[0]
    lines.append(f"stored_rows {total}")
    for source, _path in SOURCES:
        count = db.execute(
            "SELECT COUNT(*) FROM products WHERE source = ?",
            (source,),
        ).fetchone()[0]
        lines.append(f"stored_{source} {count}")
    lines.append(
        "stored_both "
        + str(
            db.execute(
                "SELECT COUNT(*) FROM products WHERE name IS NOT NULL AND brand IS NOT NULL"
            ).fetchone()[0]
        )
    )
    lines.append(
        "stored_name_only "
        + str(
            db.execute(
                "SELECT COUNT(*) FROM products WHERE name IS NOT NULL AND brand IS NULL"
            ).fetchone()[0]
        )
    )
    lines.append(
        "stored_brand_only "
        + str(
            db.execute(
                "SELECT COUNT(*) FROM products WHERE name IS NULL AND brand IS NOT NULL"
            ).fetchone()[0]
        )
    )
    lines.append(
        "stored_multi_brand "
        + str(db.execute("SELECT COUNT(*) FROM products WHERE brands_n > 1").fetchone()[0])
    )
    lines.append(
        "stored_de " + str(db.execute("SELECT COUNT(*) FROM products WHERE de = 1").fetchone()[0])
    )
    for source, counter in stats.items():
        lines.append(f"stats_{source} {counter}")
    lines.append(f"conflict_samples {conflicts}")
    out.write_text("\n".join(lines) + "\n", encoding="utf-8")
    print("\n".join(lines), flush=True)


if __name__ == "__main__":
    raise SystemExit(main())