#!/usr/bin/env python3 """Project-status oracle: read-only check of repo architecture conformance. Deterministically inspects the current repository against a standard architecture for the auto-detected project type and prints a report with ``[OK]/[WARN]/[FAIL]`` lines, an Итог summary, and Рекомендации. Read-only and stateless — no files are created or modified, no network calls beyond read-only ``gh api`` for branch protection detection. Usage: python3 .opencode/scripts/project-status.py # non-blocking (exit 0) python3 .opencode/scripts/project-status.py --check # strict (exit 1 on FAIL) python3 .opencode/scripts/project-status.py --fast # skip slow/remote checks Project types (auto-detected): fullstack — ``frontend/`` dir (SvelteKit) + ``backend/`` dir backend — ``src/api/v1/`` + ``src/db/models/`` + fastapi/uvicorn in deps cli — ``[project.scripts]`` in pyproject.toml + typer in deps bot — ``src/bot.py`` OR aiogram in deps worker — ``src/flow.py`` OR prefect in deps unknown — none of the above matched (still runs a generic check set) """ from __future__ import annotations import ast import re import subprocess import sys import tomllib from dataclasses import dataclass, field from enum import StrEnum from pathlib import Path from typing import Any # ── repo root + config ─────────────────────────────────────────────────────── def _resolve_repo_root() -> Path: """Resolve repo root via git (cwd-aware), fallback to script location.""" result = subprocess.run( ["git", "rev-parse", "--show-toplevel"], capture_output=True, text=True, check=False ) if result.returncode == 0 and result.stdout.strip(): return Path(result.stdout.strip()).resolve() return Path(__file__).resolve().parent.parent.parent REPO_ROOT = _resolve_repo_root() # Default thresholds — overridable via ``[tool.project-status]`` in pyproject.toml. DEFAULT_CONFIG: dict[str, Any] = { "route_line_limit": 50, "min_test_count": 1, "require_branch_protection": False, } def load_config() -> dict[str, Any]: """Load thresholds from ``[tool.project-status]`` in pyproject.toml. Falls back to ``DEFAULT_CONFIG`` if the section or file is missing. Uses ``tomllib`` (stdlib, Python 3.11+). Reads only — never writes. """ cfg: dict[str, Any] = dict(DEFAULT_CONFIG) pyproject = REPO_ROOT / "pyproject.toml" if not pyproject.exists(): return cfg try: with pyproject.open("rb") as f: data = tomllib.load(f) except (OSError, ValueError): return cfg section = data.get("tool", {}).get("project-status", {}) if isinstance(section, dict): for key, default in DEFAULT_CONFIG.items(): val = section.get(key, default) if isinstance(val, type(default)) or val is None: cfg[key] = val return cfg CONFIG = load_config() # ── enums + dataclasses ────────────────────────────────────────────────────── class CheckStatus(StrEnum): """Result of a single check.""" OK = "OK" WARN = "WARN" FAIL = "FAIL" class ProjectType(StrEnum): """Auto-detected project type.""" FULLSTACK = "fullstack" BACKEND = "backend" CLI = "cli" BOT = "bot" WORKER = "worker" UNKNOWN = "unknown" @dataclass(frozen=True) class CheckResult: """Result of a single check within a group.""" status: CheckStatus name: str detail: str @dataclass class GroupResult: """Result of a single check group (multiple CheckResult items).""" name: str checks: list[CheckResult] = field(default_factory=list) def overall(self) -> CheckStatus: """Roll up statuses: FAIL > WARN > OK.""" statuses = [c.status for c in self.checks] if CheckStatus.FAIL in statuses: return CheckStatus.FAIL if CheckStatus.WARN in statuses: return CheckStatus.WARN return CheckStatus.OK # ── helpers ────────────────────────────────────────────────────────────────── def run_cmd(args: list[str]) -> tuple[int, str, str]: """Run a command, return (returncode, stdout, stderr). Read-only intent.""" result = subprocess.run(args, capture_output=True, text=True, check=False) return result.returncode, result.stdout, result.stderr def path_exists(rel: str) -> bool: """True if ``REPO_ROOT / rel`` exists.""" return (REPO_ROOT / rel).exists() def read_text(rel: str) -> str | None: """Read text content of ``REPO_ROOT / rel`` or None if missing.""" p = REPO_ROOT / rel if not p.exists(): return None try: return p.read_text(encoding="utf-8-sig") except OSError: return None def parse_pyproject() -> dict[str, Any]: """Parse pyproject.toml into a dict (or empty dict on failure).""" raw = read_text("pyproject.toml") if raw is None: return {} try: return tomllib.loads(raw) except ValueError: return {} def parse_remote_url(url: str) -> tuple[str, str, str]: """Parse git remote URL into (host, org, repo). Supports both HTTPS and SSH formats and optional userinfo (insteadOf). """ ssh = re.match(r"git@([^:]+):([^/]+)/(.+?)(?:\.git)?$", url) if ssh: return ssh.group(1), ssh.group(2), ssh.group(3) https = re.match(r"https?://(?:[^/@]*@)?([^/]+)/([^/]+)/(.+?)(?:\.git)?$", url) if https: return https.group(1), https.group(2), https.group(3) raise ValueError(f"Cannot parse remote URL: {url}") def get_repo_full_name() -> str | None: """Return ``org/repo`` from git remote, or None on error (read-only).""" rc, out, _ = run_cmd(["git", "remote", "get-url", "origin"]) if rc != 0: return None try: _host, org, repo = parse_remote_url(out.strip()) except ValueError: return None else: return f"{org}/{repo}" # ── auto-detect ────────────────────────────────────────────────────────────── def _matches_backend(deps_lower: str) -> bool: """True if backend contract dirs present + fastapi/uvicorn in deps.""" return ( path_exists("src/api/v1") and path_exists("src/db/models") and ("fastapi" in deps_lower or "uvicorn" in deps_lower) ) def _detect_simple_type(deps_lower: str) -> ProjectType | None: """Detect bot/worker types by file or dep marker (or None).""" if path_exists("src/bot.py") or "aiogram" in deps_lower: return ProjectType.BOT if path_exists("src/flow.py") or "prefect" in deps_lower: return ProjectType.WORKER return None def detect_project_type() -> ProjectType: """Auto-detect project type from filesystem + pyproject.toml. Order matters: fullstack (most specific) → backend → bot → worker → cli. Falls back to ``UNKNOWN`` if nothing matches. """ if path_exists("frontend") and path_exists("backend"): return ProjectType.FULLSTACK pyproject = parse_pyproject() deps_raw = pyproject.get("project", {}).get("dependencies", []) deps_lower = " ".join(str(d).lower() for d in deps_raw) if isinstance(deps_raw, list) else "" if _matches_backend(deps_lower): return ProjectType.BACKEND simple = _detect_simple_type(deps_lower) if simple is not None: return simple scripts = pyproject.get("project", {}).get("scripts", {}) if isinstance(scripts, dict) and scripts and "typer" in deps_lower: return ProjectType.CLI return ProjectType.UNKNOWN # ── expected structure per type ────────────────────────────────────────────── STRUCTURE_EXPECTED: dict[ProjectType, list[str]] = { ProjectType.BACKEND: [ "src/api/v1", "src/db/models", "src/schemas", "src/services", "src/config/settings.py", "main.py", ], ProjectType.FULLSTACK: ["backend", "frontend"], ProjectType.CLI: ["src"], # src// — checked generically ProjectType.BOT: ["src/bot.py"], ProjectType.WORKER: ["src/flow.py"], ProjectType.UNKNOWN: [], } # ── README delimiter tags (12) — ported from create-readme.ts:140-199 ─────── README_DELIMITERS: list[str] = [ "tagline-en:start", "tagline-en:end", "tagline-ru:start", "tagline-ru:end", "summary-en:start", "summary-en:end", "features-en:start", "features-en:end", "summary-ru:start", "summary-ru:end", "features-ru:start", "features-ru:end", ] # ── check group 1: structure ───────────────────────────────────────────────── def _check_backend_lifespan() -> CheckResult: """Check main.py has a lifespan handler (backend-specific).""" main = read_text("main.py") if main is None: return CheckResult(CheckStatus.FAIL, "main.py lifespan", "main.py нет") if "lifespan" in main: return CheckResult(CheckStatus.OK, "main.py lifespan", "lifespan найден") return CheckResult(CheckStatus.WARN, "main.py lifespan", "lifespan не найден") def _check_cli_package() -> CheckResult: """Check src// with __init__.py exists (cli-specific).""" src = REPO_ROOT / "src" if src.exists() and any(p.is_dir() and (p / "__init__.py").exists() for p in src.iterdir()): return CheckResult(CheckStatus.OK, "src//", "пакет найден") return CheckResult(CheckStatus.FAIL, "src//", "пакет не найден") def _check_type_specific_structure(ptype: ProjectType) -> list[CheckResult]: """Type-specific extra checks beyond the expected dirs list.""" if ptype == ProjectType.BACKEND: return [_check_backend_lifespan()] if ptype == ProjectType.FULLSTACK and not path_exists("frontend/package.json"): return [CheckResult(CheckStatus.WARN, "frontend/package.json", "SvelteKit не обнаружен")] if ptype == ProjectType.CLI: return [_check_cli_package()] return [] def check_structure(ptype: ProjectType) -> GroupResult: """Group 1: Structure — expected dirs/files per project type.""" group = GroupResult(name="Структура") expected = STRUCTURE_EXPECTED.get(ptype, []) if not expected: group.checks.append( CheckResult(CheckStatus.WARN, "auto-detect", f"тип={ptype.value}: нет контракта") ) return group for rel in expected: status = CheckStatus.OK if path_exists(rel) else CheckStatus.FAIL detail = "существует" if status == CheckStatus.OK else "отсутствует" group.checks.append(CheckResult(status, rel, detail)) group.checks.extend(_check_type_specific_structure(ptype)) return group # ── check group 2: thin routes (AST, ≤ route_line_limit lines) ─────────────── def _route_line_count(source: str) -> int: """Count lines for a route handler function (body span in source).""" try: tree = ast.parse(source) except SyntaxError: return -1 route_methods = {"get", "post", "put", "delete", "patch"} max_lines = 0 for node in ast.walk(tree): if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): continue if _has_route_decorator(node.decorator_list, route_methods): lines = node.end_lineno - node.lineno + 1 if node.end_lineno else 0 max_lines = max(max_lines, lines) return max_lines def _has_route_decorator(decorators: list[ast.expr], route_methods: set[str]) -> bool: """True if any decorator is a route method (@router.get, @get, etc.).""" for dec in decorators: # @router.get(...) → Call(func=Attribute(attr='get')) if ( isinstance(dec, ast.Call) and isinstance(dec.func, ast.Attribute) and dec.func.attr in route_methods ): return True # @app.get / @router.get used bare → Attribute(attr='get') if isinstance(dec, ast.Attribute) and dec.attr in route_methods: return True # @get / @post (simple name) → Name(id='get') if isinstance(dec, ast.Name) and dec.id in route_methods: return True return False def _api_dirs_for(ptype: ProjectType) -> list[Path]: """Return list of api/v1 dirs to scan for routes, based on project type.""" if ptype == ProjectType.BACKEND: root = REPO_ROOT / "src" / "api" / "v1" return [root] if root.exists() else [] if ptype == ProjectType.FULLSTACK: root = REPO_ROOT / "backend" / "src" / "api" / "v1" return [root] if root.exists() else [] return [] def _scan_route_files(api_dirs: list[Path], limit: int) -> tuple[int, int, list[str]]: """Scan api dirs for route handlers; return (files_checked, longest, over_limit).""" files_checked = 0 longest = 0 over_limit: list[str] = [] for api_dir in api_dirs: for py in api_dir.rglob("*.py"): source = py.read_text(encoding="utf-8-sig", errors="ignore") n = _route_line_count(source) if n < 0: continue files_checked += 1 longest = max(longest, n) if n > limit: over_limit.append(f"{py.relative_to(REPO_ROOT)}:{n}") return files_checked, longest, over_limit def check_thin_routes(ptype: ProjectType, fast: bool = False) -> GroupResult: """Group 2: Тонкие роуты — AST parse, ≤ route_line_limit lines per handler.""" _ = fast # unused here, accepted for signature uniformity group = GroupResult(name="Тонкие роуты") if ptype not in {ProjectType.BACKEND, ProjectType.FULLSTACK}: group.checks.append( CheckResult(CheckStatus.OK, "skip", f"тип={ptype.value}: роуты не применимы") ) return group api_dirs = _api_dirs_for(ptype) if not api_dirs: group.checks.append( CheckResult(CheckStatus.WARN, "src/api/v1/", "директория роутов не найдена") ) return group limit = int(CONFIG.get("route_line_limit", 50)) files_checked, longest, over_limit = _scan_route_files(api_dirs, limit) if files_checked == 0: group.checks.append(CheckResult(CheckStatus.WARN, "AST", "роуты не найдены в src/api/v1/")) elif over_limit: group.checks.append( CheckResult( CheckStatus.FAIL, f"route ≤ {limit} lines", f"превышение: {', '.join(over_limit[:3])}", ) ) else: group.checks.append( CheckResult( CheckStatus.OK, f"route ≤ {limit} lines", f"макс={longest}, файлов={files_checked}" ) ) return group # ── check group 3: code quality (mypy/ruff/pytest presence) ────────────────── def check_quality(ptype: ProjectType) -> GroupResult: """Group 3: Качество кода — mypy/ruff/pytest configured in pyproject.toml.""" group = GroupResult(name="Качество кода") pyproject = parse_pyproject() tools = pyproject.get("tool", {}) for tool_name in ("ruff", "mypy"): if tool_name in tools: group.checks.append(CheckResult(CheckStatus.OK, tool_name, "настроен в pyproject.toml")) else: group.checks.append( CheckResult(CheckStatus.FAIL, tool_name, f"[tool.{tool_name}] отсутствует") ) pytest_cfg = pyproject.get("tool", {}).get("pytest", {}) dev_deps = pyproject.get("project", {}).get("optional-dependencies", {}).get("dev", []) dev_str = " ".join(str(d).lower() for d in dev_deps) if isinstance(dev_deps, list) else "" if pytest_cfg or "pytest" in dev_str: group.checks.append(CheckResult(CheckStatus.OK, "pytest", "настроен")) else: group.checks.append(CheckResult(CheckStatus.FAIL, "pytest", "не найден в dev-deps")) return group # ── check group 4: tests (conftest, stub-detector, no @pytest.mark.asyncio) ── def check_tests(ptype: ProjectType) -> GroupResult: """Group 4: Тесты — conftest, no @pytest.mark.asyncio, ≥1 test file.""" group = GroupResult(name="Тесты") tests_dir = REPO_ROOT / "tests" if not tests_dir.exists(): group.checks.append( CheckResult(CheckStatus.FAIL, "tests/", "директория tests/ отсутствует") ) return group if (tests_dir / "conftest.py").exists(): group.checks.append(CheckResult(CheckStatus.OK, "conftest.py", "существует")) else: group.checks.append( CheckResult( CheckStatus.WARN, "conftest.py", "отсутствует — pytest fixtures без общего конфига" ) ) test_files = list(tests_dir.glob("test_*.py")) min_tests = int(CONFIG.get("min_test_count", 1)) if len(test_files) >= min_tests: group.checks.append(CheckResult(CheckStatus.OK, "test files", f"{len(test_files)} файлов")) else: group.checks.append( CheckResult(CheckStatus.FAIL, "test files", f"{len(test_files)} (< {min_tests})") ) asyncio_marks = 0 for tf in test_files: try: content = tf.read_text(encoding="utf-8-sig") except OSError: continue asyncio_marks += content.count("@pytest.mark.asyncio") if asyncio_marks == 0: group.checks.append( CheckResult(CheckStatus.OK, "no @pytest.mark.asyncio", "asyncio_mode=auto используется") ) else: group.checks.append( CheckResult( CheckStatus.WARN, "no @pytest.mark.asyncio", f"{asyncio_marks} маркеров — не нужно при asyncio_mode=auto", ) ) stub_count = sum( 1 for tf in test_files for line in tf.read_text(encoding="utf-8-sig", errors="ignore").splitlines() if re.match(r"\s*(def test_|async def test_).*stub", line, re.IGNORECASE) ) group.checks.append( CheckResult( CheckStatus.WARN if stub_count > 0 else CheckStatus.OK, "stub-detector", f"{stub_count} stub-тестов", ) ) return group # ── check group 5: README (12 delimiter tags) ──────────────────────────────── def check_readme(ptype: ProjectType) -> GroupResult: """Group 5: README — 12 delimiter tags from create-readme.ts:140-199.""" group = GroupResult(name="README") content = read_text("README.md") if content is None: group.checks.append(CheckResult(CheckStatus.FAIL, "README.md", "отсутствует")) return group missing = [d for d in README_DELIMITERS if f"" not in content] if missing: group.checks.append( CheckResult( CheckStatus.FAIL, "12 delimiter tags", f"не хватает {len(missing)}: {', '.join(missing[:3])}", ) ) else: group.checks.append( CheckResult(CheckStatus.OK, "12 delimiter tags", "все 12 разделителей присутствуют") ) for required_text in ("# 🚀 ", "## 🇺🇸 English", "## 🇷🇺 Русский", "[English](#-english)"): if required_text in content: group.checks.append(CheckResult(CheckStatus.OK, required_text, "присутствует")) else: group.checks.append(CheckResult(CheckStatus.FAIL, required_text, "отсутствует")) if "assets/cover.png" in content: group.checks.append(CheckResult(CheckStatus.OK, "cover.png", "указан")) else: group.checks.append( CheckResult(CheckStatus.WARN, "cover.png", "не указан — slaid098.dev showcase требует") ) return group # ── check group 6: infra (branch protection, ci.yml, dependabot, LICENSE, pre-commit) ── def _check_branch_protection(repo: str) -> CheckResult: """Read-only ``gh api repos//rules/branches/main`` — returns CheckResult.""" rc, out, err = run_cmd(["gh", "api", f"repos/{repo}/rules/branches/main"]) if rc != 0: return CheckResult( CheckStatus.WARN, "branch protection", f"gh api не отвечает: {err.strip()[:60]}" ) if "pull_request" in out and "required_status_checks" in out: return CheckResult(CheckStatus.OK, "branch protection", "main защищён (PR + checks)") return CheckResult(CheckStatus.WARN, "branch protection", "правила найдены, но набор неполный") def check_infra(ptype: ProjectType, fast: bool = False) -> GroupResult: """Group 6: Infra — branch protection, ci.yml, dependabot, LICENSE, pre-commit.""" group = GroupResult(name="Infra") if path_exists(".github/workflows/ci.yml"): group.checks.append(CheckResult(CheckStatus.OK, ".github/workflows/ci.yml", "есть")) else: group.checks.append( CheckResult( CheckStatus.FAIL, ".github/workflows/ci.yml", "отсутствует — CI не настроен", ) ) if path_exists(".github/dependabot.yml"): group.checks.append(CheckResult(CheckStatus.OK, "dependabot.yml", "настроен")) else: group.checks.append( CheckResult(CheckStatus.WARN, "dependabot.yml", "обновления зависимостей вручную") ) if path_exists("LICENSE"): group.checks.append(CheckResult(CheckStatus.OK, "LICENSE", "есть")) else: group.checks.append(CheckResult(CheckStatus.FAIL, "LICENSE", "отсутствует")) if path_exists(".pre-commit-config.yaml"): group.checks.append(CheckResult(CheckStatus.OK, "pre-commit", "настроен")) else: group.checks.append( CheckResult(CheckStatus.WARN, "pre-commit", "отсутствует — quality gate только в CI") ) if fast: group.checks.append( CheckResult(CheckStatus.WARN, "branch protection", "пропущено (--fast)") ) else: repo = get_repo_full_name() if repo is None: group.checks.append( CheckResult(CheckStatus.WARN, "branch protection", "git remote недоступен") ) else: group.checks.append(_check_branch_protection(repo)) return group # ── check group 7: coverage (non-blocking) ─────────────────────────────────── def check_coverage(ptype: ProjectType) -> GroupResult: """Group 7: Coverage — non-blocking (always OK/WARN, never FAIL).""" group = GroupResult(name="Coverage") pyproject = parse_pyproject() cov = pyproject.get("tool", {}).get("coverage", {}) run_cfg = cov.get("run", {}) if isinstance(cov, dict) else {} sources = run_cfg.get("source", []) if isinstance(run_cfg, dict) else [] if sources: group.checks.append(CheckResult(CheckStatus.OK, "[tool.coverage.run]", f"source={sources}")) else: group.checks.append(CheckResult(CheckStatus.WARN, "[tool.coverage.run]", "source не задан")) pytest_opts = pyproject.get("tool", {}).get("pytest", {}).get("ini_options", {}) addopts = pytest_opts.get("addopts", "") if isinstance(pytest_opts, dict) else "" if "--cov-fail-under" in str(addopts): m = re.search(r"--cov-fail-under=(\d+)", str(addopts)) threshold = m.group(1) if m else "?" group.checks.append(CheckResult(CheckStatus.OK, "cov-fail-under", f"порог={threshold}%")) else: group.checks.append( CheckResult( CheckStatus.WARN, "cov-fail-under", "порог не задан (coverage non-blocking)" ) ) return group # ── orchestration ──────────────────────────────────────────────────────────── CHECK_GROUPS: list[str] = [ "Структура", "Тонкие роуты", "Качество кода", "Тесты", "README", "Infra", "Coverage", ] def run_all_checks(ptype: ProjectType, fast: bool = False) -> list[GroupResult]: """Run all 7 check groups, return results in order.""" return [ check_structure(ptype), check_thin_routes(ptype, fast=fast), check_quality(ptype), check_tests(ptype), check_readme(ptype), check_infra(ptype, fast=fast), check_coverage(ptype), ] STATUS_PREFIX: dict[CheckStatus, str] = { CheckStatus.OK: "[OK]", CheckStatus.WARN: "[WARN]", CheckStatus.FAIL: "[FAIL]", } def format_output(ptype: ProjectType, groups: list[GroupResult]) -> str: """Format output: 7 group blocks + Итог + Рекомендации.""" lines: list[str] = [f"Project: {ptype.value}", ""] recommendations: list[str] = [] ok_count = 0 warn_count = 0 fail_count = 0 for group in groups: overall = group.overall() prefix = STATUS_PREFIX[overall] lines.append(f"{prefix} {group.name}") for chk in group.checks: sub_prefix = STATUS_PREFIX[chk.status] lines.append(f" {sub_prefix} {chk.name}: {chk.detail}") if chk.status == CheckStatus.FAIL: fail_count += 1 recommendations.append(f"- {group.name} / {chk.name}: {chk.detail}") elif chk.status == CheckStatus.WARN: warn_count += 1 else: ok_count += 1 lines.append("") lines.append("Итог:") lines.append(f" OK: {ok_count} WARN: {warn_count} FAIL: {fail_count}") if recommendations: lines.append("") lines.append("Рекомендации:") lines.extend(recommendations) return "\n".join(lines) def main() -> None: """Entry point: parse args, run checks, print report, set exit code.""" args = sys.argv[1:] strict = "--check" in args fast = "--fast" in args ptype = detect_project_type() groups = run_all_checks(ptype, fast=fast) print(format_output(ptype, groups)) if strict and any(g.overall() == CheckStatus.FAIL for g in groups): sys.exit(1) sys.exit(0) if __name__ == "__main__": main()