#!/usr/bin/env python3
"""Inventory and import Telegram Lite desktop exports.

The live Telegram collector belongs to QazPipe. This script is a bounded
archive-recovery tool for old local Telegram Lite exports that may be deleted
after their text is safely present in Echo/QazLake.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import io
import json
import os
import re
import sys
import zipfile
from collections import Counter, defaultdict
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable

try:
    import psycopg2
    import psycopg2.extras
except ModuleNotFoundError:  # pragma: no cover - optional for inventory-only runs
    psycopg2 = None  # type: ignore[assignment]

from bs4 import BeautifulSoup

try:
    from lxml import html as lxml_html
except ModuleNotFoundError:  # pragma: no cover - production import-only mode can run without lxml
    lxml_html = None

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_SOURCE_ROOT = Path("/Users/belilovsky/Downloads/Telegram Lite")
DEFAULT_REPORT = ROOT / "reports" / "telegram_lite_archive_intake.json"
DEFAULT_STATE = ROOT / "reports" / "telegram_lite_archive_intake_state.json"

MESSAGE_FILE_RE = re.compile(r"(^|/)messages(?:\d+)?\.html$", re.IGNORECASE)
MESSAGE_ID_RE = re.compile(r"^message(-?\d+)$")
UTC_SUFFIX_RE = re.compile(r"\s+UTC[+-]\d{2}:\d{2}$")


@dataclass(frozen=True)
class ExportUnit:
    source_kind: str
    source_path: str
    export_root: str
    files: tuple[str, ...]
    fingerprint: str


@dataclass
class ArchiveRow:
    channel_name: str
    message_id: int | None
    date: datetime | None
    from_name: str | None
    text: str
    post_type: str
    is_repost: bool
    source_path: str


def _sha256_text(value: str) -> str:
    return hashlib.sha256(value.encode("utf-8", errors="replace")).hexdigest()


def _clip(value: str | None, limit: int) -> str | None:
    if value is None:
        return None
    return str(value).strip()[:limit] or None


def _clean_text(value: str) -> str:
    return re.sub(r"\s+", " ", value or "").strip()


def parse_datetime(value: str | None) -> datetime | None:
    """Parse Telegram export date while preserving the displayed local time."""
    if not value:
        return None
    cleaned = UTC_SUFFIX_RE.sub("", value.strip())
    for pattern in ("%d.%m.%Y %H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d"):
        try:
            return datetime.strptime(cleaned[:19], pattern)
        except ValueError:
            continue
    try:
        return datetime.fromisoformat(cleaned)
    except ValueError:
        return None


def _read_state(path: Path) -> set[str]:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (FileNotFoundError, json.JSONDecodeError):
        return set()
    if isinstance(data, dict):
        processed = data.get("processed_fingerprints", [])
        if isinstance(processed, list):
            return {str(item) for item in processed}
    return set()


def _write_state(path: Path, processed: set[str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = {
        "updated_at": datetime.utcnow().isoformat(timespec="seconds") + "Z",
        "processed_fingerprints": sorted(processed),
    }
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def _resolve_target_dsn(target: str, explicit_dsn: str) -> str:
    if explicit_dsn:
        return explicit_dsn.replace("postgresql+asyncpg://", "postgresql://")
    if target == "echo":
        try:
            from scripts._dsn import resolve_db_url
        except ModuleNotFoundError:  # pragma: no cover
            from _dsn import resolve_db_url  # type: ignore

        return resolve_db_url("ECHO_DB_URL").replace("postgresql+asyncpg://", "postgresql://")
    if target == "qazlake":
        for key in ("QAZPIPE_QAZLAKE_DB_URL", "QAZLAKE_DB_URL", "QAZLAKE_DSN", "QAZLAKE_WRITER_DB_URL"):
            value = os.getenv(key, "")
            if value and "echo_sounder" not in value:
                return value.replace("postgresql+asyncpg://", "postgresql://")
    raise RuntimeError("target DSN is not configured")


def _folder_fingerprint(path: Path, files: Iterable[Path]) -> str:
    parts = [str(path.resolve())]
    for file_path in sorted(files):
        stat = file_path.stat()
        parts.append(f"{file_path.name}:{stat.st_size}:{int(stat.st_mtime)}")
    return _sha256_text("|".join(parts))


def _zip_fingerprint(zip_path: Path, root: str, files: Iterable[str]) -> str:
    stat = zip_path.stat()
    parts = [str(zip_path.resolve()), str(stat.st_size), str(int(stat.st_mtime)), root]
    parts.extend(sorted(files))
    return _sha256_text("|".join(parts))


def discover_exports(source_root: Path) -> list[ExportUnit]:
    """Return Telegram export units from folders and zip archives."""
    units: list[ExportUnit] = []

    for directory in sorted(path for path in source_root.iterdir() if path.is_dir()):
        files = sorted(
            path
            for path in directory.iterdir()
            if path.is_file() and (path.name == "result.json" or MESSAGE_FILE_RE.search(path.name))
        )
        if files:
            units.append(
                ExportUnit(
                    source_kind="folder",
                    source_path=str(directory),
                    export_root=directory.name,
                    files=tuple(path.name for path in files),
                    fingerprint=_folder_fingerprint(directory, files),
                )
            )

    for zip_path in sorted(source_root.glob("*.zip")):
        try:
            with zipfile.ZipFile(zip_path) as archive:
                grouped: dict[str, list[str]] = defaultdict(list)
                for info in archive.infolist():
                    name = info.filename
                    if name.startswith("__MACOSX/") or name.endswith("/"):
                        continue
                    if name.endswith("result.json") or MESSAGE_FILE_RE.search(name):
                        root = str(Path(name).parent)
                        grouped[root].append(name)
                for root, files in sorted(grouped.items()):
                    units.append(
                        ExportUnit(
                            source_kind="zip",
                            source_path=str(zip_path),
                            export_root=root,
                            files=tuple(sorted(files)),
                            fingerprint=_zip_fingerprint(zip_path, root, files),
                        )
                    )
        except zipfile.BadZipFile:
            continue

    return units


def _extract_channel_name(soup: BeautifulSoup, fallback: str) -> str:
    header = soup.select_one(".page_header .text.bold")
    if header:
        value = _clean_text(header.get_text(" ", strip=True))
        if value:
            return value
    title = soup.find("title")
    if title:
        value = _clean_text(title.get_text(" ", strip=True))
        if value and value.lower() != "exported data":
            return value
    return fallback


def _extract_message_id(div: Any) -> int | None:
    raw = str(div.get("id") or "")
    match = MESSAGE_ID_RE.match(raw)
    if not match:
        return None
    value = int(match.group(1))
    return value if value > 0 else None


def parse_html_payload(payload: bytes, *, fallback_channel: str, source_path: str) -> list[ArchiveRow]:
    try:
        return parse_html_payload_lxml(payload, fallback_channel=fallback_channel, source_path=source_path)
    except Exception:
        pass

    soup = BeautifulSoup(payload, "html.parser")
    channel_name = _extract_channel_name(soup, fallback_channel)
    rows: list[ArchiveRow] = []
    for div in soup.find_all("div", class_="message"):
        classes = {str(item) for item in div.get("class", [])}
        if "service" in classes:
            continue

        date_title = None
        date_node = div.find("div", class_=lambda value: value and "date" in str(value).split())
        if date_node and date_node.has_attr("title"):
            date_title = date_node["title"]
        text_node = div.find("div", class_="text")
        text = _clean_text(text_node.get_text(" ", strip=True) if text_node else "")
        if not text:
            continue

        from_node = div.find("div", class_="from_name")
        forwarded = div.find("div", class_=lambda value: value and "forwarded" in str(value).split()) is not None
        rows.append(
            ArchiveRow(
                channel_name=_clip(channel_name, 100) or fallback_channel[:100],
                message_id=_extract_message_id(div),
                date=parse_datetime(str(date_title) if date_title else None),
                from_name=_clip(_clean_text(from_node.get_text(" ", strip=True)) if from_node else "", 100),
                text=text,
                post_type="telegram_lite_archive",
                is_repost=forwarded,
                source_path=source_path,
            )
        )
    return rows


def _has_class_xpath(class_name: str) -> str:
    return f"contains(concat(' ', normalize-space(@class), ' '), ' {class_name} ')"


def _first_text(node: Any, xpath: str) -> str:
    matches = node.xpath(xpath)
    if not matches:
        return ""
    value = matches[0]
    if isinstance(value, str):
        return _clean_text(value)
    return _clean_text(value.text_content())


def parse_html_payload_lxml(payload: bytes, *, fallback_channel: str, source_path: str) -> list[ArchiveRow]:
    """Fast Telegram Lite HTML parser for large multi-page exports."""
    if lxml_html is None:
        raise RuntimeError("lxml is not installed")
    doc = lxml_html.fromstring(payload)
    channel_name = _first_text(
        doc,
        f"//*[{_has_class_xpath('page_header')}]//*[contains(concat(' ', normalize-space(@class), ' '), ' text ')"
        " and contains(concat(' ', normalize-space(@class), ' '), ' bold ')]",
    )
    if not channel_name:
        channel_name = _first_text(doc, "//title")
    if not channel_name or channel_name.lower() == "exported data":
        channel_name = fallback_channel

    rows: list[ArchiveRow] = []
    message_nodes = doc.xpath(f"//*[{_has_class_xpath('message')}]")
    for node in message_nodes:
        classes = set(str(node.get("class") or "").split())
        if "service" in classes:
            continue
        text = _first_text(node, f".//*[{_has_class_xpath('text')}]")
        if not text:
            continue
        date_matches = node.xpath(f".//*[{_has_class_xpath('date')}]/@title")
        from_name = _first_text(node, f".//*[{_has_class_xpath('from_name')}]")
        forwarded = bool(node.xpath(f".//*[{_has_class_xpath('forwarded')}]"))
        rows.append(
            ArchiveRow(
                channel_name=_clip(channel_name, 100) or fallback_channel[:100],
                message_id=_extract_message_id(node),
                date=parse_datetime(str(date_matches[0]) if date_matches else None),
                from_name=_clip(from_name, 100),
                text=text,
                post_type="telegram_lite_archive",
                is_repost=forwarded,
                source_path=source_path,
            )
        )
    return rows


def _json_text(value: Any) -> str:
    if isinstance(value, str):
        return value
    if isinstance(value, list):
        parts: list[str] = []
        for item in value:
            if isinstance(item, str):
                parts.append(item)
            elif isinstance(item, dict):
                parts.append(str(item.get("text") or ""))
        return "".join(parts)
    return ""


def parse_result_json(payload: bytes, *, fallback_channel: str, source_path: str) -> list[ArchiveRow]:
    data = json.loads(payload.decode("utf-8"))
    channel_name = _clip(str(data.get("name") or fallback_channel), 100) or fallback_channel[:100]
    rows: list[ArchiveRow] = []
    for message in data.get("messages", []):
        text = _clean_text(_json_text(message.get("text")))
        if not text:
            continue
        rows.append(
            ArchiveRow(
                channel_name=channel_name,
                message_id=message.get("id") if isinstance(message.get("id"), int) else None,
                date=parse_datetime(str(message.get("date") or "")),
                from_name=_clip(str(message.get("from") or ""), 100),
                text=text,
                post_type=str(message.get("type") or "telegram_lite_archive")[:50],
                is_repost=bool(message.get("forwarded_from") or message.get("forwarded_from_id")),
                source_path=source_path,
            )
        )
    return rows


def parse_export(unit: ExportUnit) -> list[ArchiveRow]:
    rows: list[ArchiveRow] = []
    fallback = Path(unit.export_root).name or Path(unit.source_path).stem
    if unit.source_kind == "folder":
        base = Path(unit.source_path)
        for name in sorted(unit.files):
            payload = (base / name).read_bytes()
            source = f"{unit.source_path}/{name}"
            if name.endswith("result.json"):
                rows.extend(parse_result_json(payload, fallback_channel=fallback, source_path=source))
            else:
                rows.extend(parse_html_payload(payload, fallback_channel=fallback, source_path=source))
        return rows

    with zipfile.ZipFile(unit.source_path) as archive:
        for name in sorted(unit.files):
            payload = archive.read(name)
            source = f"{unit.source_path}!/{name}"
            if name.endswith("result.json"):
                rows.extend(parse_result_json(payload, fallback_channel=fallback, source_path=source))
            else:
                rows.extend(parse_html_payload(payload, fallback_channel=fallback, source_path=source))
    return rows


def _row_identity(row: ArchiveRow) -> tuple[str, int | None, str, str]:
    date_value = row.date.isoformat(sep=" ") if row.date else ""
    return (row.channel_name, row.message_id, date_value, hashlib.md5(row.text.encode("utf-8")).hexdigest())


def dedupe_rows(rows: Iterable[ArchiveRow]) -> list[ArchiveRow]:
    seen: set[tuple[str, int | None, str, str]] = set()
    unique: list[ArchiveRow] = []
    for row in rows:
        identity = _row_identity(row)
        if identity in seen:
            continue
        seen.add(identity)
        unique.append(row)
    return unique


def row_to_json(row: ArchiveRow) -> dict[str, Any]:
    return {
        "channel_name": row.channel_name,
        "message_id": row.message_id,
        "date": row.date.isoformat(sep=" ") if row.date else None,
        "from_name": row.from_name,
        "text": row.text,
        "post_type": row.post_type,
        "is_repost": row.is_repost,
        "source_path": row.source_path,
    }


def row_from_json(data: dict[str, Any]) -> ArchiveRow:
    date_value = parse_datetime(str(data.get("date") or "")) if data.get("date") else None
    return ArchiveRow(
        channel_name=_clip(str(data.get("channel_name") or ""), 100) or "unknown",
        message_id=data.get("message_id") if isinstance(data.get("message_id"), int) else None,
        date=date_value,
        from_name=_clip(str(data.get("from_name") or ""), 100),
        text=str(data.get("text") or "").strip(),
        post_type=_clip(str(data.get("post_type") or "telegram_lite_archive"), 50) or "telegram_lite_archive",
        is_repost=bool(data.get("is_repost") or False),
        source_path=str(data.get("source_path") or ""),
    )


def write_jsonl(path: Path, rows: Iterable[ArchiveRow]) -> int:
    path.parent.mkdir(parents=True, exist_ok=True)
    count = 0
    with path.open("w", encoding="utf-8") as fh:
        for row in rows:
            fh.write(json.dumps(row_to_json(row), ensure_ascii=False) + "\n")
            count += 1
    return count


def read_jsonl(path: Path) -> list[ArchiveRow]:
    rows: list[ArchiveRow] = []
    if str(path) == "-":
        for line in sys.stdin:
            if not line.strip():
                continue
            rows.append(row_from_json(json.loads(line)))
        return dedupe_rows(row for row in rows if row.text)
    with path.open("r", encoding="utf-8") as fh:
        for line in fh:
            if not line.strip():
                continue
            rows.append(row_from_json(json.loads(line)))
    return dedupe_rows(row for row in rows if row.text)


def _connect(dsn: str):
    if psycopg2 is None:
        raise RuntimeError("psycopg2 is required for database comparison or import")
    return psycopg2.connect(dsn)


def ensure_channels(conn: Any, channel_names: Iterable[str]) -> None:
    names = sorted({name for name in channel_names if name})
    if not names:
        return
    with conn.cursor() as cur:
        try:
            psycopg2.extras.execute_values(
                cur,
                "INSERT INTO channel (channel_name) VALUES %s ON CONFLICT (channel_name) DO NOTHING",
                [(name,) for name in names],
                page_size=500,
            )
        except Exception:
            conn.rollback()
            return
    conn.commit()


def insert_rows(conn: Any, rows: list[ArchiveRow], *, batch_size: int = 1_000, target: str = "echo") -> int:
    if not rows:
        return 0
    values = [
        (
            row.channel_name,
            row.message_id,
            row.date,
            row.from_name,
            row.text,
            row.post_type,
            row.is_repost,
        )
        for row in rows
        if row.date is not None and row.text
    ]
    if not values:
        return 0
    inserted = insert_rows_staged(conn, values, batch_size=batch_size, target=target)
    return inserted


def insert_rows_staged(conn: Any, values: list[tuple[Any, ...]], *, batch_size: int = 10_000, target: str = "echo") -> int:
    inserted = 0
    for idx in range(0, len(values), batch_size):
        batch = values[idx : idx + batch_size]
        with conn.cursor() as cur:
            cur.execute(
                """
                CREATE TEMP TABLE _telegram_lite_archive_stage (
                    channel_name TEXT,
                    message_id BIGINT,
                    date TIMESTAMP,
                    from_name TEXT,
                    text TEXT,
                    post_type TEXT,
                    is_repost BOOLEAN
                ) ON COMMIT DROP
                """
            )
            buffer = io.StringIO()
            writer = csv.writer(buffer)
            for row in batch:
                writer.writerow(row)
            buffer.seek(0)
            cur.copy_expert(
                """
                COPY _telegram_lite_archive_stage
                    (channel_name, message_id, date, from_name, text, post_type, is_repost)
                FROM STDIN WITH CSV
                """,
                buffer,
            )
            if target == "qazlake":
                inserted += _insert_stage_qazlake(cur)
            else:
                inserted += _insert_stage_echo(cur)
        conn.commit()
    return inserted


def _insert_stage_echo(cur: Any) -> int:
    cur.execute(
        """
        INSERT INTO telegram_messages
            (channel_name, message_id, date, from_name, text, post_type, is_repost)
        SELECT
            channel_name::varchar(100),
            message_id,
            date,
            left(COALESCE(from_name, ''), 100)::varchar(100),
            text,
            left(COALESCE(post_type, 'telegram_lite_archive'), 50)::varchar(50),
            COALESCE(is_repost, false)
        FROM _telegram_lite_archive_stage
        WHERE NULLIF(text, '') IS NOT NULL
          AND date IS NOT NULL
          AND message_id IS NOT NULL
        ON CONFLICT DO NOTHING
        RETURNING id
        """
    )
    inserted = cur.rowcount
    cur.execute(
        """
        INSERT INTO telegram_messages
            (channel_name, message_id, date, from_name, text, post_type, is_repost)
        SELECT
            channel_name::varchar(100),
            NULL,
            date,
            left(COALESCE(from_name, ''), 100)::varchar(100),
            text,
            left(COALESCE(post_type, 'telegram_lite_archive'), 50)::varchar(50),
            COALESCE(is_repost, false)
        FROM _telegram_lite_archive_stage
        WHERE NULLIF(text, '') IS NOT NULL
          AND date IS NOT NULL
          AND message_id IS NULL
        ON CONFLICT DO NOTHING
        RETURNING id
        """
    )
    return inserted + cur.rowcount


def _insert_stage_qazlake(cur: Any) -> int:
    cur.execute(
        """
        INSERT INTO telegram_messages
            (channel_name, message_id, date, from_name, text, post_type, is_repost, imported_at)
        SELECT
            stage.channel_name::varchar(255),
            stage.message_id,
            stage.date,
            left(COALESCE(stage.from_name, ''), 255)::varchar(255),
            stage.text,
            left(COALESCE(stage.post_type, 'telegram_lite_archive'), 50)::varchar(50),
            COALESCE(stage.is_repost, false),
            NOW()
        FROM _telegram_lite_archive_stage stage
        WHERE NULLIF(stage.text, '') IS NOT NULL
          AND stage.date IS NOT NULL
          AND stage.message_id IS NOT NULL
          AND NOT EXISTS (
              SELECT 1
              FROM telegram_messages existing
              WHERE existing.channel_name = stage.channel_name
                AND existing.message_id = stage.message_id
          )
        RETURNING id
        """
    )
    inserted = cur.rowcount
    cur.execute(
        """
        INSERT INTO telegram_messages
            (channel_name, message_id, date, from_name, text, post_type, is_repost, imported_at)
        SELECT
            stage.channel_name::varchar(255),
            NULL,
            stage.date,
            left(COALESCE(stage.from_name, ''), 255)::varchar(255),
            stage.text,
            left(COALESCE(stage.post_type, 'telegram_lite_archive'), 50)::varchar(50),
            COALESCE(stage.is_repost, false),
            NOW()
        FROM _telegram_lite_archive_stage stage
        WHERE NULLIF(stage.text, '') IS NOT NULL
          AND stage.date IS NOT NULL
          AND stage.message_id IS NULL
          AND NOT EXISTS (
              SELECT 1
              FROM telegram_messages existing
              WHERE existing.channel_name = stage.channel_name
                AND existing.date = stage.date
                AND md5(existing.text) = md5(stage.text)
          )
        RETURNING id
        """
    )
    return inserted + cur.rowcount


def target_summary(conn: Any, channel_names: Iterable[str]) -> dict[str, Any]:
    names = sorted({name for name in channel_names if name})
    if not names:
        return {"channels_seen": 0, "messages_seen": 0}
    with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
        cur.execute(
            """
            SELECT channel_name, count(*)::bigint AS messages
            FROM telegram_messages
            WHERE channel_name = ANY(%s)
            GROUP BY channel_name
            """,
            (names,),
        )
        rows = list(cur.fetchall())
    return {
        "channels_seen": len(rows),
        "messages_seen": int(sum(row["messages"] for row in rows)),
        "top_channels": sorted(
            ({"channel_name": row["channel_name"], "messages": int(row["messages"])} for row in rows),
            key=lambda item: item["messages"],
            reverse=True,
        )[:25],
    }


def build_report(units: list[ExportUnit], rows: list[ArchiveRow], *, skipped_processed: int) -> dict[str, Any]:
    channel_counts = Counter(row.channel_name for row in rows)
    dated = sum(1 for row in rows if row.date is not None)
    return {
        "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "exports_scanned": len(units),
        "exports_skipped_by_state": skipped_processed,
        "messages_parsed": len(rows),
        "messages_with_date": dated,
        "channels": len(channel_counts),
        "top_channels": [
            {"channel_name": name, "messages": count} for name, count in channel_counts.most_common(40)
        ],
        "exports": [asdict(unit) for unit in units],
    }


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Inventory/import Telegram Lite archive exports")
    parser.add_argument("--source-root", default=str(DEFAULT_SOURCE_ROOT), help="Telegram Lite export directory")
    parser.add_argument("--report", default=str(DEFAULT_REPORT), help="JSON report path")
    parser.add_argument("--state-file", default=str(DEFAULT_STATE), help="Resume state path")
    parser.add_argument("--target-dsn", default=os.getenv("TELEGRAM_LITE_ARCHIVE_TARGET_DSN", ""))
    parser.add_argument("--target", choices=("echo", "qazlake"), default="echo")
    parser.add_argument("--dump-jsonl", default="", help="Write parsed rows to JSONL for remote import")
    parser.add_argument("--input-jsonl", default="", help="Read rows from JSONL instead of parsing source-root")
    parser.add_argument("--apply", action="store_true", help="Insert missing rows into target-dsn")
    parser.add_argument("--rescan-processed", action="store_true", help="Ignore resume state")
    parser.add_argument("--limit-exports", type=int, default=0, help="Process at most N exports this run")
    parser.add_argument("--offset", type=int, default=0, help="Skip first N discovered exports before state filtering")
    parser.add_argument("--skip-target-summary", action="store_true", help="Avoid count queries around import")
    parser.add_argument("--import-batch-size", type=int, default=1000, help="Rows per committed database stage")
    parser.add_argument("--no-channel-upsert", action="store_true", help="Do not upsert channel names into channel table")
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    source_root = Path(args.source_root).expanduser()
    report_path = Path(args.report)
    state_path = Path(args.state_file)
    if args.input_jsonl:
        selected: list[ExportUnit] = []
        skipped_processed = 0
        rows = read_jsonl(Path(args.input_jsonl))
    elif not source_root.exists():
        raise SystemExit(f"source root does not exist: {source_root}")
    else:
        processed = _read_state(state_path)
        discovered = discover_exports(source_root)
        selected = discovered[args.offset :] if args.offset else discovered
        skipped_processed = 0
        if not args.rescan_processed:
            before = len(selected)
            selected = [unit for unit in selected if unit.fingerprint not in processed]
            skipped_processed = before - len(selected)
        if args.limit_exports and args.limit_exports > 0:
            selected = selected[: args.limit_exports]

        rows = []
        for unit in selected:
            rows.extend(parse_export(unit))
        rows = dedupe_rows(rows)

    report = build_report(selected, rows, skipped_processed=skipped_processed)
    if args.dump_jsonl:
        report["dump_jsonl"] = args.dump_jsonl
        report["dumped_rows"] = write_jsonl(Path(args.dump_jsonl), rows)

    if args.target_dsn or args.apply:
        target_dsn = _resolve_target_dsn(args.target, args.target_dsn)
        with _connect(target_dsn) as conn:
            if not args.skip_target_summary:
                report["target_before"] = target_summary(conn, (row.channel_name for row in rows))
            if args.apply:
                if args.target == "echo" and not args.no_channel_upsert:
                    ensure_channels(conn, (row.channel_name for row in rows))
                report["inserted"] = insert_rows(
                    conn,
                    rows,
                    batch_size=max(100, args.import_batch_size),
                    target=args.target,
                )
                if not args.input_jsonl:
                    processed = _read_state(state_path)
                    processed.update(unit.fingerprint for unit in selected)
                    _write_state(state_path, processed)
                if not args.skip_target_summary:
                    report["target_after"] = target_summary(conn, (row.channel_name for row in rows))
            else:
                report["inserted"] = 0
    else:
        report["inserted"] = 0

    report_path.parent.mkdir(parents=True, exist_ok=True)
    report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(
        "telegram_lite_archive_intake "
        f"exports={report['exports_scanned']} messages={report['messages_parsed']} "
        f"channels={report['channels']} inserted={report['inserted']} report={report_path}"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))