""" Action handlers for PyUCC GUI. Separates business logic from GUI construction. """ import os import time 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() self.app._set_results_columns(("name", "path")) 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() self.app._set_results_columns(("name", "path", "code", "comment", "blank", "total", "language")) 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() self.app._set_results_columns(("name", "path", "avg_cc", "max_cc", "func_count", "mi", "language")) 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...") self.app._set_results_columns(("fileA", "fileB", "added", "deleted", "modified", "unmodified", "Δ code_lines", "Δ comment_lines", "Δ blank_lines", "Δ func_count", "Δ avg_cc", "Δ mi")) 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() messagebox.showinfo("Differing", f"Baseline created: {baseline_id}") 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() # No baseline exists - create one if not baselines: create = messagebox.askyesno("Differing", "No baseline found for this project. Create baseline now?") if not create: self.app._set_phase('Idle') self.app._enable_action_buttons() return self.app._set_phase("Creating baseline...") self.app.update_idletasks() self.app._current_task_id = self.app.worker.submit( bm.create_baseline_from_dir, project, None, True, True, ignore_patterns, profile_name, 5, kind='thread', on_done=_on_create_done ) self.app.cancel_btn.config(state='normal') 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 differ = Differ(meta, project, ignore_patterns=ignore_patterns) 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, } # Diff counts fa = os.path.join(meta.project_root, a.path) if a is not None else None fb = os.path.join(project, b.path) if b is not None else None res['counts'] = Differ._diff_file_pair(fa, fb) # 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: 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), } return res # Batch buffer for GUI updates _batch_buffer = [] _batch_size = 50 # Progress handler def _on_progress(item_res): try: fileA = item_res.get('fileA') or '' fileB = 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') # Format delta values def format_delta(value, is_float=False): if value is None: return '' if is_float: formatted = f"{value:.2f}" if value > 0: return f"+{formatted}" return formatted else: if value > 0: return f"+{value}" return str(value) 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) delta_func = format_delta(metrics_delta.get('func_count') if metrics_delta else None) delta_cc = format_delta(metrics_delta.get('avg_cc') if metrics_delta else None, is_float=True) delta_mi = format_delta(metrics_delta.get('mi') if metrics_delta else None, is_float=True) # Add to batch buffer _batch_buffer.append((fileA, fileB, 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('', _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)