debian-manager/app/ui/pages/gitea.py

"""Gitea: деплой compose-проектов."""
import flet as ft

from app.core import gitea
from app.core import servers as servers_core
from app.core.ssh import SSHClient
from app.core.tasks import tasks
from app.ui.components import append_log, log_view, snack


class GiteaPage(ft.Column):
    def __init__(self, page: ft.Page, go):
        super().__init__(expand=True, scroll=ft.ScrollMode.AUTO, spacing=12)
        self._page = page

        cfg = gitea.get_config()
        self.url_tf = ft.TextField(label="URL Gitea", value=cfg["url"], width=350)
        self.token_tf = ft.TextField(label="API-токен", value=cfg["token"],
                                     password=True, width=350)
        self.status = ft.Text("", size=12)

        self.repos_dd = ft.Dropdown(label="Репозиторий", width=350, options=[])
        self.srv_dd = ft.Dropdown(label="Сервер", width=250, options=[
            ft.dropdown.Option(str(s.id), s.name) for s in servers_core.list_servers()
        ])
        self.branch_tf = ft.TextField(label="Ветка", value="main", width=150)
        self.compose_tf = ft.TextField(label="Путь к compose", value="docker-compose.yml",
                                       width=250)
        self.env_tf = ft.TextField(label=".env (опционально)", multiline=True,
                                   min_lines=3, max_lines=5, width=450,
                                   hint_text="KEY=value")
        self.log = log_view()
        self.log_container = ft.Container(
            height=200, border=ft.Border.all(1, ft.Colors.GREY_800),
            border_radius=8, padding=8, content=self.log, visible=False)
        self.bindings_col = ft.Column(spacing=4)

        self.controls = [
            ft.Text("Gitea Deploy", size=28, weight=ft.FontWeight.BOLD),
            ft.Row([self.url_tf, self.token_tf,
                    ft.FilledButton("Сохранить и проверить", on_click=self._save_cfg)]),
            self.status,
            ft.Divider(),
            ft.Row([self.repos_dd,
                    ft.IconButton(ft.Icons.REFRESH, tooltip="Загрузить репозитории",
                                  on_click=self._load_repos)]),
            ft.Row([self.srv_dd, self.branch_tf, self.compose_tf]),
            self.env_tf,
            ft.FilledButton("Деплой", icon=ft.Icons.ROCKET_LAUNCH,
                            on_click=self._deploy),
            self.log_container,
            ft.Divider(),
            ft.Text("Webhook-автодеплой (push → redeploy)", weight=ft.FontWeight.BOLD, size=16),
            ft.Text("Приложение слушает порт 9100. В Gitea: репозиторий → Настройки → Webhooks → "
                    "создать: http://<этот-ПК>:9100/webhooks/gitea, событие push.",
                    size=12, color=ft.Colors.GREY_500),
            ft.Row([
                ft.FilledButton("Добавить привязку", icon=ft.Icons.ADD,
                                on_click=self._binding_dialog),
                ft.IconButton(ft.Icons.REFRESH, on_click=lambda e: self._load_bindings()),
            ]),
            self.bindings_col,
        ]
        self._load_bindings()

    def _load_bindings(self):
        from app.core import webhook as wh
        from app.core import servers as servers_core
        servers = {s.id: s.name for s in servers_core.list_servers()}
        self.bindings_col.controls = []
        for b in wh.list_bindings():
            self.bindings_col.controls.append(ft.Card(content=ft.Container(
                padding=10,
                content=ft.Row([
                    ft.Icon(ft.Icons.LINK, color=ft.Colors.BLUE_300),
                    ft.Text(f"{b.repo}@{b.branch} → {servers.get(b.server_id, '?')}",
                            size=13, expand=True),
                    ft.Text(b.compose_path, size=11, color=ft.Colors.GREY_500),
                    ft.IconButton(ft.Icons.DELETE_OUTLINE, icon_size=16,
                                  on_click=lambda e, bid=b.id: self._del_binding(bid)),
                ]),
            )))
        if not self.bindings_col.controls:
            self.bindings_col.controls.append(
                ft.Text("привязок нет — push в репозиторий не будет триггерить деплой",
                        size=12, color=ft.Colors.GREY_500))
        self._page.update()

    def _del_binding(self, bid):
        from app.core import webhook as wh
        wh.delete_binding(bid)
        self._load_bindings()

    def _binding_dialog(self, e):
        from app.core import webhook as wh
        from app.core import servers as servers_core
        repo_tf = ft.TextField(label="Репозиторий (owner/name)", width=300,
                               hint_text="dimon/myapp")
        srv_dd = ft.Dropdown(label="Сервер", width=250, options=[
            ft.dropdown.Option(str(s.id), s.name) for s in servers_core.list_servers()
        ])
        branch_tf = ft.TextField(label="Ветка", value="main", width=150)
        compose_tf = ft.TextField(label="compose-файл", value="docker-compose.yml", width=250)
        env_tf = ft.TextField(label=".env (опционально)", multiline=True,
                              min_lines=2, max_lines=4, width=350)

        def save(e):
            if not repo_tf.value or not srv_dd.value:
                return
            wh.add_binding(repo=repo_tf.value.strip(),
                           server_id=int(srv_dd.value),
                           branch=branch_tf.value, compose_path=compose_tf.value,
                           env=env_tf.value or "")
            self._page.pop_dialog()
            self._load_bindings()
            snack(self._page, "Привязка добавлена")

        dlg = ft.AlertDialog(
            modal=True, title=ft.Text("Привязка webhook"),
            content=ft.Column([repo_tf, srv_dd, branch_tf, compose_tf, env_tf], tight=True),
            actions=[
                ft.TextButton("Отмена", on_click=lambda e: self._page.pop_dialog()),
                ft.FilledButton("Сохранить", on_click=save),
            ],
        )
        self._page.show_dialog(dlg)

    def _save_cfg(self, e):
        gitea.save_config(self.url_tf.value, self.token_tf.value)
        async def _t():
            ok, msg = await gitea.test_connection()
            self.status.value = ("✅ " if ok else "❌ ") + msg
            self.status.color = ft.Colors.GREEN_300 if ok else ft.Colors.RED_300
            self._page.update()
        self._page.run_task(_t)

    def _load_repos(self, e):
        async def _l():
            try:
                repos = await gitea.list_repos()
                self.repos_dd.options = [
                    ft.dropdown.Option(r["full_name"],
                                       f"{r['full_name']} ({r['default_branch']})")
                    for r in repos
                ]
                self._repos = {r["full_name"]: r for r in repos}
                self.status.value = f"загружено {len(repos)} репозиториев"
            except Exception as ex:
                self.status.value = f"ошибка: {ex}"
            self._page.update()
        self._page.run_task(_l)

    def _deploy(self, e):
        if not self.repos_dd.value or not self.srv_dd.value:
            snack(self._page, "Выберите репозиторий и сервер", error=True)
            return
        repo = self._repos[self.repos_dd.value]
        srv = servers_core.get_server(int(self.srv_dd.value))
        self.log_container.visible = True

        async def job(log):
            async with SSHClient(srv) as c:
                await gitea.deploy(c, repo["clone_url"], repo["name"],
                                   self.branch_tf.value, self.compose_tf.value,
                                   self.env_tf.value or None, log)
        tasks.start("gitea", f"deploy {repo['name']}", srv.id, job,
                    on_log=lambda l: (append_log(self.log, l), self._page.update()))
        snack(self._page, f"Деплой {repo['name']} запущен")