#!/usr/bin/env python3 """ Интеграция GitFlic → PT Sandbox ================================= PoC-скрипт для автоматической выгрузки релизов из GitFlic в PT Sandbox на анализ. """ import json import logging import sys import time from copy import deepcopy from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional import requests import yaml logger = logging.getLogger("sandbox_gitflic") def setup_logging(cfg: dict) -> None: """Настройка логирования из конфигурации.""" log_cfg = cfg.get("logging", {}) level_name = log_cfg.get("level", "INFO").upper() level = getattr(logging, level_name, logging.INFO) fmt = log_cfg.get("format", "%(asctime)s [%(levelname)s] %(name)s: %(message)s") handlers: List[logging.Handler] = [logging.StreamHandler(sys.stdout)] log_file = log_cfg.get("file") if log_file: handlers.append(logging.FileHandler(log_file, encoding="utf-8")) logging.basicConfig(level=level, format=fmt, handlers=handlers) logger.setLevel(level) def load_config(path: str = "config.yaml") -> dict: """Загрузить YAML-конфигурацию.""" path = Path(path) if not path.exists(): logger.error("Файл конфигурации не найден: %s", path.absolute()) sys.exit(1) with open(path, "r", encoding="utf-8") as fh: cfg = yaml.safe_load(fh) logger.info("Конфигурация загружена из %s", path) return cfg # ───────────────────────────────────────────────────────── # Управление состоянием # ───────────────────────────────────────────────────────── def load_state(path: str) -> dict: """ Загрузить состояние из JSON-файла. Формат: { "sent_releases": { ":::": { "project_id": int, "project_name": str, "release_id": int, "release_tag": str, "release_version": str, "files": [{"filename": str, "sandbox_task_id": str|null}], "sent_at": "2024-01-01T00:00:00Z" } } } """ path_obj = Path(path) if not path_obj.exists(): logger.info("Файл состояния не найден. Начинаем с пустого.") return {"sent_releases": {}} try: with open(path_obj, "r", encoding="utf-8") as fh: state = json.load(fh) logger.info("Состояние загружено из %s (%d релизов отправлено)", path, len(state.get("sent_releases", {}))) return state except (json.JSONDecodeError, OSError) as exc: logger.error("Ошибка загрузки состояния из %s: %s. Начинаем с пустого.", path, exc) return {"sent_releases": {}} def save_state(state: dict, path: str) -> None: """Сохранить состояние в JSON-файл (атомарно через rename).""" path_obj = Path(path) tmp_path = path_obj.with_suffix(".json.tmp") try: with open(tmp_path, "w", encoding="utf-8") as fh: json.dump(state, fh, ensure_ascii=False, indent=2) tmp_path.replace(path_obj) logger.debug("Состояние сохранено в %s", path) except OSError as exc: logger.error("Ошибка сохранения состояния в %s: %s", path, exc) def make_release_key(project_id: int, release_id: int, tag: str, version: str) -> str: """Сформировать ключ для идентификации релиза.""" return f"{project_id}:{release_id}:{tag}:{version}" def is_release_sent(state: dict, project_id: int, release_id: int, tag: str, version: str) -> bool: """Проверить, был ли релиз уже отправлен.""" key = make_release_key(project_id, release_id, tag, version) return key in state.get("sent_releases", {}) def mark_release_sent(state: dict, project_id: int, project_name: str, release_id: int, tag: str, version: str, files: List[Dict[str, Any]]) -> None: """Пометить релиз как отправленный.""" key = make_release_key(project_id, release_id, tag, version) if "sent_releases" not in state: state["sent_releases"] = {} state["sent_releases"][key] = { "project_id": project_id, "project_name": project_name, "release_id": release_id, "release_tag": tag, "release_version": version, "files": files, "sent_at": datetime.now(timezone.utc).isoformat(), } logger.info("Релиз помечен как отправленный: %s", key) # ───────────────────────────────────────────────────────── # API GitFlic # ───────────────────────────────────────────────────────── class GitFlicClient: """Клиент для REST API GitFlic.""" def __init__(self, cfg: dict): gf_cfg = cfg["gitflic"] self.base_url = gf_cfg["base_url"].rstrip("/") self.api_token = gf_cfg["api_token"] self.per_page = gf_cfg.get("per_page", 50) self.chunk_size = gf_cfg.get("download_chunk_size", 8 * 1024 * 1024) if not self.api_token: logger.error("API-токен GitFlic не задан в конфигурации!") sys.exit(1) self._session = requests.Session() self._session.headers.update({ "Authorization": f"Bearer {self.api_token}", "Accept": "application/json", "User-Agent": "Sandbox-GitFlic-PoC/1.0", }) def _request(self, method: str, path: str, **kwargs) -> requests.Response: """Выполнить HTTP-запрос к API GitFlic.""" url = f"{self.base_url}{path}" try: resp = self._session.request(method, url, **kwargs) resp.raise_for_status() return resp except requests.exceptions.Timeout: logger.error("Тайм-аут при запросе %s %s", method, url) raise except requests.exceptions.HTTPError as exc: logger.error("HTTP-ошибка %s: %s - %s", resp.status_code, resp.reason, resp.text[:500]) raise except requests.exceptions.RequestException as exc: logger.error("Ошибка запроса %s %s: %s", method, url, exc) raise def _paginated_get(self, path: str, params: Optional[dict] = None) -> List[dict]: """Получить все страницы пагинированного GET-запроса.""" if params is None: params = {} params.setdefault("per_page", self.per_page) page = 1 all_items: List[dict] = [] while True: params["page"] = page resp = self._request("GET", path, params=params) items = resp.json() if not isinstance(items, list): all_items = items break all_items.extend(items) total_pages = resp.headers.get("X-Total-Pages") if total_pages is not None: if page >= int(total_pages): break else: if not items: break page += 1 return all_items def get_projects(self) -> List[dict]: """GET /api/v1/projects.""" return self._paginated_get("/api/v1/projects", {"membership": True}) def get_company_projects(self, company_id: int) -> List[dict]: """GET /api/v1/companies/{id}/projects.""" return self._paginated_get(f"/api/v1/companies/{company_id}/projects") def resolve_projects(self, cfg: dict) -> List[dict]: """Определить список проектов для сканирования.""" gf_cfg = cfg["gitflic"] source = gf_cfg.get("project_source", "all") if source == "list": names = gf_cfg.get("project_list", []) ids = gf_cfg.get("project_ids", []) projects = [] for name in names: proj = self.get_project_by_path(name) if proj: projects.append(proj) else: logger.warning("Проект не найден: %s", name) for pid in ids: proj = self.get_project(pid) if proj: projects.append(proj) else: logger.warning("Проект не найден по ID: %d", pid) return projects if source == "company": cid = gf_cfg.get("company_id") if not cid: logger.error("Не указан company_id при project_source=company") return [] return self.get_company_projects(cid) if source == "team": tid = gf_cfg.get("team_id") if not tid: logger.error("Не указан team_id при project_source=team") return [] return self.get_team_projects(tid) return self.get_projects() def get_project(self, project_id: int) -> Optional[dict]: try: resp = self._request("GET", f"/api/v1/projects/{project_id}") return resp.json() except requests.RequestException: return None def get_project_by_path(self, path: str) -> Optional[dict]: try: resp = self._request("GET", f"/api/v1/projects/{path}") return resp.json() except requests.RequestException: return None def get_releases(self, project_id: int) -> List[dict]: """GET /api/v1/projects/{id}/releases.""" return self._paginated_get( f"/api/v1/projects/{project_id}/releases", {"per_page": self.per_page}, ) def get_release_assets(self, project_id: int, release_id: int) -> List[dict]: """Получить assets релиза.""" try: resp = self._request( "GET", f"/api/v1/projects/{project_id}/releases/{release_id}/assets", ) items = resp.json() if isinstance(items, list): return items if isinstance(items, dict): return items.get("assets", items.get("links", [])) return [] except requests.RequestException: logger.debug("Assets не получены: %s/%s", project_id, release_id) return [] def download_file(self, url: str, local_path: Path) -> Path: """Скачать файл по ссылке.""" if not url.startswith("http"): url = f"{self.base_url}{url}" headers = {} if url.startswith(self.base_url): headers["Authorization"] = f"Bearer {self.api_token}" logger.info("Скачивание: %s -> %s", url, local_path) resp = self._session.get(url, headers=headers, stream=True, timeout=(30, 300)) resp.raise_for_status() local_path.parent.mkdir(parents=True, exist_ok=True) with open(local_path, "wb") as fh: for chunk in resp.iter_content(chunk_size=self.chunk_size): if chunk: fh.write(chunk) logger.debug("Скачан файл: %s (%d bytes)", local_path, local_path.stat().st_size) return local_path def get_release_source_archive_url(self, project_id: int, tag_name: str) -> str: """URL архива исходного кода по тегу релиза.""" return (f"{self.base_url}/api/v1/projects/{project_id}" f"/repository/archive.zip?ref={tag_name}") def get_release_asset_download_url(self, project_id: int, release_id: int, asset_filename: str) -> str: """URL для скачивания прикреплённого файла (asset).""" return (f"{self.base_url}/api/v1/projects/{project_id}" f"/releases/{release_id}/downloads/{asset_filename}") # ───────────────────────────────────────────────────────── # Клиент PT Sandbox # ───────────────────────────────────────────────────────── class PTSandboxClient: """Клиент для REST API PT Sandbox.""" def __init__(self, cfg: dict): sb_cfg = cfg["pt_sandbox"] self.base_url = sb_cfg["base_url"].rstrip("/") self.upload_endpoint = sb_cfg.get("upload_endpoint", "/api/v1/files/upload") self.auth_header = sb_cfg.get("auth_header", "X-API-Key") self.api_key = sb_cfg.get("api_key", "") self.file_field_name = sb_cfg.get("file_field_name", "file") self.extra_fields = sb_cfg.get("extra_fields", {}) self.timeout_connect = sb_cfg.get("timeout_connect", 30) self.timeout_read = sb_cfg.get("timeout_read", 300) if not self.api_key: logger.error("API-ключ PT Sandbox не задан в конфигурации!") sys.exit(1) self._session = requests.Session() self._session.headers.update({ self.auth_header: self.api_key, "User-Agent": "Sandbox-GitFlic-PoC/1.0", }) def upload_file(self, file_path: Path) -> Optional[str]: """ Отправить файл в PT Sandbox на анализ. Возвращает ID задачи (task_id) или None при ошибке. """ url = f"{self.base_url}{self.upload_endpoint}" logger.info("Отправка файла в Sandbox: %s (%d bytes)", file_path.name, file_path.stat().st_size) try: with open(file_path, "rb") as fh: files = {self.file_field_name: (file_path.name, fh)} fields = dict(self.extra_fields) resp = self._session.post( url, files=files, data=fields, timeout=(self.timeout_connect, self.timeout_read), ) if resp.status_code in (200, 201, 202): result = resp.json() task_id = (result.get("task_id") or result.get("id") or result.get("uuid") or str(result)) logger.info("Файл %s принят Sandbox. task_id=%s", file_path.name, task_id) return str(task_id) else: logger.error("Sandbox вернул %s: %s", resp.status_code, resp.text[:500]) return None except requests.exceptions.Timeout: logger.error("Тайм-аут при отправке файла %s в Sandbox", file_path.name) return None except requests.exceptions.RequestException as exc: logger.error("Ошибка отправки файла %s в Sandbox: %s", file_path.name, exc) return None except OSError as exc: logger.error("Ошибка чтения файла %s: %s", file_path.name, exc) return None # ───────────────────────────────────────────────────────── # Вспомогательные функции # ───────────────────────────────────────────────────────── def get_temp_dir() -> Path: """Получить временную директорию для скачанных файлов.""" tmp = Path("downloads") tmp.mkdir(parents=True, exist_ok=True) return tmp # ───────────────────────────────────────────────────────── # Основная логика сканирования # ───────────────────────────────────────────────────────── def scan_new_releases(gf: GitFlicClient, sb: PTSandboxClient, state: dict, cfg: dict) -> dict: """ Основная функция: 1. Получить проекты 2. Для каждого проекта релизы 3. Отфильтровать отправленные 4. Скачать файлы (архив + assets) 5. Отправить в Sandbox 6. Обновить состояние """ projects = gf.resolve_projects(cfg) if not projects: logger.warning("Нет проектов для сканирования.") return state scan_mode = cfg["gitflic"].get("scan_mode", "latest") tmp_dir = get_temp_dir() state = deepcopy(state) for project in projects: pid = project.get("id") pname = project.get("path_with_namespace") or project.get("name", str(pid)) logger.info("Обработка проекта: %s (id=%s)", pname, pid) try: releases = gf.get_releases(pid) except requests.RequestException: logger.error("Пропускаем проект %s - ошибка получения релизов", pname) continue if not releases: continue releases.sort(key=lambda r: r.get("created_at", ""), reverse=True) if scan_mode == "latest": rel = releases[0] rel_id = rel.get("id") rel_tag = rel.get("tag_name", "") rel_version = rel.get("name", "") if is_release_sent(state, pid, rel_id, rel_tag, rel_version): logger.info("Последний релиз уже отправлен. Пропускаем.") continue releases_to_process = [rel] else: releases_to_process = [ r for r in releases if not is_release_sent(state, pid, r.get("id"), r.get("tag_name", ""), r.get("name", "")) ] if not releases_to_process: logger.info("Нет новых релизов в проекте %s", pname) continue for rel in releases_to_process: rel_id = rel.get("id") rel_tag = rel.get("tag_name", "") rel_version = rel.get("name", "") rel_key = make_release_key(pid, rel_id, rel_tag, rel_version) logger.info("Новый релиз: %s (tag=%s, version=%s)", rel_key, rel_tag, rel_version) # Собираем файлы для скачивания files_to_download: List[Dict[str, Any]] = [] local_files: List[Dict[str, Any]] = [] # 1. Архив исходного кода по тегу source_url = gf.get_release_source_archive_url(pid, rel_tag) sf = f"source_{pname.replace('/', '_')}_{rel_tag}.zip" files_to_download.append({"type": "source_archive", "url": source_url, "filename": sf}) # 2. Assets (прикреплённые файлы) try: assets = gf.get_release_assets(pid, rel_id) for asset in assets: aname = asset.get("name") or asset.get("filename", "") if not aname: continue aurl = gf.get_release_asset_download_url(pid, rel_id, aname) files_to_download.append({"type": "asset", "url": aurl, "filename": aname}) except requests.RequestException: logger.debug("Assets для %s не получены", rel_key) # Скачиваем все файлы for fi in files_to_download: safe = fi["filename"].replace("/", "_").replace("\\", "_") lpath = tmp_dir / f"{pid}_{rel_id}_{safe}" try: gf.download_file(fi["url"], lpath) local_files.append({"filename": fi["filename"], "type": fi["type"], "local_path": str(lpath)}) except requests.RequestException as exc: logger.error("Ошибка скачивания %s: %s", fi["filename"], exc) # Отправляем каждый файл в Sandbox sent_files: List[Dict[str, Any]] = [] for lf in local_files: lpath = Path(lf["local_path"]) if not lpath.exists(): logger.warning("Файл %s не найден", lpath) continue task_id = sb.upload_file(lpath) sent_files.append({"filename": lf["filename"], "type": lf["type"], "sandbox_task_id": task_id}) try: lpath.unlink() except OSError as exc: logger.warning("Не удалось удалить %s: %s", lpath, exc) mark_release_sent(state, pid, pname, rel_id, rel_tag, rel_version, sent_files) logger.info("Релиз %s обработан: %d файлов отправлено", rel_key, len(sent_files)) return state # ───────────────────────────────────────────────────────── # Точка входа # ───────────────────────────────────────────────────────── def main() -> None: """Главная функция.""" cfg = load_config() setup_logging(cfg) logger.info("=" * 60) logger.info("Запуск интеграции GitFlic -> PT Sandbox") logger.info("=" * 60) gf = GitFlicClient(cfg) sb = PTSandboxClient(cfg) loop_cfg = cfg.get("loop", {}) loop_enabled = loop_cfg.get("enabled", False) interval = loop_cfg.get("interval_seconds", 3600) state_path = cfg.get("state", {}).get("file_path", "state.json") iteration = 0 while True: iteration += 1 logger.info("Итерация #%d", iteration) state = load_state(state_path) try: new_state = scan_new_releases(gf, sb, state, cfg) save_state(new_state, state_path) except Exception as exc: logger.exception("Критическая ошибка на итерации #%d: %s", iteration, exc) try: save_state(state, state_path) except Exception: pass if not loop_enabled: logger.info("Однократный проход завершён.") break logger.info("Ожидание %d секунд до следующей итерации...", interval) try: time.sleep(interval) except KeyboardInterrupt: logger.info("Получен сигнал прерывания. Завершение.") break logger.info("Работа завершена.") if __name__ == "__main__": main()