#!/usr/bin/env python3 """ Aethero Audit System - Introspective Development Performance Analysis Systém na extrakciu a analýzu Aetheron jednotiek z vývojových logov Slovak Healthcare Worker Solo Development Efficiency Measurement """ import json import subprocess import re import os import sys from datetime import datetime, timedelta from typing import Dict, List, Any, Optional, Tuple from pathlib import Path from dataclasses import dataclass, asdict from collections import defaultdict, Counter import csv # Import existujúcich Aethero komponentov sys.path.append('/Users/_xvadur/Desktop/Aethero_github/Aethero_App') from introspective_parser_module.metrics import CognitiveMetricsAnalyzer from introspective_parser_module.models import ( ASLCognitiveTag, MentalStateEnum, EmotionToneEnum, TemporalContextEnum, AetheroIntrospectiveEntity ) @dataclass class AetheronUnit: """ Základná jednotka vývojovej produktivity - 1 Aetheron Reprezentuje kvantifikovateľný vývojový výkon """ timestamp: datetime aetheron_value: float # 1.0 = 1 Aetheron git_commit_count: int shell_commands_count: int cognitive_load_estimate: float development_rhythm_score: float efficiency_multiplier: float context_tags: List[str] @property def total_output_score(self) -> float: """Celkový výstupný skór pre daný časový úsek""" return ( self.git_commit_count * 0.4 + self.shell_commands_count * 0.2 + self.development_rhythm_score * 0.3 + self.efficiency_multiplier * 0.1 ) @dataclass class DevelopmentSession: """Reprezentácia vývojovej relácie""" start_time: datetime end_time: datetime total_aetherony: float commits: List[Dict[str, Any]] commands: List[Dict[str, Any]] cognitive_coherence: float productivity_rating: str # "vysoká", "stredná", "nízka" class AetheroAuditSystem: """ Hlavný audit systém pre analýzu vývojového výkonu Integrácia s existujúcim ASL kognitívnym systémom """ def __init__(self, git_repo_path: str = None, shell_history_path: str = None): self.git_repo_path = git_repo_path or "/Users/_xvadur/Desktop/Aethero_github" self.shell_history_path = shell_history_path or os.path.expanduser("~/.zsh_history") self.cognitive_analyzer = CognitiveMetricsAnalyzer() self.audit_session_id = datetime.now().strftime("%Y%m%d_%H%M%S") # Definícia oficiálnej Aetheron jednotky self.AETHERON_DEFINITION = { "base_unit": "1 Aetheron = 1 hodina efektívneho vývoja", "measurement_factors": { "git_commits": 0.3, "shell_commands": 0.2, "cognitive_coherence": 0.3, "time_efficiency": 0.2 }, "slovak_context": "Meranie produktivity slovenského zdravotníckeho pracovníka" } def extract_git_development_data(self, days_back: int = 30) -> List[Dict[str, Any]]: """ Extrahovanie dát z git logu za posledné dni """ since_date = (datetime.now() - timedelta(days=days_back)).strftime("%Y-%m-%d") try: # Git log s podrobnými informáciami git_cmd = [ "git", "log", f"--since={since_date}", "--pretty=format:%H|%an|%ae|%ad|%s|%b", "--date=iso", "--numstat" ] result = subprocess.run( git_cmd, cwd=self.git_repo_path, capture_output=True, text=True, check=True ) commits = self._parse_git_log_output(result.stdout) print(f"[AUDIT] Extrahovaných {len(commits)} commit-ov za posledných {days_back} dní") return commits except subprocess.CalledProcessError as e: print(f"[ERROR] Git log extraction failed: {e}") return [] def _parse_git_log_output(self, git_output: str) -> List[Dict[str, Any]]: """Parsovanie výstupu git log""" commits = [] lines = git_output.strip().split('\n') current_commit = None for line in lines: if '|' in line and len(line.split('|')) == 6: # Nový commit if current_commit: commits.append(current_commit) parts = line.split('|') # Parse date properly handling timezone date_str = parts[3].replace(' +0100', '').replace(' +0200', '') if '+' in date_str: date_str = date_str.split('+')[0] try: commit_date = datetime.fromisoformat(date_str) except: # Fallback parsing from dateutil import parser commit_date = parser.parse(parts[3]).replace(tzinfo=None) current_commit = { 'hash': parts[0], 'author': parts[1], 'email': parts[2], 'date': commit_date, 'subject': parts[4], 'body': parts[5], 'files_changed': [], 'lines_added': 0, 'lines_removed': 0 } elif current_commit and '\t' in line: # Štatistiky súborov parts = line.split('\t') if len(parts) == 3: added, removed, filename = parts current_commit['files_changed'].append(filename) if added.isdigit(): current_commit['lines_added'] += int(added) if removed.isdigit(): current_commit['lines_removed'] += int(removed) if current_commit: commits.append(current_commit) return commits def extract_shell_development_commands(self, days_back: int = 30) -> List[Dict[str, Any]]: """ Extrahovanie vývojových príkazov zo shell histórie """ if not os.path.exists(self.shell_history_path): print(f"[WARNING] Shell history súbor nenájdený: {self.shell_history_path}") return [] since_timestamp = (datetime.now() - timedelta(days=days_back)).timestamp() commands = [] try: with open(self.shell_history_path, 'r', encoding='utf-8', errors='ignore') as f: for line in f: line = line.strip() if line.startswith(': '): # Zsh history formát: ": timestamp:elapsed_time;command" match = re.match(r': (\d+):\d+;(.+)', line) if match: timestamp = int(match.group(1)) command = match.group(2) if timestamp >= since_timestamp: if self._is_development_command(command): commands.append({ 'timestamp': datetime.fromtimestamp(timestamp), 'command': command, 'category': self._categorize_command(command), 'complexity_score': self._assess_command_complexity(command) }) print(f"[AUDIT] Extrahovaných {len(commands)} vývojových príkazov") return sorted(commands, key=lambda x: x['timestamp']) except Exception as e: print(f"[ERROR] Shell history parsing failed: {e}") return [] def _is_development_command(self, command: str) -> bool: """Identifikácia, či príkaz súvisí s vývojom""" dev_keywords = [ 'git', 'npm', 'pip', 'python', 'node', 'yarn', 'pnpm', 'docker', 'kubectl', 'terraform', 'ansible', 'vim', 'nvim', 'emacs', 'code', 'subl', 'make', 'cmake', 'cargo', 'mvn', 'gradle', 'pytest', 'jest', 'mocha', 'cypress', 'cd', 'ls', 'find', 'grep', 'cat', 'tail', 'head', 'curl', 'wget', 'ssh', 'scp', 'rsync' ] return any(keyword in command.lower() for keyword in dev_keywords) def _categorize_command(self, command: str) -> str: """Kategorizácia vývojového príkazu""" if any(kw in command.lower() for kw in ['git commit', 'git push', 'git pull']): return 'version_control' elif any(kw in command.lower() for kw in ['npm install', 'pip install', 'yarn add']): return 'dependency_management' elif any(kw in command.lower() for kw in ['python', 'node', 'npm run', 'yarn']): return 'execution' elif any(kw in command.lower() for kw in ['vim', 'nvim', 'code', 'emacs']): return 'editing' elif any(kw in command.lower() for kw in ['test', 'pytest', 'jest', 'mocha']): return 'testing' elif any(kw in command.lower() for kw in ['docker', 'kubectl', 'terraform']): return 'infrastructure' else: return 'general' def _assess_command_complexity(self, command: str) -> float: """Hodnotenie zložitosti príkazu (1.0 - 10.0)""" complexity_indicators = { 'git rebase': 8.0, 'docker-compose': 7.0, 'kubectl': 6.0, 'terraform': 7.0, 'pip install': 3.0, 'npm install': 3.0, 'git commit': 4.0, 'git push': 3.0, 'python': 5.0, 'vim': 6.0, 'grep': 4.0, 'find': 5.0 } for indicator, score in complexity_indicators.items(): if indicator in command.lower(): # Úprava skóre podľa dĺžky príkazu length_multiplier = min(1.5, len(command) / 50) return min(10.0, score * length_multiplier) return 3.0 # Základné skóre pre nerozpoznané príkazy def calculate_development_sessions(self, commits: List[Dict], commands: List[Dict]) -> List[DevelopmentSession]: """ Identifikácia a analýza vývojových relácií Relácia = kontinuálny blok vývojovej aktivity """ sessions = [] all_activities = [] # Zlúčenie commit-ov a príkazov do časovej osi for commit in commits: all_activities.append({ 'timestamp': commit['date'], 'type': 'commit', 'data': commit }) for command in commands: all_activities.append({ 'timestamp': command['timestamp'], 'type': 'command', 'data': command }) all_activities.sort(key=lambda x: x['timestamp']) # Identifikácia relácií (gap viac ako 2 hodiny = nová relácia) session_gap_threshold = timedelta(hours=2) current_session_activities = [] for i, activity in enumerate(all_activities): if not current_session_activities: current_session_activities.append(activity) else: time_gap = activity['timestamp'] - current_session_activities[-1]['timestamp'] if time_gap <= session_gap_threshold: current_session_activities.append(activity) else: # Ukončenie aktuálnej relácie if len(current_session_activities) >= 3: # Minimálne 3 aktivity session = self._create_development_session(current_session_activities) sessions.append(session) current_session_activities = [activity] # Posledná relácia if len(current_session_activities) >= 3: session = self._create_development_session(current_session_activities) sessions.append(session) print(f"[AUDIT] Identifikovaných {len(sessions)} vývojových relácií") return sessions def _create_development_session(self, activities: List[Dict]) -> DevelopmentSession: """Vytvorenie DevelopmentSession z aktivít""" start_time = activities[0]['timestamp'] end_time = activities[-1]['timestamp'] commits = [a['data'] for a in activities if a['type'] == 'commit'] commands = [a['data'] for a in activities if a['type'] == 'command'] # Výpočet Aetheron jednotiek pre reláciu session_duration_hours = (end_time - start_time).total_seconds() / 3600 # Aetheron kalkulácia commit_score = len(commits) * 0.5 # 0.5 Aetheron za commit command_score = sum(cmd.get('complexity_score', 3.0) for cmd in commands) / 10 # Efektivita na základe časového pomeru efficiency_ratio = min(1.0, (commit_score + command_score) / max(session_duration_hours, 0.1)) total_aetherony = (commit_score + command_score) * efficiency_ratio # Kognitívna koherencia (simulovaná na základe aktivít) cognitive_coherence = self._estimate_cognitive_coherence(commits, commands) # Hodnotenie produktivity if total_aetherony >= 2.0: productivity_rating = "vysoká" elif total_aetherony >= 1.0: productivity_rating = "stredná" else: productivity_rating = "nízka" return DevelopmentSession( start_time=start_time, end_time=end_time, total_aetherony=total_aetherony, commits=commits, commands=commands, cognitive_coherence=cognitive_coherence, productivity_rating=productivity_rating ) def _estimate_cognitive_coherence(self, commits: List[Dict], commands: List[Dict]) -> float: """ Odhad kognitívnej koherencie na základe vzorcov v aktivitách Integrácia s existujúcim ASL systémom """ # Analýza vzorcov v commit správach commit_complexity = [] for commit in commits: subject = commit.get('subject', '') # Hodnotenie na základe ASL kritérií if any(word in subject.lower() for word in ['fix', 'bug', 'error']): commit_complexity.append(0.3) # Problémové riešenie = nižšia koherencia elif any(word in subject.lower() for word in ['feature', 'add', 'implement']): commit_complexity.append(0.8) # Nová funkcionalita = vyššia koherencia elif any(word in subject.lower() for word in ['refactor', 'clean', 'optimize']): commit_complexity.append(0.9) # Optimalizácia = najvyššia koherencia else: commit_complexity.append(0.5) # Analýza diverzity príkazov command_categories = [cmd.get('category', 'general') for cmd in commands] category_diversity = len(set(command_categories)) / max(len(command_categories), 1) base_coherence = sum(commit_complexity) / max(len(commit_complexity), 1) if commit_complexity else 0.5 diversity_bonus = category_diversity * 0.2 return min(1.0, base_coherence + diversity_bonus) def generate_aetheron_units(self, sessions: List[DevelopmentSession]) -> List[AetheronUnit]: """Generovanie detailných Aetheron jednotiek""" aetheron_units = [] for session in sessions: # Rozdelenie relácie na hodinové bloky pre presnejšie meranie session_duration = session.end_time - session.start_time hour_blocks = max(1, int(session_duration.total_seconds() / 3600)) for hour in range(hour_blocks): block_start = session.start_time + timedelta(hours=hour) block_end = min(session.start_time + timedelta(hours=hour + 1), session.end_time) # Aktivity v tomto bloku block_commits = [c for c in session.commits if block_start <= c['date'] < block_end] block_commands = [c for c in session.commands if block_start <= c['timestamp'] < block_end] if block_commits or block_commands: # Výpočet Aetheron hodnoty pre blok commit_value = len(block_commits) * 0.3 command_value = len(block_commands) * 0.1 cognitive_bonus = session.cognitive_coherence * 0.2 # Rytmus vývoja na základe frekvencií aktivít rhythm_score = min(1.0, (len(block_commits) + len(block_commands)) / 10) # Efektivitný multiplikátor efficiency = session.cognitive_coherence * rhythm_score aetheron_value = (commit_value + command_value + cognitive_bonus) * (1 + efficiency) # Kontextové tagy context_tags = [] if any('fix' in c.get('subject', '').lower() for c in block_commits): context_tags.append('debugging') if any('feature' in c.get('subject', '').lower() for c in block_commits): context_tags.append('feature_development') if any(c.get('category') == 'testing' for c in block_commands): context_tags.append('testing') unit = AetheronUnit( timestamp=block_start, aetheron_value=aetheron_value, git_commit_count=len(block_commits), shell_commands_count=len(block_commands), cognitive_load_estimate=10 - (session.cognitive_coherence * 8), development_rhythm_score=rhythm_score, efficiency_multiplier=efficiency, context_tags=context_tags ) aetheron_units.append(unit) return aetheron_units def export_audit_results(self, sessions: List[DevelopmentSession], aetheron_units: List[AetheronUnit]) -> Dict[str, str]: """Export výsledkov auditu do JSON a CSV formátov""" # JSON export json_data = { 'audit_metadata': { 'session_id': self.audit_session_id, 'generated_at': datetime.now().isoformat(), 'git_repo': self.git_repo_path, 'aetheron_definition': self.AETHERON_DEFINITION, 'total_sessions': len(sessions), 'total_aetheron_units': len(aetheron_units), 'total_aetherony_generated': sum(unit.aetheron_value for unit in aetheron_units) }, 'development_sessions': [], 'aetheron_units': [], 'summary_statistics': self._generate_summary_statistics(sessions, aetheron_units) } # Konverzia session objektov for session in sessions: json_data['development_sessions'].append({ 'start_time': session.start_time.isoformat(), 'end_time': session.end_time.isoformat(), 'duration_hours': (session.end_time - session.start_time).total_seconds() / 3600, 'total_aetherony': session.total_aetherony, 'commits_count': len(session.commits), 'commands_count': len(session.commands), 'cognitive_coherence': session.cognitive_coherence, 'productivity_rating': session.productivity_rating }) # Konverzia Aetheron jednotiek for unit in aetheron_units: json_data['aetheron_units'].append(asdict(unit)) # Zápis JSON json_filename = f"aethero_audit_{self.audit_session_id}.json" json_path = os.path.join(os.getcwd(), json_filename) with open(json_path, 'w', encoding='utf-8') as f: json.dump(json_data, f, indent=2, ensure_ascii=False, default=str) # CSV export csv_filename = f"aethero_audit_units_{self.audit_session_id}.csv" csv_path = os.path.join(os.getcwd(), csv_filename) with open(csv_path, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerow([ 'Timestamp', 'Aetheron_Value', 'Git_Commits', 'Shell_Commands', 'Cognitive_Load', 'Rhythm_Score', 'Efficiency_Multiplier', 'Context_Tags' ]) for unit in aetheron_units: writer.writerow([ unit.timestamp.isoformat(), unit.aetheron_value, unit.git_commit_count, unit.shell_commands_count, unit.cognitive_load_estimate, unit.development_rhythm_score, unit.efficiency_multiplier, ','.join(unit.context_tags) ]) return { 'json_file': json_path, 'csv_file': csv_path } def _generate_summary_statistics(self, sessions: List[DevelopmentSession], units: List[AetheronUnit]) -> Dict[str, Any]: """Generovanie súhrnných štatistík""" if not units: return {} total_aetherony = sum(unit.aetheron_value for unit in units) avg_cognitive_load = sum(unit.cognitive_load_estimate for unit in units) / len(units) avg_rhythm_score = sum(unit.development_rhythm_score for unit in units) / len(units) # Najproduktívnejšie dni daily_productivity = defaultdict(float) for unit in units: day_key = unit.timestamp.strftime('%Y-%m-%d') daily_productivity[day_key] += unit.aetheron_value # Top vývojové vzorce all_tags = [] for unit in units: all_tags.extend(unit.context_tags) tag_frequency = Counter(all_tags) return { 'total_aetherony_generated': round(total_aetherony, 2), 'average_aetherony_per_hour': round(total_aetherony / max(len(units), 1), 2), 'average_cognitive_load': round(avg_cognitive_load, 2), 'average_rhythm_score': round(avg_rhythm_score, 2), 'most_productive_day': max(daily_productivity.items(), key=lambda x: x[1])[0] if daily_productivity else None, 'productivity_by_day': dict(daily_productivity), 'top_development_patterns': dict(tag_frequency.most_common(5)), 'development_efficiency_rating': self._calculate_efficiency_rating(total_aetherony, len(sessions)) } def _calculate_efficiency_rating(self, total_aetherony: float, session_count: int) -> str: """Hodnotenie celkovej efektivity vývojára""" efficiency_score = total_aetherony / max(session_count, 1) if efficiency_score >= 3.0: return "Výnimočná - Slovak Healthcare Dev Ninja 🚀" elif efficiency_score >= 2.0: return "Vysoká - Efektívny Solo Developer 💪" elif efficiency_score >= 1.0: return "Stredná - Stabilný Vývojový Rytmus ⚡" else: return "Nízka - Potreba Optimalizácie 📈" def run_complete_audit(self, days_back: int = 30) -> Dict[str, str]: """ Spustenie kompletného audit procesu """ print(f"\n🔍 AETHERO AUDIT SYSTEM - Spúšťam analýzu za posledných {days_back} dní") print("=" * 70) # 1. Extrahovanie git dát print("📊 Extrakcia git commit histórie...") commits = self.extract_git_development_data(days_back) # 2. Extrahovanie shell dát print("💻 Extrakcia shell command histórie...") commands = self.extract_shell_development_commands(days_back) # 3. Kalkulácia relácií print("🧮 Kalkulácia vývojových relácií...") sessions = self.calculate_development_sessions(commits, commands) # 4. Generovanie Aetheron jednotiek print("⚡ Generovanie Aetheron jednotiek...") aetheron_units = self.generate_aetheron_units(sessions) # 5. Export výsledkov print("💾 Export výsledkov auditu...") file_paths = self.export_audit_results(sessions, aetheron_units) # 6. Súhrnný report total_aetherony = sum(unit.aetheron_value for unit in aetheron_units) print(f"\n✅ AUDIT DOKONČENÝ") print(f"📈 Celkovo vygenerovaných: {total_aetherony:.2f} Aetheron jednotiek") print(f"🕐 Počet vývojových relácií: {len(sessions)}") print(f"📝 Analyzovaných commit-ov: {len(commits)}") print(f"💻 Analyzovaných príkazov: {len(commands)}") print(f"\n📁 Súbory:") print(f" JSON: {file_paths['json_file']}") print(f" CSV: {file_paths['csv_file']}") return file_paths def main(): """Hlavná funkcia pre spustenie auditu""" audit_system = AetheroAuditSystem() # Parametrizácia cez argumenty import argparse parser = argparse.ArgumentParser(description='Aethero Development Audit System') parser.add_argument('--days', type=int, default=30, help='Počet dní na analýzu (default: 30)') parser.add_argument('--git-repo', type=str, help='Cesta k git repozitáru') parser.add_argument('--shell-history', type=str, help='Cesta k shell history súboru') args = parser.parse_args() if args.git_repo: audit_system.git_repo_path = args.git_repo if args.shell_history: audit_system.shell_history_path = args.shell_history # Spustenie auditu file_paths = audit_system.run_complete_audit(args.days) print(f"\n🎯 Aethero Audit dokončený! Súbory uložené:") for file_type, path in file_paths.items(): print(f" {file_type.upper()}: {path}") if __name__ == "__main__": main()