Tg Cli

TG_API_ID=123456
TG_API_HASH=your_telegram_app_hash
TG_SESSION_NAME=tg_cli

name: Bug Report
description: Report a bug or unexpected behavior
labels: ["bug"]
body:
  - type: input
    id: version
    attributes:
      label: Version
      description: "Run `tg --version` or `pip show kabi-tg-cli | grep Version`"
      placeholder: "e.g. 1.0.0"
    validations:
      required: true
  - type: dropdown
    id: os
    attributes:
      label: Operating System
      options:
        - macOS
        - Linux
        - Windows
        - Other
    validations:
      required: true
  - type: textarea
    id: description
    attributes:
      label: What happened?
      description: Describe the bug clearly.
    validations:
      required: true
  - type: textarea
    id: expected
    attributes:
      label: Expected behavior
      description: What did you expect to happen?
    validations:
      required: true
  - type: textarea
    id: reproduce
    attributes:
      label: Steps to reproduce
      description: "Commands or steps to reproduce the issue. Use `tg -v <command>` for debug output."
      render: bash
  - type: textarea
    id: logs
    attributes:
      label: Error output / logs
      description: Paste any error messages or verbose output here.
      render: text

name: Feature Request
description: Suggest a new feature or improvement
labels: ["enhancement"]
body:
  - type: input
    id: version
    attributes:
      label: Current version
      description: "Run `tg --version` or `pip show kabi-tg-cli | grep Version`"
      placeholder: "e.g. 1.0.0"
    validations:
      required: false
  - type: textarea
    id: description
    attributes:
      label: Describe the feature
      description: What would you like to see added or changed?
    validations:
      required: true
  - type: textarea
    id: use_case
    attributes:
      label: Use case
      description: Why do you need this feature? How would you use it?

name: CI

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]
  workflow_call:

jobs:
  lint-and-test:
    name: Lint and test (Python ${{ matrix.python-version }})
    runs-on: ubuntu-latest
    strategy:
      fail-fast: false
      matrix:
        python-version: ["3.10", "3.12", "3.14"]
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}

      - name: Install uv
        uses: astral-sh/setup-uv@v6

      - name: Install dependencies
        run: uv sync --group dev

      - name: Run ruff
        run: uv run ruff check .

      - name: Run tests
        run: uv run python -m pytest -q

  build:
    name: Build package
    runs-on: ubuntu-latest
    needs: lint-and-test
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install uv
        uses: astral-sh/setup-uv@v6

      - name: Install dependencies
        run: uv sync --group dev

      - name: Build distribution
        run: uv build

      - name: Check distribution metadata
        run: uv run twine check dist/*

name: Publish to PyPI

# Trigger: push a version tag like v0.2.0
on:
  push:
    tags:
      - "v*"

jobs:
  verify:
    uses: ./.github/workflows/ci.yml

  publish:
    name: Publish to PyPI
    needs: verify
    runs-on: ubuntu-latest
    environment: pypi
    permissions:
      contents: read
      id-token: write  # Required for Trusted Publisher (OIDC)
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install uv
        uses: astral-sh/setup-uv@v6

      - name: Build package
        run: uv build

      - name: Publish to PyPI
        uses: pypa/gh-action-pypi-publish@release/v1

__pycache__/
*.pyc
*.pyo
*.egg-info/
dist/
build/
.venv/
.env
*.session
*.session-journal
*.session-shm
*.session-wal
data/
.pytest_cache/

[Unit]
Description=tg-cli refresh local Telegram cache
After=network-online.target
Wants=network-online.target

[Service]
Type=oneshot
WorkingDirectory=/path/to/tg-cli
Environment=TG_API_ID=123456
Environment=TG_API_HASH=your_telegram_app_hash
ExecStart=/path/to/uv run tg refresh --yaml

[Unit]
Description=Run tg-cli refresh every 15 minutes

[Timer]
OnBootSec=2m
OnUnitActiveSec=15m
Unit=tg-refresh.service

[Install]
WantedBy=timers.target

# Refresh the local Telegram cache every 15 minutes.
# Adjust the path to uv and the working directory for your environment.
*/15 * * * * cd /path/to/tg-cli && /path/to/uv run tg refresh --yaml >> /tmp/tg-refresh.log 2>&1

Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/

TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION

1. Definitions.

"License" shall mean the terms and conditions for use, reproduction, and
distribution as defined by Sections 1 through 9 of this document.

"Licensor" shall mean the copyright owner or entity authorized by the copyright
owner that is granting the License.

"Legal Entity" shall mean the union of the acting entity and all other entities
that control, are controlled by, or are under common control with that entity.
For the purposes of this definition, "control" means (i) the power, direct or
indirect, to cause the direction or management of such entity, whether by
contract or otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.

"You" (or "Your") shall mean an individual or Legal Entity exercising
permissions granted by this License.

"Source" form shall mean the preferred form for making modifications, including
but not limited to software source code, documentation source, and configuration
files.

"Object" form shall mean any form resulting from mechanical transformation or
translation of a Source form, including but not limited to compiled object code,
generated documentation, and conversions to other media types.

"Work" shall mean the work of authorship, whether in Source or Object form,
made available under the License, as indicated by a copyright notice that is
included in or attached to the work (an example is provided in the Appendix
below).

"Derivative Works" shall mean any work, whether in Source or Object form, that
is based on (or derived from) the Work and for which the editorial revisions,
annotations, elaborations, or other modifications represent, as a whole, an
original work of authorship. For the purposes of this License, Derivative Works
shall not include works that remain separable from, or merely link (or bind by
name) to the interfaces of, the Work and Derivative Works thereof.

"Contribution" shall mean any work of authorship, including the original version
of the Work and any modifications or additions to that Work or Derivative Works
thereof, that is intentionally submitted to Licensor for inclusion in the Work
by the copyright owner or by an individual or Legal Entity authorized to submit
on behalf of the copyright owner. For the purposes of this definition,
"submitted" means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems, and
issue tracking systems that are managed by, or on behalf of, the Licensor for
the purpose of discussing and improving the Work, but excluding communication
that is conspicuously marked or otherwise designated in writing by the copyright
owner as "Not a Contribution."

"Contributor" shall mean Licensor and any individual or Legal Entity on behalf
of whom a Contribution has been received by Licensor and subsequently
incorporated within the Work.

2. Grant of Copyright License. Subject to the terms and conditions of this
License, each Contributor hereby grants to You a perpetual, worldwide,
non-exclusive, no-charge, royalty-free, irrevocable copyright license to
reproduce, prepare Derivative Works of, publicly display, publicly perform,
sublicense, and distribute the Work and such Derivative Works in Source or
Object form.

3. Grant of Patent License. Subject to the terms and conditions of this License,
each Contributor hereby grants to You a perpetual, worldwide, non-exclusive,
no-charge, royalty-free, irrevocable (except as stated in this section) patent
license to make, have made, use, offer to sell, sell, import, and otherwise
transfer the Work, where such license applies only to those patent claims
licensable by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s) with the Work
to which such Contribution(s) was submitted. If You institute patent litigation
against any entity (including a cross-claim or counterclaim in a lawsuit)
alleging that the Work or a Contribution incorporated within the Work constitutes
direct or contributory patent infringement, then any patent licenses granted to
You under this License for that Work shall terminate as of the date such
litigation is filed.

4. Redistribution. You may reproduce and distribute copies of the Work or
Derivative Works thereof in any medium, with or without modifications, and in
Source or Object form, provided that You meet the following conditions:

(a) You must give any other recipients of the Work or Derivative Works a copy of
this License; and

(b) You must cause any modified files to carry prominent notices stating that
You changed the files; and

(c) You must retain, in the Source form of any Derivative Works that You
distribute, all copyright, patent, trademark, and attribution notices from the
Source form of the Work, excluding those notices that do not pertain to any part
of the Derivative Works; and

(d) If the Work includes a "NOTICE" text file as part of its distribution, then
any Derivative Works that You distribute must include a readable copy of the
attribution notices contained within such NOTICE file, excluding those notices
that do not pertain to any part of the Derivative Works, in at least one of the
following places: within a NOTICE text file distributed as part of the
Derivative Works; within the Source form or documentation, if provided along
with the Derivative Works; or, within a display generated by the Derivative
Works, if and wherever such third-party notices normally appear. The contents of
the NOTICE file are for informational purposes only and do not modify the
License. You may add Your own attribution notices within Derivative Works that
You distribute, alongside or as an addendum to the NOTICE text from the Work,
provided that such additional attribution notices cannot be construed as
modifying the License.

You may add Your own copyright statement to Your modifications and may provide
additional or different license terms and conditions for use, reproduction, or
distribution of Your modifications, or for any such Derivative Works as a whole,
provided Your use, reproduction, and distribution of the Work otherwise complies
with the conditions stated in this License.

5. Submission of Contributions. Unless You explicitly state otherwise, any
Contribution intentionally submitted for inclusion in the Work by You to the
Licensor shall be under the terms and conditions of this License, without any
additional terms or conditions. Notwithstanding the above, nothing herein shall
supersede or modify the terms of any separate license agreement you may have
executed with Licensor regarding such Contributions.

6. Trademarks. This License does not grant permission to use the trade names,
trademarks, service marks, or product names of the Licensor, except as required
for reasonable and customary use in describing the origin of the Work and
reproducing the content of the NOTICE file.

7. Disclaimer of Warranty. Unless required by applicable law or agreed to in
writing, Licensor provides the Work (and each Contributor provides its
Contributions) on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied, including, without limitation, any warranties
or conditions of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any risks
associated with Your exercise of permissions under this License.

8. Limitation of Liability. In no event and under no legal theory, whether in
tort (including negligence), contract, or otherwise, unless required by
applicable law (such as deliberate and grossly negligent acts) or agreed to in
writing, shall any Contributor be liable to You for damages, including any
direct, indirect, special, incidental, or consequential damages of any character
arising as a result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill, work stoppage,
computer failure or malfunction, or any and all other commercial damages or
losses), even if such Contributor has been advised of the possibility of such
damages.

9. Accepting Warranty or Additional Liability. While redistributing the Work or
Derivative Works thereof, You may choose to offer, and charge a fee for,
acceptance of support, warranty, indemnity, or other liability obligations
and/or rights consistent with this License. However, in accepting such
obligations, You may act only on Your own behalf and on Your sole responsibility,
not on behalf of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability incurred by, or
claims asserted against, such Contributor by reason of your accepting any such
warranty or additional liability.

END OF TERMS AND CONDITIONS

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "kabi-tg-cli"
version = "0.6.0"
description = "Telethon-powered Telegram CLI for local sync, search, and agent-friendly retrieval"
readme = "README.md"
license = { file = "LICENSE" }
requires-python = ">=3.10"
authors = [{ name = "jackwener", email = "jakevingoo@gmail.com" }]
keywords = ["telegram", "tg", "cli", "telethon", "search", "agent"]
classifiers = [
    "Development Status :: 3 - Alpha",
    "Environment :: Console",
    "Intended Audience :: Developers",
    "License :: OSI Approved :: Apache Software License",
    "Programming Language :: Python :: 3",
    "Programming Language :: Python :: 3.10",
    "Programming Language :: Python :: 3.11",
    "Programming Language :: Python :: 3.12",
    "Programming Language :: Python :: 3.13",
    "Programming Language :: Python :: 3.14",
    "Topic :: Communications :: Chat",
    "Topic :: Utilities",
]
dependencies = [
    "telethon>=1.36",
    "click>=8.0",
    "rich>=13.0",
    "python-dotenv>=1.0",
    "pyyaml>=6.0",
]

[project.urls]
Homepage = "https://github.com/jackwener/tg-cli"
Repository = "https://github.com/jackwener/tg-cli"
Issues = "https://github.com/jackwener/tg-cli/issues"


[project.scripts]
tg = "tg_cli.cli.main:cli"

[tool.hatch.build.targets.wheel]
packages = ["src/tg_cli"]

[tool.hatch.build.targets.sdist]
include = [
    "src/tg_cli",
    "README.md",
    "LICENSE",
    "CHANGELOG.md",
    "CONTRIBUTING.md",
    "SKILL.md",
    ".env.example",
    "examples",
    "tests",
]

[tool.pytest.ini_options]
addopts = "-v"

[tool.ruff]
line-length = 100

[tool.ruff.lint]
select = ["E", "F", "I", "B", "UP"]

[dependency-groups]
dev = [
    "pytest>=9.0.2",
    "pytest-asyncio>=1.3.0",
    "ruff>=0.11.0",
    "twine>=6.1.0",
]

"""tg-cli — Telegram CLI tool for syncing chats and local analysis."""

"""CLI package for tg-cli."""

"""Shared chat resolution helpers for CLI commands."""

from rich.table import Table

from ..console import console
from ..db import MessageDB
from ._output import emit_error


def _parse_chat(chat: str) -> str | int:
    """Parse a chat argument: return int if numeric, else the original string."""
    try:
        return int(chat)
    except ValueError:
        return chat


def resolve_chat_id_or_print(
    db: MessageDB,
    chat: str | None,
    *,
    allow_missing: bool = False,
) -> int | None:
    """Resolve a user-supplied chat filter and print helpful errors."""
    if not chat:
        return None

    matches = db.find_chats(chat)
    if not matches:
        if allow_missing:
            return None
        if emit_error("chat_not_found", f"Chat '{chat}' not found in database."):
            raise SystemExit(1) from None
        console.print(f"[red]Chat '{chat}' not found in database.[/red]")
        return None

    if len(matches) == 1:
        return matches[0]["chat_id"]

    table = Table(title=f"Ambiguous chat: {chat}")
    table.add_column("Chat ID", style="dim")
    table.add_column("Chat Name", style="bold")
    table.add_column("Messages", justify="right")
    for match in matches[:10]:
        table.add_row(
            str(match["chat_id"]),
            match.get("chat_name") or "—",
            str(match.get("msg_count") or 0),
        )

    if emit_error(
        "chat_ambiguous",
        f"Chat '{chat}' matches multiple local chats.",
        details={"query": chat, "matches": matches[:10]},
    ):
        raise SystemExit(1) from None
    console.print(f"[red]Chat '{chat}' matches multiple local chats.[/red]")
    console.print(table)
    console.print("[yellow]Use a more specific name or the numeric chat ID.[/yellow]")
    return None

"""Shared structured output helpers for CLI commands."""

from __future__ import annotations

import json
import os
import sys
from collections.abc import Callable
from typing import Any

import click
import yaml

_OUTPUT_ENV = "OUTPUT"
_SCHEMA_VERSION = "1"


def default_structured_format(*, as_json: bool, as_yaml: bool) -> str | None:
    """Resolve explicit flags first, then fall back to env and TTY defaults."""
    if as_json and as_yaml:
        raise click.UsageError("Use only one of --json or --yaml.")
    if as_yaml:
        return "yaml"
    if as_json:
        return "json"
    output_mode = os.getenv(_OUTPUT_ENV, "auto").strip().lower()
    if output_mode == "yaml":
        return "yaml"
    if output_mode == "json":
        return "json"
    if output_mode == "rich":
        return None
    if not sys.stdout.isatty():
        return "yaml"
    return None


def structured_output_options(command: Callable) -> Callable:
    """Add --json/--yaml flags to a click command."""
    command = click.option("--yaml", "as_yaml", is_flag=True, help="Output as YAML")(command)
    command = click.option("--json", "as_json", is_flag=True, help="Output as JSON")(command)
    return command


def emit_structured(data: Any, *, as_json: bool, as_yaml: bool) -> bool:
    """Emit structured output and return True when a structured format was used."""
    fmt = default_structured_format(as_json=as_json, as_yaml=as_yaml)
    if fmt is None:
        return False
    click.echo(dump_structured(_normalize_success_payload(data), fmt=fmt))
    return True


def dump_structured(data: Any, *, fmt: str) -> str:
    """Serialize structured data to JSON or YAML text."""
    if fmt == "json":
        return json.dumps(data, ensure_ascii=False, indent=2, default=str)
    if fmt == "yaml":
        return yaml.safe_dump(
            data,
            allow_unicode=True,
            sort_keys=False,
            default_flow_style=False,
        )
    raise ValueError(f"Unsupported structured format: {fmt}")


def success_payload(data: Any) -> dict[str, Any]:
    """Wrap structured success data in the shared agent schema."""
    return {
        "ok": True,
        "schema_version": _SCHEMA_VERSION,
        "data": data,
    }


def error_payload(code: str, message: str, *, details: Any | None = None) -> dict[str, Any]:
    """Wrap structured error data in the shared agent schema."""
    error = {
        "code": code,
        "message": message,
    }
    if details is not None:
        error["details"] = details
    return {
        "ok": False,
        "schema_version": _SCHEMA_VERSION,
        "error": error,
    }


def _normalize_success_payload(data: Any) -> Any:
    """Wrap plain structured data in the shared agent success schema."""
    if isinstance(data, dict) and data.get("schema_version") == _SCHEMA_VERSION and "ok" in data:
        return data
    return success_payload(data)


def emit_error(
    code: str,
    message: str,
    *,
    as_json: bool | None = None,
    as_yaml: bool | None = None,
    details: Any | None = None,
) -> bool:
    """Emit a structured error when the active output mode is machine-readable."""
    if as_json is None or as_yaml is None:
        ctx = click.get_current_context(silent=True)
        params = ctx.params if ctx is not None else {}
        as_json = bool(params.get("as_json", False)) if as_json is None else as_json
        as_yaml = bool(params.get("as_yaml", False)) if as_yaml is None else as_yaml

    fmt = default_structured_format(as_json=bool(as_json), as_yaml=bool(as_yaml))
    if fmt is None:
        return False
    click.echo(dump_structured(error_payload(code, message, details=details), fmt=fmt))
    return True

"""Shared sync helpers for CLI commands."""

from __future__ import annotations

from collections.abc import Callable

from ..client import connect, fetch_history, sync_all
from ..db import MessageDB
from ._chat import _parse_chat


async def sync_all_dialogs(
    *,
    limit: int,
    on_chat_done: Callable[[str, int, int], None] | None = None,
    delay: float = 1.0,
    max_chats: int | None = None,
) -> dict[str, int]:
    """Sync all dialogs available to the current Telegram account."""
    with MessageDB() as db:
        async with connect() as client:
            return await sync_all(
                client,
                db,
                limit_per_chat=limit,
                on_chat_done=on_chat_done,
                delay=delay,
                max_chats=max_chats,
            )


async def sync_chat_dialog(
    chat: str,
    *,
    limit: int,
    on_progress: Callable[[int], None] | None = None,
) -> int:
    """Sync a single chat into the local database."""
    with MessageDB() as db:
        chat_id = db.resolve_chat_id(chat)
        last_id = db.get_last_msg_id(chat_id) if chat_id else 0
        async with connect() as client:
            return await fetch_history(
                client,
                _parse_chat(chat),
                limit=limit,
                db=db,
                on_progress=on_progress,
                min_id=last_id or 0,
            )

"""Data commands — export, purge."""

import click

from ..console import console
from ..db import MessageDB
from ._chat import resolve_chat_id_or_print
from ._output import default_structured_format, dump_structured, error_payload


@click.group("data")
def data_group():
    """Data management commands (registered at top-level)."""


@data_group.command("export")
@click.argument("chat")
@click.option("-f", "--format", "fmt", type=click.Choice(["text", "json", "yaml"]), default="text")
@click.option("-o", "--output", "output_file", help="Output file path")
@click.option("--hours", type=int, help="Only export last N hours")
def export(chat: str, fmt: str, output_file: str | None, hours: int | None):
    """Export messages from CHAT to text, JSON, or YAML."""
    with MessageDB() as db:
        chat_id = resolve_chat_id_or_print(db, chat)
        if chat_id is None:
            return

        if hours:
            msgs = db.get_recent(chat_id=chat_id, hours=hours, limit=100000)
        else:
            msgs = db.get_recent(chat_id=chat_id, hours=None, limit=100000)

    if not msgs:
        structured_fmt = (
            fmt
            if fmt in {"json", "yaml"}
            else default_structured_format(as_json=False, as_yaml=False)
        )
        if structured_fmt in {"json", "yaml"} and output_file is None:
            payload = error_payload("no_messages", f"No messages found for '{chat}'.")
            click.echo(dump_structured(payload, fmt=structured_fmt))
            raise SystemExit(1) from None
        console.print(f"[yellow]No messages found for '{chat}'.[/yellow]")
        return

    if fmt in {"json", "yaml"}:
        content = dump_structured(msgs, fmt=fmt)
    else:
        lines = []
        for msg in msgs:
            ts = (msg.get("timestamp") or "")[:19]
            sender = msg.get("sender_name") or "Unknown"
            text = msg.get("content") or ""
            lines.append(f"[{ts}] {sender}: {text}")
        content = "\n".join(lines)

    if output_file:
        with open(output_file, "w", encoding="utf-8") as f:
            f.write(content)
        console.print(f"[green]✓[/green] Exported {len(msgs)} messages to {output_file}")
    else:
        console.print(content)


@data_group.command("purge")
@click.argument("chat")
@click.option("-y", "--yes", is_flag=True, help="Skip confirmation")
def purge(chat: str, yes: bool):
    """Delete all stored messages for CHAT."""
    with MessageDB() as db:
        chat_id = resolve_chat_id_or_print(db, chat)
        if chat_id is None:
            return

        if not yes:
            count = db.count(chat_id)
            if not click.confirm(f"Delete {count} messages from chat {chat_id}?"):
                return

        deleted = db.delete_chat(chat_id)
    console.print(f"[green]✓[/green] Deleted {deleted} messages")

"""tg-cli — Telegram CLI entry point."""

import logging

import click

from .data import data_group
from .query import query_group
from .tg import tg_group


def _setup_logging(verbose: bool):
    level = logging.DEBUG if verbose else logging.WARNING
    logging.basicConfig(
        level=level,
        format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
        datefmt="%H:%M:%S",
    )


@click.group()
@click.version_option(package_name="kabi-tg-cli")
@click.option("-v", "--verbose", is_flag=True, help="Enable debug logging")
def cli(verbose: bool):
    """tg — Telegram CLI for syncing chats, searching messages, and local analysis."""
    _setup_logging(verbose)


# Register ALL commands at top-level (flat structure, no `tg tg` nonsense)
for group in (tg_group, query_group, data_group):
    for name, cmd in group.commands.items():
        cli.add_command(cmd, name)

"""Query commands — search, stats, top, timeline, today, filter."""

import asyncio
from collections import defaultdict

import click
from rich.table import Table

from ..console import console
from ..db import MessageDB
from ._chat import resolve_chat_id_or_print
from ._output import emit_error, emit_structured, structured_output_options
from ._sync import sync_all_dialogs, sync_chat_dialog


@click.group("query")
def query_group():
    """Query and analysis commands (registered at top-level)."""


def _maybe_sync_first(chat: str | None, sync_first: bool, sync_limit: int) -> None:
    """Refresh local cache before running a query command."""
    if not sync_first:
        return

    if chat:
        with MessageDB() as db:
            matches = db.find_chats(chat)
        if len(matches) > 1:
            return
        asyncio.run(sync_chat_dialog(chat, limit=sync_limit))
        return

    asyncio.run(sync_all_dialogs(limit=sync_limit))


@query_group.command("search")
@click.argument("keyword")
@click.option("-c", "--chat", help="Filter by chat name")
@click.option("-s", "--sender", help="Filter by sender name")
@click.option("--hours", type=int, help="Only search messages within N hours")
@click.option("--regex", "use_regex", is_flag=True, help="Treat KEYWORD as a regex pattern")
@click.option("--sync-first", is_flag=True, help="Refresh local cache before searching")
@click.option(
    "--sync-limit",
    default=5000,
    show_default=True,
    help="Max messages per chat when using --sync-first",
)
@click.option("-n", "--limit", default=50, help="Max results")
@structured_output_options
def search(
    keyword: str,
    chat: str | None,
    sender: str | None,
    hours: int | None,
    use_regex: bool,
    sync_first: bool,
    sync_limit: int,
    limit: int,
    as_json: bool,
    as_yaml: bool,
):
    """Search messages by KEYWORD with optional chat, sender, and time filters."""
    import re

    _maybe_sync_first(chat, sync_first, sync_limit)

    with MessageDB() as db:
        chat_id = resolve_chat_id_or_print(db, chat)
        if chat and chat_id is None:
            return
        try:
            if use_regex:
                results = db.search_regex(
                    keyword, chat_id=chat_id, sender=sender, hours=hours, limit=limit
                )
            else:
                results = db.search(
                    keyword,
                    chat_id=chat_id,
                    sender=sender,
                    hours=hours,
                    limit=limit,
                )
        except re.error as exc:
            if emit_error("invalid_regex", f"Invalid regex pattern: {exc}"):
                raise SystemExit(1) from None
            console.print(f"[red]Invalid regex pattern: {exc}[/red]")
            return

    if results and emit_structured(results, as_json=as_json, as_yaml=as_yaml):
        return

    if not results:
        if emit_structured([], as_json=as_json, as_yaml=as_yaml):
            return
        console.print("[yellow]No messages found.[/yellow]")
        return

    for msg in results:
        ts = (msg.get("timestamp") or "")[:19]
        sender = msg.get("sender_name") or "Unknown"
        chat_name = msg.get("chat_name") or ""
        content = (msg.get("content") or "")[:200]
        console.print(
            f"[dim]{ts}[/dim] [cyan]{chat_name}[/cyan] | [bold]{sender}[/bold]: {content}"
        )

    filters = []
    if chat:
        filters.append(f"chat={chat}")
    if sender:
        filters.append(f"sender={sender}")
    if hours:
        filters.append(f"hours={hours}")
    if use_regex:
        filters.append("mode=regex")
    suffix = f" ({', '.join(filters)})" if filters else ""
    console.print(f"\n[dim]Found {len(results)} messages{suffix}[/dim]")


@query_group.command("recent")
@click.option("-c", "--chat", help="Filter by chat name")
@click.option("-s", "--sender", help="Filter by sender name")
@click.option("--hours", type=int, default=24, show_default=True, help="Only show last N hours")
@click.option(
    "--sync-first",
    is_flag=True,
    help="Refresh local cache before reading recent messages",
)
@click.option(
    "--sync-limit",
    default=5000,
    show_default=True,
    help="Max messages per chat when using --sync-first",
)
@click.option("-n", "--limit", default=50, help="Max messages")
@structured_output_options
def recent(
    chat: str | None,
    sender: str | None,
    hours: int,
    sync_first: bool,
    sync_limit: int,
    limit: int,
    as_json: bool,
    as_yaml: bool,
):
    """Show recent messages for browsing without a keyword search."""

    _maybe_sync_first(chat, sync_first, sync_limit)

    with MessageDB() as db:
        chat_id = resolve_chat_id_or_print(db, chat)
        if chat and chat_id is None:
            return
        msgs = db.get_recent(chat_id=chat_id, sender=sender, hours=hours, limit=limit)

    if msgs and emit_structured(msgs, as_json=as_json, as_yaml=as_yaml):
        return

    if not msgs:
        if emit_structured([], as_json=as_json, as_yaml=as_yaml):
            return
        console.print("[yellow]No recent messages found.[/yellow]")
        return

    for msg in msgs:
        ts = (msg.get("timestamp") or "")[:19]
        sender_name = msg.get("sender_name") or "Unknown"
        chat_name = msg.get("chat_name") or ""
        content = (msg.get("content") or "")[:200].replace("\n", " ")
        console.print(
            f"[dim]{ts}[/dim] [cyan]{chat_name}[/cyan] | [bold]{sender_name}[/bold]: {content}"
        )

    filters = [f"hours={hours}"]
    if chat:
        filters.append(f"chat={chat}")
    if sender:
        filters.append(f"sender={sender}")
    console.print(f"\n[dim]Showing {len(msgs)} recent messages ({', '.join(filters)})[/dim]")


@query_group.command("stats")
@click.option("--sync-first", is_flag=True, help="Refresh local cache before calculating stats")
@click.option(
    "--sync-limit",
    default=5000,
    show_default=True,
    help="Max messages per chat when using --sync-first",
)
@structured_output_options
def stats(sync_first: bool, sync_limit: int, as_json: bool, as_yaml: bool):
    """Show message statistics per chat."""
    _maybe_sync_first(None, sync_first, sync_limit)

    with MessageDB() as db:
        chats = db.get_chats()
        total = db.count()

    if emit_structured({"total": total, "chats": chats}, as_json=as_json, as_yaml=as_yaml):
        return

    table = Table(title=f"Message Stats (Total: {total})")
    table.add_column("Chat ID", style="dim")
    table.add_column("Chat Name", style="bold")
    table.add_column("Messages", justify="right")
    table.add_column("First Message", style="dim")
    table.add_column("Last Message", style="dim")

    for c in chats:
        table.add_row(
            str(c["chat_id"]),
            c["chat_name"] or "—",
            str(c["msg_count"]),
            (c["first_msg"] or "")[:19],
            (c["last_msg"] or "")[:19],
        )

    console.print(table)


@query_group.command("top")
@click.option("-c", "--chat", help="Filter by chat name")
@click.option("--hours", type=int, help="Only count messages within N hours")
@click.option(
    "--sync-first",
    is_flag=True,
    help="Refresh local cache before calculating top senders",
)
@click.option(
    "--sync-limit",
    default=5000,
    show_default=True,
    help="Max messages per chat when using --sync-first",
)
@click.option("-n", "--limit", default=20, help="Top N senders")
@structured_output_options
def top(
    chat: str | None,
    hours: int | None,
    sync_first: bool,
    sync_limit: int,
    limit: int,
    as_json: bool,
    as_yaml: bool,
):
    """Show most active senders."""
    _maybe_sync_first(chat, sync_first, sync_limit)

    with MessageDB() as db:
        chat_id = resolve_chat_id_or_print(db, chat)
        if chat and chat_id is None:
            return
        results = db.top_senders(chat_id=chat_id, hours=hours, limit=limit)

    if results and emit_structured(results, as_json=as_json, as_yaml=as_yaml):
        return

    if not results:
        if emit_structured([], as_json=as_json, as_yaml=as_yaml):
            return
        console.print("[yellow]No sender data found.[/yellow]")
        return

    table = Table(title="Top Senders")
    table.add_column("#", style="dim", justify="right")
    table.add_column("Sender", style="bold")
    table.add_column("Messages", justify="right")
    table.add_column("First", style="dim")
    table.add_column("Last", style="dim")

    for i, r in enumerate(results, 1):
        table.add_row(
            str(i),
            r["sender_name"],
            str(r["msg_count"]),
            (r["first_msg"] or "")[:10],
            (r["last_msg"] or "")[:10],
        )

    console.print(table)


@query_group.command("timeline")
@click.option("-c", "--chat", help="Filter by chat name")
@click.option("--hours", type=int, help="Only show last N hours")
@click.option("--by", "granularity", type=click.Choice(["day", "hour"]), default="day")
@click.option(
    "--sync-first",
    is_flag=True,
    help="Refresh local cache before building the timeline",
)
@click.option(
    "--sync-limit",
    default=5000,
    show_default=True,
    help="Max messages per chat when using --sync-first",
)
@structured_output_options
def timeline(
    chat: str | None,
    hours: int | None,
    granularity: str,
    sync_first: bool,
    sync_limit: int,
    as_json: bool,
    as_yaml: bool,
):
    """Show message activity over time as a bar chart."""
    _maybe_sync_first(chat, sync_first, sync_limit)

    with MessageDB() as db:
        chat_id = resolve_chat_id_or_print(db, chat)
        if chat and chat_id is None:
            return
        results = db.timeline(chat_id=chat_id, hours=hours, granularity=granularity)

    if results and emit_structured(results, as_json=as_json, as_yaml=as_yaml):
        return

    if not results:
        if emit_structured([], as_json=as_json, as_yaml=as_yaml):
            return
        console.print("[yellow]No timeline data.[/yellow]")
        return

    max_count = max(r["msg_count"] for r in results)
    bar_width = 40

    for r in results:
        period = r["period"]
        count = r["msg_count"]
        bar_len = int(count / max_count * bar_width) if max_count > 0 else 0
        bar = "█" * bar_len
        console.print(f"[dim]{period}[/dim] {bar} [bold]{count}[/bold]")


@query_group.command("today")
@click.option("-c", "--chat", help="Filter by chat name")
@click.option(
    "--sync-first",
    is_flag=True,
    help="Refresh local cache before reading today's messages",
)
@click.option(
    "--sync-limit",
    default=5000,
    show_default=True,
    help="Max messages per chat when using --sync-first",
)
@structured_output_options
def today(chat: str | None, sync_first: bool, sync_limit: int, as_json: bool, as_yaml: bool):
    """Show today's messages, grouped by chat."""
    from datetime import datetime

    _maybe_sync_first(chat, sync_first, sync_limit)

    with MessageDB() as db:
        chat_id = resolve_chat_id_or_print(db, chat)
        if chat and chat_id is None:
            return
        msgs = db.get_today(chat_id=chat_id)
        latest_ts = db.get_latest_timestamp(chat_id=chat_id)

    if msgs and emit_structured(msgs, as_json=as_json, as_yaml=as_yaml):
        return

    if not msgs:
        if emit_structured([], as_json=as_json, as_yaml=as_yaml):
            return
        console.print("[yellow]No messages today.[/yellow]")
        if latest_ts:
            latest_local = datetime.fromisoformat(latest_ts).astimezone()
            console.print(
                "[dim]Latest local message is from "
                f"{latest_local.strftime('%Y-%m-%d %H:%M:%S %Z')}. "
                "Run 'tg refresh' to refresh.[/dim]"
            )
        else:
            console.print("[dim]Local database is empty. Run 'tg refresh' first.[/dim]")
        return

    # Group by chat
    grouped: dict[str, list[dict]] = defaultdict(list)
    for m in msgs:
        grouped[m.get("chat_name") or "Unknown"].append(m)

    for chat_name, chat_msgs in sorted(grouped.items(), key=lambda x: -len(x[1])):
        console.print(f"\n[bold cyan]═══ {chat_name} ({len(chat_msgs)} msgs) ═══[/bold cyan]")
        for m in chat_msgs:
            ts = (m.get("timestamp") or "")[11:19]
            sender = m.get("sender_name") or "Unknown"
            content = (m.get("content") or "")[:200].replace("\n", " ")
            console.print(f"  [dim]{ts}[/dim] [bold]{sender[:15]}[/bold]: {content}")

    console.print(f"\n[green]Total: {len(msgs)} messages today[/green]")


@query_group.command("filter")
@click.argument("keywords")
@click.option("-c", "--chat", help="Filter by chat name")
@click.option("--hours", type=int, help="Only search last N hours (default: today)")
@click.option("--sync-first", is_flag=True, help="Refresh local cache before filtering")
@click.option(
    "--sync-limit",
    default=5000,
    show_default=True,
    help="Max messages per chat when using --sync-first",
)
@structured_output_options
def filter_msgs(
    keywords: str,
    chat: str | None,
    hours: int | None,
    sync_first: bool,
    sync_limit: int,
    as_json: bool,
    as_yaml: bool,
):
    """Filter messages by KEYWORDS (comma-separated, OR logic).

    Examples:
        tg filter "Rust,Golang,Java"
        tg filter "招聘,remote,远程" --hours 48
        tg filter "Rust" --chat "牛油果" --json
    """
    import re

    keyword_list = [k.strip() for k in keywords.split(",") if k.strip()]
    if not keyword_list:
        if emit_error("invalid_keywords", "Please provide at least one keyword."):
            raise SystemExit(1) from None
        console.print("[red]Please provide at least one keyword.[/red]")
        return

    _maybe_sync_first(chat, sync_first, sync_limit)

    with MessageDB() as db:
        chat_id = resolve_chat_id_or_print(db, chat)
        if chat and chat_id is None:
            return

        if hours:
            msgs = db.get_recent(chat_id=chat_id, hours=hours, limit=100000)
        else:
            msgs = db.get_today(chat_id=chat_id)

    # Filter messages containing ANY of the keywords (case-insensitive)
    pattern = re.compile("|".join(re.escape(k) for k in keyword_list), re.IGNORECASE)
    matched = [m for m in msgs if m.get("content") and pattern.search(m["content"])]

    if not matched:
        if emit_structured([], as_json=as_json, as_yaml=as_yaml):
            return
        console.print(f"[yellow]No messages matching: {', '.join(keyword_list)}[/yellow]")
        return

    if emit_structured(matched, as_json=as_json, as_yaml=as_yaml):
        return

    # Group by chat
    grouped: dict[str, list[dict]] = defaultdict(list)
    for m in matched:
        grouped[m.get("chat_name") or "Unknown"].append(m)

    for chat_name, chat_msgs in sorted(grouped.items(), key=lambda x: -len(x[1])):
        console.print(f"\n[bold cyan]═══ {chat_name} ({len(chat_msgs)} matches) ═══[/bold cyan]")
        for m in chat_msgs:
            ts = (m.get("timestamp") or "")[:19]
            sender = m.get("sender_name") or "Unknown"
            content = (m.get("content") or "")[:300].replace("\n", " ")
            # Highlight keywords
            for kw in keyword_list:
                content = re.sub(
                    re.escape(kw),
                    f"[bold red]{kw}[/bold red]",
                    content,
                    flags=re.IGNORECASE,
                )
            console.print(
                f"  [dim]{ts}[/dim] [bold]{sender[:15]}[/bold]: ",
                end="",
            )
            console.print(content, markup=True, highlight=False)

    console.print(
        f"\n[green]Found {len(matched)} messages matching "
        f"'{', '.join(keyword_list)}' "
        f"(from {len(msgs)} total)[/green]"
    )

"""Telegram subcommands — send, edit, delete, and more."""

import asyncio
import time

import click
from rich.progress import Progress, SpinnerColumn, TextColumn
from rich.table import Table

from ..client import connect, fetch_history, get_chat_info, list_chats, listen
from ..console import console
from ..db import MessageDB
from ._chat import _parse_chat, resolve_chat_id_or_print
from ._output import (
    default_structured_format,
    dump_structured,
    emit_structured,
    error_payload,
    structured_output_options,
    success_payload,
)
from ._sync import sync_all_dialogs, sync_chat_dialog


def _telegram_user_payload(me) -> dict[str, str | int]:
    """Normalize Telegram user info for structured agent output."""
    name = " ".join(part for part in [me.first_name, me.last_name] if part).strip()
    return {
        "id": me.id,
        "name": name,
        "username": me.username or "",
        "first_name": me.first_name or "",
        "last_name": me.last_name or "",
        "phone": me.phone or "",
    }


@click.group("tg")
def tg_group():
    """Telegram operations — connect, fetch, sync, listen."""
    pass


@tg_group.command("chats")
@click.option("--type", "chat_type", help="Filter by type: user, group, supergroup, channel")
@structured_output_options
def tg_chats(chat_type: str | None, as_json: bool, as_yaml: bool):
    """List joined Telegram chats."""

    async def _run():
        async with connect() as client:
            return await list_chats(client, chat_type)

    chats = asyncio.run(_run())
    if emit_structured(chats, as_json=as_json, as_yaml=as_yaml):
        return

    table = Table(title="Telegram Chats")
    table.add_column("ID", style="dim")
    table.add_column("Name", style="bold")
    table.add_column("Type", style="cyan")
    table.add_column("Unread", justify="right")

    for c in chats:
        table.add_row(str(c["id"]), c["name"], c["type"], str(c["unread"]))

    console.print(table)
    console.print(f"\nTotal: {len(chats)} chats")


@tg_group.command("history")
@click.argument("chat")
@click.option("-n", "--limit", default=1000, help="Max messages to fetch")
@structured_output_options
def tg_history(chat: str, limit: int, as_json: bool, as_yaml: bool):
    """Fetch historical messages from CHAT (name, username, or numeric ID)."""

    async def _run():
        with MessageDB() as db:
            async with connect() as client:
                with Progress(
                    SpinnerColumn(),
                    TextColumn("[progress.description]{task.description}"),
                    console=console,
                ) as progress:
                    task = progress.add_task(f"Fetching messages from {chat}...", total=None)

                    def on_progress(count: int):
                        progress.update(task, description=f"Stored {count} messages...")

                    count = await fetch_history(
                        client, _parse_chat(chat), limit=limit, db=db, on_progress=on_progress
                    )
                return count

    count = asyncio.run(_run())
    payload = {"stored": count, "chat": chat}
    if emit_structured(payload, as_json=as_json, as_yaml=as_yaml):
        return
    console.print(f"\n[green]\u2713[/green] Stored {count} messages from {chat}")


@tg_group.command("sync")
@click.argument("chat")
@click.option("-n", "--limit", default=5000, help="Max messages per sync")
@structured_output_options
def tg_sync(chat: str, limit: int, as_json: bool, as_yaml: bool):
    """Incremental sync — fetch only new messages from CHAT."""

    async def _run():
        with MessageDB() as db:
            # Resolve chat_id to get last_msg_id
            chat_id = resolve_chat_id_or_print(db, chat, allow_missing=True)
            matches = db.find_chats(chat)
            if len(matches) > 1:
                resolve_chat_id_or_print(db, chat)
                return None
            last_id = db.get_last_msg_id(chat_id) if chat_id else 0
        if last_id:
            console.print(f"Syncing from msg_id > {last_id}...")

        with Progress(
            SpinnerColumn(),
            TextColumn("[progress.description]{task.description}"),
            console=console,
        ) as progress:
            task_id = progress.add_task(f"Syncing {chat}...", total=None)

            def on_progress(count: int):
                progress.update(task_id, description=f"Stored {count} new messages...")

            return await sync_chat_dialog(chat, limit=limit, on_progress=on_progress)

    count = asyncio.run(_run())
    if count is None:
        return
    payload = {"synced": count, "chat": chat}
    if emit_structured(payload, as_json=as_json, as_yaml=as_yaml):
        return
    console.print(f"\n[green]\u2713[/green] Synced {count} new messages from {chat}")


@tg_group.command("sync-all")
@click.option("-n", "--limit", default=5000, help="Max messages per chat")
@click.option(
    "--delay",
    default=1.0,
    show_default=True,
    help="Seconds between chat syncs (anti-ban). Set 0 to disable.",
)
@click.option(
    "--max-chats",
    default=None,
    type=int,
    help="Max number of chats to sync per run (default: all)",
)
@structured_output_options
def tg_sync_all(limit: int, delay: float, max_chats: int | None, as_json: bool, as_yaml: bool):
    """Sync all currently available Telegram dialogs with a single connection."""

    async def _run():
        on_chat_done = None
        if not as_json and not as_yaml:
            console.print("Syncing all available chats...")

            def _on_chat_done(name: str, new_count: int, total: int):
                if new_count > 0:
                    console.print(f"  [green]✓[/green] {name}: +{new_count} (total: {total})")
                else:
                    console.print(f"  [dim]✓ {name}: no new messages[/dim]")

            on_chat_done = _on_chat_done

        return await sync_all_dialogs(
            limit=limit, on_chat_done=on_chat_done, delay=delay, max_chats=max_chats
        )

    results = asyncio.run(_run())
    total_new = sum(results.values())
    payload = {"new_messages": total_new, "chats": len(results), "results": results}
    if emit_structured(payload, as_json=as_json, as_yaml=as_yaml):
        return
    console.print(f"\n[green]✓[/green] Synced {total_new} new messages across {len(results)} chats")


@tg_group.command("refresh")
@click.option("-n", "--limit", default=5000, help="Max messages per chat")
@click.option(
    "--delay",
    default=1.0,
    show_default=True,
    help="Seconds between chat syncs (anti-ban). Set 0 to disable.",
)
@click.option(
    "--max-chats",
    default=None,
    type=int,
    help="Max number of chats to sync per run (default: all)",
)
@structured_output_options
def tg_refresh(limit: int, delay: float, max_chats: int | None, as_json: bool, as_yaml: bool):
    """Refresh the local cache from all current Telegram dialogs."""

    async def _run():
        on_chat_done = None
        if not as_json and not as_yaml:
            console.print("Refreshing local cache...")

            def _on_chat_done(name: str, new_count: int, total: int):
                if new_count > 0:
                    console.print(f"  [green]✓[/green] {name}: +{new_count} (total: {total})")
                else:
                    console.print(f"  [dim]✓ {name}: no new messages[/dim]")

            on_chat_done = _on_chat_done

        return await sync_all_dialogs(
            limit=limit, on_chat_done=on_chat_done, delay=delay, max_chats=max_chats
        )

    results = asyncio.run(_run())
    total_new = sum(results.values())
    updated = [
        name
        for name, count in sorted(results.items(), key=lambda item: (-item[1], item[0]))
        if count > 0
    ]
    payload = {
        "new_messages": total_new,
        "chats": len(results),
        "updated_chats": updated,
        "results": results,
    }
    if emit_structured(payload, as_json=as_json, as_yaml=as_yaml):
        return

    console.print(f"\n[green]✓[/green] Refreshed {len(results)} chats, {total_new} new messages.")
    if updated:
        console.print(f"[dim]Most recently updated: {', '.join(updated[:5])}[/dim]")


@tg_group.command("listen")
@click.argument("chats", nargs=-1)
@click.option("--persist", is_flag=True, help="Reconnect automatically if the connection drops")
@click.option(
    "--retry-seconds",
    default=5,
    show_default=True,
    help="Reconnect delay when using --persist",
)
def tg_listen(chats: tuple[str, ...], persist: bool, retry_seconds: int):
    """Real-time listener for new messages. Optionally specify CHATS to filter."""
    parsed: list[str | int] | None = None
    if chats:
        parsed = []
        for c in chats:
            try:
                parsed.append(int(c))
            except ValueError:
                parsed.append(c)

    async def _run_once():
        async with connect() as client:
            return await listen(client, chats=parsed)

    while True:
        try:
            result = asyncio.run(_run_once())
        except click.ClickException:
            raise
        except Exception as exc:
            if not persist:
                raise
            console.print(
                f"[yellow]Listener disconnected: {exc}. Retrying in {retry_seconds}s...[/yellow]"
            )
            time.sleep(retry_seconds)
            continue

        if not persist or result == "stopped":
            break

        console.print(
            f"[yellow]Listener disconnected. Reconnecting in {retry_seconds}s...[/yellow]"
        )
        time.sleep(retry_seconds)


@tg_group.command("info")
@click.argument("chat")
@structured_output_options
def tg_info(chat: str, as_json: bool, as_yaml: bool):
    """Show detailed info about CHAT."""

    async def _run():
        async with connect() as client:
            return await get_chat_info(client, _parse_chat(chat))

    info = asyncio.run(_run())
    if not info:
        console.print(f"[red]Could not find chat: {chat}[/red]")
        return

    if emit_structured(info, as_json=as_json, as_yaml=as_yaml):
        return

    table = Table(title="Chat Info", show_header=False)
    table.add_column("Field", style="bold")
    table.add_column("Value")

    for k, v in info.items():
        table.add_row(k, v)

    console.print(table)


@tg_group.command("whoami")
@structured_output_options
def tg_whoami(as_json: bool, as_yaml: bool):
    """Show current logged-in user info."""

    async def _run():
        async with connect() as client:
            me = await client.get_me()
            return me

    fmt = default_structured_format(as_json=as_json, as_yaml=as_yaml)
    try:
        me = asyncio.run(_run())
    except Exception as exc:
        if fmt is not None:
            click.echo(dump_structured(error_payload("auth_error", str(exc)), fmt=fmt))
            raise SystemExit(1) from None
        raise click.ClickException(str(exc)) from exc

    info = _telegram_user_payload(me)

    if emit_structured(success_payload({"user": info}), as_json=as_json, as_yaml=as_yaml):
        return

    name = " ".join(p for p in [me.first_name, me.last_name] if p)
    table = Table(title=f"👤 {name}")
    table.add_column("Field", style="bold cyan")
    table.add_column("Value", style="green")
    table.add_row("ID", str(me.id))
    table.add_row("Name", name)
    if me.username:
        table.add_row("Username", f"@{me.username}")
    if me.phone:
        table.add_row("Phone", f"+{me.phone}")

    console.print(table)


@tg_group.command("status")
@structured_output_options
def tg_status(as_json: bool, as_yaml: bool):
    """Show Telegram authentication status."""

    async def _run():
        async with connect() as client:
            me = await client.get_me()
            return {
                "authenticated": True,
                "id": me.id,
                "first_name": me.first_name or "",
                "last_name": me.last_name or "",
                "username": me.username or "",
                "phone": me.phone or "",
            }

    fmt = default_structured_format(as_json=as_json, as_yaml=as_yaml)
    try:
        info = asyncio.run(_run())
    except Exception as exc:
        if fmt is not None:
            click.echo(dump_structured(error_payload("auth_error", str(exc)), fmt=fmt))
            raise SystemExit(1) from None
        raise click.ClickException(str(exc)) from exc

    user = {key: value for key, value in info.items() if key != "authenticated"}
    if emit_structured(
        success_payload({"authenticated": True, "user": user}),
        as_json=as_json,
        as_yaml=as_yaml,
    ):
        return

    name = " ".join(part for part in [info["first_name"], info["last_name"]] if part).strip()
    console.print(f"[green]✓[/green] Authenticated as [bold]{name or info['id']}[/bold]")
    if info["username"]:
        console.print(f"[dim]@{info['username']}[/dim]")


@tg_group.command("send")
@click.argument("chat")
@click.argument("message")
@click.option("-r", "--reply", type=int, default=None, help="Message ID to reply to")
@click.option("--no-preview", is_flag=True, help="Disable link preview")
@structured_output_options
def tg_send(
    chat: str,
    message: str,
    reply: int | None,
    no_preview: bool,
    as_json: bool,
    as_yaml: bool,
):
    """Send a MESSAGE to CHAT (name, username, or numeric ID)."""

    async def _run():
        async with connect() as client:
            msg = await client.send_message(
                _parse_chat(chat),
                message,
                reply_to=reply,
                link_preview=not no_preview,
            )
            return msg

    msg = asyncio.run(_run())
    payload = {"sent": True, "msg_id": msg.id, "chat": chat}
    if reply is not None:
        payload["reply_to"] = reply
    if emit_structured(payload, as_json=as_json, as_yaml=as_yaml):
        return
    console.print(f"[green]\u2713[/green] Message sent (id: {msg.id})")


@tg_group.command("edit")
@click.argument("chat")
@click.argument("msg_id", type=int)
@click.argument("new_text")
@click.option("--no-preview", is_flag=True, help="Disable link preview")
@structured_output_options
def tg_edit(chat: str, msg_id: int, new_text: str, no_preview: bool, as_json: bool, as_yaml: bool):
    """Edit a previously sent message. CHAT MSG_ID NEW_TEXT."""

    async def _run():
        async with connect() as client:
            return await client.edit_message(
                _parse_chat(chat),
                msg_id,
                new_text,
                link_preview=not no_preview,
            )

    asyncio.run(_run())
    payload = {"edited": True, "msg_id": msg_id, "chat": chat}
    if emit_structured(payload, as_json=as_json, as_yaml=as_yaml):
        return
    console.print(f"[green]\u2713[/green] Message {msg_id} edited")


@tg_group.command("delete")
@click.argument("chat")
@click.argument("msg_ids", nargs=-1, type=int, required=True)
@structured_output_options
def tg_delete(chat: str, msg_ids: tuple[int, ...], as_json: bool, as_yaml: bool):
    """Delete one or more messages. CHAT MSG_ID [MSG_ID ...]."""

    async def _run():
        async with connect() as client:
            await client.delete_messages(_parse_chat(chat), list(msg_ids))

    asyncio.run(_run())
    payload = {"deleted": True, "msg_ids": list(msg_ids), "chat": chat}
    if emit_structured(payload, as_json=as_json, as_yaml=as_yaml):
        return
    console.print(f"[green]\u2713[/green] Deleted {len(msg_ids)} message(s)")