This repository has been archived on 2026-08-19. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
ROCm-Research-Archive/_TestScripts/Scripts and Tests/pjstats.py
T
2026-08-20 00:45:43 +02:00

1907 lines
79 KiB
Python

#!/usr/bin/env python3
import gc
import os
import re
import sys
import json
import time
import argparse
import statistics
import subprocess
import platform
from pathlib import Path
from collections import defaultdict
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, wait, FIRST_COMPLETED
import fnmatch
VERSION = "2.1.2"
# Windows Terminal Fix for ANSI codes
if sys.platform == "win32":
os.system("")
# Ensure UTF-8 output on Windows
if sys.stdout.encoding != 'utf-8':
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
if sys.stderr.encoding != 'utf-8':
sys.stderr.reconfigure(encoding='utf-8', errors='replace')
HELP_TEXT = """pjstats - Enhanced project statistics with multi-threaded analysis.
Usage examples:
python3 pjstats.py
python3 pjstats.py /path/to/project --top 10 --json > pjstats.json
python3 pjstats.py --ext .py .js --min-code-lines 10
python3 pjstats.py --exclude language_model .git
python3 pjstats.py --no-color --no-git
python3 pjstats.py --workers 4 --no-animation
Options:
path Project root (default: .)
--json Output JSON summary
--top N Top N files to display (default: 5)
--ext Limit to specific extensions (e.g. .py .js)
--min-code-lines Ignore files with fewer code lines than this
--exclude Directories to ignore recursively (relative to project root)
--include-hidden Include hidden directories/files
--no-color Disable color output
--no-git Disable Git info detection
--no-animation Disable startup animation and progress bars
--bypass-excluded Bypass all ignore rules (.gitignore, built-in excludes, etc.)
--workers N Number of threads for parallel analysis (0=auto)
Notes:
- Files named Dockerfile.* are treated as Dockerfiles.
- Languages without a comment token (e.g., Markdown, JSON) are shown as N/A in the Code Quality Snapshot.
- The installer chooses a sensible default install directory for your OS (Unix: ~/.local/bin, Windows: %LOCALAPPDATA%/Programs/pjstats or %USERPROFILE%/Scripts) if `--install-dir` is not specified.
"""
# --------------------------------------------------
# Configuration
# --------------------------------------------------
# files to check for ignore patterns
IGNORE_FILES_TO_CHECK = ['.gitignore', '.ignore', '.dockerignore', '.npmignore']
CODE_LANGUAGES = {
# JavaScript / TypeScript
".js": ("JavaScript", "//", "code"),
".jsx": ("JavaScript (JSX)", "//", "code"),
".ts": ("TypeScript", "//", "code"),
".tsx": ("TypeScript (TSX)", "//", "code"),
".mjs": ("JavaScript (ESM)", "//", "code"),
".cjs": ("JavaScript (CJS)", "//", "code"),
# Python
".py": ("Python", "#", "code"),
".pyw": ("Python", "#", "code"),
".pyi": ("Python Stub", "#", "code"),
# Shell
".sh": ("Shell", "#", "code"),
".bash": ("Bash", "#", "code"),
".zsh": ("Zsh", "#", "code"),
".fish": ("Fish", "#", "code"),
".ksh": ("KornShell", "#", "code"),
# PowerShell
".ps1": ("PowerShell", "#", "code"),
".psm1": ("PowerShell Module", "#", "code"),
".psd1": ("PowerShell Data", "#", "code"),
# Batch / CMD
".bat": ("Batch", "REM", "code"),
".cmd": ("Batch", "REM", "code"),
# Web
".html": ("HTML", "<!--", "code"),
".htm": ("HTML", "<!--", "code"),
".css": ("CSS", "/*", "code"),
".scss": ("SCSS", "//", "code"),
".sass": ("Sass", "//", "code"),
".less": ("Less", "//", "code"),
".styl": ("Stylus", "//", "code"),
# Web Frameworks
".vue": ("Vue", "//", "code"),
".svelte": ("Svelte", "//", "code"),
".astro": ("Astro", "//", "code"),
# Dart / Flutter
".dart": ("Dart", "//", "code"),
# C-Family
".c": ("C", "//", "code"),
".h": ("C Header", "//", "code"),
".cpp": ("C++", "//", "code"),
".hpp": ("C++ Header", "//", "code"),
".cc": ("C++", "//", "code"),
".cxx": ("C++", "//", "code"),
".hxx": ("C++ Header", "//", "code"),
".c++": ("C++", "//", "code"),
".h++": ("C++ Header", "//", "code"),
".cs": ("C#", "//", "code"),
".m": ("Objective-C", "//", "code"),
".mm": ("Objective-C++", "//", "code"),
# Java / JVM
".java": ("Java", "//", "code"),
".kt": ("Kotlin", "//", "code"),
".kts": ("Kotlin Script", "//", "code"),
".groovy": ("Groovy", "//", "code"),
".scala": ("Scala", "//", "code"),
".clj": ("Clojure", ";;", "code"),
".cljs": ("ClojureScript", ";;", "code"),
".cljc": ("Clojure", ";;", "code"),
# Apple / iOS
".swift": ("Swift", "//", "code"),
# Build systems
".gradle": ("Gradle", "//", "code"),
".cmake": ("CMake", "#", "code"),
# Go
".go": ("Go", "//", "code"),
# Rust
".rs": ("Rust", "//", "code"),
# PHP
".php": ("PHP", "//", "code"),
# Ruby
".rb": ("Ruby", "#", "code"),
".rake": ("Ruby", "#", "code"),
".gemspec": ("Ruby", "#", "code"),
# Lua
".lua": ("Lua", "--", "code"),
# SQL
".sql": ("SQL", "--", "code"),
# R
".r": ("R", "#", "code"),
".R": ("R", "#", "code"),
".rmd": ("R Markdown", "#", "code"),
# Perl
".pl": ("Perl", "#", "code"),
".pm": ("Perl", "#", "code"),
".t": ("Perl Test", "#", "code"),
# Elixir / Erlang
".ex": ("Elixir", "#", "code"),
".exs": ("Elixir Script", "#", "code"),
".erl": ("Erlang", "%", "code"),
".hrl": ("Erlang Header", "%", "code"),
# Haskell
".hs": ("Haskell", "--", "code"),
".lhs": ("Literate Haskell", "--", "code"),
# OCaml
".ml": ("OCaml", "(*", "code"),
".mli": ("OCaml Interface", "(*", "code"),
# F#
".fs": ("F#", "//", "code"),
".fsx": ("F# Script", "//", "code"),
".fsi": ("F# Signature", "//", "code"),
# Nim
".nim": ("Nim", "#", "code"),
".nims": ("Nim Script", "#", "code"),
# Zig
".zig": ("Zig", "//", "code"),
# V
".v": ("V", "//", "code"),
# Crystal
".cr": ("Crystal", "#", "code"),
# Julia
".jl": ("Julia", "#", "code"),
# Elm
".elm": ("Elm", "--", "code"),
# PureScript
".purs": ("PureScript", "--", "code"),
# Fortran
".f": ("Fortran", "!", "code"),
".f90": ("Fortran", "!", "code"),
".f95": ("Fortran", "!", "code"),
".f03": ("Fortran", "!", "code"),
".f08": ("Fortran", "!", "code"),
".for": ("Fortran", "!", "code"),
# COBOL
".cob": ("COBOL", "*", "code"),
".cbl": ("COBOL", "*", "code"),
# Pascal / Delphi
".pas": ("Pascal", "//", "code"),
".dpr": ("Delphi", "//", "code"),
".dpk": ("Delphi", "//", "code"),
# D
".d": ("D", "//", "code"),
# Ada
".adb": ("Ada", "--", "code"),
".ads": ("Ada", "--", "code"),
# Lisp / Scheme / Racket
".lisp": ("Lisp", ";;", "code"),
".lsp": ("Lisp", ";;", "code"),
".cl": ("Common Lisp", ";;", "code"),
".scm": ("Scheme", ";;", "code"),
".ss": ("Scheme", ";;", "code"),
".rkt": ("Racket", ";;", "code"),
# Prolog
".pro": ("Prolog", "%", "code"),
# Tcl
".tcl": ("Tcl", "#", "code"),
# VB.NET / VBScript
".vb": ("VB.NET", "'", "code"),
".vbs": ("VBScript", "'", "code"),
# Assembly
".asm": ("Assembly", ";", "code"),
".s": ("Assembly", ";", "code"),
# VHDL / SystemVerilog
".vhd": ("VHDL", "--", "code"),
".vhdl": ("VHDL", "--", "code"),
".sv": ("SystemVerilog", "//", "code"),
".svh": ("SystemVerilog Header", "//", "code"),
# CoffeeScript
".coffee": ("CoffeeScript", "#", "code"),
# Handlebars
".hbs": ("Handlebars", "{{!--", "code"),
# Solidity
".sol": ("Solidity", "//", "code"),
# Terraform / HCL
".tf": ("Terraform", "#", "code"),
".tfvars": ("Terraform Vars", "#", "code"),
".hcl": ("HCL", "#", "code"),
# Nix
".nix": ("Nix", "#", "code"),
# GraphQL
".graphql": ("GraphQL", "#", "code"),
".gql": ("GraphQL", "#", "code"),
# Protocol Buffers / Thrift
".proto": ("Protobuf", "//", "code"),
".thrift": ("Thrift", "//", "code"),
# AWK
".awk": ("AWK", "#", "code"),
# Gleam
".gleam": ("Gleam", "//", "code"),
# Mojo
".mojo": ("Mojo", "#", "code"),
# Odin
".odin": ("Odin", "//", "code"),
# Wren
".wren": ("Wren", "//", "code"),
# ReScript / Reason
".res": ("ReScript", "//", "code"),
".resi": ("ReScript Interface", "//", "code"),
".re": ("Reason", "//", "code"),
".rei": ("Reason Interface", "//", "code"),
# Apex (Salesforce)
".cls": ("Apex", "//", "code"),
".trigger": ("Apex Trigger", "//", "code"),
# CUDA
".cu": ("CUDA", "//", "code"),
".cuh": ("CUDA Header", "//", "code"),
# Hack
".hack": ("Hack", "//", "code"),
# Jsonnet
".jsonnet": ("Jsonnet", "//", "code"),
".libsonnet": ("Jsonnet", "//", "code"),
# Dhall
".dhall": ("Dhall", "--", "code"),
# Starlark (Bazel)
".bzl": ("Starlark", "#", "code"),
".star": ("Starlark", "#", "code"),
}
CONFIG_LANGUAGES = {
".json": ("JSON", None, "config"),
".jsonc": ("JSON with Comments", "//", "config"),
".json5": ("JSON5", "//", "config"),
".yml": ("YAML", "#", "config"),
".yaml": ("YAML", "#", "config"),
# Flutter / Dart config
".arb": ("ARB (Flutter i18n)", None, "config"),
".pubspec": ("Pubspec", "#", "config"),
# Build / Project config
".xml": ("XML", "<!--", "config"),
".plist": ("Property List", "<!--", "config"),
".iml": ("IntelliJ Module", "<!--", "config"),
".properties": ("Properties", "#", "config"),
".env": ("Environment", "#", "config"),
".toml": ("TOML", "#", "config"),
".ini": ("INI", ";", "config"),
".cfg": ("Config", "#", "config"),
".conf": ("Config", "#", "config"),
# Lock files
".lock": ("Lock File", "#", "config"),
# Editor / VCS config
".editorconfig": ("EditorConfig", "#", "config"),
".gitignore": ("Git Ignore", "#", "config"),
".gitattributes": ("Git Attributes", "#", "config"),
".dockerignore": ("Docker Ignore", "#", "config"),
# Data formats
".csv": ("CSV", None, "config"),
".tsv": ("TSV", None, "config"),
# EDN (Clojure data)
".edn": ("EDN", ";;", "config"),
}
DOC_LANGUAGES = {
".md": ("Markdown", None, "docs"),
".rst": ("reStructuredText", None, "docs"),
".txt": ("Text", None, "docs"),
".adoc": ("AsciiDoc", None, "docs"),
".tex": ("LaTeX", "%", "docs"),
".org": ("Org Mode", None, "docs"),
".wiki": ("Wiki", None, "docs"),
".rdoc": ("RDoc", None, "docs"),
}
IGNORE_DIRECTORIES = {
".git", "node_modules", "__pycache__", "dist",
"build", ".venv", "backup", "backups", ".idea", ".vscode", ".pytest_cache",
".env", ".eggs", "venv", "env", ".tox", ".mypy_cache", ".ruff_cache",
".claude", ".openai", "logs", "log", "tmp", "temp", ".githooks",
# Flutter / Dart specific
".dart_tool", ".pub-cache", ".pub", "ephemeral", ".symlinks",
"Pods", "DerivedData", "xcuserdata", ".gradle",
# Additional
"vendor", "target", "out", "bin", "obj", "packages", "cache",
"site-packages", "virtualenv", ".cache", ".next", ".nuxt",
".svn", ".hg", "bower_components", ".terraform",
".serverless", "__MACOSX", "coverage", ".coverage",
".nyc_output", "htmlcov", "_build", ".build", ".sass-cache",
}
IGNORE_FILE_PATTERNS = (
"package-lock.json",
"yarn.lock",
"pnpm-lock.yaml",
"composer.lock",
"Gemfile.lock",
"Cargo.lock",
"poetry.lock",
".min.js",
".min.css",
".map",
".bundle.",
".chunk.",
)
# --------------------------------------------------
# Terminal Styling
# --------------------------------------------------
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
BLUE = "\033[38;5;75m"
GREEN = "\033[38;5;114m"
ORANGE = "\033[38;5;215m"
RED = "\033[38;5;203m"
CYAN = "\033[38;5;117m"
MAGENTA = "\033[38;5;183m"
YELLOW = "\033[38;5;229m"
WHITE = "\033[38;5;255m"
PURPLE = "\033[38;5;141m"
PINK = "\033[38;5;218m"
TEAL = "\033[38;5;80m"
# Animation-specific colors
HIDE_CURSOR = "\033[?25l"
SHOW_CURSOR = "\033[?25h"
GRAD = ["\033[38;5;51m", "\033[38;5;50m", "\033[38;5;49m", "\033[38;5;48m", "\033[38;5;47m", "\033[38;5;46m"]
GOLD = "\033[38;5;220m"
DARKGRAY = "\033[38;5;240m"
GRAY = "\033[38;5;245m"
LOGO = r"""
██████╗ ██╗███████╗████████╗ █████╗ ████████╗███████╗
██╔══██╗ ██║██╔════╝╚══██╔══╝██╔══██╗╚══██╔══╝██╔════╝
██████╔╝ ██║███████╗ ██║ ███████║ ██║ ███████╗
██╔═══╝ ██ ██║╚════██║ ██║ ██╔══██║ ██║ ╚════██║
██║ ╚█████╔╝███████║ ██║ ██║ ██║ ██║ ███████║
╚═╝ ╚════╝ ╚══════╝ ╚═╝ ╚═╝ ╚═╝ ╚═╝ ╚══════╝
"""
# --------------------------------------------------
# Animation & Progress
# --------------------------------------------------
def get_term_width():
try:
return os.get_terminal_size().columns
except Exception:
return 80
def clear_screen():
sys.stdout.write("\033[2J\033[H")
sys.stdout.flush()
def print_centered(text, color=""):
width = get_term_width()
clean = re.sub(r'\033\[[0-9;]*m', '', text)
padding = (width - len(clean)) // 2
print(" " * max(0, padding) + color + text + RESET)
def animate_logo():
sys.stdout.write(HIDE_CURSOR)
clear_screen()
lines = LOGO.strip().split('\n')
width = get_term_width()
print()
for i, line in enumerate(lines):
color = GRAD[i % len(GRAD)]
padding = (width - len(line)) // 2
sys.stdout.write(" " * max(0, padding) + color + line + RESET + "\n")
sys.stdout.flush()
time.sleep(0.05)
print()
print_centered("━━━ Project Statistics Analyzer ━━━", GRAY)
time.sleep(0.1)
print_centered(f"v{VERSION} │ Multi-Threaded Analytics", DARKGRAY)
print()
time.sleep(0.2)
def loading_animation(text, duration=0.5, style="dots"):
frames = {
"dots": ["", "", "", "", "", "", "", "", "", ""],
"blocks": ["", "", "", "", "", "", "", "", "", "", "", "", "", ""],
"arrows": ["", "", "", "", "", "", "", ""],
"pulse": ["", "", "", "", "", "", "", ""],
}
chars = frames.get(style, frames["dots"])
start = time.time()
i = 0
while time.time() - start < duration:
frame = chars[i % len(chars)]
sys.stdout.write(f"\r {CYAN}{frame}{RESET} {GRAY}{text}{RESET} ")
sys.stdout.flush()
time.sleep(0.08)
i += 1
sys.stdout.write(f"\r {GREEN}{RESET} {GRAY}{text}{RESET} \n")
sys.stdout.flush()
def startup_sequence(root_path):
animate_logo()
loading_animation("Initializing analyzer", 0.3, "dots")
loading_animation("Loading language definitions", 0.2, "pulse")
loading_animation(f"Target: {root_path}", 0.2, "arrows")
print()
width = min(60, get_term_width() - 4)
print_centered("" * width, DARKGRAY)
print()
def scan_progress(current, total, filename, width=40):
pct = (current / total * 100) if total else 0
filled = int(width * current / total) if total else 0
spinners = ["", "", "", "", "", "", "", "", "", ""]
spinner = spinners[current % len(spinners)]
if pct < 25:
bar_color = CYAN
elif pct < 50:
bar_color = BLUE
elif pct < 75:
bar_color = TEAL
else:
bar_color = GREEN
bar_str = f"{bar_color}{'' * filled}{DARKGRAY}{'' * (width - filled)}{RESET}"
line1 = f" {CYAN}{spinner}{RESET} {bar_str} {GOLD}{pct:5.1f}%{RESET} {DARKGRAY}{RESET} {GRAY}{current:,}/{total:,}{RESET}"
ext = Path(filename).suffix.lower()
icon = "📄"
if ext in CODE_LANGUAGES: icon = "💻"
elif ext in CONFIG_LANGUAGES: icon = "⚙️"
elif ext in DOC_LANGUAGES: icon = "📝"
max_fn = get_term_width() - 10
if len(filename) > max_fn:
filename = "..." + filename[-(max_fn - 3):]
line2 = f" {icon} {GRAY}{filename}{RESET}"
sys.stdout.write(f"\033[2A\033[2K{line1}\n\033[2K{line2}\n")
sys.stdout.flush()
def clear_scan_progress():
sys.stdout.write("\033[2A\033[2K\033[2K")
sys.stdout.write(f" {GREEN}{RESET} {GRAY}Scan complete{RESET}\n\n")
sys.stdout.flush()
# --------------------------------------------------
# Helpers / Metrics
# --------------------------------------------------
def horizontal_bar(percentage, width=24, color=None, bg_color=None):
filled = int(width * percentage / 100)
bar_filled = "" * filled
bar_empty = "" * (width - filled)
if color:
return f"{color}{bar_filled}{RESET}{DIM}{bar_empty}{RESET}"
return bar_filled + bar_empty
def format_number_short(n):
"""Format large numbers with K/M/B suffixes."""
if n >= 1_000_000_000:
return f"{n / 1_000_000_000:.1f}B"
if n >= 1_000_000:
return f"{n / 1_000_000:.1f}M"
if n >= 1_000:
return f"{n / 1_000:.1f}K"
return str(n)
def should_ignore_file(path: Path, root: Path, ignore_patterns=IGNORE_FILE_PATTERNS, ignore_globs=None, negation_globs=None):
"""Return True if file should be ignored based on name patterns or glob patterns.
- ignore_patterns: simple substrings (legacy behavior)
- ignore_globs: list of glob patterns relative to root (supports wildcards and paths)
- negation_globs: list of patterns that un-ignore files (overrides ignore_globs)
"""
name = path.name
# Calculate relative path for pattern matching
try:
rel = str(path.relative_to(root)).replace('\\', '/')
except ValueError:
# If can't get relative path, try resolved
try:
rel = str(path.resolve().relative_to(root.resolve())).replace('\\', '/')
except ValueError:
return True # Can't determine path, ignore it
# Check if file matches a negation pattern (should NOT be ignored)
if negation_globs:
for pat in negation_globs:
if fnmatch.fnmatch(rel, pat) or fnmatch.fnmatch(name, pat):
return False # Explicitly un-ignored
# Check simple substring patterns
if any(pattern in name for pattern in ignore_patterns):
return True
# Check glob patterns
if ignore_globs:
for pat in ignore_globs:
# try match against relative path and filename
if fnmatch.fnmatch(rel, pat) or fnmatch.fnmatch(name, pat):
return True
return False
_ALL_LANGUAGES = {}
_ALL_LANGUAGES.update(CODE_LANGUAGES)
_ALL_LANGUAGES.update(CONFIG_LANGUAGES)
_ALL_LANGUAGES.update(DOC_LANGUAGES)
def detect_language(extension):
return _ALL_LANGUAGES.get(extension)
# --------------------------------------------------
# Pre-compiled regex patterns (module-level for performance)
# --------------------------------------------------
_RE_PY_FUNC = re.compile(r"^\s*(async\s+def|def)\s+\w+")
_RE_PY_CLASS = re.compile(r"^\s*class\s+\w+")
_RE_JS_FUNC = re.compile(r"\bfunction\b|=>")
_RE_JS_CLASS = re.compile(r"^\s*(export\s+)?(default\s+)?(abstract\s+)?class\s+\w+")
_RE_C_FAMILY = re.compile(r"^\s*(?:public|private|protected|static|final|abstract|async|override)?\s*(?:void|int|double|float|bool|String|var|dynamic|Future|Stream|List|Map|Set|[A-Z]\w*(?:<[^>]+>)?)?\s+\w+\s*\([^)]*\)\s*(?:async)?\s*[{:]?")
_RE_DART_FUNC = re.compile(r"^\s*(?:Future|Stream|void|bool|int|double|String|dynamic|var|[A-Z]\w*(?:<[^>]+>)?)\s+\w+\s*\(")
_RE_JAVA_METHOD = re.compile(r"^\s*(?:@\w+\s*)*(?:public|private|protected)?\s*(?:static)?\s*(?:final)?\s*(?:void|int|long|float|double|boolean|char|byte|short|String|[A-Z]\w*(?:<[^>]+>)?)\s+\w+\s*\(")
_RE_SWIFT_FUNC = re.compile(r"^\s*(?:@\w+\s*)*(?:public|private|internal|fileprivate|open)?\s*(?:static|class)?\s*func\s+\w+")
_RE_GO_FUNC = re.compile(r"^\s*func\s+(?:\([^)]+\)\s*)?\w+\s*\(")
_RE_RUST_FUNC = re.compile(r"^\s*(?:pub\s+)?(?:async\s+)?fn\s+\w+")
_RE_RUBY_DEF = re.compile(r"^\s*def\s+\w+")
_RE_RUBY_CLASS = re.compile(r"^\s*(class|module)\s+\w+")
_RE_TODO = re.compile(r"\b(TODO|FIXME)\b", re.IGNORECASE)
_RE_GENERAL_CLASS = re.compile(r"^\s*(?:export\s+)?(?:abstract\s+)?(?:public|private|protected|internal|open|sealed|data|final)?\s*(?:abstract\s+)?(?:class|struct|enum|interface|mixin|extension|typedef|protocol|trait)\s+\w+")
_RE_RUST_CLASS = re.compile(r"^\s*(?:pub\s+)?(?:struct|enum|trait|impl)\s+\w+")
_RE_GO_CLASS = re.compile(r"^\s*type\s+\w+\s+(?:struct|interface)\b")
_RE_ELIXIR_FUNC = re.compile(r"^\s*(def|defp|defmacro|defmacrop)\s+\w+")
_RE_ELIXIR_CLASS = re.compile(r"^\s*defmodule\s+\w+")
_RE_ERLANG_FUNC = re.compile(r"^[a-z]\w*\s*\(")
_RE_HASKELL_FUNC = re.compile(r"^[a-z]\w*\s+::")
_RE_HASKELL_DATA = re.compile(r"^\s*(data|newtype|class|type)\s+[A-Z]")
_RE_PERL_FUNC = re.compile(r"^\s*sub\s+\w+")
_RE_PERL_CLASS = re.compile(r"^\s*package\s+\w+")
_RE_LUA_FUNC = re.compile(r"(^\s*(local\s+)?function\s+\w+|=\s*function\s*\()")
_RE_PS_FUNC = re.compile(r"^\s*function\s+\w+", re.IGNORECASE)
_RE_PS_CLASS = re.compile(r"^\s*class\s+\w+", re.IGNORECASE)
_RE_SOLIDITY_FUNC = re.compile(r"^\s*(function|modifier|constructor|receive|fallback)\s+\w*")
_RE_SOLIDITY_CLASS = re.compile(r"^\s*(contract|library|interface)\s+\w+")
_RE_LISP_FUNC = re.compile(r"^\s*\(def(n|un|macro|multi|method|record|type|protocol)\s+")
_RE_LISP_IMPORT = re.compile(r"^\s*\((?:require|import|use|ns)\s+")
# Import patterns
_RE_IMPORT_PY = re.compile(r"^\s*(import|from)\s+")
_RE_IMPORT_JS = re.compile(r"^\s*(import\s+|const\s+.*=\s*require\(|require\()")
_RE_IMPORT_DART = re.compile(r"^\s*import\s+['\"]")
_RE_IMPORT_JAVA = re.compile(r"^\s*import\s+")
_RE_IMPORT_GO = re.compile(r"^\s*import\s+")
_RE_IMPORT_RUST = re.compile(r"^\s*(use|extern\s+crate)\s+")
_RE_IMPORT_C = re.compile(r"^\s*#\s*include\s+")
_RE_IMPORT_RUBY = re.compile(r"^\s*(require|require_relative|load)\s+")
_RE_IMPORT_ELIXIR = re.compile(r"^\s*(import|use|alias|require)\s+")
_RE_IMPORT_HASKELL = re.compile(r"^\s*import\s+")
_RE_IMPORT_PERL = re.compile(r"^\s*(use|require)\s+")
_RE_IMPORT_LUA = re.compile(r"^\s*(require|dofile|loadfile)\s*[\(\"']")
_RE_IMPORT_PS = re.compile(r"^\s*(Import-Module|using\s+(module|namespace|assembly))\s+", re.IGNORECASE)
# Language family sets
_LANGS_JS = frozenset({"JavaScript", "JavaScript (JSX)", "JavaScript (ESM)", "JavaScript (CJS)",
"TypeScript", "TypeScript (TSX)", "Shell", "Bash", "Zsh", "Groovy", "Scala", "PHP",
"CoffeeScript", "Vue", "Svelte", "Astro"})
_LANGS_C_FAMILY = frozenset({"C", "C Header", "C++", "C++ Header", "C#", "Objective-C", "Objective-C++",
"Java", "Kotlin", "Kotlin Script", "Gradle", "CUDA", "CUDA Header", "D",
"Hack", "Apex", "Apex Trigger"})
_LANGS_C_INCLUDE = frozenset({"C", "C Header", "C++", "C++ Header", "Objective-C", "Objective-C++", "CUDA", "CUDA Header"})
_LANGS_DART = frozenset({"Dart"})
_LANGS_SWIFT = frozenset({"Swift"})
_LANGS_GO = frozenset({"Go"})
_LANGS_RUST = frozenset({"Rust"})
_LANGS_RUBY = frozenset({"Ruby"})
_LANGS_ELIXIR = frozenset({"Elixir", "Elixir Script"})
_LANGS_ERLANG = frozenset({"Erlang", "Erlang Header"})
_LANGS_HASKELL = frozenset({"Haskell", "Literate Haskell", "Elm", "PureScript"})
_LANGS_PERL = frozenset({"Perl", "Perl Test"})
_LANGS_LUA = frozenset({"Lua"})
_LANGS_PS = frozenset({"PowerShell", "PowerShell Module", "PowerShell Data"})
_LANGS_SOLIDITY = frozenset({"Solidity"})
_LANGS_LISP = frozenset({"Clojure", "ClojureScript", "Lisp", "Common Lisp", "Scheme", "Racket"})
_LANGS_PYTHON = frozenset({"Python", "Python Stub"})
# --------------------------------------------------
# Analysis
# --------------------------------------------------
def load_ignore_files(root: Path):
"""Load ignore patterns from .gitignore/.ignore and return a list of glob patterns, directory names to ignore, and negation patterns."""
root = Path(root)
globs = []
dir_names = set()
negation_globs = [] # patterns that start with ! (un-ignore)
negation_dirs = set() # directory negations
for fname in IGNORE_FILES_TO_CHECK:
fpath = root / fname
if not fpath.exists():
continue
try:
with fpath.open('r', encoding='utf-8', errors='ignore') as fh:
for line in fh:
line = line.strip()
if not line or line.startswith('#'):
continue
# handle negations (lines starting with !)
if line.startswith('!'):
pattern = line[1:] # remove the !
if pattern.endswith('/'):
negation_dirs.add(pattern.rstrip('/'))
else:
negation_globs.append(pattern)
continue
# directory pattern
if line.endswith('/'):
dir_names.add(line.rstrip('/'))
else:
globs.append(line)
except Exception:
continue
return list(globs), dir_names, list(negation_globs), negation_dirs
def analyze_file(path: Path):
info = detect_language(path.suffix.lower())
if not info:
# fallback for special filenames (Dockerfile, Dockerfile.*, Makefile) or shebang
name = path.name
if name in SPECIAL_FILES:
info = SPECIAL_FILES[name]
elif name.lower().startswith('dockerfile'):
# catch Dockerfile, Dockerfile.bot, Dockerfile.native, etc.
info = ('Dockerfile', '#', 'code')
else:
try:
with path.open('r', encoding='utf-8', errors='ignore') as fh:
first = fh.readline()
if first.startswith('#!'):
info = ('Shell', '#', 'code')
except Exception:
pass
if not info:
return None
language, comment_token, category = info
# Skip extremely large files (minified bundles, generated code, blobs)
try:
size = path.stat().st_size
except OSError:
return None
if size > 1_000_000: # 1MB cap — files above this are overwhelmingly minified/generated
# Still report the file but with minimal metadata (no line-by-line analysis)
return {
"path": str(path),
"relpath": str(path),
"language": language,
"category": category,
"code_lines": 0,
"code_chars": 0,
"comment_lines": 0,
"empty_lines": 0,
"total_lines": 0,
"size_bytes": size,
"function_count": 0,
"class_count": 0,
"todo_count": 0,
"longest_line": 0,
"avg_line_length": 0,
"avg_func_length": 0,
"max_nesting_depth": 0,
"long_lines": 0,
"import_count": 0,
"comment_ratio": 0,
"has_comment": bool(comment_token),
}
# Single-pass analysis: line counting + function/class/import detection
code = comments = empty = 0
code_chars = 0
func_count = class_count = todo_count = 0
longest_line = 0
avg_len_acc = 0
max_nesting = 0
current_nesting = 0
long_lines = 0
import_count = 0
total_lines = 0
# Select the right import matcher for this language
import_matcher = None
if language in _LANGS_PYTHON:
import_matcher = _RE_IMPORT_PY
elif language in _LANGS_DART:
import_matcher = _RE_IMPORT_DART
elif language in _LANGS_JS:
import_matcher = _RE_IMPORT_JS
elif language in _LANGS_C_FAMILY:
import_matcher = _RE_IMPORT_C if language in _LANGS_C_INCLUDE else _RE_IMPORT_JAVA
elif language in _LANGS_GO:
import_matcher = _RE_IMPORT_GO
elif language in _LANGS_RUST:
import_matcher = _RE_IMPORT_RUST
elif language in _LANGS_RUBY:
import_matcher = _RE_IMPORT_RUBY
elif language in _LANGS_SWIFT:
import_matcher = _RE_IMPORT_JAVA
elif language in _LANGS_ELIXIR:
import_matcher = _RE_IMPORT_ELIXIR
elif language in _LANGS_HASKELL:
import_matcher = _RE_IMPORT_HASKELL
elif language in _LANGS_PERL:
import_matcher = _RE_IMPORT_PERL
elif language in _LANGS_LUA:
import_matcher = _RE_IMPORT_LUA
elif language in _LANGS_PS:
import_matcher = _RE_IMPORT_PS
elif language in _LANGS_LISP:
import_matcher = _RE_LISP_IMPORT
# Select the right func/class matchers for this language
func_matcher = class_matcher = None
func_is_search = False # use .search() instead of .match()
func_matcher2 = None # secondary func matcher (for dart/c-family)
if language in _LANGS_PYTHON:
func_matcher = _RE_PY_FUNC
class_matcher = _RE_PY_CLASS
elif language in _LANGS_JS:
func_matcher = _RE_JS_FUNC
func_is_search = True
class_matcher = _RE_JS_CLASS
elif language in _LANGS_DART:
func_matcher = _RE_DART_FUNC
func_matcher2 = _RE_C_FAMILY
class_matcher = _RE_GENERAL_CLASS
elif language in _LANGS_C_FAMILY:
func_matcher = _RE_JAVA_METHOD
func_matcher2 = _RE_C_FAMILY
class_matcher = _RE_GENERAL_CLASS
elif language in _LANGS_SWIFT:
func_matcher = _RE_SWIFT_FUNC
class_matcher = _RE_GENERAL_CLASS
elif language in _LANGS_GO:
func_matcher = _RE_GO_FUNC
class_matcher = _RE_GO_CLASS
elif language in _LANGS_RUST:
func_matcher = _RE_RUST_FUNC
class_matcher = _RE_RUST_CLASS
elif language in _LANGS_RUBY:
func_matcher = _RE_RUBY_DEF
class_matcher = _RE_RUBY_CLASS
elif language in _LANGS_ELIXIR:
func_matcher = _RE_ELIXIR_FUNC
class_matcher = _RE_ELIXIR_CLASS
elif language in _LANGS_ERLANG:
func_matcher = _RE_ERLANG_FUNC
elif language in _LANGS_HASKELL:
func_matcher = _RE_HASKELL_FUNC
class_matcher = _RE_HASKELL_DATA
elif language in _LANGS_PERL:
func_matcher = _RE_PERL_FUNC
class_matcher = _RE_PERL_CLASS
elif language in _LANGS_LUA:
func_matcher = _RE_LUA_FUNC
func_is_search = True
elif language in _LANGS_PS:
func_matcher = _RE_PS_FUNC
class_matcher = _RE_PS_CLASS
elif language in _LANGS_SOLIDITY:
func_matcher = _RE_SOLIDITY_FUNC
class_matcher = _RE_SOLIDITY_CLASS
elif language in _LANGS_LISP:
func_matcher = _RE_LISP_FUNC
else:
class_matcher = _RE_GENERAL_CLASS
_todo_search = _RE_TODO.search
_LINE_LEN_CAP = 10_000 # skip regex on very long lines (minified code)
_MAX_LINES = 100_000 # bail out on files with excessive line counts
try:
with path.open("r", encoding="utf-8", errors="ignore") as fh:
for line in fh:
total_lines += 1
if total_lines > _MAX_LINES:
break
stripped = line.strip()
# Line counting (code / comment / empty)
if not stripped:
empty += 1
elif comment_token and stripped.startswith(comment_token):
comments += 1
else:
code += 1
code_chars += len(stripped)
# Line length metrics
l_len = len(line) - 1 if line.endswith('\n') else len(line)
avg_len_acc += l_len
if l_len > longest_line:
longest_line = l_len
if l_len > 120:
long_lines += 1
# Skip regex analysis on extremely long lines (minified code)
if l_len > _LINE_LEN_CAP:
continue
# Nesting depth (brace-based)
if stripped and not stripped.startswith(('//', '#', '/*', '*', '<!--')):
delta = (stripped.count('{') + stripped.count('(')) - (stripped.count('}') + stripped.count(')'))
current_nesting += delta
if current_nesting < 0:
current_nesting = 0
if current_nesting > max_nesting:
max_nesting = current_nesting
# TODO/FIXME detection
if _todo_search(line):
todo_count += 1
# Import detection
if import_matcher and import_matcher.match(line):
import_count += 1
# Function/class detection
if class_matcher and class_matcher.match(line):
class_count += 1
elif func_matcher:
if func_is_search:
if func_matcher.search(line):
func_count += 1
else:
if func_matcher.match(line):
func_count += 1
elif func_matcher2 and func_matcher2.match(line):
func_count += 1
except Exception:
return None
avg_line_len = (avg_len_acc / total_lines) if total_lines else 0.0
total = code + comments + empty
comment_ratio = (comments / code * 100) if code else 0
avg_func_length = (code / func_count) if func_count else 0
return {
"path": str(path),
"relpath": str(path),
"language": language,
"category": category,
"code_lines": code,
"code_chars": code_chars,
"comment_lines": comments,
"empty_lines": empty,
"total_lines": total,
"size_bytes": size,
"function_count": func_count,
"class_count": class_count,
"todo_count": todo_count,
"longest_line": longest_line,
"avg_line_length": avg_line_len,
"avg_func_length": avg_func_length,
"max_nesting_depth": max_nesting,
"long_lines": long_lines,
"import_count": import_count,
"comment_ratio": comment_ratio,
"has_comment": bool(comment_token),
}
def get_git_info(root: Path, commit_msg_max: int = 120):
"""Return git info dict if 'root' is inside a git worktree, otherwise None.
commit_msg_max: max characters for the single-line commit message shown (truncated).
"""
try:
p = subprocess.run(["git", "rev-parse", "--is-inside-work-tree"], cwd=str(root), capture_output=True, text=True)
if p.returncode != 0 or p.stdout.strip() != "true":
return None
branch = subprocess.run(["git", "rev-parse", "--abbrev-ref", "HEAD"], cwd=str(root), capture_output=True, text=True).stdout.strip()
commit_out = subprocess.run(["git", "log", "-1", "--pretty=format:%h;%an;%ad;%B", "--date=short"], cwd=str(root), capture_output=True, text=True)
latest = None
if commit_out.returncode == 0 and commit_out.stdout:
# use %B for full raw body, then split fields manually
raw = commit_out.stdout.strip()
# the format we used here is: <hash>;<author>;<date>;<body...>
parts = raw.split(";", 3)
if len(parts) == 4:
short, author, date, msg = parts
# collapse multi-line messages to single line and trim
single = " ".join(line.strip() for line in msg.splitlines()).strip()
if len(single) > commit_msg_max:
short_msg = single[:commit_msg_max - 1].rstrip() + ""
else:
short_msg = single
latest = {"hash": short, "author": author, "date": date, "message": msg, "message_short": short_msg}
# commit count
commit_count = 0
ccount = subprocess.run(["git", "rev-list", "--count", "HEAD"], cwd=str(root), capture_output=True, text=True)
if ccount.returncode == 0 and ccount.stdout.strip().isdigit():
commit_count = int(ccount.stdout.strip())
# files changed in last commit
files_changed = []
files_out = subprocess.run(["git", "diff-tree", "--no-commit-id", "--name-only", "-r", "HEAD"], cwd=str(root), capture_output=True, text=True)
if files_out.returncode == 0 and files_out.stdout:
files_changed = [l for l in files_out.stdout.splitlines() if l.strip()]
# stats (insertions/deletions) for last commit
insertions = deletions = 0
stats_out = subprocess.run(["git", "show", "--numstat", "-1", "HEAD"], cwd=str(root), capture_output=True, text=True)
if stats_out.returncode == 0 and stats_out.stdout:
for line in stats_out.stdout.splitlines():
parts = line.split()
if len(parts) >= 3:
try:
ins = int(parts[0]) if parts[0].isdigit() else 0
dels = int(parts[1]) if parts[1].isdigit() else 0
insertions += ins
deletions += dels
except Exception:
pass
# remote URL
remote = subprocess.run(["git", "config", "--get", "remote.origin.url"], cwd=str(root), capture_output=True, text=True)
remote_url = remote.stdout.strip() if remote.returncode == 0 and remote.stdout else None
status_out = subprocess.run(["git", "status", "--porcelain"], cwd=str(root), capture_output=True, text=True)
changed = untracked = 0
dirty = False
if status_out.returncode == 0:
lines = [l for l in status_out.stdout.splitlines() if l.strip()]
changed = sum(1 for l in lines if not l.startswith("??"))
untracked = sum(1 for l in lines if l.startswith("??"))
dirty = bool(lines)
ahead = behind = 0
ub = subprocess.run(["git", "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}"], cwd=str(root), capture_output=True, text=True)
if ub.returncode == 0:
upstream = ub.stdout.strip()
ab = subprocess.run(["git", "rev-list", "--left-right", "--count", f"{upstream}...HEAD"], cwd=str(root), capture_output=True, text=True)
if ab.returncode == 0 and ab.stdout:
behind_s, ahead_s = ab.stdout.strip().split()
behind = int(behind_s)
ahead = int(ahead_s)
return {
"branch": branch,
"dirty": dirty,
"changed": changed,
"untracked": untracked,
"ahead": ahead,
"behind": behind,
"latest_commit": latest,
"commit_count": commit_count,
"last_files_changed": files_changed,
"last_commit_insertions": insertions,
"last_commit_deletions": deletions,
"remote_url": remote_url,
}
except Exception:
return None
# --------------------------------------------------
# CLI
# --------------------------------------------------
SPECIAL_FILES = {
'Dockerfile': ("Dockerfile", "#", "code"),
'Makefile': ("Makefile", "#", "code"),
}
def parse_args():
p = argparse.ArgumentParser(description="pjstats -- enhanced project stats", epilog=HELP_TEXT, formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("path", nargs="?", default='.', help="Project root path")
p.add_argument("--json", action="store_true", help="Output JSON summary")
p.add_argument("--top", type=int, default=5, help="Top N files to display")
p.add_argument("--ext", nargs="*", help="Limit to specific extensions (e.g. .py .js)")
p.add_argument("--min-code-lines", type=int, default=0, help="Ignore files with fewer code lines than this")
p.add_argument("--exclude", nargs="*", default=[], help="Directories to ignore recursively (relative to project root)")
p.add_argument("--include-hidden", action="store_true", help="Include hidden directories/files")
p.add_argument("--no-color", action="store_true", help="Disable color output")
p.add_argument("--no-git", action="store_true", help="Disable Git info detection")
p.add_argument("--no-animation", action="store_true", help="Disable startup animation and progress bars")
p.add_argument("--bypass-excluded", action="store_true", help="Bypass all ignore rules (.gitignore, built-in excludes, hidden dirs, etc.) and scan everything")
p.add_argument("--workers", type=int, default=0, help="Number of threads for parallel analysis (0=auto)")
p.add_argument("--commit-msg-max", type=int, default=120, help="Max characters to show for commit message (single-line, truncated)")
p.add_argument("--install", action="store_true", help="Install pjstats to a directory in your PATH (auto-detected per OS if not provided)")
p.add_argument("--uninstall", action="store_true", help="Remove installed pjstats from install directory")
p.add_argument("--install-dir", default=None, help="Installation directory (auto-detected per OS if not set)")
return p.parse_args()
def default_install_dir():
"""Return a sensible default install directory depending on the OS."""
system = platform.system()
if system == 'Windows':
# prefer %LOCALAPPDATA%/Programs/pjstats, fallback to %USERPROFILE%/Scripts
local = os.environ.get('LOCALAPPDATA')
if local:
return str(Path(local) / 'Programs' / 'pjstats')
up = os.environ.get('USERPROFILE')
if up:
return str(Path(up) / 'Scripts')
return str(Path.home() / 'Scripts')
else:
# Unix-like default
return str(Path.home() / '.local' / 'bin')
def install_self(install_dir: str):
install_dir = Path(os.path.expanduser(install_dir))
install_dir.mkdir(parents=True, exist_ok=True)
src = Path(__file__).resolve()
dest = install_dir / 'pjstats'
try:
# normalize line endings while installing to avoid CRLF shebang issues
with src.open('r', encoding='utf-8', errors='ignore') as fh:
text = fh.read()
with dest.open('w', encoding='utf-8', newline='\n') as fh:
fh.write(text.replace('\r\n', '\n').replace('\r', '\n'))
dest.chmod(0o755)
print(f"Installed pjstats -> {dest}")
# warn if install_dir not in PATH
path_env = os.environ.get('PATH','')
if str(install_dir) not in path_env.split(os.pathsep):
if platform.system() == 'Windows':
print(f"Note: {install_dir} is not in your PATH. Consider adding it to your User PATH or copy to a directory already on PATH.")
else:
print(f"Note: {install_dir} is not in your PATH. Add it to your PATH (e.g. export PATH=\"{install_dir}:$PATH\") to use 'pjstats' directly.")
return True
except Exception as exc:
print(f"Failed to install: {exc}")
return False
def uninstall_self(install_dir: str):
install_dir = Path(os.path.expanduser(install_dir))
dest = install_dir / 'pjstats'
try:
if dest.exists():
dest.unlink()
print(f"Removed {dest}")
return True
else:
print(f"pjstats is not installed at {dest}")
return False
except Exception as exc:
print(f"Failed to uninstall: {exc}")
return False
def compute_quality_score(data):
"""Compute a 0-100 quality score using smooth curves for realistic assessment.
Uses continuous penalties proportional to how far each metric deviates
from ideal values, rather than hard step thresholds.
"""
score = 100.0
# 1. Comment ratio (max -30 pts) - smooth curve with over-commenting penalty
if data.get('has_comment', True) and data['code_lines'] > 0:
cr = data['comment_lines'] / data['code_lines'] * 100
if cr < 1:
score -= 30
elif cr < 5:
# 30 -> 18 as cr goes 0 -> 5
score -= 30 - (cr / 5) * 12
elif cr < 10:
# 18 -> 8 as cr goes 5 -> 10
score -= 18 - ((cr - 5) / 5) * 10
elif cr < 20:
# 8 -> 0 as cr goes 10 -> 20
score -= 8 - ((cr - 10) / 10) * 8
# Penalty for excessive commenting (>60% is unusual)
if cr > 60:
score -= min(10, (cr - 60) / 10 * 5)
# 2. Lines per function (max -20 pts) - smooth curve
if data['function_count'] > 0:
lpf = data['code_lines'] / data['function_count']
if lpf > 100:
score -= 20
elif lpf > 60:
score -= 10 + (lpf - 60) / 40 * 10
elif lpf > 30:
score -= (lpf - 30) / 30 * 10
# Too many tiny functions penalty (< 3 lines avg with many funcs)
if lpf < 3 and data['function_count'] > 5:
score -= 5
# 3. TODO/FIXME density per 1000 lines (max -10 pts) - proportional
td = data.get('todo_density', 0)
if td > 0:
score -= min(10, td * 0.5)
# 4. Max nesting depth (max -15 pts) - smooth from depth 3+
mnd = data.get('max_nesting_depth', 0)
if mnd > 3:
score -= min(15, (mnd - 3) * 2.5)
# 5. Long line ratio (max -10 pts) - proportional
llr = data.get('long_line_ratio', 0)
if llr > 0:
score -= min(10, llr * 0.7)
# 6. Large file count (max -10 pts) - proportional to % of files > 300 lines
if data['files'] > 0:
lf_pct = data.get('large_file_count', 0) / data['files'] * 100
if lf_pct > 0:
score -= min(10, lf_pct * 0.2)
# 7. File size consistency (max -5 pts) - coefficient of variation
med = data.get('median_code_lines', 0)
sd = data.get('stddev_code_lines', 0)
if med > 0:
cv = sd / med
if cv > 0.5:
score -= min(5, (cv - 0.5) * 3.3)
return max(0, min(100, round(score, 1)))
def score_color(score, no_color=False):
"""Return an ANSI color for a quality score."""
if no_color:
return ""
if score >= 80:
return GREEN
elif score >= 60:
return YELLOW
elif score >= 40:
return ORANGE
else:
return RED
def score_grade(score):
"""Return a letter grade for a quality score."""
if score >= 90:
return "A+"
elif score >= 85:
return "A"
elif score >= 80:
return "A-"
elif score >= 75:
return "B+"
elif score >= 70:
return "B"
elif score >= 65:
return "B-"
elif score >= 60:
return "C+"
elif score >= 55:
return "C"
elif score >= 50:
return "C-"
elif score >= 45:
return "D+"
elif score >= 40:
return "D"
else:
return "F"
def print_summary(result, root: Path, args):
if args.no_color:
global BOLD, DIM, BLUE, GREEN, ORANGE, RED, RESET, CYAN, MAGENTA, YELLOW, WHITE, PURPLE, PINK, TEAL
BOLD = DIM = BLUE = GREEN = ORANGE = RED = RESET = CYAN = MAGENTA = YELLOW = WHITE = PURPLE = PINK = TEAL = ""
language_stats = result["language_stats"]
category_totals = result["category_totals"]
per_file = result["per_file"]
total_files = len(per_file)
total_lines = sum(f["total_lines"] for f in per_file)
print(f"\n{BOLD}{CYAN}PROJECT OVERVIEW{RESET}")
print(f"{DIM}{root}{RESET}\n")
if not args.no_git:
git = get_git_info(root, args.commit_msg_max)
if git:
s = f"{PURPLE}Git:{RESET} branch {YELLOW}{git['branch']}{RESET}"
if git.get('dirty'):
s += f" | {ORANGE}modified:{git.get('changed',0)}{RESET} {DIM}untracked:{git.get('untracked',0)}{RESET}"
if git.get('ahead') or git.get('behind'):
ahead_c = GREEN if git.get('ahead', 0) > 0 else DIM
behind_c = RED if git.get('behind', 0) > 0 else DIM
s += f" | {ahead_c}ahead:{git.get('ahead',0)}{RESET} {behind_c}behind:{git.get('behind',0)}{RESET}"
if git.get('commit_count'):
s += f" | {CYAN}commits:{git.get('commit_count')}{RESET}"
cm = git.get('latest_commit') or {}
if cm:
s += f" | latest {MAGENTA}{cm.get('hash')}{RESET} by {BLUE}{cm.get('author')}{RESET} on {DIM}{cm.get('date')}{RESET}"
# insertion/deletion summary for last commit
ins = git.get('last_commit_insertions', 0) or 0
dels = git.get('last_commit_deletions', 0) or 0
files_changed = git.get('last_files_changed') or []
if files_changed:
s += f" | last {GREEN}+{ins}{RESET}/{RED}-{dels}{RESET} in {WHITE}{len(files_changed)}{RESET} files"
print(s)
# show the commit message on its own (single-line, truncated)
if cm and cm.get('message_short'):
print(f" {DIM}Message:{RESET} {cm.get('message_short')}")
# print a concise list of files changed in the last commit (first 6)
if files_changed:
show = files_changed[:6]
more = len(files_changed) - len(show)
files_line = f"{DIM}, {RESET}".join(f"{TEAL}{f}{RESET}" for f in show)
if more > 0:
files_line += f" {DIM}(+{more} more){RESET}"
print(f" {DIM}Changed files:{RESET} {files_line}")
total_chars = sum(f.get("code_chars", 0) for f in per_file)
total_funcs = sum(f.get("function_count", 0) for f in per_file)
total_classes = sum(f.get("class_count", 0) for f in per_file)
total_todos = sum(f.get("todo_count", 0) for f in per_file)
print(f"{DIM}Files scanned :{RESET} {CYAN}{total_files}{RESET}")
print(f"{DIM}Total lines :{RESET} {CYAN}{total_lines:,}{RESET}")
print(f"{DIM}Total characters (code) :{RESET} {CYAN}{total_chars:,}{RESET}")
print(f"{DIM}Functions / Methods :{RESET} {MAGENTA}{total_funcs:,}{RESET}")
print(f"{DIM}Classes / Structs :{RESET} {PURPLE}{total_classes:,}{RESET}")
print(f"{DIM}TODO / FIXME markers :{RESET} {ORANGE}{total_todos:,}{RESET}")
print()
# Category colors
cat_colors = {"code": BLUE, "config": ORANGE, "docs": MAGENTA}
print(f"{BOLD}{CYAN}CONTENT DISTRIBUTION{RESET}")
for category, lines in category_totals.items():
percent = (lines / total_lines * 100) if total_lines else 0
label = category.capitalize()
cat_color = cat_colors.get(category, WHITE)
print(
f"{cat_color}{label:<15}{RESET} "
f"{WHITE}{lines:>8,}{RESET} {DIM}lines{RESET} "
f"{YELLOW}{percent:>5.1f}%{RESET} "
f"{horizontal_bar(percent, color=cat_color)}"
)
print(f"\n{BOLD}{CYAN}LANGUAGE IMPACT (actual code){RESET}")
code_total = sum(v["code_lines"] for v in language_stats.values())
chars_total = sum(v["code_chars"] for v in language_stats.values())
# Color mapping for language groups
def get_lang_color(lang):
if "Python" in lang or "Mojo" in lang: return YELLOW
if "Dart" in lang or "Flutter" in lang: return TEAL
if "Java" in lang or "Kotlin" in lang or "Gradle" in lang: return ORANGE
if "JavaScript" in lang or "TypeScript" in lang or "CoffeeScript" in lang: return YELLOW
if "Swift" in lang: return ORANGE
if "C++" in lang or "C#" in lang or lang == "C" or "Header" in lang or "CUDA" in lang: return BLUE
if "Objective" in lang: return BLUE
if "Markdown" in lang or lang in ("reStructuredText", "AsciiDoc", "LaTeX", "Org Mode"): return MAGENTA
if "SQL" in lang: return CYAN
if "Shell" in lang or "Bash" in lang or "Zsh" in lang or "Fish" in lang or "KornShell" in lang: return GREEN
if "PowerShell" in lang or "Batch" in lang: return GREEN
if "Rust" in lang: return RED
if "Go" in lang or lang == "V" or "Zig" in lang or "Odin" in lang: return CYAN
if "Ruby" in lang or "Crystal" in lang: return RED
if "Elixir" in lang or "Erlang" in lang: return PURPLE
if "Haskell" in lang or "Elm" in lang or "PureScript" in lang or "OCaml" in lang: return PURPLE
if "Clojure" in lang or "Lisp" in lang or "Scheme" in lang or "Racket" in lang: return PURPLE
if "Lua" in lang: return BLUE
if "PHP" in lang: return PURPLE
if "Perl" in lang: return PINK
if "Scala" in lang or "Groovy" in lang: return RED
if "F#" in lang: return BLUE
if "Nim" in lang or "Julia" in lang: return PURPLE
if "Solidity" in lang: return ORANGE
if "Terraform" in lang or "HCL" in lang: return PURPLE
if "Vue" in lang or "Svelte" in lang or "Astro" in lang: return GREEN
if "Fortran" in lang or "COBOL" in lang or "Pascal" in lang or "Ada" in lang: return BLUE
if "Assembly" in lang or "VHDL" in lang or "SystemVerilog" in lang: return BLUE
if "Protobuf" in lang or "Thrift" in lang or "GraphQL" in lang: return CYAN
if lang in ("D", "Hack", "Gleam", "Wren", "Dhall"): return TEAL
if "ReScript" in lang or "Reason" in lang: return RED
if "VB" in lang: return BLUE
if "Nix" in lang or "Starlark" in lang or "Jsonnet" in lang: return CYAN
return WHITE
for language, data in sorted(language_stats.items(), key=lambda x: x[1]["code_lines"], reverse=True):
if data["code_lines"] == 0:
continue
percent = (data["code_lines"] / code_total * 100) if code_total else 0
lang_color = get_lang_color(language)
chars_str = format_number_short(data['code_chars'])
print(
f"{lang_color}{language:<18}{RESET} "
f"{WHITE}{data['code_lines']:>8,}{RESET} {DIM}lines{RESET} "
f"{PURPLE}{chars_str:>7}{RESET} {DIM}chars{RESET} "
f"{YELLOW}{percent:>5.1f}%{RESET} "
f"{horizontal_bar(percent, color=lang_color)}"
)
print(f"\n{BOLD}{CYAN}DECLARATIONS{RESET}")
# Sort by total declarations (functions + classes) descending
decl_items = [(lang, data) for lang, data in language_stats.items()
if (data.get('function_count', 0) + data.get('class_count', 0)) > 0]
decl_items.sort(key=lambda x: x[1].get('function_count', 0) + x[1].get('class_count', 0), reverse=True)
if decl_items:
# Dynamic column widths
dfunc_w = max(len(str(d.get('function_count', 0))) for _, d in decl_items)
dclass_w = max(len(str(d.get('class_count', 0))) for _, d in decl_items)
dtotal_w = max(len(str(d.get('function_count', 0) + d.get('class_count', 0))) for _, d in decl_items)
for language, data in decl_items:
funcs = data.get('function_count', 0)
classes = data.get('class_count', 0)
total_decl = funcs + classes
lang_color = get_lang_color(language)
print(
f"{lang_color}{language:<18}{RESET} "
f"{MAGENTA}{funcs:>{dfunc_w}}{RESET} {DIM}funcs{RESET} "
f"{PURPLE}{classes:>{dclass_w}}{RESET} {DIM}classes{RESET} "
f"{DIM}={RESET} "
f"{WHITE}{total_decl:>{dtotal_w}}{RESET} {DIM}total{RESET}"
)
# Totals row
sum_funcs = sum(d.get('function_count', 0) for _, d in decl_items)
sum_classes = sum(d.get('class_count', 0) for _, d in decl_items)
sum_total = sum_funcs + sum_classes
print(f"{DIM}{'' * 18}{RESET} {DIM}{'' * (dfunc_w + 6)} {'' * (dclass_w + 8)} {'' * 2} {'' * (dtotal_w + 6)}{RESET}")
print(
f"{BOLD}{'Total':<18}{RESET} "
f"{MAGENTA}{sum_funcs:>{dfunc_w}}{RESET} {DIM}funcs{RESET} "
f"{PURPLE}{sum_classes:>{dclass_w}}{RESET} {DIM}classes{RESET} "
f"{DIM}={RESET} "
f"{WHITE}{BOLD}{sum_total:>{dtotal_w}}{RESET} {DIM}total{RESET}"
)
print(f"\n{BOLD}{CYAN}CODE QUALITY SNAPSHOT{RESET}")
# compute dynamic column widths based on values
code_items = [(l, d) for l, d in language_stats.items() if d['code_lines'] > 0]
if code_items:
files_w = max(len(str(d['files'])) for _, d in code_items)
code_w = max(len(f"{d['code_lines']:,}") for _, d in code_items)
med_w = max(len(str(d.get('median_code_lines', 0))) for _, d in code_items)
sd_w = max(len(f"{d.get('stddev_code_lines', 0):.1f}") for _, d in code_items)
avg_size_w = max(len(f"{(d.get('avg_size_kb',0)):.1f}") for _, d in code_items)
lpf_w = max(len(f"{d.get('avg_func_length', 0):.0f}") for _, d in code_items)
else:
files_w = code_w = med_w = sd_w = avg_size_w = lpf_w = 4
# Sort by quality score (descending)
def get_quality_score(item):
_, data = item
return data.get('quality_score', 0)
for language, data in sorted(code_items, key=get_quality_score, reverse=True):
comment_ratio = (data["comment_lines"] / data["code_lines"] * 100) if data["code_lines"] else 0
qs = data.get('quality_score', 0)
grade = score_grade(qs)
sc = score_color(qs, args.no_color)
avg_size = data.get("avg_size_kb", 0)
med = data.get("median_code_lines", 0)
sd = data.get("stddev_code_lines", 0)
lpf = data.get("avg_func_length", 0)
lang_color = get_lang_color(language)
# Color the comment ratio based on health
if not data.get('has_comment', True):
ratio_color = DIM
elif comment_ratio < 10:
ratio_color = RED
elif comment_ratio < 20:
ratio_color = YELLOW
else:
ratio_color = GREEN
print(
f"{sc}{grade:>2}{RESET} "
f"{lang_color}{language:<18}{RESET} "
f"{DIM}Files:{RESET}{CYAN}{data['files']:>{files_w}}{RESET} {DIM}|{RESET} "
f"{DIM}Code:{RESET}{WHITE}{data['code_lines']:>{code_w},}{RESET} {DIM}|{RESET} "
f"{DIM}L/F:{RESET}{WHITE}{lpf:>{lpf_w}.0f}{RESET} {DIM}|{RESET} "
f"{DIM}Comment:{RESET}{ratio_color}{comment_ratio:>4.1f}%{RESET} {DIM}|{RESET} "
f"{DIM}Size:{RESET}{WHITE}{avg_size:>{avg_size_w}.1f}{RESET}{DIM}kb{RESET} {DIM}|{RESET} "
f"{DIM}Score:{RESET} {sc}{qs:.0f}{RESET}"
)
# ---- CODE HEALTH REPORT ----
print(f"\n{BOLD}{MAGENTA}CODE HEALTH REPORT{RESET}")
# Compute overall quality score (weighted average by code lines)
total_code_lines = sum(d['code_lines'] for _, d in code_items)
if total_code_lines > 0:
overall_score = sum(d.get('quality_score', 0) * d['code_lines'] for _, d in code_items) / total_code_lines
else:
overall_score = 0
overall_grade = score_grade(overall_score)
osc = score_color(overall_score, args.no_color)
# Overall totals
total_funcs = sum(d['function_count'] for _, d in code_items)
total_classes = sum(d.get('class_count', 0) for _, d in code_items)
total_comments = sum(d['comment_lines'] for _, d in code_items)
total_todos = sum(d.get('todo_count', 0) for _, d in code_items)
total_imports = sum(d.get('import_count', 0) for _, d in code_items)
total_long_lines = sum(d.get('long_lines', 0) for _, d in code_items)
total_large_files = sum(d.get('large_file_count', 0) for _, d in code_items)
overall_cr = (total_comments / total_code_lines * 100) if total_code_lines else 0
overall_lpf = (total_code_lines / total_funcs) if total_funcs else 0
overall_td = (total_todos / total_code_lines * 1000) if total_code_lines else 0
max_nesting_all = max((d.get('max_nesting_depth', 0) for _, d in code_items), default=0)
all_total_lines = sum(d['code_lines'] + d['comment_lines'] + d['empty_lines'] for _, d in code_items)
overall_llr = (total_long_lines / all_total_lines * 100) if all_total_lines else 0
# Grade display
print(f"\n {BOLD}Overall Grade:{RESET} {osc}{BOLD}{overall_grade}{RESET} {DIM}({osc}{overall_score:.0f}{RESET}{DIM}/100){RESET}")
print(f" {BOLD}Quality Bar:{RESET} {horizontal_bar(overall_score, width=30, color=osc)}")
# Breakdown metrics
cr_color = GREEN if overall_cr >= 20 else (YELLOW if overall_cr >= 10 else RED)
lpf_color = GREEN if overall_lpf <= 30 else (YELLOW if overall_lpf <= 50 else RED)
td_color = GREEN if overall_td <= 5 else (YELLOW if overall_td <= 10 else RED)
nesting_color = GREEN if max_nesting_all <= 5 else (YELLOW if max_nesting_all <= 7 else RED)
llr_color = GREEN if overall_llr <= 3 else (YELLOW if overall_llr <= 8 else RED)
lf_total_files = sum(d['files'] for _, d in code_items)
lf_pct = (total_large_files / lf_total_files * 100) if lf_total_files else 0
lf_color = GREEN if lf_pct <= 15 else (YELLOW if lf_pct <= 30 else RED)
print(f"\n {BOLD}Metrics Breakdown:{RESET}")
print(f" {DIM}Comment Ratio:{RESET} {cr_color}{overall_cr:>5.1f}%{RESET} {DIM}({total_comments:,} comment lines / {total_code_lines:,} code lines){RESET}")
print(f" {DIM}Lines per Func:{RESET} {lpf_color}{overall_lpf:>5.1f}{RESET} {DIM}({total_code_lines:,} code lines / {total_funcs:,} functions){RESET}")
print(f" {DIM}TODO Density:{RESET} {td_color}{overall_td:>5.1f}{RESET} {DIM}({total_todos} TODOs per 1K lines){RESET}")
print(f" {DIM}Max Nesting:{RESET} {nesting_color}{max_nesting_all:>5}{RESET} {DIM}(deepest brace nesting across project){RESET}")
print(f" {DIM}Long Lines:{RESET} {llr_color}{overall_llr:>5.1f}%{RESET} {DIM}({total_long_lines:,} lines > 120 chars){RESET}")
print(f" {DIM}Large Files:{RESET} {lf_color}{total_large_files:>5}{RESET} {DIM}({lf_pct:.0f}% files > 300 lines){RESET}")
print(f" {DIM}Imports:{RESET} {WHITE}{total_imports:>5,}{RESET} {DIM}(total import statements){RESET}")
# Per-language quality bar chart (sorted by score)
print(f"\n {BOLD}Per-Language Scores:{RESET}")
for language, data in sorted(code_items, key=get_quality_score, reverse=True):
qs = data.get('quality_score', 0)
grade = score_grade(qs)
sc = score_color(qs, args.no_color)
lang_color = get_lang_color(language)
bar = horizontal_bar(qs, width=20, color=sc)
print(f" {lang_color}{language:<18}{RESET} {bar} {sc}{grade:>2}{RESET} {DIM}({qs:.0f}){RESET}")
print(f"\n{BOLD}{CYAN}TOP FILES ({args.top}){RESET}")
# prepare filtered list
filtered = [f for f in per_file if f['code_lines'] >= args.min_code_lines]
# compute widths for predictable columns
code_w = max((len(f"{f['code_lines']:,}") for f in filtered), default=6)
lang_w = max((len(f['language']) for f in filtered), default=8)
size_w = max((len(f"{f['size_bytes']/1024:.1f}" ) for f in filtered), default=6)
func_w = max((len(str(f['function_count'])) for f in filtered), default=4)
todo_w = max((len(str(f['todo_count'])) for f in filtered), default=3)
comment_w = max((len(f"{f['comment_ratio']:.1f}") for f in filtered), default=4)
# largest by code
largest_by_code = sorted(filtered, key=lambda x: x['code_lines'], reverse=True)[:args.top]
print(f"\n{BOLD}{PURPLE}Largest by code lines{RESET}")
for f in largest_by_code:
size_kb = f['size_bytes'] / 1024
lang_color = get_lang_color(f['language'])
print(f"{WHITE}{f['code_lines']:{code_w},}{RESET} {DIM}|{RESET} {lang_color}{f['language']:<{lang_w}}{RESET} {DIM}|{RESET} {CYAN}{size_kb:>{size_w}.1f}{RESET}{DIM}kb{RESET} {DIM}|{RESET} {DIM}funcs:{RESET}{MAGENTA}{f['function_count']:{func_w}}{RESET} {DIM}todos:{RESET}{ORANGE}{f['todo_count']:{todo_w}}{RESET} {DIM}|{RESET} {DIM}comment:{RESET}{YELLOW}{f['comment_ratio']:{comment_w}.1f}%{RESET} {TEAL}{Path(f['relpath']).relative_to(root)}{RESET}")
# Most TODOs
most_todos = sorted(filtered, key=lambda x: x['todo_count'], reverse=True)[:args.top]
print(f"\n{BOLD}{ORANGE}Files with most TODO/FIXME{RESET}")
for f in most_todos:
if f['todo_count'] == 0:
continue
lang_color = get_lang_color(f['language'])
print(f"{ORANGE}{f['todo_count']:{todo_w}}{RESET} {DIM}TODOs |{RESET} {lang_color}{f['language']:<{lang_w}}{RESET} {DIM}|{RESET} {DIM}code:{RESET}{WHITE}{f['code_lines']:{code_w},}{RESET} {DIM}|{RESET} {DIM}comment:{RESET}{YELLOW}{f['comment_ratio']:{comment_w}.1f}%{RESET} {TEAL}{Path(f['relpath']).relative_to(root)}{RESET}")
# Lowest comment ratio (but with at least some code)
low_comments = sorted((f for f in filtered if f.get('has_comment', True)), key=lambda x: x['comment_ratio'])[:args.top]
print(f"\n{BOLD}{RED}Files with lowest comment ratio{RESET} {DIM}(languages that support comments){RESET}")
for f in low_comments:
lang_color = get_lang_color(f['language'])
ratio_color = RED if f['comment_ratio'] < 5 else (YELLOW if f['comment_ratio'] < 10 else GREEN)
print(f"{ratio_color}{f['comment_ratio']:{comment_w}.1f}%{RESET} {DIM}|{RESET} {lang_color}{f['language']:<{lang_w}}{RESET} {DIM}|{RESET} {DIM}code:{RESET}{WHITE}{f['code_lines']:{code_w},}{RESET} {DIM}|{RESET} {DIM}todos:{RESET}{ORANGE}{f['todo_count']:{todo_w}}{RESET} {TEAL}{Path(f['relpath']).relative_to(root)}{RESET}")
elapsed = result.get('elapsed', 0)
workers = result.get('workers', 1)
total_files = len(result.get('per_file', []))
total_lines = sum(f['total_lines'] for f in result.get('per_file', []))
fps = f"{total_files / elapsed:,.0f} files/s" if elapsed > 0 else "N/A"
lps_raw = total_lines / elapsed if elapsed > 0 else 0
if lps_raw >= 1_000_000:
lps = f"{lps_raw / 1_000_000:.1f}M lines/s"
elif lps_raw >= 1_000:
lps = f"{lps_raw / 1_000:.1f}K lines/s"
else:
lps = f"{lps_raw:.0f} lines/s"
print(f"\n{GREEN}Analysis complete ✔{RESET} {DIM}{elapsed:.2f}s │ {workers} threads │ {fps}{lps} │ v{VERSION}{RESET}\n")
# --------------------------------------------------
# Runner
# --------------------------------------------------
def run(root: Path, args):
t0 = time.time()
language_stats = defaultdict(lambda: {
"files": 0,
"code_lines": 0,
"code_chars": 0,
"comment_lines": 0,
"empty_lines": 0,
"function_count": 0,
"class_count": 0,
"todo_count": 0,
"import_count": 0,
"long_lines": 0,
"max_nesting_depth": 0,
"nesting_depths": [],
"func_lengths": [],
"longest_lines": [],
"sizes": [],
"per_file_code_lines": [],
"has_comment": False,
})
category_totals = {
"code": 0,
"config": 0,
"docs": 0
}
per_file = []
_bypass = args.bypass_excluded
# load dynamic ignore patterns from .gitignore/.ignore files
if _bypass:
ignore_dirs = set()
ignore_globs = []
negation_globs = []
negation_dirs = set()
else:
ignore_dirs = set(IGNORE_DIRECTORIES)
ignore_globs, extra_dir_names, negation_globs, negation_dirs = load_ignore_files(root)
for d in extra_dir_names:
ignore_dirs.add(Path(d).name)
exclude_paths = [] if _bypass else [ (root / e).resolve() for e in (args.exclude or []) ]
def is_negated(rel_path: str, name: str) -> bool:
"""Check if a path matches a negation pattern (should NOT be ignored)."""
for neg_dir in negation_dirs:
if fnmatch.fnmatch(rel_path, neg_dir) or fnmatch.fnmatch(rel_path, neg_dir + '/*'):
return True
if rel_path == neg_dir or rel_path.startswith(neg_dir + '/'):
return True
for pat in negation_globs:
if fnmatch.fnmatch(rel_path, pat) or fnmatch.fnmatch(name, pat):
return True
return False
show_anim = not args.json and not args.no_color and not args.no_animation
has_excludes = bool(exclude_paths)
root_str = str(root.resolve())
_include_hidden = args.include_hidden or _bypass
_ext_filter = args.ext
_spinners = ("", "", "", "", "", "", "", "", "", "")
# ── Phase 1: Collect all file paths ──────────────────────────────────
all_file_paths = []
_last_anim = time.monotonic()
for current_path, directory_names, file_names in os.walk(root, followlinks=False):
current_path_obj = Path(current_path)
# Skip if current path is a symlink pointing outside the project
if current_path_obj.is_symlink():
try:
if not str(current_path_obj.resolve()).startswith(root_str):
directory_names[:] = []
continue
except (OSError, ValueError):
directory_names[:] = []
continue
# if current path is within an excluded path, skip descending
if has_excludes:
try:
current_res = current_path_obj.resolve()
if any(ex == current_res or ex in current_res.parents for ex in exclude_paths):
directory_names[:] = []
continue
except (OSError, ValueError):
directory_names[:] = []
continue
# prune directory_names to avoid descending into ignores/excludes/hidden
new_dirs = []
for d in directory_names:
dpath = current_path_obj / d
# Skip symlinks that might point outside the project
if dpath.is_symlink():
try:
if not str(dpath.resolve()).startswith(root_str):
continue
except (OSError, ValueError):
continue
# Calculate relative path for pattern matching
try:
rel = str(dpath.relative_to(root)).replace('\\', '/')
except ValueError:
continue
path_is_negated = is_negated(rel, d)
if d in ignore_dirs and not path_is_negated:
continue
if has_excludes:
try:
dpath_res = dpath.resolve()
if any(ex == dpath_res or ex in dpath_res.parents for ex in exclude_paths):
continue
except (OSError, ValueError):
continue
if not _include_hidden and d.startswith('.'):
continue
if not path_is_negated:
if any(fnmatch.fnmatch(rel, pat) for pat in ignore_globs):
continue
new_dirs.append(d)
directory_names[:] = new_dirs
for file in file_names:
file_path = Path(current_path) / file
if has_excludes:
try:
file_res = file_path.resolve()
except (OSError, ValueError):
continue
if any(ex == file_res or ex in file_res.parents for ex in exclude_paths):
continue
if not _bypass and should_ignore_file(file_path, root, IGNORE_FILE_PATTERNS, ignore_globs, negation_globs):
continue
if _ext_filter and file_path.suffix.lower() not in _ext_filter:
continue
all_file_paths.append(file_path)
if show_anim:
_now = time.monotonic()
if _now - _last_anim >= 0.15:
_last_anim = _now
sp = _spinners[len(all_file_paths) % len(_spinners)]
sys.stdout.write(f"\r {CYAN}{sp}{RESET} {GRAY}Discovering files... {GOLD}{len(all_file_paths):,}{RESET} ")
sys.stdout.flush()
total_to_scan = len(all_file_paths)
# ── Phase 2: Analyze files in parallel with progress ─────────────────
if show_anim:
sys.stdout.write(f"\r {GREEN}{RESET} {GRAY}Discovered {GOLD}{total_to_scan:,}{GRAY} files to analyze{RESET} \n")
sys.stdout.flush()
sys.stdout.write(HIDE_CURSOR)
print() # line 1 for progress bar
print() # line 2 for filename
cpu = os.cpu_count() or 1
workers = args.workers if args.workers > 0 else min(cpu, 16)
# Use processes for large workloads (true parallelism), threads for small ones (low overhead)
use_processes = total_to_scan > 500
Executor = ProcessPoolExecutor if use_processes else ThreadPoolExecutor
# Disable cyclic GC during analysis — the growing results list causes
# increasingly expensive gen2 collections that stall all threads/processes.
gc.disable()
try:
with Executor(max_workers=workers) as executor:
# Bounded submit + wait(FIRST_COMPLETED) loop:
# - Each task is a single file → one slow file only blocks one worker
# - At most (workers*4) futures in flight → O(1) per wait() call
# - Workers always have queued work → full utilization
results = []
_last_progress = time.monotonic()
processed = 0
_MAX_INFLIGHT = workers * 4
_pending = set()
_path_map = {} # future -> path
_path_iter = iter(all_file_paths)
def _submit_up_to_limit():
"""Keep _pending at _MAX_INFLIGHT futures."""
while len(_pending) < _MAX_INFLIGHT:
fp = next(_path_iter, None)
if fp is None:
return
fut = executor.submit(analyze_file, fp)
_pending.add(fut)
_path_map[fut] = fp
_submit_up_to_limit()
while _pending:
finished, _pending = wait(_pending, return_when=FIRST_COMPLETED)
for fut in finished:
processed += 1
fp = _path_map.pop(fut)
if show_anim:
_now = time.monotonic()
if _now - _last_progress >= 0.1 or processed == total_to_scan:
_last_progress = _now
try:
rel = str(fp.relative_to(root))
except ValueError:
rel = str(fp)
scan_progress(processed, total_to_scan, rel)
try:
info = fut.result()
except Exception:
info = None
if info is not None:
results.append(info)
_submit_up_to_limit()
finally:
gc.enable()
gc.collect()
if show_anim:
clear_scan_progress()
sys.stdout.write(SHOW_CURSOR)
# ── Phase 3: Accumulate results ──────────────────────────────────────
for info in results:
if info['code_lines'] < args.min_code_lines:
continue
per_file.append(info)
lang = info['language']
language_stats[lang]['files'] += 1
language_stats[lang]['code_lines'] += info['code_lines']
language_stats[lang]['code_chars'] += info['code_chars']
language_stats[lang]['comment_lines'] += info['comment_lines']
language_stats[lang]['empty_lines'] += info['empty_lines']
language_stats[lang]['function_count'] += info['function_count']
language_stats[lang]['class_count'] += info['class_count']
language_stats[lang]['todo_count'] += info['todo_count']
language_stats[lang]['import_count'] += info['import_count']
language_stats[lang]['long_lines'] += info['long_lines']
if info['max_nesting_depth'] > language_stats[lang]['max_nesting_depth']:
language_stats[lang]['max_nesting_depth'] = info['max_nesting_depth']
language_stats[lang]['nesting_depths'].append(info['max_nesting_depth'])
if info['function_count'] > 0:
language_stats[lang]['func_lengths'].append(info['avg_func_length'])
language_stats[lang]['longest_lines'].append(info['longest_line'])
language_stats[lang]['sizes'].append(info['size_bytes'])
language_stats[lang]['per_file_code_lines'].append(info['code_lines'])
language_stats[lang]['has_comment'] = language_stats[lang].get('has_comment', False) or bool(info.get('has_comment'))
category_totals[info['category']] += info['total_lines']
# ── Phase 4: Post-process language stats ─────────────────────────────
for lang, data in language_stats.items():
sizes = data['sizes']
pl = data['per_file_code_lines']
data['avg_size_kb'] = (sum(sizes) / len(sizes) / 1024) if sizes else 0
data['median_code_lines'] = int(statistics.median(pl)) if pl else 0
data['stddev_code_lines'] = statistics.pstdev(pl) if pl else 0
data['max_file_lines'] = max(pl) if pl else 0
data['large_file_count'] = sum(1 for x in pl if x > 300)
data['avg_func_length'] = (data['code_lines'] / data['function_count']) if data['function_count'] else 0
nd = data.get('nesting_depths', [])
data['avg_nesting_depth'] = (sum(nd) / len(nd)) if nd else 0
ll = data.get('longest_lines', [])
data['max_longest_line'] = max(ll) if ll else 0
data['todo_density'] = (data['todo_count'] / data['code_lines'] * 1000) if data['code_lines'] else 0
total_lines = data['code_lines'] + data['comment_lines'] + data['empty_lines']
data['long_line_ratio'] = (data['long_lines'] / total_lines * 100) if total_lines else 0
data['quality_score'] = compute_quality_score(data)
elapsed = time.time() - t0
result = {
'language_stats': language_stats,
'category_totals': category_totals,
'per_file': per_file,
'elapsed': elapsed,
'workers': workers,
}
if args.json:
out = {
'root': str(root),
'summary': {
'files': len(per_file),
'total_lines': sum(f['total_lines'] for f in per_file),
'elapsed': round(elapsed, 3),
'workers': workers,
},
'languages': {k: {"files": v['files'], "code_lines": v['code_lines'], "comment_lines": v['comment_lines']} for k, v in language_stats.items()},
'top_files': {
'largest_by_code': [ {k: f[k] for k in ('path','code_lines','size_bytes','function_count','todo_count')} for f in sorted(per_file, key=lambda x: x['code_lines'], reverse=True)[:args.top] ],
},
}
if not args.no_git:
git = get_git_info(root, args.commit_msg_max)
if git:
out['git'] = git
print(json.dumps(out, indent=2))
return result
print_summary(result, root, args)
return result
if __name__ == '__main__':
args = parse_args()
root = Path(args.path or '.').resolve()
if args.ext:
args.ext = [e if e.startswith('.') else '.' + e for e in args.ext]
# handle install/uninstall requests
if args.install or args.uninstall:
chosen = args.install_dir if args.install_dir else default_install_dir()
install_dir = os.path.expanduser(chosen)
if args.install:
ok = install_self(install_dir)
sys.exit(0 if ok else 2)
if args.uninstall:
ok = uninstall_self(install_dir)
sys.exit(0 if ok else 2)
# Startup animation (skip for JSON output, no-color, or no-animation)
if not args.json and not args.no_color and not args.no_animation:
try:
startup_sequence(root)
except Exception:
pass # graceful fallback if terminal doesn't support ANSI
try:
run(root, args)
except KeyboardInterrupt:
sys.stdout.write(SHOW_CURSOR)
print(f"\n{RED}Interrupted.{RESET}")
sys.exit(1)