turns-00030.parquet:25550
c97022818ab9f14f598193c4degenerate_repetitionAbsentFinal dense release
USER
# -*- coding: utf-8 -*-
import console
import aiohttp
import asyncio
import os
import json
import time
import re
import logging
import importlib
import ast
import sys
import concurrent.futures
import difflib
import datetime
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field, asdict
from tenacity import retry, stop_after_attempt, wait_exponential
@dataclass
class CodeVersion:
version: int
timestamp: str
content: str
@dataclass
class Aufgabe:
id: str
beschreibung: str
prioritaet: int
status: str = 'offen'
abhaengigkeiten: List[str] = field(default_factory=list)
fortschritt: int = 0
checkliste: List[str] = field(default_factory=list)
class TaskManagementSystem:
def __init__(self):
self.aufgaben: Dict[str, Aufgabe] = {}
self.versionen: Dict[str, List[CodeVersion]] = {}
self.aktuelles_projekt: str = ""
def aufgabe_hinzufuegen(self, aufgabe: Aufgabe):
self.aufgaben[aufgabe.id] = aufgabe
def aufgabe_aktualisieren(self, aufgabe_id: str, **kwargs):
if aufgabe_id in self.aufgaben:
for key, value in kwargs.items():
setattr(self.aufgaben[aufgabe_id], key, value)
def aufgaben_anzeigen(self):
console.clear()
console.set_font('Menlo', 14)
console.set_color(0, 0.5, 1) # Hellblau für Überschrift
print("=== Aufgabenliste ===\n")
console.set_color()
for aufgabe in self.aufgaben.values():
if aufgabe.status == 'abgeschlossen':
console.set_color(0, 0.8, 0) # Grün für abgeschlossene Aufgaben
elif aufgabe.status == 'in_bearbeitung':
console.set_color(1, 0.8, 0) # Orange für Aufgaben in Bearbeitung
else:
console.set_color(0.5, 0.5, 0.5) # Grau für offene Aufgaben
print(f"[{aufgabe.id}] {aufgabe.beschreibung} (Priorität: {aufgabe.prioritaet}, Status: {aufgabe.status})")
print(f" Fortschritt: {aufgabe.fortschritt}%")
if aufgabe.abhaengigkeiten:
print(f" Abhängigkeiten: {', '.join(aufgabe.abhaengigkeiten)}")
if aufgabe.checkliste:
print(" Checkliste:")
for item in aufgabe.checkliste:
print(f" - {item}")
print()
console.set_color() # Zurücksetzen der Farbe
def version_speichern(self):
if not self.aktuelles_projekt:
print("Kein aktives Projekt zum Speichern der Version")
return
content = self.get_current_project_content()
version = len(self.versionen.get(self.aktuelles_projekt, [])) + 1
timestamp = datetime.datetime.now().isoformat()
new_version = CodeVersion(version=version, timestamp=timestamp, content=content)
if self.aktuelles_projekt not in self.versionen:
self.versionen[self.aktuelles_projekt] = []
self.versionen[self.aktuelles_projekt].append(new_version)
print(f"Version {version} für Projekt {self.aktuelles_projekt} gespeichert")
def get_current_project_content(self):
# In einem echten Szenario würden wir hier den aktuellen Projektinhalt lesen
return "Dummy-Projektinhalt"
def versionen_vergleichen(self, version1: int, version2: int):
if self.aktuelles_projekt not in self.versionen:
print(f"Projekt {self.aktuelles_projekt} nicht gefunden")
return
versions = self.versionen[self.aktuelles_projekt]
v1 = next((v for v in versions if v.version == version1), None)
v2 = next((v for v in versions if v.version == version2), None)
if not v1 or not v2:
print("Eine oder beide Versionen nicht gefunden")
return
diff = difflib.unified_diff(
v1.content.splitlines(keepends=True),
v2.content.splitlines(keepends=True),
fromfile=f'Version {v1.version}',
tofile=f'Version {v2.version}'
)
print(''.join(diff))
def auf_version_zuruecksetzen(self, version: int):
if self.aktuelles_projekt not in self.versionen:
print(f"Projekt {self.aktuelles_projekt} nicht gefunden")
return
versions = self.versionen[self.aktuelles_projekt]
target_version = next((v for v in versions if v.version == version), None)
if not target_version:
print(f"Version {version} nicht gefunden")
return
# Hier würden wir den Projektinhalt auf den Stand der Zielversion zurücksetzen
print(f"Projekt {self.aktuelles_projekt} auf Version {version} zurückgesetzt")
def zustand_speichern(self):
state = {
'aufgaben': {k: asdict(v) for k, v in self.aufgaben.items()},
'versionen': {k: [asdict(v) for v in versions] for k, versions in self.versionen.items()},
'aktuelles_projekt': self.aktuelles_projekt
}
with open('task_management_state.json', 'w') as f:
json.dump(state, f)
def zustand_laden(self):
if os.path.exists('task_management_state.json'):
with open('task_management_state.json', 'r') as f:
state = json.load(f)
self.aufgaben = {k: Aufgabe(**v) for k, v in state['aufgaben'].items()}
self.versionen = {k: [CodeVersion(**v) for v in versions] for k, versions in state['versionen'].items()}
self.aktuelles_projekt = state['aktuelles_projekt']
class SelbstverbessernderPairProgrammer:
def __init__(self, api_schluessel: str):
self.api_schluessel = api_schluessel
self.gemini_url = 'https://generativelanguage.googleapis.com/v1beta/models/gemini-pro:generateContent'
self.headers = {
'Content-Type': 'application/json',
'x-goog-api-key': api_schluessel
}
self.projekt_pfad = os.getcwd()
self.projekt_struktur: Dict[str, Any] = {}
self.erstellte_dateien: List[str] = []
self.aufgaben_liste: List[Aufgabe] = []
self.task_management = TaskManagementSystem()
self.setup_logging()
def setup_logging(self):
logging.basicConfig(filename='projekt_log.txt', level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s')
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
logging.getLogger().addHandler(console_handler)
def konsolen_ausgabe(self, nachricht: str, farbe: str = 'standard', stil: str = 'normal'):
farben = {
'rot': (1, 0, 0),
'gruen': (0, 1, 0),
'blau': (0, 0, 1),
'gelb': (1, 1, 0),
'cyan': (0, 1, 1),
'magenta': (1, 0, 1),
'standard': (0.8, 0.8, 0.8)
}
stile = {
'normal': '',
'fett': '\033[1m',
'unterstrichen': '\033[4m',
'invers': '\033[7m'
}
console.set_color(*farben.get(farbe, farben['standard']))
print(f"{stile.get(stil, '')}{nachricht}{stile['normal']}")
console.set_color()
@retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=4, max=10))
async def gemini_api_aufrufen(self, prompt: str) -> Optional[Dict[str, Any]]:
try:
async with aiohttp.ClientSession() as session:
payload = {
'contents': [{'parts': [{'text': prompt}]}],
'generationConfig': {
'temperature': 0.7,
'topK': 40,
'topP': 0.95,
'maxOutputTokens': 1024,
}
}
async with session.post(self.gemini_url, headers=self.headers, json=payload, timeout=30) as response:
response.raise_for_status()
return await response.json()
except aiohttp.ClientError as e:
self.konsolen_ausgabe(f"API-Fehler: {e}", 'rot', 'fett')
logging.error(f"API-Fehler: {e}")
return None
def text_aus_antwort_extrahieren(self, response: Optional[Dict[str, Any]]) -> Optional[str]:
if response and 'candidates' in response:
for candidate in response['candidates']:
if 'content' in candidate:
for part in candidate['content'].get('parts', []):
if 'text' in part:
return part['text']
return None
async def projekt_anforderungen_analysieren(self, projekt_ziel: str) -> str:
self.konsolen_ausgabe("Analysiere Projektanforderungen...", 'cyan', 'fett')
prompt = f"""Analysiere detailliert die folgenden Projektanforderungen für: {projekt_ziel}
Berücksichtige dabei:
1. Genaues Projektziel
2. Zielplattform(en)
3. Hauptfunktionen und Features
4. Allgemeine Anforderungen (z.B. Benutzerfreundlichkeit, Leistung)
5. Technische Spezifikationen
6. Potenzielle Herausforderungen
7. Benötigte Ressourcen und Bibliotheken
8. Ähnliche existierende Projekte oder Best Practices
Gib eine strukturierte, detaillierte Analyse mit Beispielen und relevanten Informationen zurück."""
response = await self.gemini_api_aufrufen(prompt)
analysis = self.text_aus_antwort_extrahieren(response)
if analysis:
self.konsolen_ausgabe("Projektanalyse:", 'gruen')
print(analysis)
logging.info("Detaillierte Projektanalyse abgeschlossen")
return analysis
logging.warning("Keine detaillierte Analyse möglich")
return "Keine detaillierte Analyse möglich."
async def entwicklungsplan_erstellen(self, analyse: str) -> Dict[str, Any]:
self.konsolen_ausgabe("Erstelle detaillierten Entwicklungsplan...", 'cyan', 'fett')
prompt = f"""Basierend auf dieser Analyse, erstelle einen umfassenden Entwicklungsplan:
{analyse}
Der Plan sollte folgendes enthalten:
1. Hauptphasen des Projekts
2. Detaillierte Aufgaben für jede Phase
3. Geschätzter Zeitrahmen für jede Aufgabe
4. Abhängigkeiten zwischen Aufgaben
5. Benötigte Ressourcen für jede Aufgabe
6. Meilensteine und Überprüfungspunkte
7. Potenzielle Risiken und Gegenmaßnahmen
Gib den Plan als JSON-Objekt zurück mit den Schlüsseln:
"phasen", "aufgaben", "meilensteine", "risiken"."""
response = await self.gemini_api_aufrufen(prompt)
plan_text = self.text_aus_antwort_extrahieren(response)
if plan_text:
plan = json.loads(plan_text)
self.konsolen_ausgabe("Detaillierter Entwicklungsplan erstellt:", 'gruen')
self.plan_anzeigen(plan)
logging.info("Detaillierter Entwicklungsplan erstellt")
return plan
logging.warning("Kein detaillierter Plan erstellt")
return {}
def plan_anzeigen(self, plan: Dict[str, Any]):
console.clear()
console.set_font('Menlo', 14)
console.set_color(0, 0.5, 1) # Hellblau für Überschriften
print("=== Entwicklungsplan ===\n")
console.set_color(0, 0.8, 0) # Grün für Phasen
print("Phasen:")
for phase in plan.get('phasen', []):
print(f" - {phase}")
console.set_color(1, 0.8, 0) # Orange für Aufgaben
print("\nAufgaben:")
for aufgabe in plan.get('aufgaben', []):
print(f" - {aufgabe['beschreibung']} (Priorität: {aufgabe['prioritaet']})")
if 'abhaengigkeiten' in aufgabe:
print(f" Abhängigkeiten: {', '.join(aufgabe['abhaengigkeiten'])}")
console.set_color(0.5, 0, 0.5) # Lila für Meilensteine
print("\nMeilensteine:")
for meilenstein in plan.get('meilensteine', []):
print(f" - {meilenstein}")
console.set_color(1, 0, 0) # Rot für Risiken
print("\nRisiken:")
for risiko in plan.get('risiken', []):
print(f" - {risiko}")
console.set_color() # Zurücksetzen der Farbe
async def projekt_struktur_erstellen(self, plan: Dict[str, Any]):
self.konsolen_ausgabe("Erstelle spezifische Projektstruktur...", 'cyan', 'fett')
prompt = f"""Basierend auf diesem Entwicklungsplan, erstelle eine detaillierte und spezifische Datei- und Ordnerstruktur für das Projekt:
{json.dumps(plan, indent=2)}
Beachte dabei:
1. Erstelle eine main.py als Hauptdatei
2. Organisiere Module und Pakete sinnvoll
3. Berücksichtige Ressourcen-, Konfigurations- und Testdateien
4. Folge Best Practices für die Projektstrukturierung in Python
Gib die Struktur als JSON-Objekt zurück, mit Ordnern als verschachtelte Objekte und Dateien als Schlüssel mit leeren Strings."""
response = await self.gemini_api_aufrufen(prompt)
structure_text = self.text_aus_antwort_extrahieren(response)
if structure_text:
structure = self.json_extrahieren(structure_text)
if structure:
self.projekt_struktur = structure
self.konsolen_ausgabe("Spezifische Projektstruktur:", 'gruen')
print(json.dumps(structure, indent=2, ensure_ascii=False))
await self.struktur_implementieren(structure)
logging.info("Spezifische Projektstruktur erstellt und implementiert")
else:
self.konsolen_ausgabe("Konnte keine gültige JSON-Struktur extrahieren. Verwende Standard-Struktur.", 'gelb')
logging.warning("Keine gültige JSON-Struktur. Standard-Struktur verwendet")
await self.standard_struktur_verwenden()
else:
self.konsolen_ausgabe("Keine Struktur erstellt. Verwende Standard-Struktur.", 'gelb')
logging.warning("Keine Struktur erstellt. Standard-Struktur verwendet")
await self.standard_struktur_verwenden()
def json_extrahieren(self, text: str) -> Optional[Dict[str, Any]]:
try:
return json.loads(text)
except json.JSONDecodeError:
json_match = re.search(r'\{.*\}', text, re.DOTALL)
if json_match:
try:
return json.loads(json_match.group())
except json.JSONDecodeError:
pass
logging.error("JSON-Extraktion fehlgeschlagen")
return None
async def standard_struktur_verwenden(self):
self.projekt_struktur = {
"main.py": "",
"src": {
"__init__.py": "",
"core.py": "",
"utils.py": ""
},
"tests": {
"__init__.py": "",
"test_main.py": ""
},
"README.md": ""
}
self.konsolen_ausgabe("Standard-Projektstruktur:", 'gruen')
print(json.dumps(self.projekt_struktur, indent=2, ensure_ascii=False))
await self.struktur_implementieren(self.projekt_struktur)
logging.info("Standard-Projektstruktur implementiert")
async def struktur_implementieren(self, struktur: Dict[str, Any], eltern_pfad: str = ''):
for item, content in struktur.items():
path = os.path.join(eltern_pfad, item) if eltern_pfad else item
if isinstance(content, dict):
self.verzeichnis_sicherstellen(path)
await self.struktur_implementieren(content, path)
else:
await self.sicheres_dateischreiben(path, content if content else '')
self.erstellte_dateien.append(path)
logging.info(f"Datei erstellt: {path}")
def verzeichnis_sicherstellen(self, pfad: str):
if pfad and not os.path.exists(pfad):
os.makedirs(pfad)
logging.info(f"Verzeichnis erstellt: {pfad}")
async def sicheres_dateischreiben(self, datei_pfad: str, inhalt: str) -> bool:
try:
self.verzeichnis_sicherstellen(os.path.dirname(datei_pfad))
temp_pfad = datei_pfad + '.temp'
with open(temp_pfad, 'w', encoding='utf-8') as f:
f.write(inhalt)
if await self.datei_validieren(temp_pfad):
os.replace(temp_pfad, datei_pfad)
self.konsolen_ausgabe(f"Datei {datei_pfad} erfolgreich geschrieben und validiert.", 'gruen')
return True
else:
os.remove(temp_pfad)
self.konsolen_ausgabe(f"Fehler: Validierung fehlgeschlagen für {datei_pfad}", 'rot')
return False
except Exception as e:
self.konsolen_ausgabe(f"Fehler beim Schreiben der Datei {datei_pfad}: {e}", 'rot')
logging.error(f"Fehler beim Schreiben der Datei {datei_pfad}: {e}")
return False
async def datei_validieren(self, datei_pfad: str) -> bool:
_, erweiterung = os.path.splitext(datei_pfad)
if erweiterung == '.py':
return await self.python_datei_validieren(datei_pfad)
elif erweiterung == '.md':
return await self.markdown_datei_validieren(datei_pfad)
else:
return await self.allgemeine_datei_validieren(datei_pfad)
async def python_datei_validieren(self, datei_pfad: str) -> bool:
try:
with open(datei_pfad, 'r', encoding='utf-8') as file:
content = file.read()
ast.parse(content)
return True
except SyntaxError as e:
self.konsolen_ausgabe(f"Syntaxfehler in {datei_pfad}: {e}", 'rot')
logging.error(f"Syntaxfehler in {datei_pfad}: {e}")
return False
async def markdown_datei_validieren(self, datei_pfad: str) -> bool:
try:
with open(datei_pfad, 'r', encoding='utf-8') as file:
content = file.read()
if '```python' in content:
self.konsolen_ausgabe(f"Warnung: {datei_pfad} enthält Python-Code-Blöcke", 'gelb')
return True
except Exception as e:
self.konsolen_ausgabe(f"Fehler beim Lesen der Markdown-Datei {datei_pfad}: {e}", 'rot')
logging.error(f"Fehler beim Lesen der Markdown-Datei {datei_pfad}: {e}")
return False
async def allgemeine_datei_validieren(self, datei_pfad: str) -> bool:
try:
with open(datei_pfad, 'r', encoding='utf-8') as file:
file.read()
return True
except Exception as e:
self.konsolen_ausgabe(f"Fehler beim Lesen der Datei {datei_pfad}: {e}", 'rot')
logging.error(f"Fehler beim Lesen der Datei {datei_pfad}: {e}")
return False
async def aufgaben_extrahieren(self, plan: Dict[str, Any]) -> List[Aufgabe]:
aufgaben = []
for aufgabe_dict in plan.get('aufgaben', []):
aufgabe = Aufgabe(
id=f"task_{len(aufgaben) + 1}",
beschreibung=aufgabe_dict['beschreibung'],
prioritaet=aufgabe_dict['prioritaet'],
abhaengigkeiten=aufgabe_dict.get('abhaengigkeiten', [])
)
aufgaben.append(aufgabe)
self.task_management.aufgabe_hinzufuegen(aufgabe)
self.aufgaben_liste = aufgaben
self.konsolen_ausgabe(f"{len(aufgaben)} Aufgaben extrahiert:", 'gruen')
for aufgabe in aufgaben:
print(f"- [{aufgabe.prioritaet}] {aufgabe.beschreibung}")
logging.info(f"{len(aufgaben)} Aufgaben extrahiert")
return aufgaben
async def passende_datei_finden(self, aufgabe: Aufgabe) -> str:
prompt = f"""Für welche Datei in dieser Struktur ist die Aufgabe am besten geeignet?
Aufgabe: {aufgabe.beschreibung}
Struktur: {json.dumps(self.projekt_struktur, indent=2)}
Berücksichtige:
1. Die Funktion und den Zweck jeder Datei
2. Die logische Organisation des Projekts
3. Best Practices für Python-Projektstrukturen
Gib nur den Dateipfad zurück."""
response = await self.gemini_api_aufrufen(prompt)
datei_pfad = self.text_aus_antwort_extrahieren(response)
if datei_pfad and datei_pfad.strip():
datei_pfad = datei_pfad.strip()
self.konsolen_ausgabe(f"Passende Datei für Aufgabe gefunden: {datei_pfad}", 'gruen')
logging.info(f"Passende Datei für Aufgabe gefunden: {datei_pfad}")
return datei_pfad
self.konsolen_ausgabe(f"Keine passende Datei für Aufgabe gefunden: {aufgabe.beschreibung}", 'gelb')
logging.warning(f"Keine passende Datei für Aufgabe gefunden: {aufgabe.beschreibung}")
return "main.py" # Fallback auf eine Standarddatei
async def code_entwickeln(self, aufgabe: Aufgabe, datei_pfad: str) -> str:
self.konsolen_ausgabe(f"Entwickle Code für: {aufgabe.beschreibung}", 'cyan')
current_content = await self.datei_inhalt_lesen(datei_pfad)
prompt = f"""Entwickle Python-Code für: {aufgabe.beschreibung}
Datei: {datei_pfad}
Aktueller Inhalt:
{current_content}
Beachte:
1. Implementiere nur den Code für diese spezifische Aufgabe
2. Berücksichtige den bestehenden Code und füge den neuen nahtlos ein
3. Folge Python Best Practices und PEP 8
4. Kommentiere den Code angemessen
5. Behandle mögliche Fehler und Ausnahmen
6. Gib den Code in einem Block zwischen ''' Python und ''' zurück
Beispielformat:
''' Python
# Dein Code hier
'''"""
response = await self.gemini_api_aufrufen(prompt)
code = self.text_aus_antwort_extrahieren(response)
if code:
self.konsolen_ausgabe("Generierter Code:", 'gruen')
print(code)
logging.info(f"Code für {aufgabe.beschreibung} entwickelt")
return code
logging.warning(f"Kein Code für {aufgabe.beschreibung} generiert")
return "''' Python\n# Kein Code generiert.\n'''"
async def code_implementieren(self, datei_pfad: str, code: str):
try:
code_match = re.search(r"'''(?:\s*Python)?\s*(.*?)'''", code, re.DOTALL | re.IGNORECASE)
if code_match:
new_code = code_match.group(1).strip()
else:
new_code = code.strip()
current_content = await self.datei_inhalt_lesen(datei_pfad)
if new_code in current_content:
self.konsolen_ausgabe(f"Code in {datei_pfad} bereits vorhanden.", 'gelb')
return
insert_position = self.finde_einfuegeposition(current_content)
updated_content = current_content[:insert_position] + "\n\n" + new_code + current_content[insert_position:]
if await self.sicheres_dateischreiben(datei_pfad, updated_content):
self.konsolen_ausgabe(f"Code in {datei_pfad} implementiert.", 'gruen')
logging.info(f"Code in {datei_pfad} implementiert")
self.task_management.version_speichern()
else:
self.konsolen_ausgabe(f"Fehler beim Implementieren des Codes in {datei_pfad}", 'rot')
logging.error(f"Fehler beim Implementieren des Codes in {datei_pfad}")
except Exception as e:
self.konsolen_ausgabe(f"Unerwarteter Fehler beim Implementieren: {e}", 'rot')
logging.error(f"Unerwarteter Fehler beim Implementieren von {datei_pfad}: {e}")
await self.fehler_beheben(datei_pfad, str(e))
def finde_einfuegeposition(self, content: str) -> int:
try:
tree = ast.parse(content)
last_node = max(tree.body, key=lambda node: node.end_lineno) if tree.body else None
return last_node.end_lineno if last_node else len(content.split('\n'))
except SyntaxError:
return len(content.split('\n'))
async def datei_inhalt_lesen(self, datei_pfad: str) -> str:
if os.path.exists(datei_pfad):
with open(datei_pfad, 'r', encoding='utf-8') as f:
return f.read()
return ""
async def code_reparieren(self, datei_pfad: str, fehlermeldung: str):
self.konsolen_ausgabe(f"Versuche, Code in {datei_pfad} zu reparieren...", 'cyan')
current_code = await self.datei_inhalt_lesen(datei_pfad)
prompt = f"""Repariere den folgenden Python-Code, der diesen Fehler verursacht:
Fehler: {fehlermeldung}
Code:
{current_code}
Gib nur den reparierten Code zurück, ohne zusätzliche Erklärungen."""
response = await self.gemini_api_aufrufen(prompt)
repaired_code = self.text_aus_antwort_extrahieren(response)
if repaired_code:
if await self.sicheres_dateischreiben(datei_pfad, repaired_code):
self.konsolen_ausgabe(f"Code in {datei_pfad} repariert.", 'gruen')
logging.info(f"Code in {datei_pfad} repariert")
self.task_management.version_speichern()
else:
self.konsolen_ausgabe(f"Reparierter Code enthält immer noch Fehler.", 'rot')
logging.error(f"Reparierter Code für {datei_pfad} enthält Fehler")
else:
self.konsolen_ausgabe(f"Konnte Code in {datei_pfad} nicht reparieren.", 'rot')
logging.error(f"Code-Reparatur für {datei_pfad} fehlgeschlagen")
async def code_refaktorisieren(self, datei_pfad: str):
self.konsolen_ausgabe(f"Refaktorisiere Code in: {datei_pfad}", 'cyan')
current_code = await self.datei_inhalt_lesen(datei_pfad)
prompt = f"""Refaktorisiere diesen Python-Code:
{current_code}
Beachte dabei:
1. Verbessere die Lesbarkeit und Effizienz
2. Folge Python Best Practices und PEP 8
3. Optimiere die Struktur und Organisation
4. Füge oder verbessere Kommentare wo nötig
5. Behandle mögliche Fehler und Ausnahmen besser
Gib nur den verbesserten Code zurück, ohne zusätzliche Erklärungen."""
response = await self.gemini_api_aufrufen(prompt)
refactored_code = self.text_aus_antwort_extrahieren(response)
if refactored_code:
if await self.sicheres_dateischreiben(datei_pfad, refactored_code):
self.konsolen_ausgabe(f"Code in {datei_pfad} refaktorisiert.", 'gruen')
logging.info(f"Code in {datei_pfad} refaktorisiert")
self.task_management.version_speichern()
else:
self.konsolen_ausgabe(f"Fehler beim Refaktorisieren von {datei_pfad}", 'rot')
logging.error(f"Fehler beim Refaktorisieren von {datei_pfad}")
else:
self.konsolen_ausgabe("Refaktorisierung fehlgeschlagen.", 'gelb')
logging.warning(f"Refaktorisierung für {datei_pfad} fehlgeschlagen")
async def projektstruktur_validieren(self):
self.konsolen_ausgabe("Validiere Projektstruktur...", 'cyan', 'fett')
for root, dirs, files in os.walk(self.projekt_pfad):
for file in files:
datei_pfad = os.path.join(root, file)
if not await self.datei_validieren(datei_pfad):
self.konsolen_ausgabe(f"Fehlerhafte Datei gefunden: {datei_pfad}", 'rot')
await self.fehlerhafte_datei_behandeln(datei_pfad)
async def fehlerhafte_datei_behandeln(self, datei_pfad: str):
self.konsolen_ausgabe(f"Versuche, fehlerhafte Datei zu reparieren: {datei_pfad}", 'gelb')
_, erweiterung = os.path.splitext(datei_pfad)
if erweiterung == '.py':
await self.code_reparieren(datei_pfad, "Syntaxfehler")
elif erweiterung == '.md':
await self.markdown_datei_bereinigen(datei_pfad)
else:
self.konsolen_ausgabe(f"Kann Datei nicht automatisch reparieren: {datei_pfad}", 'rot')
async def markdown_datei_bereinigen(self, datei_pfad: str):
try:
with open(datei_pfad, 'r', encoding='utf-8') as file:
content = file.read()
bereinigter_inhalt = re.sub(r'```python.*?```', '', content, flags=re.DOTALL)
bereinigter_inhalt = re.sub(r'[^\x00-\x7F]+', '', bereinigter_inhalt)
if await self.sicheres_dateischreiben(datei_pfad, bereinigter_inhalt):
self.konsolen_ausgabe(f"Markdown-Datei bereinigt: {datei_pfad}", 'gruen')
else:
self.konsolen_ausgabe(f"Fehler beim Bereinigen der Markdown-Datei: {datei_pfad}", 'rot')
except Exception as e:
self.konsolen_ausgabe(f"Fehler beim Bereinigen der Markdown-Datei {datei_pfad}: {e}", 'rot')
logging.error(f"Fehler beim Bereinigen der Markdown-Datei {datei_pfad}: {e}")
async def module_kompatibilitaet_pruefen(self):
self.konsolen_ausgabe("Prüfe die Kompatibilität aller Module...", 'cyan', 'fett')
fehlerhafte_module = []
for datei_pfad in self.erstellte_dateien:
if datei_pfad.endswith('.py'):
modulname = os.path.splitext(os.path.basename(datei_pfad))[0]
try:
if modulname in sys.modules:
importlib.reload(sys.modules[modulname])
else:
importlib.import_module(modulname)
self.konsolen_ausgabe(f"Modul {modulname} erfolgreich geladen.", 'gruen')
logging.info(f"Modul {modulname} erfolgreich geladen.")
except Exception as e:
fehlerhafte_module.append((modulname, str(e)))
self.konsolen_ausgabe(f"Fehler beim Laden des Moduls {modulname}: {e}", 'rot')
logging.error(f"Fehler beim Laden des Moduls {modulname}: {e}")
if not fehlerhafte_module:
self.konsolen_ausgabe("Alle Module sind kompatibel.", 'gruen', 'fett')
logging.info("Alle Module sind kompatibel.")
else:
self.konsolen_ausgabe(f"Fehlerhafte Module: {len(fehlerhafte_module)}", 'rot', 'fett')
for modulname, fehler in fehlerhafte_module:
self.konsolen_ausgabe(f"- {modulname}: {fehler}", 'rot')
await self.code_reparieren(f"{modulname}.py", fehler)
logging.warning(f"{len(fehlerhafte_module)} fehlerhafte Module gefunden.")
async def unit_tests_erstellen_und_ausfuehren(self):
self.konsolen_ausgabe("Erstelle und führe Unit-Tests aus...", 'cyan', 'fett')
for datei_pfad in self.erstellte_dateien:
if datei_pfad.endswith('.py') and not datei_pfad.endswith('test.py'):
test_datei = datei_pfad.replace('.py', '_test.py')
modulname = os.path.splitext(os.path.basename(datei_pfad))[0]
# Generiere Unit-Tests mit Gemini
prompt = f"Erstelle Unit-Tests für das folgende Python-Modul:\n{await self.datei_inhalt_lesen(datei_pfad)}"
response = await self.gemini_api_aufrufen(prompt)
test_code = self.text_aus_antwort_extrahieren(response)
if test_code:
if await self.sicheres_dateischreiben(test_datei, test_code):
self.konsolen_ausgabe(f"Unit-Tests für {modulname} erstellt.", 'gruen')
# Führe Tests aus
try:
import unittest
test_suite = unittest.defaultTestLoader.discover(os.path.dirname(test_datei), pattern=f'{modulname}_test.py')
test_runner = unittest.TextTestRunner()
result = test_runner.run(test_suite)
if result.wasSuccessful():
self.konsolen_ausgabe(f"Alle Tests für {modulname} erfolgreich.", 'gruen')
else:
self.konsolen_ausgabe(f"Einige Tests für {modulname} fehlgeschlagen.", 'rot')
logging.warning(f"Tests für {modulname} fehlgeschlagen")
except Exception as e:
self.konsolen_ausgabe(f"Fehler beim Ausführen der Tests für {modulname}: {e}", 'rot')
logging.error(f"Fehler bei Unit-Tests für {modulname}: {e}")
else:
self.konsolen_ausgabe(f"Fehler beim Erstellen der Unit-Tests für {modulname}", 'rot')
else:
self.konsolen_ausgabe(f"Konnte keine Unit-Tests für {modulname} generieren.", 'gelb')
logging.warning(f"Keine Unit-Tests für {modulname} generiert")
async def abschlusspruefung(self):
self.konsolen_ausgabe("Führe finale Überprüfung durch...", 'cyan', 'fett')
for datei_pfad in self.erstellte_dateien:
if datei_pfad.endswith('.py'):
await self.datei_validieren(datei_pfad)
await self.code_refaktorisieren(datei_pfad)
await self.projektstruktur_validieren()
await self.module_kompatibilitaet_pruefen()
await self.unit_tests_erstellen_und_ausfuehren()
self.konsolen_ausgabe("Finale Überprüfung abgeschlossen.", 'gruen')
logging.info("Abschlussprüfung durchgeführt")
async def fortschritt_anzeigen(self):
self.task_management.aufgaben_anzeigen()
async def benutzer_feedback_einholen(self, nachricht: str) -> str:
return console.input(nachricht + " ")
async def aufgabe_bearbeiten(self, aufgabe: Aufgabe, datei_pfad: str):
aufgabe.status = 'in_bearbeitung'
self.task_management.aufgabe_aktualisieren(aufgabe.id, status='in_bearbeitung')
self.konsolen_ausgabe(f"Bearbeite Aufgabe: {aufgabe.beschreibung}", 'cyan')
code = await self.code_entwickeln(aufgabe, datei_pfad)
await self.code_implementieren(datei_pfad, code)
aufgabe.fortschritt = 100
aufgabe.status = 'abgeschlossen'
self.task_management.aufgabe_aktualisieren(aufgabe.id, fortschritt=100, status='abgeschlossen')
await self.fortschritt_anzeigen()
await asyncio.sleep(1) # Kurze Pause für bessere Lesbarkeit der Ausgabe
async def ausfuehren(self, projekt_ziel: str):
self.task_management.aktuelles_projekt = projekt_ziel
self.task_management.zustand_laden()
console.clear()
self.konsolen_ausgabe("Self-Improving Pair Programmer gestartet.", 'blau', 'fett')
logging.info(f"Projekt gestartet: {projekt_ziel}")
try:
analyse = await self.projekt_anforderungen_analysieren(projekt_ziel)
plan = await self.entwicklungsplan_erstellen(analyse)
await self.projekt_struktur_erstellen(plan)
aufgaben = await self.aufgaben_extrahieren(plan)
while aufgaben:
await self.fortschritt_anzeigen()
verfuegbare_aufgaben = [a for a in aufgaben if not a.abhaengigkeiten]
if not verfuegbare_aufgaben:
self.konsolen_ausgabe("Keine verfügbaren Aufgaben. Überprüfe Abhängigkeiten.", 'gelb')
break
aktuelle_aufgabe = min(verfuegbare_aufgaben, key=lambda a: a.prioritaet)
datei_pfad = await self.passende_datei_finden(aktuelle_aufgabe)
if datei_pfad:
await self.aufgabe_bearbeiten(aktuelle_aufgabe, datei_pfad)
aufgaben.remove(aktuelle_aufgabe)
# Aktualisiere Abhängigkeiten
for aufgabe in aufgaben:
if aktuelle_aufgabe.id in aufgabe.abhaengigkeiten:
aufgabe.abhaengigkeiten.remove(aktuelle_aufgabe.id)
else:
self.konsolen_ausgabe(f"Keine passende Datei für Aufgabe gefunden: {aktuelle_aufgabe.beschreibung}", 'gelb')
aufgaben.remove(aktuelle_aufgabe)
self.task_management.zustand_speichern()
feedback = await self.benutzer_feedback_einholen("Möchten Sie fortfahren? (j/n):")
if feedback.lower() != 'j':
break
await self.abschlusspruefung()
self.konsolen_ausgabe("Projekt erfolgreich abgeschlossen!", 'gruen', 'fett')
logging.info("Projekt erfolgreich abgeschlossen")
except Exception as e:
self.konsolen_ausgabe(f"Ein kritischer Fehler ist aufgetreten: {e}", 'rot', 'fett')
logging.error(f"Kritischer Fehler während der Projektausführung: {e}", exc_info=True)
await self.fehler_beheben("main.py", str(e))
finally:
self.task_management.zustand_speichern()
async def fehler_beheben(self, datei_pfad: str, fehlermeldung: str):
self.konsolen_ausgabe(f"Versuche, Fehler in {datei_pfad} zu beheben...", 'cyan')
await self.code_reparieren(datei_pfad, fehlermeldung)
if not await self.datei_validieren(datei_pfad):
self.konsolen_ausgabe(f"Konnte Fehler in {datei_pfad} nicht beheben. Manuelle Überprüfung erforderlich.", 'rot')
logging.error(f"Fehler in {datei_pfad} konnte nicht behoben werden")
# Verwendung
if __name__ == "__main__":
api_schluessel = 'AIzaSyDUaG1ti_VHM89oPD7hLmeWzb-nmcP8gmI'
programmer = SelbstverbessernderPairProgrammer(api_schluessel)
# Erstellen und Ausführen des Event Loops
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(programmer.ausfuehren("Entwickle eine moderne foto app für pythonista 3 auf dem iphone 13"))
finally:
loop.close()
ASSISTANT
Ihr Python-Skript für ein selbstverbesserndes Pair-Programming-System ist beeindruckend umfangreich und enthält viele fortschrittliche Funktionen zur Automatisierung der Softwareentwicklung. Im Folgenden finden Sie eine detaillierte Analyse Ihres Codes mit Hinweisen zu Stärken, potenziellen Verbesserungen und Sicherheitsaspekten.
## 1. **Übersicht und Funktionalität**
Das Skript implementiert ein komplexes Task-Management-System, das in der Lage ist:
- Projektanforderungen zu analysieren.
- Einen Entwicklungsplan zu erstellen.
- Eine detaillierte Datei- und Ordnerstruktur zu generieren.
- Aufgaben zu extrahieren und zu bearbeiten.
- Code zu entwickeln, zu implementieren, zu reparieren und zu refaktorisieren.
- Projektversionen zu verwalten und zu vergleichen.
- Unit-Tests zu erstellen und auszuführen.
- Abschlussprüfungen durchzuführen.
Dies ermöglicht eine weitgehend automatisierte Softwareentwicklung mit Unterstützung durch die Gemini API von Google.
## 2. **Stärken des Codes**
- **Modularität und Struktur:** Der Einsatz von `@dataclass` für `CodeVersion` und `Aufgabe` sowie die klar getrennten Klassen `TaskManagementSystem` und `SelbstverbessernderPairProgrammer` fördern eine gute Code-Organisation und Wiederverwendbarkeit.
- **Asynchrone Programmierung:** Die Nutzung von `asyncio` und `aiohttp` ermöglicht effiziente Netzwerkoperationen und verbessert die Reaktionsfähigkeit des Systems.
- **Fehlerbehandlung und Logging:** Umfangreiche Fehlerbehandlung mit konsolennotizen und einem robusten Logging-System erleichtert das Debugging und die Wartung.
- **Versionskontrolle:** Die Implementierung eines einfachen Versionskontrollsystems erlaubt die Nachverfolgung und Rücksetzung von Projektversionen.
- **Interaktive Konsole:** Die Nutzung des `console`-Moduls zur farbigen und formatieren Ausgabe verbessert die Benutzererfahrung.
## 3. **Sicherheitsbedenken**
### **API-Schlüssel im Klartext**
Der API-Schlüssel ist direkt im Code fest codiert:
```python
api_schluessel = 'AIzaSyDUaG1ti_VHM89oPD7hLmeWzb-nmcP8gmI'
```
**Empfehlungen:**
- **Umgebung Variablen:** Speichern Sie sensible Informationen wie API-Schlüssel in Umgebungsvariablen und laden Sie diese zur Laufzeit. Dies verhindert, dass Schlüssel versehentlich in Versionskontrollsysteme gelangen.
```python
import os
api_schluessel = os.getenv('GEMINI_API_SCHLUESSEL')
```
- **.env-Datei:** Nutzen Sie eine `.env`-Datei in Kombination mit dem `python-dotenv`-Paket, um Umgebungsvariablen zu verwalten.
```python
from dotenv import load_dotenv
load_dotenv()
api_schluessel = os.getenv('GEMINI_API_SCHLUESSEL')
```
- **Zugriffsrechte:** Stellen Sie sicher, dass die Datei mit den Umgebungsvariablen nicht öffentlich zugänglich ist und in `.gitignore` aufgeführt wird.
### **Fehlerhafte Eingabedaten**
Beim Laden und Speichern von Zuständen sowie beim Verarbeiten von JSON-Daten sollten Sie immer auf die Validität der Eingaben achten, um Sicherheitslücken wie Code-Injektionen zu vermeiden.
**Empfehlung:**
- **Strikte Validierung:** Validieren und sanitizen Sie alle externen Datenquellen bevor Sie sie verwenden oder speichern.
## 4. **Abhängigkeiten und Module**
### **Nicht standardmäßige Module**
Sie verwenden das `console`-Modul, das nicht zu den Standardbibliotheken von Python gehört. Stellen Sie sicher, dass es installiert ist und korrekt funktioniert.
**Empfehlung:**
- **Dokumentation:** Dokumentieren Sie alle benötigten Abhängigkeiten in einer `requirements.txt`-Datei, um die Installation zu erleichtern.
```plaintext
aiohttp
asyncio
tenacity
dataclasses
etc.
```
- **Virtual Environment:** Nutzen Sie virtuelle Umgebungen (`venv` oder `conda`), um Abhängigkeiten isoliert zu verwalten.
## 5. **Potenzielle Verbesserungen und Optimierungen**
### **Asynchrone Dateiverarbeitung**
Der aktuelle Code verwendet synchrone Dateizugriffe, was die Vorteile der Asynchronität teilweise beeinträchtigt.
**Empfehlung:**
- **Async File I/O:** Nutzen Sie asynchrone Bibliotheken wie `aiofiles` für Dateioperationen, um die Effizienz weiter zu steigern.
```python
import aiofiles
async with aiofiles.open(datei_pfad, 'r', encoding='utf-8') as f:
content = await f.read()
```
### **Async Event Loop Handling**
Im Hauptblock verwenden Sie `asyncio.get_event_loop()`, was in neueren Python-Versionen durch `asyncio.run()` ersetzt werden kann.
**Empfehlung:**
- **Modernes Event Loop Management:**
```python
if __name__ == "__main__":
api_schluessel = os.getenv('GEMINI_API_SCHLUESSEL')
programmer = SelbstverbessernderPairProgrammer(api_schluessel)
asyncio.run(programmer.ausfuehren("Entwickle eine moderne Foto-App für Pythonista 3 auf dem iPhone 13"))
```
### **Fehlerbehebung und Wiederherstellung**
Die Fehlerbehebungsfunktionen sind gut implementiert, könnten aber durch automatisierte Tests unterstützt werden, um die Zuverlässigkeit zu erhöhen.
### **Code-Duplikation vermeiden**
Einige Teile des Codes, wie das Extrahieren von Text aus API-Antworten, könnten in Hilfsfunktionen ausgelagert werden, um die Wiederverwendbarkeit zu erhöhen.
**Beispiel:**
```python
def extract_text(response: Optional[Dict[str, Any]]) -> Optional[str]:
if response and 'candidates' in response:
for candidate in response['candidates']:
if 'content' in candidate:
for part in candidate['content'].get('parts', []):
if 'text' in part:
return part['text']
return None
```
### **Logging-Verbesserungen**
Der aktuelle Logging-Mechanismus ist funktional, könnte jedoch durch unterschiedliche Log-Level zu verschiedenen Zwecken erweitert werden.
**Empfehlung:**
- **Log Rotation:** Implementieren Sie eine Log-Rotation, um das Log-File vor übermäßigem Wachstum zu schützen.
```python
from logging.handlers import RotatingFileHandler
handler = RotatingFileHandler('projekt_log.txt', maxBytes=1048576, backupCount=5)
logging.getLogger().addHandler(handler)
```
### **Tests und Qualitätssicherung**
Obwohl das Skript Unit-Tests erstellt und ausführt, könnten zusätzliche Tests direkt im Entwicklungsprozess integriert werden, um die Codequalität kontinuierlich zu überprüfen.
## 6. **Dokumentation und Benutzerführung**
### **Inline-Kommentare**
Obwohl der Code größtenteils selbstdokumentierend ist, könnten zusätzliche Kommentare, insbesondere bei komplexen Logikabschnitten, die Verständlichkeit erhöhen.
### **README und Anleitungen**
Stellen Sie sicher, dass ein ausführliches `README.md` vorhanden ist, das den Zweck des Projekts, Installationsanweisungen, Abhängigkeiten und grundlegende Nutzungsbeispiele beschreibt.
## 7. **Sonstige Empfehlungen**
### **Internationalisierung**
Da der Code und die Konsolenausgaben auf Deutsch sind, stellen Sie sicher, dass alle Benutzer dieselbe Sprache verwenden oder bieten Sie eine Möglichkeit zur Sprachwahl.
### **Benutzerinteraktion**
Die Funktion `benutzer_feedback_einholen` nutzt `console.input`, was möglicherweise auf Plattformen oder in bestimmten Umgebungen nicht funktioniert.
**Empfehlung:**
- **Plattformkompatibilität sicherstellen:** Testen Sie die Benutzerinteraktion in verschiedenen Umgebungen (z.B. Windows, macOS, Linux) und passen Sie sie bei Bedarf an.
### **Performance-Optimierungen**
Bei großen Projekten oder vielen Aufgaben könnte die Performance beeinträchtigt werden. Überwachen Sie die Ressourcen und optimieren Sie gegebenenfalls die Verarbeitungsschritte.
## 8. **Schlussbemerkung**
Ihr Skript ist ein hoch entwickeltes Beispiel für die Automatisierung von Softwareentwicklungsprozessen unter Verwendung moderner Technologien und Methoden. Mit den oben genannten Verbesserungen können Sie die Sicherheit, Effizienz und Benutzerfreundlichkeit Ihres Systems weiter steigern. Achten Sie besonders auf die sicheren Umgangs mit sensiblen Daten wie API-Schlüsseln und erleichtern Sie die Wartbarkeit durch klare Strukturierung und Dokumentation.
Falls Sie spezifische Fragen zu bestimmten Teilen des Codes haben oder weitere Unterstützung benötigen, stehe ich Ihnen gerne zur Verfügung!