조회 결과를 fetchmany로 나누어 저장하고 모든 ID를 검산하기

"""한 SQLite 조회 커서를 fetchmany로 끝까지 읽고 모든 저장 ID를 검산한다.

Python 3.11+, 표준 라이브러리만 사용. 입력 SQL은 코드의 고정 상수다.
ID 검산은 전체 ID를 list/set에 쌓지 않고 같은 읽기 스냅샷과 스트리밍 비교한다.
"""
import argparse
import csv
import hashlib
import io
import json
from pathlib import Path
import platform
import signal
import sqlite3
import sys

COLUMNS = ("id", "item", "quantity")
SELECT_SQL = "SELECT id, item, quantity FROM records ORDER BY id"
ID_SQL = "SELECT id FROM records ORDER BY id"
COUNT_SQL = "SELECT COUNT(*), MIN(id), MAX(id) FROM records"
FAILURE_EVIDENCE_BYTES = 16384
HARD_LIMITS = {
    "batch_size": 1000,
    "max_rows": 1000000,
    "max_parts": 1000,
    "max_output_bytes": 50 * 1024 * 1024,
    "max_row_bytes": 65536,
}
MAX_SOURCE_BYTES = 64 * 1024 * 1024


class ExportError(Exception):
    def __init__(self, code, message):
        super().__init__(message)
        self.code = code


class SignalInterruption(Exception):
    def __init__(self, signum):
        self.signum = signum
        super().__init__(f"signal {signum}")


def sha256_file(path):
    digest = hashlib.sha256()
    with Path(path).open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def check_options(options):
    for key, maximum in HARD_LIMITS.items():
        value = options[key]
        if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= maximum:
            raise ExportError("invalid_option", f"{key} must be an integer in 1..{maximum}")
    if options["max_row_bytes"] < 32:
        raise ExportError("invalid_option", "max_row_bytes must be at least 32")


def open_readonly(database, max_row_bytes):
    path = Path(database).resolve(strict=True)
    # as_uri는 공백, ?, # 등을 인코딩한다. mode=ro이면 없는 DB를 만들지 않는다.
    connect_options = {"uri": True, "isolation_level": None, "timeout": 5.0}
    if hasattr(sqlite3, "LEGACY_TRANSACTION_CONTROL"):
        # 3.12+에서는 isolation_level 모드를 명시한다. 3.11에는 이 인자가 없다.
        connect_options["autocommit"] = sqlite3.LEGACY_TRANSACTION_CONTROL
    conn = sqlite3.connect(path.as_uri() + "?mode=ro", **connect_options)
    try:
        # 읽기 전용 URI와 함께 적용하는 방어. 이 설정은 원본 파일을 수정하지 않는다.
        conn.execute("PRAGMA query_only=ON")
        # SQLite가 한 값/행을 무제한으로 materialize하지 못하게 한다.
        # 스키마 문자열/레코드 오버헤드 여유와 CSV 행 상한은 구분한다.
        conn.setlimit(sqlite3.SQLITE_LIMIT_LENGTH, max(max_row_bytes + 4096, 8192))
        conn.setlimit(sqlite3.SQLITE_LIMIT_COLUMN, 32)
        conn.setlimit(sqlite3.SQLITE_LIMIT_SQL_LENGTH, 8192)
        return conn
    except BaseException:
        conn.close()
        raise


class OutputBudget:
    """CSV + 로그 + 성공 manifest 실제 UTF-8 바이트를 함께 센다."""
    def __init__(self, limit):
        self.limit = limit
        self.used = 0

    def write(self, handle, data):
        if isinstance(data, str):
            data = data.encode("utf-8")
        if self.used + len(data) > self.limit:
            raise ExportError("output_limit", f"output would exceed {self.limit} bytes")
        handle.write(data)
        self.used += len(data)


def csv_line(values):
    buffer = io.StringIO(newline="")
    csv.writer(buffer).writerow(values)
    return buffer.getvalue().encode("utf-8")


def check_source_schema(conn):
    schema_cursor = conn.execute("PRAGMA table_info(records)")
    try:
        fields = schema_cursor.fetchall()
    finally:
        schema_cursor.close()
    if len(fields) != 3 or tuple(f[1] for f in fields) != COLUMNS:
        raise ExportError("source_schema", "records must have exactly id,item,quantity columns")
    if fields[0][2].upper() != "INTEGER" or fields[0][5] != 1:
        raise ExportError("source_schema", "id must be the unique INTEGER PRIMARY KEY")
    if any(f[5] for f in fields[1:]):
        raise ExportError("source_schema", "id must be the sole primary key")


def validate_saved_ids(conn, paths, expected, max_row_bytes):
    """파일별 ID를 같은 스냅샷의 별도 ID 커서와 한 행씩 비교한다.

    COUNT/MIN/MAX는 보조 검산이다. 임의 ID 집합의 동등성을 이것만으로 주장하지 않는다.
    커서 행을 fetchone으로 소비하므로 ID 비교 상태는 O(1)이다.
    """
    reference = conn.execute(ID_SQL)
    previous = None
    minimum = None
    count = 0
    old_field_limit = csv.field_size_limit()
    csv.field_size_limit(max_row_bytes)
    try:
        for path in paths:
            with Path(path).open("r", encoding="utf-8-sig", newline="") as handle:
                reader = csv.reader(handle)
                if next(reader, None) != list(COLUMNS):
                    raise ExportError("csv_header", f"unexpected CSV header: {Path(path).name}")
                for row in reader:
                    if len(row) != len(COLUMNS):
                        raise ExportError("csv_columns", f"unexpected CSV column count: {Path(path).name}")
                    if len(csv_line(row)) > max_row_bytes:
                        raise ExportError("row_bytes_limit", "saved CSV row exceeds configured byte limit")
                    try:
                        current = int(row[0])
                    except ValueError as exc:
                        raise ExportError("id_type", f"non-integer saved ID at row {count + 1}") from exc
                    if str(current) != row[0]:
                        raise ExportError("id_type", f"non-canonical saved ID at row {count + 1}")
                    if previous is not None and current <= previous:
                        raise ExportError("id_order", f"duplicate or out-of-order ID at row {count + 1}: {current}")
                    wanted = reference.fetchone()
                    if wanted is None or type(wanted[0]) is not int or current != wanted[0]:
                        raise ExportError("id_mismatch", f"source/output ID mismatch at row {count + 1}: {current}")
                    if minimum is None:
                        minimum = current
                    previous = current
                    count += 1
                    if count > expected["count"]:
                        raise ExportError("id_count", "saved rows exceed snapshot count")
        if reference.fetchone() is not None:
            raise ExportError("id_missing", "source IDs remain after the last saved CSV row")
        if (count, minimum, previous) != (expected["count"], expected["min_id"], expected["max_id"]):
            raise ExportError("id_count", "saved count/min/max differ from the snapshot")
        return {"strictly_increasing": True, "independent_stream_equal": True, "count": count, "min_id": minimum, "max_id": previous}
    except csv.Error as exc:
        raise ExportError("csv_error", str(exc)) from exc
    finally:
        csv.field_size_limit(old_field_limit)
        reference.close()


def write_failure(output, error, before, after, exported, parts, used):
    # 실패 보고는 정상 출력 예산과 별도로 최대 16 KiB만 허용한다.
    # 디렉터리를 새로 만든 실행만 사용하고, existing failure.json은 절대 덮어쓰지 않는다.
    report = {
        "status": "failed", "code": error.code, "message": str(error)[:2000],
        "source_sha256_before": before, "source_sha256_after": after,
        "source_hash_scope": "main_database_file_only", "exported_rows_written": exported,
        "completed_parts": len(parts), "last_completed_files": [p["file"] for p in parts[-10:]],
        "normal_output_bytes": used,
        "warning": "CSV files are failure evidence, not a verified complete export. A current file may be partial. Use a new output directory when retrying.",
    }
    payload = (json.dumps(report, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
    if len(payload) > FAILURE_EVIDENCE_BYTES:
        payload = (json.dumps({"status": "failed", "code": error.code, "message": "failure evidence truncated"}) + "\n").encode()
    with (output / "failure.json").open("xb") as handle:
        handle.write(payload)


def export_database(database, output, *, batch_size=10, max_rows=100000,
                    max_parts=1000, max_output_bytes=1024 * 1024, max_row_bytes=65536,
                    on_snapshot=None, on_batch=None, on_finished_read=None):
    """실행 API. on_*는 결정적 검수용 콜백이며 CLI에는 노출하지 않는다."""
    options = {"batch_size": batch_size, "max_rows": max_rows, "max_parts": max_parts,
               "max_output_bytes": max_output_bytes, "max_row_bytes": max_row_bytes}
    check_options(options)
    if sys.version_info < (3, 11):
        raise ExportError("python_version", "Python 3.11 or later is required")
    database = Path(database)
    output = Path(output)
    if not database.is_file():
        raise ExportError("source_missing", "source database must be an existing regular file")
    database = database.resolve(strict=True)
    source_bytes = database.stat().st_size
    for suffix in ("-wal", "-shm"):
        sidecar = Path(str(database) + suffix)
        if sidecar.exists():
            source_bytes += sidecar.stat().st_size
    if source_bytes > MAX_SOURCE_BYTES:
        raise ExportError("source_limit", f"database and WAL/SHM exceed {MAX_SOURCE_BYTES} bytes")
    if output.exists() or output.is_symlink():
        raise ExportError("output_exists", "output path already exists; choose a new directory")
    if not output.parent.is_dir():
        raise ExportError("output_parent", "output parent directory must already exist")
    try:
        output.mkdir(exist_ok=False)
    except FileExistsError as exc:
        raise ExportError("output_exists", "output path was created concurrently; choose a new directory") from exc
    except OSError as exc:
        raise ExportError("io_error", str(exc)) from exc
    budget = OutputBudget(max_output_bytes)
    before = None
    after = None
    parts = []
    exported = 0
    conn = None
    log = None
    try:
        before = sha256_file(database)
        log = (output / "run.log").open("xb")
        budget.write(log, "READ_ONLY mode=ro transaction=BEGIN query=ORDER_BY_UNIQUE_ID\n")
        conn = open_readonly(database, max_row_bytes)
        conn.execute("BEGIN")
        # BEGIN만으로는 스냅샷이 고정되지 않는다. 이 첫 SELECT가 읽기를 시작한다.
        count_cursor = conn.execute(COUNT_SQL)
        try:
            count, min_id, max_id = count_cursor.fetchone()
        finally:
            count_cursor.close()
        check_source_schema(conn)
        snapshot = {"count": count, "min_id": min_id, "max_id": max_id}
        budget.write(log, f"SNAPSHOT count={count} min_id={min_id} max_id={max_id}\n")
        if count > max_rows:
            raise ExportError("row_limit", f"snapshot has {count} rows; max_rows={max_rows}")
        required_parts = (count + batch_size - 1) // batch_size
        if required_parts > max_parts:
            raise ExportError("part_limit", f"snapshot requires {required_parts} CSV files; max_parts={max_parts}")
        if on_snapshot is not None:
            on_snapshot(dict(snapshot))
        # 이 SELECT는 단 한 번 실행한다. LIMIT/OFFSET 페이지 조회가 아니다.
        cursor = conn.execute(SELECT_SQL)
        previous = None
        while True:
            rows = cursor.fetchmany(batch_size)
            # 결과가 소진되면 빈 list. 마지막 짧은 배치는 반드시 저장한다.
            if not rows:
                break
            index = len(parts) + 1
            if index > max_parts or exported + len(rows) > max_rows:
                raise ExportError("resource_limit", "stream exceeded configured row/part limit")
            filename = f"part_{index:04d}.csv"
            path = output / filename
            first_id = None
            last_id = None
            with path.open("xb") as handle:
                budget.write(handle, b"\xef\xbb\xbf")
                budget.write(handle, csv_line(COLUMNS))
                for row in rows:
                    if len(row) != 3 or type(row[0]) is not int:
                        raise ExportError("id_type", "source id must be an integer")
                    ident = row[0]
                    if previous is not None and ident <= previous:
                        raise ExportError("id_order", "source IDs must be strictly increasing")
                    if type(row[1]) is not str or type(row[2]) is not int:
                        raise ExportError("source_type", "item must be TEXT and quantity must be INTEGER")
                    # CSV 수식/Excel 실행 보안 대신 합성 item만 제공한다. 실제 데이터 사용 시 따로 검토한다.
                    line = csv_line(row)
                    if len(line) > max_row_bytes:
                        raise ExportError("row_bytes_limit", f"CSV row for id={ident} exceeds {max_row_bytes} bytes")
                    budget.write(handle, line)
                    if first_id is None:
                        first_id = ident
                    last_id = previous = ident
                    exported += 1
            parts.append({"file": filename, "rows": len(rows), "first_id": first_id,
                          "last_id": last_id, "bytes": path.stat().st_size, "sha256": sha256_file(path)})
            budget.write(log, f"BATCH {index:04d} rows={len(rows)} first_id={first_id} last_id={last_id}\n")
            log.flush()
            if on_batch is not None:
                on_batch(index, path)
        cursor.close()
        id_validation = validate_saved_ids(conn, (output / p["file"] for p in parts), snapshot, max_row_bytes)
        if exported != count:
            raise ExportError("id_count", "exported rows differ from snapshot count")
        for part in parts:
            path = output / part["file"]
            if path.stat().st_size != part["bytes"] or sha256_file(path) != part["sha256"]:
                raise ExportError("output_changed", f"CSV changed after it was written: {part['file']}")
        budget.write(log, f"ID_COMPARE OK count={exported} min_id={min_id} max_id={max_id}\n")
        conn.rollback()  # 읽기 스냅샷 종료. 파일/레코드를 쓰지 않았다.
        conn.close()
        conn = None
        if on_finished_read is not None:
            on_finished_read()
        after = sha256_file(database)
        if before != after:
            raise ExportError("source_changed", "main database file SHA-256 changed during the run")
        budget.write(log, "SOURCE_SHA256 UNCHANGED scope=MAIN_DATABASE_FILE_ONLY\n")
        budget.write(log, f"READ_AND_ID_VERIFY_COMPLETE rows={exported} parts={len(parts)}\n")
        log.close()
        log = None
        report = {
            "status": "success", "query": SELECT_SQL, "batch_size": batch_size,
            "snapshot": snapshot, "exported_rows": exported, "parts": parts,
            "id_validation": id_validation,
            "source_sha256_before": before, "source_sha256_after": after,
            "source_hash_scope": "main_database_file_only",
            "source_hash_warning": "Main-file equality does not prove no concurrent database writer. WAL commits can change logical data without changing this file; this export is checked against its own read transaction snapshot.",
            "resource_limits": options, "normal_output_bytes_before_manifest": budget.used,
            "log_sha256": sha256_file(output / "run.log"),
            "versions": {"python": platform.python_version(), "sqlite": sqlite3.sqlite_version, "system": platform.platform()},
        }
        payload = (json.dumps(report, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
        # 예산 초과라면 성공 marker 자체도 만들지 않는다.
        if budget.used + len(payload) > budget.limit:
            raise ExportError("output_limit", "manifest would exceed the combined output byte limit")
        with (output / "manifest.json").open("xb") as handle:
            budget.write(handle, payload)
        return report
    except (Exception, KeyboardInterrupt) as exc:
        if conn is not None:
            try:
                conn.rollback()
            finally:
                conn.close()
            conn = None
        if log is not None:
            log.close()
            log = None
        try:
            after = sha256_file(database)
        except OSError:
            after = None
        if isinstance(exc, ExportError):
            error = exc
        elif isinstance(exc, (KeyboardInterrupt, SignalInterruption)):
            error = ExportError("interrupted", "export interrupted; retained files are failure evidence")
        elif isinstance(exc, sqlite3.Error):
            error = ExportError("sqlite_error", str(exc))
        elif isinstance(exc, OSError):
            error = ExportError("io_error", str(exc))
        else:
            error = ExportError("unexpected_error", str(exc))
        try:
            write_failure(output, error, before, after, exported, parts, budget.used)
        except OSError as evidence_error:
            print(f"WARNING failure report could not be written: {evidence_error}", file=sys.stderr)
        raise error from exc


def main():
    parser = argparse.ArgumentParser(description="조회 결과를 fetchmany로 나누어 저장하고 모든 ID를 검산합니다.")
    parser.add_argument("--database", type=Path, default=Path("data/source.sqlite3"))
    parser.add_argument("--output", type=Path, required=True, help="아직 존재하지 않는 출력 디렉터리")
    parser.add_argument("--batch-size", type=int, default=10)
    parser.add_argument("--max-rows", type=int, default=100000)
    parser.add_argument("--max-parts", type=int, default=1000)
    parser.add_argument("--max-output-bytes", type=int, default=1024 * 1024)
    parser.add_argument("--max-row-bytes", type=int, default=65536)
    args = parser.parse_args()
    old_term = signal.getsignal(signal.SIGTERM)
    def stop(signum, frame):
        raise SignalInterruption(signum)
    signal.signal(signal.SIGTERM, stop)
    try:
        report = export_database(**vars(args))
    except (ExportError, OSError) as exc:
        code = exc.code if isinstance(exc, ExportError) else "io_error"
        print(f"ERROR {code}: {exc}", file=sys.stderr)
        return 130 if code == "interrupted" else 2
    finally:
        signal.signal(signal.SIGTERM, old_term)
    print(f"SUCCESS rows={report['exported_rows']} parts={len(report['parts'])} id_compare=OK source_sha=UNCHANGED")
    print(f"OUTPUT {args.output}")
    return 0


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

"""두 검수 완료 출력의 전체 ID 순서를 bounded-memory로 비교한다."""
import argparse
import csv
import json
from pathlib import Path
import re
import sys

from export_batches import COLUMNS, ExportError, HARD_LIMITS, csv_line, sha256_file

MAX_MANIFEST_BYTES = 1024 * 1024


def load_manifest(directory):
    directory = Path(directory)
    path = directory / "manifest.json"
    if (directory / "failure.json").exists() or not path.is_file():
        raise ExportError("compare_incomplete", "both inputs need success manifest.json and no failure.json")
    if path.stat().st_size > MAX_MANIFEST_BYTES:
        raise ExportError("compare_manifest", "manifest exceeds 1 MiB")
    try:
        report = json.loads(path.read_text(encoding="utf-8"))
    except (ValueError, UnicodeError, RecursionError) as exc:
        raise ExportError("compare_manifest", "invalid manifest JSON") from exc
    if not isinstance(report, dict) or report.get("status") != "success":
        raise ExportError("compare_manifest", "manifest status must be success")
    parts = report.get("parts")
    snapshot = report.get("snapshot")
    if not isinstance(parts, list) or len(parts) > HARD_LIMITS["max_parts"] or not isinstance(snapshot, dict):
        raise ExportError("compare_manifest", "invalid or excessive declared parts/snapshot")
    count = snapshot.get("count")
    if type(count) is not int or not 0 <= count <= HARD_LIMITS["max_rows"]:
        raise ExportError("compare_manifest", "invalid snapshot count")
    if count == 0:
        if snapshot.get("min_id") is not None or snapshot.get("max_id") is not None or parts:
            raise ExportError("compare_manifest", "empty snapshot must have no parts/min/max")
    elif type(snapshot.get("min_id")) is not int or type(snapshot.get("max_id")) is not int:
        raise ExportError("compare_manifest", "snapshot min/max must be integers")
    validation = report.get("id_validation", {})
    if not isinstance(validation, dict) or validation.get("independent_stream_equal") is not True:
        raise ExportError("compare_manifest", "original snapshot ID validation is required")
    if report.get("exported_rows") != count:
        raise ExportError("compare_manifest", "exported_rows differs from snapshot")
    total_rows = 0
    total_bytes = 0
    for index, part in enumerate(parts, 1):
        if not isinstance(part, dict) or part.get("file") != f"part_{index:04d}.csv":
            raise ExportError("compare_manifest", "declared files must be the numbered local CSV sequence")
        rows = part.get("rows")
        size = part.get("bytes")
        digest = part.get("sha256")
        if type(rows) is not int or not 1 <= rows <= HARD_LIMITS["batch_size"]:
            raise ExportError("compare_manifest", "invalid part row count")
        if type(size) is not int or size < 1 or not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", digest):
            raise ExportError("compare_manifest", "invalid part bytes/SHA-256")
        total_rows += rows
        total_bytes += size
        if total_rows > HARD_LIMITS["max_rows"] or total_bytes > HARD_LIMITS["max_output_bytes"]:
            raise ExportError("compare_manifest", "declared input rows/bytes exceed hard limits")
        csv_path = directory / part["file"]
        if not csv_path.is_file() or csv_path.is_symlink() or csv_path.stat().st_size != size:
            raise ExportError("compare_hash", "CSV file is missing, symlinked, or size differs")
        if sha256_file(csv_path) != digest:
            raise ExportError("compare_hash", "CSV SHA-256 differs from success manifest")
    if total_rows != count:
        raise ExportError("compare_manifest", "declared part rows differ from snapshot count")
    return report


def iter_ids(directory, report):
    """manifest에 적힌 순서를 쓴다. glob 결과 순서나 전체 ID 목록을 믿지 않는다."""
    previous = None
    count = 0
    for part in report["parts"]:
        rows_in_part = 0
        with (Path(directory) / part["file"]).open(encoding="utf-8-sig", newline="") as handle:
            reader = csv.reader(handle)
            if next(reader, None) != list(COLUMNS):
                raise ExportError("compare_csv", "invalid CSV header")
            for row in reader:
                if len(row) != len(COLUMNS) or len(csv_line(row)) > HARD_LIMITS["max_row_bytes"]:
                    raise ExportError("compare_csv", "invalid or excessive CSV row")
                try:
                    ident = int(row[0])
                except ValueError as exc:
                    raise ExportError("compare_csv", "saved ID must be an integer") from exc
                if str(ident) != row[0] or previous is not None and ident <= previous:
                    raise ExportError("compare_csv", "IDs must be canonical and strictly increasing")
                if rows_in_part == 0 and ident != part.get("first_id"):
                    raise ExportError("compare_manifest", "first ID differs from declared part")
                previous = ident
                rows_in_part += 1
                count += 1
                if count > report["snapshot"]["count"] or rows_in_part > part["rows"]:
                    raise ExportError("compare_csv", "CSV exceeds declared row count")
                yield ident
        if rows_in_part != part["rows"] or previous != part.get("last_id"):
            raise ExportError("compare_manifest", "part count/last ID differs from declared part")
    if count != report["snapshot"]["count"]:
        raise ExportError("compare_csv", "CSV count differs from snapshot")


def compare_exports(first, second):
    left_report = load_manifest(first)
    right_report = load_manifest(second)
    if left_report["snapshot"] != right_report["snapshot"]:
        raise ExportError("compare_snapshot", "snapshot count/min/max differ between exports")
    # 두 커서 ID를 한 개씩만 유지한다. 해시가 아닌 실제 ID 순서 동등성 검사다.
    old_limit = csv.field_size_limit()
    csv.field_size_limit(HARD_LIMITS["max_row_bytes"])
    left = iter_ids(first, left_report)
    right = iter_ids(second, right_report)
    sentinel = object()
    count = 0
    minimum = None
    maximum = None
    try:
        while True:
            a = next(left, sentinel)
            b = next(right, sentinel)
            if a is sentinel and b is sentinel:
                break
            if a is sentinel or b is sentinel or a != b:
                raise ExportError("compare_ids", f"ID sequences differ at row {count + 1}")
            if minimum is None:
                minimum = a
            maximum = a
            count += 1
        snapshot = left_report["snapshot"]
        if (count, minimum, maximum) != (snapshot["count"], snapshot["min_id"], snapshot["max_id"]):
            raise ExportError("compare_ids", "actual IDs differ from declared snapshot count/min/max")
        return {"count": count, "min_id": minimum, "max_id": maximum}
    except csv.Error as exc:
        raise ExportError("compare_csv", str(exc)) from exc
    finally:
        left.close()
        right.close()
        csv.field_size_limit(old_limit)


def main():
    parser = argparse.ArgumentParser(description="두 검수 완료 출력의 모든 ID를 한 행씩 비교합니다.")
    parser.add_argument("--first", type=Path, required=True)
    parser.add_argument("--second", type=Path, required=True)
    args = parser.parse_args()
    try:
        result = compare_exports(args.first, args.second)
    except (ExportError, OSError) as exc:
        code = exc.code if isinstance(exc, ExportError) else "io_error"
        print(f"ERROR {code}: {exc}", file=sys.stderr)
        return 2
    print(f"MATCH count={result['count']} min_id={result['min_id']} max_id={result['max_id']} all_ids_equal=YES")
    return 0


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