946 lines
34 KiB
Python
946 lines
34 KiB
Python
"""
|
|
Action handlers for PyUCC GUI.
|
|
Separates business logic from GUI construction.
|
|
"""
|
|
import os
|
|
import time
|
|
import types
|
|
from pathlib import Path
|
|
from tkinter import messagebox
|
|
import tkinter as tk
|
|
from tkinter import ttk
|
|
|
|
from ..core.scanner import find_source_files
|
|
from ..core.differ import Differ, BaselineManager
|
|
from ..config import settings as app_settings
|
|
|
|
|
|
class ActionHandlers:
|
|
"""Handles all main actions (scan, countings, metrics, differ) for the GUI."""
|
|
|
|
def __init__(self, gui_app):
|
|
"""
|
|
Args:
|
|
gui_app: Reference to the main GUI application instance
|
|
"""
|
|
self.app = gui_app
|
|
|
|
def handle_scan(self):
|
|
"""Execute scan action: find source files matching profile filters."""
|
|
self.app.log("Action: Scan started", level='INFO')
|
|
self.app._set_phase("Scanning...")
|
|
self.app._current_mode = 'scan'
|
|
|
|
# Disable action buttons during operation
|
|
self.app._disable_action_buttons()
|
|
|
|
# Reset UI
|
|
self._reset_progress_ui()
|
|
tooltips = {
|
|
"name": "File name",
|
|
"path": "Relative path from project root"
|
|
}
|
|
self.app._set_results_columns(("name", "path"), tooltips=tooltips)
|
|
self._clear_results()
|
|
self.app.update_idletasks()
|
|
|
|
# Get paths and filters
|
|
paths, allowed_exts, ignore_patterns, pr = self.app._resolve_profile_and_filters()
|
|
if not paths:
|
|
messagebox.showwarning("Missing path", "Please select a folder or profile to analyze first.")
|
|
return
|
|
|
|
# Submit scan task
|
|
def _scan_task(paths):
|
|
files = []
|
|
for p in paths:
|
|
pth = Path(p)
|
|
try:
|
|
if pth.is_dir():
|
|
files.extend(find_source_files(pth, allowed_extensions=allowed_exts, ignore_patterns=ignore_patterns))
|
|
elif pth.is_file():
|
|
files.append(pth)
|
|
except Exception:
|
|
continue
|
|
# Deduplicate
|
|
seen = set()
|
|
unique = []
|
|
for f in files:
|
|
s = str(f)
|
|
if s not in seen:
|
|
seen.add(s)
|
|
unique.append(f)
|
|
return unique
|
|
|
|
self.app._current_task_id = self.app.worker.submit(_scan_task, paths, kind='thread', on_done=self.app._scan_done)
|
|
self.app.cancel_btn.config(state='normal')
|
|
|
|
def handle_countings(self):
|
|
"""Execute countings action: analyze code lines (code/comment/blank)."""
|
|
self.app.log("Action: Countings started", level='INFO')
|
|
self.app._set_phase("Counting...")
|
|
self.app._current_mode = 'countings'
|
|
|
|
# Disable action buttons during operation
|
|
self.app._disable_action_buttons()
|
|
|
|
# Reset UI
|
|
self._reset_progress_ui()
|
|
tooltips = {
|
|
"name": "File name",
|
|
"path": "Relative path from project root",
|
|
"code": "Code lines: executable lines excluding comments and blanks",
|
|
"comment": "Comment lines: lines containing only comments",
|
|
"blank": "Blank lines: empty lines",
|
|
"total": "Total physical lines: sum of code + comment + blank",
|
|
"language": "Detected programming language"
|
|
}
|
|
self.app._set_results_columns(("name", "path", "code", "comment", "blank", "total", "language"), tooltips=tooltips)
|
|
self._clear_results()
|
|
self.app.update_idletasks()
|
|
|
|
paths, allowed_exts, ignore_patterns, pr = self.app._resolve_profile_and_filters()
|
|
if not paths:
|
|
messagebox.showwarning("Missing path", "Please select a folder or profile to analyze first.")
|
|
return
|
|
|
|
# Gather files first, then analyze
|
|
def _gather_files(paths):
|
|
return self._gather_source_files(paths, allowed_exts, ignore_patterns)
|
|
|
|
def _on_gather_done(results):
|
|
files = results or []
|
|
if not files:
|
|
messagebox.showinfo("Countings", "No files found to analyze.")
|
|
self.app._set_phase('Idle')
|
|
self.app._enable_action_buttons()
|
|
return
|
|
|
|
from ..core.countings_impl import analyze_file_counts
|
|
|
|
self.app._total_files = len(files)
|
|
self.app._processed_files = 0
|
|
try:
|
|
self.app.progress['maximum'] = max(1, self.app._total_files)
|
|
self.app.progress['value'] = 0
|
|
self.app._lbl_files.config(text=f"Files: 0/{self.app._total_files}")
|
|
except Exception:
|
|
pass
|
|
|
|
self.app._set_phase(f"Analyzing {self.app._total_files} files...")
|
|
self.app.update_idletasks()
|
|
|
|
self.app._current_task_id = self.app.worker.map_iterable(
|
|
func=analyze_file_counts,
|
|
items=files,
|
|
kind='thread',
|
|
on_progress=self.app._on_count_progress,
|
|
on_done=self.app._on_count_done
|
|
)
|
|
|
|
self.app._set_phase("Gathering files...")
|
|
self.app.update_idletasks()
|
|
self.app._current_task_id = self.app.worker.submit(_gather_files, paths, kind='thread', on_done=_on_gather_done)
|
|
self.app.cancel_btn.config(state='normal')
|
|
|
|
def handle_metrics(self):
|
|
"""Execute metrics action: compute complexity metrics (CC, MI)."""
|
|
self.app.log("Action: Metrics started", level='INFO')
|
|
self.app._set_phase("Computing metrics...")
|
|
self.app._current_mode = 'metrics'
|
|
|
|
# Disable action buttons during operation
|
|
self.app._disable_action_buttons()
|
|
|
|
# Reset UI
|
|
self._reset_progress_ui()
|
|
tooltips = {
|
|
"name": "File name",
|
|
"path": "Relative path from project root",
|
|
"avg_cc": "Avg Cyclomatic Complexity: average complexity across all functions",
|
|
"max_cc": "Max Cyclomatic Complexity: highest complexity found in a single function",
|
|
"func_count": "Function Count: total number of functions/methods",
|
|
"mi": "Maintainability Index: 0-100 scale (higher is better)",
|
|
"language": "Detected programming language"
|
|
}
|
|
self.app._set_results_columns(("name", "path", "avg_cc", "max_cc", "func_count", "mi", "language"), tooltips=tooltips)
|
|
self._clear_results()
|
|
self.app.update_idletasks()
|
|
|
|
paths, allowed_exts, ignore_patterns, pr = self.app._resolve_profile_and_filters()
|
|
if not paths:
|
|
messagebox.showwarning("Missing path", "Please select a folder or profile to analyze first.")
|
|
return
|
|
|
|
# Gather files first, then analyze
|
|
def _gather_files(paths):
|
|
return self._gather_source_files(paths, allowed_exts, ignore_patterns)
|
|
|
|
def _on_gather_done(results):
|
|
files = results or []
|
|
if not files:
|
|
messagebox.showinfo("Metrics", "No files found to analyze.")
|
|
self.app._set_phase('Idle')
|
|
self.app._enable_action_buttons()
|
|
return
|
|
|
|
try:
|
|
from ..core.metrics import analyze_file_metrics as analyzer
|
|
except Exception:
|
|
messagebox.showerror("Metrics", "Metrics analyzer not available")
|
|
self.app._set_phase('Idle')
|
|
self.app._enable_action_buttons()
|
|
return
|
|
|
|
self.app._total_files = len(files)
|
|
self.app._processed_files = 0
|
|
try:
|
|
self.app.progress['maximum'] = max(1, self.app._total_files)
|
|
self.app.progress['value'] = 0
|
|
self.app._lbl_files.config(text=f"Files: 0/{self.app._total_files}")
|
|
except Exception:
|
|
pass
|
|
|
|
self.app._set_phase(f"Computing metrics for {self.app._total_files} files...")
|
|
self.app.update_idletasks()
|
|
|
|
self.app._current_task_id = self.app.worker.map_iterable(
|
|
func=analyzer,
|
|
items=files,
|
|
kind='thread',
|
|
on_progress=self.app._on_metrics_progress,
|
|
on_done=self.app._on_metrics_done
|
|
)
|
|
|
|
self.app._set_phase("Gathering files...")
|
|
self.app.update_idletasks()
|
|
self.app._current_task_id = self.app.worker.submit(_gather_files, paths, kind='thread', on_done=_on_gather_done)
|
|
self.app.cancel_btn.config(state='normal')
|
|
|
|
def handle_differ(self):
|
|
"""Execute differ action: compare current code with baseline."""
|
|
self.app.log("Action: Differing started", level='INFO')
|
|
self.app._set_phase("Differing...")
|
|
# Show columns as Current (codebase) and Baseline
|
|
tooltips = {
|
|
"Current": "Current file path",
|
|
"Baseline": "Baseline file path",
|
|
"added": "Added files: present in current but not in baseline",
|
|
"deleted": "Deleted files: present in baseline but not in current",
|
|
"modified": "Modified files: content changed between baseline and current",
|
|
"unmodified": "Unmodified files: identical in both versions",
|
|
"Δ code_lines": "Delta Code Lines: change in executable lines (+positive / -negative)",
|
|
"Δ comment_lines": "Delta Comment Lines: change in comment lines",
|
|
"Δ blank_lines": "Delta Blank Lines: change in empty lines",
|
|
"Δ func_count": "Current Function Count (delta vs baseline)\nShows absolute value with change in parentheses",
|
|
"Δ avg_cc": "Current Avg Cyclomatic Complexity (delta vs baseline)\nShows absolute value with change in parentheses",
|
|
"Δ mi": "Current Maintainability Index (delta vs baseline)\nShows absolute value with change in parentheses"
|
|
}
|
|
self.app._set_results_columns(("Current", "Baseline", "added", "deleted", "modified", "unmodified", "Δ code_lines", "Δ comment_lines", "Δ blank_lines", "Δ func_count", "Δ avg_cc", "Δ mi"), tooltips=tooltips)
|
|
self._clear_results()
|
|
self.app._current_mode = 'differ'
|
|
|
|
# Disable action buttons during operation
|
|
self.app._disable_action_buttons()
|
|
|
|
try:
|
|
self.app._lbl_files.config(text="Files: 0/0")
|
|
self.app.progress['maximum'] = 1
|
|
self.app.progress['value'] = 0
|
|
except Exception:
|
|
pass
|
|
|
|
self.app.update_idletasks()
|
|
|
|
paths, allowed_exts, ignore_patterns, pr = self.app._resolve_profile_and_filters()
|
|
if not paths:
|
|
messagebox.showwarning("Missing path", "Please select a folder or profile to analyze first.")
|
|
return
|
|
|
|
project = paths[0]
|
|
self.app._set_phase("Loading baselines...")
|
|
self.app.update_idletasks()
|
|
|
|
bm = BaselineManager(project)
|
|
all_baselines = bm.list_baselines()
|
|
|
|
profile_name = pr.get('name') if pr else None
|
|
max_keep = app_settings.get_max_keep()
|
|
|
|
# Filter baselines by current profile
|
|
baselines = []
|
|
for baseline_id in all_baselines:
|
|
try:
|
|
meta = bm.load_metadata(baseline_id)
|
|
baseline_profile = meta.profile if hasattr(meta, 'profile') else None
|
|
# Match baselines with same profile (or both None)
|
|
if baseline_profile == profile_name:
|
|
baselines.append(baseline_id)
|
|
except Exception:
|
|
# Skip baselines that can't be loaded
|
|
pass
|
|
|
|
# Callback handlers
|
|
def _on_create_done(baseline_id):
|
|
try:
|
|
self.app._set_phase('Idle')
|
|
self.app._current_task_id = None
|
|
self.app._enable_action_buttons()
|
|
# After baseline is created, export the differ `result` (if available) to CSV and text report inside baseline folder
|
|
try:
|
|
from ..utils.csv_exporter import export_rows_to_csv
|
|
from ..utils.report_generator import generate_differ_report
|
|
bdir = bm._baseline_dir(baseline_id)
|
|
csv_path = os.path.join(bdir, "diff.csv")
|
|
report_path = os.path.join(bdir, "diff_report.txt")
|
|
result = getattr(self.app, '_current_results', None)
|
|
if result:
|
|
headers = [c for c in self.app.results_tree['columns']]
|
|
# Build rows from result['pairs'] to ensure consistency with computed data
|
|
def _row_from_pair(p):
|
|
baseline_name = p.get('fileA') or ''
|
|
current_name = p.get('fileB') or ''
|
|
counts = p.get('counts') or {}
|
|
cd = p.get('countings_delta')
|
|
md = p.get('metrics_delta')
|
|
# ensure simple deltas when missing
|
|
if cd is None:
|
|
base = p.get('baseline_countings')
|
|
cur = p.get('current_countings')
|
|
if base is None and cur is not None:
|
|
cd = {
|
|
'physical_lines': cur.get('physical_lines', 0),
|
|
'code_lines': cur.get('code_lines', 0),
|
|
'comment_lines': cur.get('comment_lines', 0),
|
|
'blank_lines': cur.get('blank_lines', 0),
|
|
}
|
|
elif cur is None and base is not None:
|
|
cd = {
|
|
'physical_lines': 0 - base.get('physical_lines', 0),
|
|
'code_lines': 0 - base.get('code_lines', 0),
|
|
'comment_lines': 0 - base.get('comment_lines', 0),
|
|
'blank_lines': 0 - base.get('blank_lines', 0),
|
|
}
|
|
# Format numeric values similarly to UI (no sign formatting in CSV values)
|
|
return (
|
|
current_name,
|
|
baseline_name,
|
|
counts.get('added',0),
|
|
counts.get('deleted',0),
|
|
counts.get('modified',0),
|
|
counts.get('unmodified',0),
|
|
(cd.get('code_lines') if cd else ''),
|
|
(cd.get('comment_lines') if cd else ''),
|
|
(cd.get('blank_lines') if cd else ''),
|
|
(md.get('func_count') if md else ''),
|
|
(md.get('avg_cc') if md else ''),
|
|
(md.get('mi') if md else ''),
|
|
)
|
|
rows = (_row_from_pair(p) for p in result.get('pairs', []))
|
|
export_rows_to_csv(csv_path, headers, rows)
|
|
|
|
# Generate text report
|
|
profile_config = {
|
|
'name': profile_name,
|
|
'root': project,
|
|
'paths': paths if paths else [project],
|
|
'languages': allowed_exts if allowed_exts else [],
|
|
'exclude_patterns': ignore_patterns
|
|
}
|
|
generate_differ_report(result, profile_config, baseline_id, report_path)
|
|
except Exception:
|
|
# non-fatal: continue even if export fails
|
|
pass
|
|
messagebox.showinfo("Differing", f"Baseline created: {baseline_id}\nReports saved: diff.csv, diff_report.txt")
|
|
except Exception:
|
|
self.app._set_phase('Idle')
|
|
self.app._current_task_id = None
|
|
self.app._enable_action_buttons()
|
|
|
|
def _on_diff_done(result):
|
|
try:
|
|
total = result.get('total', {})
|
|
self.app.log(f"Differ finished: added={total.get('added',0)} deleted={total.get('deleted',0)} modified={total.get('modified',0)}", level='INFO')
|
|
self.app.export_btn.config(state='normal')
|
|
|
|
# Create new baseline after successful diff
|
|
try:
|
|
self.app._set_phase('Creating baseline...')
|
|
self.app._current_task_id = self.app.worker.submit(
|
|
bm.create_baseline_from_dir,
|
|
project, None, True, True, ignore_patterns, profile_name, max_keep,
|
|
kind='thread',
|
|
on_done=_on_create_done
|
|
)
|
|
self.app.cancel_btn.config(state='normal')
|
|
except Exception:
|
|
self.app._set_phase('Idle')
|
|
self.app._current_task_id = None
|
|
self.app._enable_action_buttons()
|
|
except Exception as e:
|
|
messagebox.showerror('Differ Error', str(e))
|
|
self.app._enable_action_buttons()
|
|
|
|
# If no baseline exists, treat baseline as empty (new project) and run differ
|
|
if not baselines:
|
|
self.app.log("No baseline found: treating baseline as empty (new project)", level='INFO')
|
|
# Create an in-memory minimal baseline metadata object
|
|
meta = types.SimpleNamespace(baseline_id="__empty__", project_root=project, files=[], profile=profile_name)
|
|
# Setup differ roots: baseline root empty string, current root is project
|
|
self.app._differ_baseline_root = ''
|
|
self.app._differ_current_root = project
|
|
# Build differ and pairs using empty baseline
|
|
differ = Differ(meta, project, ignore_patterns=ignore_patterns, baseline_files_dir=meta.project_root)
|
|
current_files = differ.build_current_file_list()
|
|
pairs = differ.match_files(differ.baseline.files, current_files)
|
|
if not pairs:
|
|
messagebox.showinfo("Differing", "No files to compare")
|
|
self.app._set_phase('Idle')
|
|
self.app._enable_action_buttons()
|
|
return
|
|
# Process pairs in parallel (reuse existing worker function)
|
|
self._run_differ_worker(differ, pairs, project, meta, _on_diff_done)
|
|
return
|
|
|
|
# Select baseline to compare with
|
|
chosen = self._select_baseline_dialog(baselines)
|
|
if not chosen:
|
|
self.app._set_phase('Idle')
|
|
self.app._enable_action_buttons()
|
|
return
|
|
|
|
try:
|
|
meta = bm.load_metadata(chosen)
|
|
self.app.log(f"Loaded baseline: {chosen}", level='INFO')
|
|
except Exception as e:
|
|
messagebox.showerror("Differing", f"Failed to load baseline metadata: {e}")
|
|
self.app._set_phase('Idle')
|
|
self.app._enable_action_buttons()
|
|
return
|
|
|
|
# Setup baseline and current roots for diff viewer
|
|
self._setup_differ_roots(bm, chosen, meta, project)
|
|
|
|
# Build file lists and start diff
|
|
baseline_files_dir = bm.get_baseline_files_dir(chosen)
|
|
differ = Differ(meta, project, ignore_patterns=ignore_patterns, baseline_files_dir=baseline_files_dir)
|
|
|
|
self.app._set_phase("Scanning current files...")
|
|
self.app.update_idletasks()
|
|
|
|
try:
|
|
current_files = differ.build_current_file_list()
|
|
except Exception:
|
|
messagebox.showerror('Differing', 'Failed to scan current project files')
|
|
self.app._set_phase('Idle')
|
|
self.app._enable_action_buttons()
|
|
return
|
|
|
|
self.app._set_phase("Building pairs...")
|
|
self.app.update_idletasks()
|
|
|
|
try:
|
|
pairs = differ.match_files(differ.baseline.files, current_files)
|
|
except Exception:
|
|
messagebox.showerror('Differing', 'Failed to build file pairs')
|
|
self.app._set_phase('Idle')
|
|
self.app._enable_action_buttons()
|
|
return
|
|
|
|
if not pairs:
|
|
messagebox.showinfo("Differing", "No files to compare")
|
|
self.app._set_phase('Idle')
|
|
self.app._enable_action_buttons()
|
|
return
|
|
|
|
# Process pairs in parallel
|
|
self._run_differ_worker(differ, pairs, project, meta, _on_diff_done)
|
|
|
|
def _run_differ_worker(self, differ, pairs, project, meta, on_diff_done):
|
|
"""Start background worker to process differ pairs."""
|
|
from ..core.countings_impl import analyze_file_counts
|
|
from ..core.metrics import analyze_file_metrics
|
|
|
|
self.app._processed_files = 0
|
|
try:
|
|
self.app.progress['maximum'] = max(1, len(pairs))
|
|
self.app.progress['value'] = 0
|
|
self.app._lbl_files.config(text=f"Files: 0/{len(pairs)}")
|
|
except Exception:
|
|
pass
|
|
|
|
self.app._set_phase(f"Comparing {len(pairs)} file pairs...")
|
|
self.app.update_idletasks()
|
|
|
|
# Process pair function
|
|
def _process_pair(pair):
|
|
a, b = pair
|
|
res = {
|
|
'fileA': a.path if a is not None else None,
|
|
'fileB': b.path if b is not None else None,
|
|
'counts': None,
|
|
'file_meta': None,
|
|
'baseline_countings': None,
|
|
'current_countings': None,
|
|
'countings_delta': None,
|
|
'baseline_metrics': None,
|
|
'current_metrics': None,
|
|
'metrics_delta': None,
|
|
}
|
|
|
|
# Determine file status: added, deleted, or modified
|
|
# Check if both files exist and have same SHA1 hash
|
|
if a is None and b is not None:
|
|
# File added
|
|
res['counts'] = {'added': 1, 'deleted': 0, 'modified': 0, 'unmodified': 0}
|
|
elif a is not None and b is None:
|
|
# File deleted
|
|
res['counts'] = {'added': 0, 'deleted': 1, 'modified': 0, 'unmodified': 0}
|
|
elif a is not None and b is not None:
|
|
# Both exist: check if modified (compare SHA1 hash)
|
|
# Compute current file SHA1 if not already present
|
|
if not hasattr(b, 'sha1') or b.sha1 is None:
|
|
try:
|
|
import hashlib
|
|
file_path = Path(os.path.join(project, b.path))
|
|
h = hashlib.sha1()
|
|
with file_path.open("rb") as f:
|
|
for chunk in iter(lambda: f.read(8192), b""):
|
|
h.update(chunk)
|
|
b.sha1 = h.hexdigest()
|
|
except Exception:
|
|
b.sha1 = None
|
|
|
|
# Compare hashes
|
|
if hasattr(a, 'sha1') and a.sha1 and b.sha1 and a.sha1 == b.sha1:
|
|
# Identical files
|
|
res['counts'] = {'added': 0, 'deleted': 0, 'modified': 0, 'unmodified': 1}
|
|
else:
|
|
# Modified file
|
|
res['counts'] = {'added': 0, 'deleted': 0, 'modified': 1, 'unmodified': 0}
|
|
else:
|
|
# Both None (shouldn't happen)
|
|
res['counts'] = {'added': 0, 'deleted': 0, 'modified': 0, 'unmodified': 0}
|
|
|
|
# Extract baseline countings and metrics
|
|
if a is not None:
|
|
if hasattr(a, 'countings') and a.countings:
|
|
res['baseline_countings'] = a.countings
|
|
if hasattr(a, 'metrics') and a.metrics:
|
|
res['baseline_metrics'] = a.metrics
|
|
|
|
# Extract current countings and metrics
|
|
if b is not None:
|
|
# Compute on-the-fly if missing
|
|
need_countings = not (hasattr(b, 'countings') and b.countings)
|
|
need_metrics = not (hasattr(b, 'metrics') and b.metrics)
|
|
|
|
if need_countings:
|
|
try:
|
|
counts_result = analyze_file_counts(Path(os.path.join(project, b.path)))
|
|
b.countings = {
|
|
'physical_lines': counts_result.get('physical_lines', 0),
|
|
'code_lines': counts_result.get('code_lines', 0),
|
|
'comment_lines': counts_result.get('comment_lines', 0),
|
|
'blank_lines': counts_result.get('blank_lines', 0),
|
|
}
|
|
except Exception:
|
|
b.countings = None
|
|
|
|
if need_metrics:
|
|
try:
|
|
metrics_result = analyze_file_metrics(Path(os.path.join(project, b.path)))
|
|
b.metrics = {
|
|
'func_count': metrics_result.get('func_count', 0),
|
|
'avg_cc': metrics_result.get('avg_cc', 0.0),
|
|
'max_cc': metrics_result.get('max_cc', 0),
|
|
'mi': metrics_result.get('mi', 0.0),
|
|
}
|
|
except Exception:
|
|
b.metrics = None
|
|
|
|
file_entry = {
|
|
'path': b.path,
|
|
'size': b.size,
|
|
'mtime': b.mtime,
|
|
'sha1': b.sha1,
|
|
}
|
|
if hasattr(b, 'countings') and b.countings:
|
|
file_entry['countings'] = b.countings
|
|
res['current_countings'] = b.countings
|
|
else:
|
|
file_entry['countings'] = None
|
|
|
|
if hasattr(b, 'metrics') and b.metrics:
|
|
file_entry['metrics'] = b.metrics
|
|
res['current_metrics'] = b.metrics
|
|
else:
|
|
file_entry['metrics'] = None
|
|
|
|
res['file_meta'] = file_entry
|
|
|
|
# Compute deltas
|
|
if res['baseline_countings'] and res['current_countings']:
|
|
res['countings_delta'] = {
|
|
'physical_lines': res['current_countings']['physical_lines'] - res['baseline_countings'].get('physical_lines', 0),
|
|
'code_lines': res['current_countings']['code_lines'] - res['baseline_countings'].get('code_lines', 0),
|
|
'comment_lines': res['current_countings']['comment_lines'] - res['baseline_countings'].get('comment_lines', 0),
|
|
'blank_lines': res['current_countings']['blank_lines'] - res['baseline_countings'].get('blank_lines', 0),
|
|
}
|
|
|
|
if res['baseline_metrics'] and res['current_metrics']:
|
|
res['metrics_delta'] = {
|
|
'func_count': res['current_metrics']['func_count'] - res['baseline_metrics'].get('func_count', 0),
|
|
'avg_cc': res['current_metrics']['avg_cc'] - res['baseline_metrics'].get('avg_cc', 0.0),
|
|
'max_cc': res['current_metrics']['max_cc'] - res['baseline_metrics'].get('max_cc', 0),
|
|
'mi': res['current_metrics']['mi'] - res['baseline_metrics'].get('mi', 0.0),
|
|
}
|
|
|
|
# Compute metrics delta when one side missing (treat missing as zero)
|
|
if res.get('metrics_delta') is None:
|
|
base_m = res.get('baseline_metrics')
|
|
cur_m = res.get('current_metrics')
|
|
if base_m is None and cur_m is not None:
|
|
# additions: delta == current
|
|
res['metrics_delta'] = {
|
|
'func_count': cur_m.get('func_count', 0),
|
|
'avg_cc': cur_m.get('avg_cc', 0.0),
|
|
'max_cc': cur_m.get('max_cc', 0),
|
|
'mi': cur_m.get('mi', 0.0),
|
|
}
|
|
elif cur_m is None and base_m is not None:
|
|
# deletions: delta == 0 - baseline
|
|
res['metrics_delta'] = {
|
|
'func_count': 0 - base_m.get('func_count', 0),
|
|
'avg_cc': 0.0 - base_m.get('avg_cc', 0.0),
|
|
'max_cc': 0 - base_m.get('max_cc', 0),
|
|
'mi': 0.0 - base_m.get('mi', 0.0),
|
|
}
|
|
|
|
return res
|
|
|
|
# Batch buffer for GUI updates
|
|
_batch_buffer = []
|
|
_batch_size = 50
|
|
|
|
# Progress handler
|
|
def _on_progress(item_res):
|
|
try:
|
|
# Columns: Current (codebase) and Baseline
|
|
baseline = item_res.get('fileA') or ''
|
|
current = item_res.get('fileB') or ''
|
|
counts = item_res.get('counts') or {}
|
|
|
|
countings_delta = item_res.get('countings_delta')
|
|
metrics_delta = item_res.get('metrics_delta')
|
|
|
|
# If deltas are not provided, compute simple deltas when one side is missing
|
|
if countings_delta is None:
|
|
base = item_res.get('baseline_countings')
|
|
cur = item_res.get('current_countings')
|
|
if base is None and cur is not None:
|
|
# additions: delta == current
|
|
countings_delta = {
|
|
'physical_lines': cur.get('physical_lines', 0),
|
|
'code_lines': cur.get('code_lines', 0),
|
|
'comment_lines': cur.get('comment_lines', 0),
|
|
'blank_lines': cur.get('blank_lines', 0),
|
|
}
|
|
elif cur is None and base is not None:
|
|
# deletions: delta == 0 - baseline
|
|
countings_delta = {
|
|
'physical_lines': 0 - base.get('physical_lines', 0),
|
|
'code_lines': 0 - base.get('code_lines', 0),
|
|
'comment_lines': 0 - base.get('comment_lines', 0),
|
|
'blank_lines': 0 - base.get('blank_lines', 0),
|
|
}
|
|
|
|
# Format delta values with explicit sign (include + for zero/positive)
|
|
def format_delta(value, is_float=False):
|
|
if value is None:
|
|
return ''
|
|
if is_float:
|
|
formatted = f"{abs(value):.2f}"
|
|
sign = '+' if value >= 0 else '-'
|
|
return f"{sign}{formatted}"
|
|
else:
|
|
ival = int(value)
|
|
formatted = str(abs(ival))
|
|
sign = '+' if ival >= 0 else '-'
|
|
return f"{sign}{formatted}"
|
|
|
|
# Format metrics: show absolute current value with delta in parentheses (only when baseline exists)
|
|
def format_metric(current_val, delta_val, baseline_exists, is_float=False):
|
|
if current_val is None:
|
|
return ''
|
|
# Show delta in parentheses when baseline exists (even if delta is zero)
|
|
if baseline_exists and delta_val is not None:
|
|
# Show absolute value with delta in parentheses
|
|
if is_float:
|
|
abs_str = f"{current_val:.2f}"
|
|
delta_str = format_delta(delta_val, is_float=True)
|
|
else:
|
|
abs_str = str(int(current_val))
|
|
delta_str = format_delta(delta_val, is_float=False)
|
|
return f"{abs_str} ({delta_str})"
|
|
else:
|
|
# No baseline: show only absolute value
|
|
if is_float:
|
|
return f"{current_val:.2f}"
|
|
else:
|
|
return str(int(current_val))
|
|
|
|
delta_code = format_delta(countings_delta.get('code_lines') if countings_delta else None)
|
|
delta_comment = format_delta(countings_delta.get('comment_lines') if countings_delta else None)
|
|
delta_blank = format_delta(countings_delta.get('blank_lines') if countings_delta else None)
|
|
|
|
# Metrics: show absolute current value with delta in parentheses (only when baseline exists)
|
|
# Check if baseline exists by verifying baseline_id is not "__empty__"
|
|
baseline_exists = meta.baseline_id != "__empty__"
|
|
cur_metrics = item_res.get('current_metrics')
|
|
delta_func = format_metric(
|
|
cur_metrics.get('func_count') if cur_metrics else None,
|
|
metrics_delta.get('func_count') if metrics_delta else None,
|
|
baseline_exists
|
|
)
|
|
delta_cc = format_metric(
|
|
cur_metrics.get('avg_cc') if cur_metrics else None,
|
|
metrics_delta.get('avg_cc') if metrics_delta else None,
|
|
baseline_exists,
|
|
is_float=True
|
|
)
|
|
delta_mi = format_metric(
|
|
cur_metrics.get('mi') if cur_metrics else None,
|
|
metrics_delta.get('mi') if metrics_delta else None,
|
|
baseline_exists,
|
|
is_float=True
|
|
)
|
|
|
|
# Add to batch buffer: Current, Baseline, then counts/deltas
|
|
_batch_buffer.append((current, baseline, counts.get('added',0), counts.get('deleted',0), counts.get('modified',0), counts.get('unmodified',0), delta_code, delta_comment, delta_blank, delta_func, delta_cc, delta_mi))
|
|
|
|
# Insert batch when buffer is full
|
|
if len(_batch_buffer) >= _batch_size:
|
|
for row_values in _batch_buffer:
|
|
self.app.results_tree.insert('', 'end', values=row_values)
|
|
_batch_buffer.clear()
|
|
|
|
# Update progress (less frequent)
|
|
try:
|
|
self.app._processed_files = getattr(self.app, '_processed_files', 0) + 1
|
|
if self.app._processed_files % 10 == 0 or self.app._processed_files == len(pairs):
|
|
self.app._lbl_files.config(text=f"Files: {self.app._processed_files}/{len(pairs)}")
|
|
self.app.progress['maximum'] = max(1, len(pairs))
|
|
self.app.progress['value'] = self.app._processed_files
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
# Done handler
|
|
def _on_done(all_results):
|
|
# Flush remaining batch buffer
|
|
if _batch_buffer:
|
|
for row_values in _batch_buffer:
|
|
self.app.results_tree.insert('', 'end', values=row_values)
|
|
_batch_buffer.clear()
|
|
|
|
# Compute totals
|
|
total = {"added":0, "deleted":0, "modified":0, "unmodified":0}
|
|
current_files_list = []
|
|
for it in all_results:
|
|
c = it.get('counts', {})
|
|
total['added'] += c.get('added',0)
|
|
total['deleted'] += c.get('deleted',0)
|
|
total['modified'] += c.get('modified',0)
|
|
total['unmodified'] += c.get('unmodified',0)
|
|
fm = it.get('file_meta')
|
|
if fm:
|
|
current_files_list.append(fm)
|
|
|
|
result = {"baseline_id": meta.baseline_id, "compared_at": time.time(), "total": total, "pairs": all_results}
|
|
|
|
# Build summary statistics
|
|
try:
|
|
baseline_counts = {"physical_lines": 0, "code_lines": 0, "comment_lines": 0, "blank_lines": 0, "file_count": 0}
|
|
baseline_metrics = {"file_count": 0, "total_func_count": 0, "avg_avg_cc": 0.0, "avg_mi": 0.0}
|
|
baseline_metrics_count = 0
|
|
|
|
for fm in meta.files:
|
|
if hasattr(fm, 'countings') and fm.countings:
|
|
baseline_counts["physical_lines"] += fm.countings.get("physical_lines", 0)
|
|
baseline_counts["code_lines"] += fm.countings.get("code_lines", 0)
|
|
baseline_counts["comment_lines"] += fm.countings.get("comment_lines", 0)
|
|
baseline_counts["blank_lines"] += fm.countings.get("blank_lines", 0)
|
|
baseline_counts["file_count"] += 1
|
|
if hasattr(fm, 'metrics') and fm.metrics:
|
|
baseline_metrics["total_func_count"] += fm.metrics.get("func_count", 0)
|
|
baseline_metrics["avg_avg_cc"] += fm.metrics.get("avg_cc", 0.0)
|
|
baseline_metrics["avg_mi"] += fm.metrics.get("mi", 0.0)
|
|
baseline_metrics["file_count"] += 1
|
|
baseline_metrics_count += 1
|
|
|
|
if baseline_metrics_count > 0:
|
|
baseline_metrics["avg_avg_cc"] /= baseline_metrics_count
|
|
baseline_metrics["avg_mi"] /= baseline_metrics_count
|
|
|
|
current_counts = {"physical_lines": 0, "code_lines": 0, "comment_lines": 0, "blank_lines": 0, "file_count": 0}
|
|
current_metrics = {"file_count": 0, "total_func_count": 0, "avg_avg_cc": 0.0, "avg_mi": 0.0}
|
|
current_metrics_count = 0
|
|
|
|
for fm in current_files_list:
|
|
c = fm.get('countings')
|
|
if c:
|
|
current_counts["physical_lines"] += c.get("physical_lines", 0)
|
|
current_counts["code_lines"] += c.get("code_lines", 0)
|
|
current_counts["comment_lines"] += c.get("comment_lines", 0)
|
|
current_counts["blank_lines"] += c.get("blank_lines", 0)
|
|
current_counts["file_count"] += 1
|
|
m = fm.get('metrics')
|
|
if m:
|
|
current_metrics["total_func_count"] += m.get("func_count", 0)
|
|
current_metrics["avg_avg_cc"] += m.get("avg_cc", 0.0)
|
|
current_metrics["avg_mi"] += m.get("mi", 0.0)
|
|
current_metrics["file_count"] += 1
|
|
current_metrics_count += 1
|
|
|
|
if current_metrics_count > 0:
|
|
current_metrics["avg_avg_cc"] /= current_metrics_count
|
|
current_metrics["avg_mi"] /= current_metrics_count
|
|
|
|
result["summary"] = {
|
|
"baseline": {"countings": baseline_counts, "metrics": baseline_metrics},
|
|
"current": {"countings": current_counts, "metrics": current_metrics}
|
|
}
|
|
except Exception:
|
|
pass
|
|
|
|
self.app._current_results = result
|
|
on_diff_done(result)
|
|
|
|
self.app._current_task_id = self.app.worker.map_iterable(_process_pair, pairs, kind='thread', on_progress=_on_progress, on_done=_on_done)
|
|
self.app.cancel_btn.config(state='normal')
|
|
|
|
def _setup_differ_roots(self, bm, chosen, meta, project):
|
|
"""Setup baseline and current roots for diff viewer."""
|
|
import zipfile
|
|
|
|
baseline_dir = bm._baseline_dir(chosen)
|
|
baseline_snapshot_dir = os.path.join(baseline_dir, "files")
|
|
baseline_zip = os.path.join(baseline_dir, "files.zip")
|
|
|
|
if os.path.isdir(baseline_snapshot_dir):
|
|
self.app._differ_baseline_root = baseline_snapshot_dir
|
|
elif os.path.exists(baseline_zip):
|
|
try:
|
|
with zipfile.ZipFile(baseline_zip, 'r') as zip_ref:
|
|
zip_ref.extractall(baseline_snapshot_dir)
|
|
self.app._differ_baseline_root = baseline_snapshot_dir
|
|
except Exception as e:
|
|
self.app.log(f"Failed to extract baseline snapshot: {e}", level='WARNING')
|
|
self.app._differ_baseline_root = meta.project_root
|
|
else:
|
|
self.app.log("No baseline snapshot found, using project_root (may be inaccurate)", level='WARNING')
|
|
self.app._differ_baseline_root = meta.project_root
|
|
|
|
self.app._differ_current_root = project
|
|
|
|
def _select_baseline_dialog(self, baselines):
|
|
"""Show modal dialog to select a baseline."""
|
|
sorted_b = sorted(baselines)
|
|
|
|
dlg = tk.Toplevel(self.app)
|
|
dlg.title("Select baseline to compare")
|
|
dlg.transient(self.app)
|
|
dlg.grab_set()
|
|
|
|
# Center dialog
|
|
dlg.update_idletasks()
|
|
pw = self.app.winfo_width()
|
|
ph = self.app.winfo_height()
|
|
px = self.app.winfo_rootx()
|
|
py = self.app.winfo_rooty()
|
|
dw = dlg.winfo_reqwidth()
|
|
dh = dlg.winfo_reqheight()
|
|
x = px + (pw - dw) // 2
|
|
y = py + (ph - dh) // 2
|
|
dlg.geometry(f"+{x}+{y}")
|
|
|
|
ttk.Label(dlg, text="Select a baseline:").grid(row=0, column=0, padx=12, pady=(12,4), sticky='w')
|
|
listbox = tk.Listbox(dlg, height=min(10, len(sorted_b)), width=60, exportselection=False)
|
|
for item in sorted_b:
|
|
listbox.insert('end', item)
|
|
listbox.grid(row=1, column=0, padx=12, pady=(0,12))
|
|
listbox.select_set(len(sorted_b)-1) # Select latest
|
|
|
|
selected = {'id': None}
|
|
|
|
def _on_ok():
|
|
sel = listbox.curselection()
|
|
if not sel:
|
|
messagebox.showwarning('Select baseline', 'Please select a baseline to compare')
|
|
return
|
|
selected['id'] = listbox.get(sel[0])
|
|
dlg.destroy()
|
|
|
|
def _on_cancel():
|
|
selected['id'] = None
|
|
dlg.destroy()
|
|
|
|
btn_frame = ttk.Frame(dlg)
|
|
btn_frame.grid(row=2, column=0, pady=(0,12), sticky='e', padx=12)
|
|
ok_btn = ttk.Button(btn_frame, text='OK', command=_on_ok)
|
|
ok_btn.grid(row=0, column=0, padx=(0,8))
|
|
cancel_btn = ttk.Button(btn_frame, text='Cancel', command=_on_cancel)
|
|
cancel_btn.grid(row=0, column=1)
|
|
|
|
def _on_dbl(evt):
|
|
idx = listbox.curselection()
|
|
if idx:
|
|
selected['id'] = listbox.get(idx[0])
|
|
dlg.destroy()
|
|
|
|
listbox.bind('<Double-Button-1>', _on_dbl)
|
|
self.app.wait_window(dlg)
|
|
|
|
return selected.get('id')
|
|
|
|
def _gather_source_files(self, paths, allowed_exts, ignore_patterns):
|
|
"""Gather source files from given paths."""
|
|
files = []
|
|
for p in paths:
|
|
pth = Path(p)
|
|
try:
|
|
if pth.is_dir():
|
|
files.extend(find_source_files(pth, allowed_extensions=allowed_exts, ignore_patterns=ignore_patterns))
|
|
elif pth.is_file():
|
|
files.append(pth)
|
|
except Exception:
|
|
continue
|
|
|
|
# Deduplicate
|
|
seen = set()
|
|
unique = []
|
|
for f in files:
|
|
s = str(f)
|
|
if s not in seen:
|
|
seen.add(s)
|
|
unique.append(f)
|
|
return unique
|
|
|
|
def _reset_progress_ui(self):
|
|
"""Reset progress counters and UI elements."""
|
|
self.app._processed_files = 0
|
|
self.app._total_files = 0
|
|
try:
|
|
self.app.progress['maximum'] = 1
|
|
self.app.progress['value'] = 0
|
|
self.app._lbl_files.config(text="Files: 0/0")
|
|
except Exception:
|
|
pass
|
|
|
|
def _clear_results(self):
|
|
"""Clear results tree."""
|
|
for c in self.app.results_tree.get_children(""):
|
|
self.app.results_tree.delete(c)
|