#!/usr/bin/env python3
"""
MA.csv -> public_directory.people_addresses importer

SAFE DEFAULT:
    Without --commit, this script performs a dry run only.

Designed for:
    Python 3.9+
    MariaDB
    mysql-connector-python
"""

import argparse
import csv
import datetime
import hashlib
import json
import re
import sys
from pathlib import Path
from typing import Dict, Optional, Tuple

try:
    import mysql.connector
except ImportError:
    print("ERROR: mysql-connector-python is not installed.")
    print("Use the existing Public Directory API virtual environment.")
    sys.exit(1)


PROJECT_ROOT = Path(
    "/home/mostincr/maillists.mostincrediblevalue.com/data/public-directory"
)

REPORT_DIR = PROJECT_ROOT / "import-reports"

DEFAULT_ENV_FILE = Path("/etc/public-directory-api.env")

MASTER_TABLE = "people_addresses"

STAGING_TABLE = "import_ma_staging"
UNIQUE_TABLE = "import_ma_unique"


TARGET_COLUMNS = [
    "first_name",
    "middle_name",
    "last_name",
    "full_name",
    "title",
    "street_number",
    "street",
    "unit",
    "city",
    "state",
    "zip_code",
    "county",
    "country",
]


COLUMN_LIMITS = {
    "first_name": 40,
    "middle_name": 40,
    "last_name": 60,
    "full_name": 150,
    "title": 20,
    "street_number": 20,
    "street": 100,
    "unit": 30,
    "city": 100,
    "state": 10,
    "zip_code": 20,
    "county": 100,
    "country": 20,
}


REQUIRED_SOURCE_COLUMNS = {
    "buyer_first_name",
    "buyer_last_name",
    "buyer_full_name",
    "full_address",
    "city",
    "state",
    "zip",
    "county",
}


# =========================================================
# TEXT NORMALIZATION
# =========================================================

def clean_text(value: Optional[str]) -> Optional[str]:
    if value is None:
        return None

    value = str(value).strip()

    if not value:
        return None

    value = re.sub(r"\s+", " ", value)

    return value.upper()


def normalize_zip(value: Optional[str]) -> Optional[str]:
    """
    Preserve leading-zero ZIP codes.

    Examples:

        02038       -> 02038
        2038        -> 02038
        2038.0      -> 02038
        020381234   -> 02038-1234
        02038-1234  -> 02038-1234
    """

    if value is None:
        return None

    z = str(value).strip()

    if not z:
        return None

    if re.fullmatch(r"\d+\.0", z):
        z = z[:-2]

    z = re.sub(r"\s+", "", z)

    if re.fullmatch(r"\d{1,5}", z):
        return z.zfill(5)

    if re.fullmatch(r"\d{9}", z):
        return z[:5] + "-" + z[5:]

    if re.fullmatch(r"\d{5}-\d{4}", z):
        return z

    return clean_text(z)


# =========================================================
# ADDRESS PARSING
# =========================================================

def parse_address(
    full_address: Optional[str],
) -> Tuple[
    Optional[str],
    Optional[str],
    Optional[str],
    Optional[str],
]:
    """
    Returns:

        street_number
        street
        unit
        parsing_error

    This parser is intentionally conservative.
    """

    address = clean_text(full_address)

    if not address:
        return None, None, None, "missing address"

    unit = None

    # Examples:
    #
    # 12 MAIN ST APT 3
    # 12 MAIN ST APARTMENT 3
    # 12 MAIN ST UNIT 3
    # 12 MAIN ST STE 200
    # 12 MAIN ST SUITE 200
    # 12 MAIN ST #3

    unit_pattern = re.compile(
        r"\s+(APT(?:ARTMENT)?|UNIT|STE|SUITE|#)\s*"
        r"([A-Z0-9\-]+)\s*$",
        re.IGNORECASE,
    )

    unit_match = unit_pattern.search(address)

    if unit_match:
        designator = unit_match.group(1).upper()
        unit_number = unit_match.group(2).upper()

        if designator == "APARTMENT":
            designator = "APT"

        if designator == "SUITE":
            designator = "STE"

        if designator == "#":
            unit = "#" + unit_number
        else:
            unit = "{} {}".format(
                designator,
                unit_number,
            )

        address = address[:unit_match.start()].strip()

    # Handles:
    #
    # 12 MAIN ST
    # 12A MAIN ST
    # 12-14 MAIN ST

    street_match = re.match(
        r"^([0-9]+(?:-[0-9]+)?[A-Z]?)\s+(.+)$",
        address,
        re.IGNORECASE,
    )

    if not street_match:
        return (
            None,
            address,
            unit,
            "unable to safely parse street number",
        )

    street_number = clean_text(
        street_match.group(1)
    )

    street = clean_text(
        street_match.group(2)
    )

    if not street:
        return (
            street_number,
            None,
            unit,
            "missing street name",
        )

    return (
        street_number,
        street,
        unit,
        None,
    )


# =========================================================
# NAME HANDLING
# =========================================================

def derive_middle_name(
    first_name: Optional[str],
    last_name: Optional[str],
    full_name: Optional[str],
) -> Optional[str]:

    first_name = clean_text(first_name)
    last_name = clean_text(last_name)
    full_name = clean_text(full_name)

    if not first_name:
        return None

    if not last_name:
        return None

    if not full_name:
        return None

    parts = full_name.split()

    first_parts = first_name.split()
    last_parts = last_name.split()

    if len(parts) <= (
        len(first_parts) + len(last_parts)
    ):
        return None

    if parts[:len(first_parts)] != first_parts:
        return None

    if parts[-len(last_parts):] != last_parts:
        return None

    middle = parts[
        len(first_parts):
        len(parts) - len(last_parts)
    ]

    if not middle:
        return None

    suffixes = {
        "JR",
        "SR",
        "II",
        "III",
        "IV",
        "V",
    }

    for item in middle:
        if item.rstrip(".") in suffixes:
            return None

    return " ".join(middle)


# =========================================================
# DUPLICATE FINGERPRINT
# =========================================================

def make_fingerprint(
    row: Dict[str, Optional[str]]
) -> str:
    """
    Fingerprint represents:

        same person
        +
        same address

    Different people at the same address remain separate.

    Same person at different addresses remains separate.
    """

    fields = [
        row.get("full_name") or "",
        row.get("street_number") or "",
        row.get("street") or "",
        row.get("unit") or "",
        row.get("city") or "",
        row.get("state") or "",
        row.get("zip_code") or "",
    ]

    canonical = "\x1f".join(
        re.sub(
            r"\s+",
            " ",
            str(value).strip().upper(),
        )
        for value in fields
    )

    return hashlib.sha256(
        canonical.encode("utf-8")
    ).hexdigest()


# =========================================================
# ENVIRONMENT FILE
# =========================================================

def load_env_file(path: Path) -> Dict[str, str]:

    env = {}

    if not path.exists():
        raise RuntimeError(
            "Environment file not found: {}".format(
                path
            )
        )

    with path.open(
        "r",
        encoding="utf-8",
    ) as file_handle:

        for raw_line in file_handle:

            line = raw_line.strip()

            if not line:
                continue

            if line.startswith("#"):
                continue

            if "=" not in line:
                continue

            key, value = line.split(
                "=",
                1,
            )

            key = key.strip()
            value = value.strip()

            if (
                len(value) >= 2
                and value[0] == value[-1]
                and value[0] in ("'", '"')
            ):
                value = value[1:-1]

            env[key] = value

    return env


# =========================================================
# DATABASE CONNECTION
# =========================================================

def connect_database(
    env: Dict[str, str]
):

    required = [
        "PD_DB_HOST",
        "PD_DB_PORT",
        "PD_DB_USER",
        "PD_DB_PASSWORD",
        "PD_DB_NAME",
    ]

    missing = [
        key
        for key in required
        if not env.get(key)
    ]

    if missing:
        raise RuntimeError(
            "Missing database environment variables: {}".format(
                ", ".join(missing)
            )
        )

    return mysql.connector.connect(
        host=env["PD_DB_HOST"],
        port=int(env["PD_DB_PORT"]),
        user=env["PD_DB_USER"],
        password=env["PD_DB_PASSWORD"],
        database=env["PD_DB_NAME"],
        autocommit=False,
        charset="utf8mb4",
        use_unicode=True,
    )


# =========================================================
# FIELD LENGTH VALIDATION
# =========================================================

def validate_lengths(
    row: Dict[str, Optional[str]]
) -> Optional[str]:

    problems = []

    for column, limit in COLUMN_LIMITS.items():

        value = row.get(column)

        if (
            value is not None
            and len(value) > limit
        ):
            problems.append(
                "{} length {} exceeds {}".format(
                    column,
                    len(value),
                    limit,
                )
            )

    if problems:
        return "; ".join(problems)

    return None


# =========================================================
# SOURCE ROW TRANSFORMATION
# =========================================================

def transform_source_row(
    source: Dict[str, str]
) -> Tuple[
    Dict[str, Optional[str]],
    Optional[str],
]:

    first_name = clean_text(
        source.get("buyer_first_name")
    )

    last_name = clean_text(
        source.get("buyer_last_name")
    )

    full_name = clean_text(
        source.get("buyer_full_name")
    )

    if not full_name:

        components = [
            value
            for value in [
                first_name,
                last_name,
            ]
            if value
        ]

        if components:
            full_name = " ".join(components)

    middle_name = derive_middle_name(
        first_name,
        last_name,
        full_name,
    )

    (
        street_number,
        street,
        unit,
        address_error,
    ) = parse_address(
        source.get("full_address")
    )

    city = clean_text(
        source.get("city")
    )

    state = clean_text(
        source.get("state")
    )

    zip_code = normalize_zip(
        source.get("zip")
    )

    county = clean_text(
        source.get("county")
    )

    row = {
        "first_name": first_name,
        "middle_name": middle_name,
        "last_name": last_name,
        "full_name": full_name,
        "title": None,
        "street_number": street_number,
        "street": street,
        "unit": unit,
        "city": city,
        "state": state,
        "zip_code": zip_code,
        "county": county,
        "country": "USA",
    }

    errors = []

    if not first_name:
        errors.append(
            "missing first name"
        )

    if not last_name:
        errors.append(
            "missing last name"
        )

    if not full_name:
        errors.append(
            "missing full name"
        )

    raw_address = source.get(
        "full_address",
        "",
    )

    if not str(raw_address).strip():
        errors.append(
            "missing address"
        )

    if address_error:
        errors.append(
            address_error
        )

    if not city:
        errors.append(
            "missing city"
        )

    if not state:
        errors.append(
            "missing state"
        )

    if not zip_code:
        errors.append(
            "missing ZIP"
        )

    elif not re.fullmatch(
        r"\d{5}(?:-\d{4})?",
        zip_code,
    ):
        errors.append(
            "malformed ZIP"
        )

    length_error = validate_lengths(
        row
    )

    if length_error:
        errors.append(
            length_error
        )

    if errors:

        # Remove duplicate error messages
        errors = list(
            dict.fromkeys(errors)
        )

        return (
            row,
            "; ".join(errors),
        )

    return row, None


# =========================================================
# MASTER SCHEMA CHECK
# =========================================================

def verify_master_schema(cursor):

    cursor.execute(
        """
        SELECT
            COLUMN_NAME,
            COLUMN_TYPE
        FROM information_schema.COLUMNS
        WHERE TABLE_SCHEMA = DATABASE()
          AND TABLE_NAME = %s
        ORDER BY ORDINAL_POSITION
        """,
        (MASTER_TABLE,),
    )

    actual = {
        name: column_type
        for name, column_type
        in cursor.fetchall()
    }

    required = (
        {"id"}
        | set(TARGET_COLUMNS)
    )

    missing = (
        required
        - set(actual)
    )

    if missing:
        raise RuntimeError(
            "Master table is missing required columns: {}".format(
                ", ".join(
                    sorted(missing)
                )
            )
        )


# =========================================================
# TEMPORARY STAGING TABLE
# =========================================================

def create_staging_table(cursor):

    cursor.execute(
        """
        DROP TEMPORARY TABLE IF EXISTS
        import_ma_staging
        """
    )

    cursor.execute(
        """
        CREATE TEMPORARY TABLE import_ma_staging (

            source_row BIGINT UNSIGNED NOT NULL,

            fingerprint CHAR(64) NOT NULL,

            first_name VARCHAR(40) NULL,
            middle_name VARCHAR(40) NULL,
            last_name VARCHAR(60) NULL,
            full_name VARCHAR(150) NULL,
            title VARCHAR(20) NULL,

            street_number VARCHAR(20) NULL,
            street VARCHAR(100) NULL,
            unit VARCHAR(30) NULL,

            city VARCHAR(100) NULL,
            state VARCHAR(10) NULL,
            zip_code VARCHAR(20) NULL,
            county VARCHAR(100) NULL,
            country VARCHAR(20) NULL,

            KEY idx_stage_fingerprint (
                fingerprint
            ),

            KEY idx_stage_zip (zip_code)

        )
        ENGINE=InnoDB
        DEFAULT CHARSET=utf8mb4
        COLLATE=utf8mb4_unicode_ci
        """
    )


# =========================================================
# INSERT STAGING BATCH
# =========================================================

def insert_stage_batch(
    cursor,
    batch,
):

    sql = """
        INSERT INTO import_ma_staging (

            source_row,
            fingerprint,

            first_name,
            middle_name,
            last_name,
            full_name,
            title,

            street_number,
            street,
            unit,

            city,
            state,
            zip_code,
            county,
            country

        )
        VALUES (

            %s,
            %s,

            %s,
            %s,
            %s,
            %s,
            %s,

            %s,
            %s,
            %s,

            %s,
            %s,
            %s,
            %s,
            %s

        )
    """

    cursor.executemany(
        sql,
        batch,
    )


# =========================================================
# CREATE DEDUPLICATED STAGING TABLE
# =========================================================

def create_unique_stage_table(cursor):

    cursor.execute(
        """
        DROP TEMPORARY TABLE IF EXISTS
        import_ma_unique
        """
    )

    cursor.execute(
        """
        CREATE TEMPORARY TABLE import_ma_unique AS

        SELECT s.*

        FROM import_ma_staging s

        INNER JOIN (

            SELECT
                fingerprint,
                MIN(source_row) AS keep_row

            FROM import_ma_staging

            GROUP BY fingerprint

        ) duplicates

          ON duplicates.fingerprint =
             s.fingerprint

         AND duplicates.keep_row =
             s.source_row
        """
    )

    # IMPORTANT:
    #
    # Do not create the original very large composite
    # UTF8MB4 index here. It can exceed MariaDB's
    # maximum key length.
    #
    # ZIP + a prefix of full_name is sufficient to
    # narrow duplicate candidates before comparing
    # the complete address.

    cursor.execute(
        """
        ALTER TABLE import_ma_unique

        ADD KEY idx_unique_zip (zip_code)
        """
    )


# =========================================================
# MASTER DUPLICATE CONDITION
# =========================================================

def master_duplicate_condition() -> str:

    return """
        m.full_name = s.full_name

        AND COALESCE(
            m.street_number,
            ''
        ) = COALESCE(
            s.street_number,
            ''
        )

        AND COALESCE(
            m.street,
            ''
        ) = COALESCE(
            s.street,
            ''
        )

        AND COALESCE(
            m.unit,
            ''
        ) = COALESCE(
            s.unit,
            ''
        )

        AND COALESCE(
            m.city,
            ''
        ) = COALESCE(
            s.city,
            ''
        )

        AND COALESCE(
            m.state,
            ''
        ) = COALESCE(
            s.state,
            ''
        )

        AND COALESCE(
            m.zip_code,
            ''
        ) = COALESCE(
            s.zip_code,
            ''
        )
    """


# =========================================================
# DUPLICATE ANALYSIS
# =========================================================

def analyze_duplicates(cursor):

    cursor.execute(
        """
        SELECT COUNT(*)
        FROM import_ma_staging
        """
    )

    valid_stage_rows = (
        cursor.fetchone()[0]
    )

    cursor.execute(
        """
        SELECT
            COALESCE(
                SUM(cnt - 1),
                0
            )

        FROM (

            SELECT
                COUNT(*) AS cnt

            FROM import_ma_staging

            GROUP BY fingerprint

            HAVING COUNT(*) > 1

        ) duplicate_groups
        """
    )

    source_duplicates = int(
        cursor.fetchone()[0]
        or 0
    )

    cursor.execute(
        """
        SELECT COUNT(*)
        FROM import_ma_unique
        """
    )

    unique_source_rows = (
        cursor.fetchone()[0]
    )

    duplicate_sql = """
        SELECT COUNT(*)

        FROM import_ma_unique s

        WHERE EXISTS (

            SELECT 1

            FROM people_addresses m

            WHERE
                m.zip_code = s.zip_code

                AND m.full_name =
                    s.full_name

                AND {}

            LIMIT 1
        )
    """.format(
        master_duplicate_condition()
    )

    cursor.execute(
        duplicate_sql
    )

    existing_duplicates = (
        cursor.fetchone()[0]
    )

    candidate_new = (
        unique_source_rows
        - existing_duplicates
    )

    return {
        "valid_stage_rows":
            valid_stage_rows,

        "source_duplicates":
            source_duplicates,

        "unique_source_rows":
            unique_source_rows,

        "existing_duplicates":
            existing_duplicates,

        "candidate_new":
            candidate_new,
    }


# =========================================================
# PRODUCTION INSERT
# =========================================================

def insert_new_rows(cursor):

    sql = """
        INSERT INTO people_addresses (

            first_name,
            middle_name,
            last_name,
            full_name,
            title,

            street_number,
            street,
            unit,

            city,
            state,
            zip_code,
            county,
            country

        )

        SELECT

            s.first_name,
            s.middle_name,
            s.last_name,
            s.full_name,
            s.title,

            s.street_number,
            s.street,
            s.unit,

            s.city,
            s.state,
            s.zip_code,
            s.county,
            s.country

        FROM import_ma_unique s

        WHERE NOT EXISTS (

            SELECT 1

            FROM people_addresses m

            WHERE
                m.zip_code = s.zip_code

                AND m.full_name =
                    s.full_name

                AND {}

            LIMIT 1
        )
    """.format(
        master_duplicate_condition()
    )

    cursor.execute(sql)

    return cursor.rowcount


# =========================================================
# MAIN
# =========================================================

def main():

    parser = argparse.ArgumentParser(
        description=(
            "Safely merge MA.csv "
            "into Public Directory"
        )
    )

    mode = (
        parser.add_mutually_exclusive_group()
    )

    mode.add_argument(
        "--dry-run",
        action="store_true",
        help=(
            "Analyze only. "
            "This is the default."
        ),
    )

    mode.add_argument(
        "--commit",
        action="store_true",
        help=(
            "Actually insert new records."
        ),
    )

    parser.add_argument(
        "--input",
        required=True,
        help="Path to MA.csv",
    )

    parser.add_argument(
        "--env-file",
        default=str(
            DEFAULT_ENV_FILE
        ),
        help=(
            "Database environment file"
        ),
    )

    parser.add_argument(
        "--batch-size",
        type=int,
        default=5000,
    )

    args = parser.parse_args()

    commit_mode = bool(
        args.commit
    )

    input_path = Path(
        args.input
    )

    if not input_path.exists():

        print(
            "ERROR: input file not found: {}".format(
                input_path
            )
        )

        return 1

    REPORT_DIR.mkdir(
        parents=True,
        exist_ok=True,
    )

    timestamp = (
        datetime.datetime.now().strftime(
            "%Y%m%d-%H%M%S"
        )
    )

    rejected_path = (
        REPORT_DIR
        / "ma-import-rejected-{}.csv".format(
            timestamp
        )
    )

    report_path = (
        REPORT_DIR
        / "ma-import-report-{}.json".format(
            timestamp
        )
    )

    env = load_env_file(
        Path(args.env_file)
    )

    print("=" * 72)

    print(
        "PUBLIC DIRECTORY — MA CSV IMPORT"
    )

    print("=" * 72)

    print(
        "Input: {}".format(
            input_path
        )
    )

    print(
        "Mode: {}".format(
            "COMMIT"
            if commit_mode
            else "DRY RUN"
        )
    )

    print(
        "Database: {}".format(
            env.get("PD_DB_NAME")
        )
    )

    print(
        "Master table: {}".format(
            MASTER_TABLE
        )
    )

    print()

    conn = connect_database(
        env
    )

    cursor = conn.cursor()

    try:

        # ---------------------------------------------
        # Verify production schema
        # ---------------------------------------------

        verify_master_schema(
            cursor
        )

        cursor.execute(
            """
            SELECT COUNT(*)
            FROM people_addresses
            """
        )

        before_count = (
            cursor.fetchone()[0]
        )

        print(
            "Master records before analysis: {:,}".format(
                before_count
            )
        )

        # ---------------------------------------------
        # Temporary staging
        # ---------------------------------------------

        create_staging_table(
            cursor
        )

        counters = {

            "source_rows": 0,
            "valid_rows": 0,
            "rejected_rows": 0,

            "missing_first_name": 0,
            "missing_last_name": 0,
            "missing_address": 0,

            "missing_city": 0,
            "missing_state": 0,
            "missing_zip": 0,

            "malformed_zip": 0,

            "address_parse_failure": 0,
        }

        rejected_fields = [

            "source_row",
            "reason",

            "buyer_first_name",
            "buyer_last_name",
            "buyer_full_name",

            "full_address",

            "city",
            "state",
            "zip",
            "county",
        ]

        # ---------------------------------------------
        # Parse CSV
        # ---------------------------------------------

        with rejected_path.open(
            "w",
            newline="",
            encoding="utf-8",
        ) as rejected_file:

            rejected_writer = (
                csv.DictWriter(
                    rejected_file,
                    fieldnames=rejected_fields,
                )
            )

            rejected_writer.writeheader()

            with input_path.open(
                "r",
                newline="",
                encoding="utf-8-sig",
                errors="replace",
            ) as csv_file:

                reader = csv.DictReader(
                    csv_file
                )

                if not reader.fieldnames:

                    raise RuntimeError(
                        "CSV has no header row"
                    )

                headers = {
                    header.strip()
                    for header
                    in reader.fieldnames
                    if header
                }

                missing_headers = (
                    REQUIRED_SOURCE_COLUMNS
                    - headers
                )

                if missing_headers:

                    raise RuntimeError(
                        "CSV is missing required columns: {}".format(
                            ", ".join(
                                sorted(
                                    missing_headers
                                )
                            )
                        )
                    )

                batch = []

                for (
                    source_row,
                    source,
                ) in enumerate(
                    reader,
                    start=2,
                ):

                    counters[
                        "source_rows"
                    ] += 1

                    (
                        transformed,
                        error,
                    ) = transform_source_row(
                        source
                    )

                    # ---------------------------------
                    # Rejected record
                    # ---------------------------------

                    if error:

                        counters[
                            "rejected_rows"
                        ] += 1

                        lower_error = (
                            error.lower()
                        )

                        if (
                            "missing first name"
                            in lower_error
                        ):
                            counters[
                                "missing_first_name"
                            ] += 1

                        if (
                            "missing last name"
                            in lower_error
                        ):
                            counters[
                                "missing_last_name"
                            ] += 1

                        if (
                            "missing address"
                            in lower_error
                        ):
                            counters[
                                "missing_address"
                            ] += 1

                        if (
                            "missing city"
                            in lower_error
                        ):
                            counters[
                                "missing_city"
                            ] += 1

                        if (
                            "missing state"
                            in lower_error
                        ):
                            counters[
                                "missing_state"
                            ] += 1

                        if (
                            "missing zip"
                            in lower_error
                        ):
                            counters[
                                "missing_zip"
                            ] += 1

                        if (
                            "malformed zip"
                            in lower_error
                        ):
                            counters[
                                "malformed_zip"
                            ] += 1

                        if (
                            "unable to safely parse street number"
                            in lower_error
                        ):
                            counters[
                                "address_parse_failure"
                            ] += 1

                        rejected_writer.writerow(
                            {
                                "source_row":
                                    source_row,

                                "reason":
                                    error,

                                "buyer_first_name":
                                    source.get(
                                        "buyer_first_name",
                                        "",
                                    ),

                                "buyer_last_name":
                                    source.get(
                                        "buyer_last_name",
                                        "",
                                    ),

                                "buyer_full_name":
                                    source.get(
                                        "buyer_full_name",
                                        "",
                                    ),

                                "full_address":
                                    source.get(
                                        "full_address",
                                        "",
                                    ),

                                "city":
                                    source.get(
                                        "city",
                                        "",
                                    ),

                                "state":
                                    source.get(
                                        "state",
                                        "",
                                    ),

                                "zip":
                                    source.get(
                                        "zip",
                                        "",
                                    ),

                                "county":
                                    source.get(
                                        "county",
                                        "",
                                    ),
                            }
                        )

                        continue

                    # ---------------------------------
                    # Valid record
                    # ---------------------------------

                    counters[
                        "valid_rows"
                    ] += 1

                    fingerprint = (
                        make_fingerprint(
                            transformed
                        )
                    )

                    batch.append(
                        (
                            source_row,
                            fingerprint,

                            transformed[
                                "first_name"
                            ],

                            transformed[
                                "middle_name"
                            ],

                            transformed[
                                "last_name"
                            ],

                            transformed[
                                "full_name"
                            ],

                            transformed[
                                "title"
                            ],

                            transformed[
                                "street_number"
                            ],

                            transformed[
                                "street"
                            ],

                            transformed[
                                "unit"
                            ],

                            transformed[
                                "city"
                            ],

                            transformed[
                                "state"
                            ],

                            transformed[
                                "zip_code"
                            ],

                            transformed[
                                "county"
                            ],

                            transformed[
                                "country"
                            ],
                        )
                    )

                    if (
                        len(batch)
                        >= args.batch_size
                    ):

                        insert_stage_batch(
                            cursor,
                            batch,
                        )

                        batch = []

                    if (
                        counters[
                            "source_rows"
                        ]
                        % 10000
                        == 0
                    ):

                        print(
                            "Processed={:,} "
                            "Valid={:,} "
                            "Rejected={:,}".format(

                                counters[
                                    "source_rows"
                                ],

                                counters[
                                    "valid_rows"
                                ],

                                counters[
                                    "rejected_rows"
                                ],
                            )
                        )

                if batch:

                    insert_stage_batch(
                        cursor,
                        batch,
                    )

        # ---------------------------------------------
        # Deduplicate incoming CSV
        # ---------------------------------------------

        create_unique_stage_table(
            cursor
        )

        # ---------------------------------------------
        # Compare with production master
        # ---------------------------------------------

        duplicate_stats = (
            analyze_duplicates(
                cursor
            )
        )

        report = {

            "timestamp":
                timestamp,

            "mode":
                (
                    "commit"
                    if commit_mode
                    else "dry-run"
                ),

            "input_file":
                str(input_path),

            "database":
                env.get(
                    "PD_DB_NAME"
                ),

            "master_table":
                MASTER_TABLE,

            "before_count":
                before_count,

            **counters,

            **duplicate_stats,
        }

        # ---------------------------------------------
        # Display analysis
        # ---------------------------------------------

        print()

        print("=" * 72)
        print("ANALYSIS")
        print("=" * 72)

        print(
            "Source rows:              {:,}".format(
                counters[
                    "source_rows"
                ]
            )
        )

        print(
            "Valid rows:               {:,}".format(
                counters[
                    "valid_rows"
                ]
            )
        )

        print(
            "Rejected rows:            {:,}".format(
                counters[
                    "rejected_rows"
                ]
            )
        )

        print(
            "Duplicates inside CSV:    {:,}".format(
                duplicate_stats[
                    "source_duplicates"
                ]
            )
        )

        print(
            "Already in master:        {:,}".format(
                duplicate_stats[
                    "existing_duplicates"
                ]
            )
        )

        print(
            "Candidate new records:    {:,}".format(
                duplicate_stats[
                    "candidate_new"
                ]
            )
        )

        # ---------------------------------------------
        # DRY RUN
        # ---------------------------------------------

        if not commit_mode:

            print()

            print(
                "DRY RUN COMPLETE."
            )

            print(
                "No records were added "
                "to people_addresses."
            )

            conn.rollback()

            report[
                "inserted_rows"
            ] = 0

            report[
                "after_count"
            ] = before_count

            report[
                "status"
            ] = "dry-run-complete"

        # ---------------------------------------------
        # COMMIT MODE
        # ---------------------------------------------

        else:

            print()

            print(
                "COMMIT MODE: "
                "inserting candidate records..."
            )

            inserted_rows = (
                insert_new_rows(
                    cursor
                )
            )

            conn.commit()

            cursor.execute(
                """
                SELECT COUNT(*)
                FROM people_addresses
                """
            )

            after_count = (
                cursor.fetchone()[0]
            )

            expected_difference = (
                inserted_rows
            )

            actual_difference = (
                after_count
                - before_count
            )

            report[
                "inserted_rows"
            ] = inserted_rows

            report[
                "after_count"
            ] = after_count

            report[
                "expected_difference"
            ] = expected_difference

            report[
                "actual_difference"
            ] = actual_difference

            if (
                actual_difference
                != expected_difference
            ):

                report[
                    "status"
                ] = (
                    "COMMITTED-WITH-"
                    "RECONCILIATION-ERROR"
                )

                print()

                print(
                    "WARNING: "
                    "COUNT RECONCILIATION FAILED."
                )

            else:

                report[
                    "status"
                ] = "committed"

                print()

                print(
                    "Inserted: {:,}".format(
                        inserted_rows
                    )
                )

                print(
                    "Master before: {:,}".format(
                        before_count
                    )
                )

                print(
                    "Master after:  {:,}".format(
                        after_count
                    )
                )

            # Update optimizer statistics.
            # This does NOT rebuild all indexes.

            cursor.execute(
                """
                ANALYZE TABLE
                people_addresses
                """
            )

            cursor.fetchall()

        # ---------------------------------------------
        # Write JSON report
        # ---------------------------------------------

        with report_path.open(
            "w",
            encoding="utf-8",
        ) as report_file:

            json.dump(
                report,
                report_file,
                indent=2,
                sort_keys=True,
            )

        print()

        print(
            "Rejected/review CSV:"
        )

        print(
            rejected_path
        )

        print()

        print(
            "Import report:"
        )

        print(
            report_path
        )

        return 0

    except Exception as exc:

        conn.rollback()

        print()

        print(
            "IMPORT FAILED:"
        )

        print(
            str(exc)
        )

        return 1

    finally:

        cursor.close()
        conn.close()


if __name__ == "__main__":
    sys.exit(
        main()
    )