from __future__ import annotations

from decimal import Decimal
import hashlib
import sqlite3
import tempfile
from pathlib import Path
import unittest

from openpyxl import Workbook

from bank_importer.db import applied_versions, connect, migrate, rollback, utc_now
from bank_importer.importing import import_statement


ROOT = Path(__file__).resolve().parents[1]
SAMPLES = ROOT / "流水模板"
SAMPLE_FILE = SAMPLES / "中国建设银行账户流水.xls"


class PersistenceTestCase(unittest.TestCase):
    def setUp(self) -> None:
        self.temp_dir = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp_dir.cleanup)
        root = Path(self.temp_dir.name)
        self.db_path = root / "app.db"
        self.storage = root / "files"
        self.connection = connect(self.db_path)
        self.addCleanup(self.connection.close)
        migrate(self.connection)

    def import_sample(self) -> object:
        return import_statement(
            self.connection,
            self.storage,
            SAMPLE_FILE.name,
            SAMPLE_FILE.read_bytes(),
        )


class MigrationTests(PersistenceTestCase):
    def test_migrate_creates_schema_and_is_idempotent(self) -> None:
        first = applied_versions(self.connection)
        self.assertEqual([1, 2, 3, 4, 5], first)
        self.assertEqual([], migrate(self.connection))
        self.assertEqual(first, applied_versions(self.connection))
        tables = {
            row["name"]
            for row in self.connection.execute(
                "SELECT name FROM sqlite_master WHERE type = 'table'"
            )
        }
        for table in (
            "companies",
            "bank_accounts",
            "account_aliases",
            "master_data_changes",
            "source_files",
            "import_batches",
            "sheet_batches",
            "sheet_reviews",
            "source_rows",
            "import_exceptions",
            "users",
            "sessions",
            "login_attempts",
            "audit_log",
            "personal_transit_mappings",
            "canonical_transfer_events",
            "transfer_match_decisions",
            "transfer_decision_observations",
            "transfer_decision_participants",
            "transfer_match_candidates",
            "current_transfer_decisions",
            "transfer_observation_claims",
            "schema_migrations",
        ):
            self.assertIn(table, tables)

    def test_rollback_removes_schema_and_forward_rebuilds_it(self) -> None:
        self.assertEqual([5, 4, 3, 2, 1], rollback(self.connection, 0))
        self.assertEqual([], applied_versions(self.connection))
        remaining = self.connection.execute(
            "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'source_rows'"
        ).fetchone()
        self.assertIsNone(remaining)
        self.assertEqual([1, 2, 3, 4, 5], migrate(self.connection))
        self.assertEqual([1, 2, 3, 4, 5], applied_versions(self.connection))

    def test_rollback_to_4_keeps_bank_evidence_and_drops_event_layer(self) -> None:
        self.import_sample()
        row_count = self.connection.execute(
            "SELECT COUNT(*) AS n FROM source_rows"
        ).fetchone()["n"]
        self.assertGreater(row_count, 0)
        self.assertEqual([5], rollback(self.connection, 4))
        # The pre-migration evidence and schema are untouched.
        self.assertEqual(
            row_count,
            self.connection.execute("SELECT COUNT(*) AS n FROM source_rows").fetchone()["n"],
        )
        self.assertEqual(
            "parsed",
            self.connection.execute(
                "SELECT status FROM import_batches LIMIT 1"
            ).fetchone()["status"],
        )
        remaining = self.connection.execute(
            "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'transfer_match_decisions'"
        ).fetchone()
        self.assertIsNone(remaining)

    def test_event_layer_views_exist_after_migration(self) -> None:
        view = self.connection.execute(
            "SELECT name FROM sqlite_master WHERE type = 'view' AND name = 'eligible_intercompany_events'"
        ).fetchone()
        self.assertIsNotNone(view)

    def test_decision_log_immutability_triggers(self) -> None:
        self.import_sample()
        source_row_id = self.connection.execute(
            "SELECT id FROM source_rows LIMIT 1"
        ).fetchone()["id"]
        with self.connection:
            self.connection.execute(
                "INSERT INTO canonical_transfer_events (lifecycle, created_at) VALUES ('active', ?)",
                (utc_now(),),
            )
            event_id = self.connection.execute(
                "SELECT id FROM canonical_transfer_events LIMIT 1"
            ).fetchone()["id"]
            self.connection.execute(
                """
                INSERT INTO transfer_match_decisions (
                    event_id, revision, classification, pairing, mode, rule_version,
                    locked, created_at
                ) VALUES (?, 1, 'unresolved', 'not_applicable', 'auto', 'transfer-match-v1', 0, ?)
                """,
                (event_id, utc_now()),
            )
            decision_id = self.connection.execute(
                "SELECT id FROM transfer_match_decisions LIMIT 1"
            ).fetchone()["id"]
            self.connection.execute(
                """
                INSERT INTO transfer_match_candidates (
                    decision_id, source_row_id, rule_tier, rule_version, created_at
                ) VALUES (?, ?, 'R1', 'transfer-match-v1', ?)
                """,
                (decision_id, source_row_id, utc_now()),
            )
        for statement in (
            "UPDATE transfer_match_decisions SET classification = 'external'",
            "DELETE FROM transfer_match_decisions",
            "UPDATE transfer_match_candidates SET rule_tier = 'M1'",
            "DELETE FROM transfer_match_candidates",
            "DELETE FROM canonical_transfer_events",
        ):
            with self.subTest(statement=statement):
                with self.assertRaises(sqlite3.IntegrityError):
                    self.connection.execute(statement)
                self.connection.rollback()

    def test_observation_claims_source_row_is_unique(self) -> None:
        self.import_sample()
        source_row_id = self.connection.execute(
            "SELECT id FROM source_rows LIMIT 1"
        ).fetchone()["id"]
        with self.connection:
            self.connection.execute(
                "INSERT INTO canonical_transfer_events (lifecycle, created_at) VALUES ('active', ?)",
                (utc_now(),),
            )
            event_id = self.connection.execute(
                "SELECT id FROM canonical_transfer_events LIMIT 1"
            ).fetchone()["id"]
            self.connection.execute(
                """
                INSERT INTO transfer_match_decisions (
                    event_id, revision, classification, pairing, mode, locked, created_at
                ) VALUES (?, 1, 'unresolved', 'not_applicable', 'auto', 0, ?)
                """,
                (event_id, utc_now()),
            )
            decision_id = self.connection.execute(
                "SELECT id FROM transfer_match_decisions LIMIT 1"
            ).fetchone()["id"]
            self.connection.execute(
                """
                INSERT INTO transfer_observation_claims (source_row_id, event_id, decision_id)
                VALUES (?, ?, ?)
                """,
                (source_row_id, event_id, decision_id),
            )
            with self.assertRaises(sqlite3.IntegrityError):
                self.connection.execute(
                    """
                    INSERT INTO transfer_observation_claims (source_row_id, event_id, decision_id)
                    VALUES (?, ?, ?)
                    """,
                    (source_row_id, event_id, decision_id),
                )
            self.connection.rollback()


class ImportPersistenceTests(PersistenceTestCase):
    def test_import_persists_batch_sheets_and_rows(self) -> None:
        result = self.import_sample()
        self.assertEqual("parsed", result.status)

        batch = self.connection.execute(
            "SELECT status FROM import_batches WHERE id = ?", (result.batch_id,)
        ).fetchone()
        self.assertEqual("parsed", batch["status"])

        sheets = self.connection.execute(
            "SELECT * FROM sheet_batches WHERE import_batch_id = ?", (result.batch_id,)
        ).fetchall()
        self.assertEqual(1, len(sheets))
        self.assertEqual("中国建设银行", sheets[0]["bank_name"])
        self.assertEqual("ccb-account-detail-v1", sheets[0]["template_id"])
        self.assertEqual(1, sheets[0]["template_version"])

        rows = self.connection.execute(
            "SELECT * FROM source_rows WHERE sheet_batch_id = ?", (sheets[0]["id"],)
        ).fetchall()
        self.assertEqual(sheets[0]["transaction_count"], len(rows))

    def test_amounts_keep_decimal_precision(self) -> None:
        result = self.import_sample()
        rows = self.connection.execute(
            """
            SELECT income, expense FROM source_rows
            WHERE sheet_batch_id IN (
                SELECT id FROM sheet_batches WHERE import_batch_id = ?
            )
            """,
            (result.batch_id,),
        ).fetchall()
        self.assertTrue(rows)
        for row in rows:
            self.assertEqual(row["income"], str(Decimal(row["income"])))
            self.assertEqual(row["expense"], str(Decimal(row["expense"])))

    def test_every_row_traces_back_to_file_sheet_and_template(self) -> None:
        result = self.import_sample()
        row = self.connection.execute(
            """
            SELECT r.source_row, s.sheet_name, s.template_id, s.template_version, f.sha256
            FROM source_rows r
            JOIN sheet_batches s ON s.id = r.sheet_batch_id
            JOIN import_batches b ON b.id = s.import_batch_id
            JOIN source_files f ON f.id = b.source_file_id
            WHERE b.id = ?
            LIMIT 1
            """,
            (result.batch_id,),
        ).fetchone()
        self.assertIsNotNone(row)
        self.assertEqual(
            hashlib.sha256(SAMPLE_FILE.read_bytes()).hexdigest(), row["sha256"]
        )
        self.assertEqual("ccb-account-detail-v1", row["template_id"])
        self.assertEqual(1, row["template_version"])
        self.assertGreater(row["source_row"], 0)
        self.assertTrue(row["sheet_name"])

    def test_source_file_is_stored_immutably_by_content_hash(self) -> None:
        self.import_sample()
        stored = self.connection.execute(
            "SELECT storage_path FROM source_files"
        ).fetchone()
        stored_path = Path(stored["storage_path"])
        self.assertTrue(stored_path.is_file())
        self.assertEqual(SAMPLE_FILE.read_bytes(), stored_path.read_bytes())
        self.assertIn(hashlib.sha256(SAMPLE_FILE.read_bytes()).hexdigest(), stored_path.name)

    def test_data_survives_reconnect(self) -> None:
        result = self.import_sample()
        self.connection.close()

        reopened = connect(self.db_path)
        self.addCleanup(reopened.close)
        batch = reopened.execute(
            "SELECT status FROM import_batches WHERE id = ?", (result.batch_id,)
        ).fetchone()
        self.assertEqual("parsed", batch["status"])
        files = reopened.execute("SELECT COUNT(*) AS n FROM source_files").fetchone()
        self.assertEqual(1, files["n"])
        rows = reopened.execute("SELECT COUNT(*) AS n FROM source_rows").fetchone()
        self.assertGreater(rows["n"], 0)


class IdempotencyTests(PersistenceTestCase):
    def test_duplicate_upload_creates_no_second_facts(self) -> None:
        first = self.import_sample()
        second = self.import_sample()

        self.assertEqual("parsed", first.status)
        self.assertEqual("duplicate", second.status)
        self.assertEqual(first.batch_id, second.batch_id)
        self.assertTrue(second.duplicate_same_company)

        files = self.connection.execute("SELECT COUNT(*) AS n FROM source_files").fetchone()
        self.assertEqual(1, files["n"])
        sheets = self.connection.execute("SELECT COUNT(*) AS n FROM sheet_batches").fetchone()
        self.assertEqual(1, sheets["n"])
        rows = self.connection.execute("SELECT COUNT(*) AS n FROM source_rows").fetchone()
        self.assertEqual(first.batches[0].transactions.__len__(), rows["n"])

        duplicate = self.connection.execute(
            "SELECT status, duplicate_of_id FROM import_batches WHERE status = 'duplicate'"
        ).fetchone()
        self.assertEqual(first.batch_id, duplicate["duplicate_of_id"])

    def test_cross_company_duplicate_never_returns_other_companys_batch(self) -> None:
        now = utc_now()
        with self.connection:
            self.connection.execute(
                "INSERT INTO companies (name, created_at, updated_at) VALUES ('甲公司', ?, ?)",
                (now, now),
            )
            self.connection.execute(
                "INSERT INTO companies (name, created_at, updated_at) VALUES ('乙公司', ?, ?)",
                (now, now),
            )
        company_a, company_b = [
            row["id"]
            for row in self.connection.execute("SELECT id FROM companies ORDER BY id").fetchall()
        ]

        first = import_statement(
            self.connection, self.storage, SAMPLE_FILE.name,
            SAMPLE_FILE.read_bytes(), company_id=company_a,
        )
        self.assertEqual("parsed", first.status)

        second = import_statement(
            self.connection, self.storage, SAMPLE_FILE.name,
            SAMPLE_FILE.read_bytes(), company_id=company_b,
        )
        self.assertEqual("duplicate", second.status)
        self.assertFalse(second.duplicate_same_company)
        # The returned batch id is the uploader's own duplicate batch, never
        # the other company's original batch.
        self.assertNotEqual(first.batch_id, second.batch_id)
        self.assertNotEqual(second.batch_id, first.batch_id)
        own = self.connection.execute(
            "SELECT company_id FROM import_batches WHERE id = ?", (second.batch_id,)
        ).fetchone()
        self.assertEqual(company_b, own["company_id"])

    def test_repeated_upload_under_a_different_filename_is_still_duplicate(self) -> None:
        first = self.import_sample()
        second = import_statement(
            self.connection,
            self.storage,
            "改名后的流水.xls",
            SAMPLE_FILE.read_bytes(),
        )
        self.assertEqual("duplicate", second.status)
        self.assertEqual(first.batch_id, second.batch_id)
        files = self.connection.execute("SELECT COUNT(*) AS n FROM source_files").fetchone()
        self.assertEqual(1, files["n"])

    def test_immutability_triggers_block_updates_and_deletes(self) -> None:
        self.import_sample()
        for statement in (
            "UPDATE source_rows SET income = '0'",
            "DELETE FROM source_rows",
            "UPDATE source_files SET sha256 = 'x'",
            "DELETE FROM source_files",
            "UPDATE sheet_batches SET bank_name = 'x'",
            "DELETE FROM sheet_batches",
        ):
            with self.subTest(statement=statement):
                with self.assertRaises(sqlite3.IntegrityError):
                    self.connection.execute(statement)
                self.connection.rollback()

    def test_row_level_unique_constraint_blocks_duplicate_rows(self) -> None:
        self.import_sample()
        row = self.connection.execute(
            "SELECT sheet_batch_id, source_row FROM source_rows LIMIT 1"
        ).fetchone()
        with self.assertRaises(sqlite3.IntegrityError):
            self.connection.execute(
                """
                INSERT INTO source_rows (
                    sheet_batch_id, source_row, transaction_at, income, expense, created_at
                ) VALUES (?, ?, '2026-01-01T00:00:00', '1', '0', '2026-01-01T00:00:00Z')
                """,
                (row["sheet_batch_id"], row["source_row"]),
            )
        self.connection.rollback()


class FailedImportTests(PersistenceTestCase):
    def _unknown_template_bytes(self) -> bytes:
        workbook = Workbook()
        sheet = workbook.active
        sheet.title = "流水"
        sheet.append(["日期", "金额", "备注"])
        sheet.append(["2026-01-01", "100.00", "测试"])
        target = Path(self.temp_dir.name) / "unknown.xlsx"
        workbook.save(target)
        return target.read_bytes()

    def test_parse_failure_keeps_exception_batch_without_rows(self) -> None:
        result = import_statement(
            self.connection, self.storage, "未知银行.xlsx", self._unknown_template_bytes()
        )
        self.assertEqual("exception", result.status)
        self.assertIsNotNone(result.message)
        self.assertIn("未知银行.xlsx", result.message)
        self.assertNotIn(result.sha256, result.message)

        batch = self.connection.execute(
            "SELECT status FROM import_batches WHERE id = ?", (result.batch_id,)
        ).fetchone()
        self.assertEqual("exception", batch["status"])

        exceptions = self.connection.execute(
            "SELECT * FROM import_exceptions WHERE import_batch_id = ?",
            (result.batch_id,),
        ).fetchall()
        self.assertEqual(1, len(exceptions))
        self.assertEqual("parse", exceptions[0]["stage"])

        rows = self.connection.execute("SELECT COUNT(*) AS n FROM source_rows").fetchone()
        self.assertEqual(0, rows["n"])

        # The source file is still preserved as evidence for later diagnosis.
        files = self.connection.execute("SELECT COUNT(*) AS n FROM source_files").fetchone()
        self.assertEqual(1, files["n"])

    def test_reupload_after_failure_is_tracked_as_duplicate(self) -> None:
        content = self._unknown_template_bytes()
        first = import_statement(self.connection, self.storage, "未知银行.xlsx", content)
        second = import_statement(self.connection, self.storage, "未知银行.xlsx", content)
        self.assertEqual("exception", first.status)
        self.assertEqual("duplicate", second.status)
        self.assertEqual(first.batch_id, second.batch_id)


if __name__ == "__main__":
    unittest.main()