Floke/Brancheneinstufung2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(timetracking): Complete and verify time tracking implementation [2f488f42]

Browse Source 
Implemented a full time tracking feature. The system now displays the previously recorded time in hh:mm format when a session starts. When a work unit is completed, the invested time is automatically calculated, added to the total in Notion, and included in the status report. Various bugs were fixed during this process.
This commit is contained in:
Floke
2026-01-27 08:24:44 +00:00
parent 5f391a2caa
commit fff89ff012
1 changed files with 126 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

187
dev_session.py
View File

@@ -5,6 +5,8 @@ import re
from typing import List, Dict, Optional, Tuple from typing import List, Dict, Optional, Tuple
from getpass import getpass from getpass import getpass
from dotenv import load_dotenv from dotenv import load_dotenv
import argparse
import shutil
load_dotenv() load_dotenv()
@@ -43,10 +45,11 @@ def find_database_by_title(token: str, title: str) -> Optional[str]:
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
print(f"Fehler bei der Suche nach der Notion-Datenbank '{title}': {e}") print(f"Fehler bei der Suche nach der Notion-Datenbank '{title}': {e}")
try: if e.response is not None:
print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}") try:
except: print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}")
print(f"Antwort des Servers: {e.response.text}") except:
print(f"Antwort des Servers: {e.response.text}")
return None return None
def query_notion_database(token: str, database_id: str, filter_payload: Dict = None) -> List[Dict]: def query_notion_database(token: str, database_id: str, filter_payload: Dict = None) -> List[Dict]:
@@ -68,10 +71,11 @@ def query_notion_database(token: str, database_id: str, filter_payload: Dict = N
return response.json().get("results", []) return response.json().get("results", [])
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
print(f"Fehler bei der Abfrage der Notion-Datenbank {database_id}: {e}") print(f"Fehler bei der Abfrage der Notion-Datenbank {database_id}: {e}")
try: if e.response is not None:
print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}") try:
except: print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}")
print(f"Antwort des Servers: {e.response.text}") except:
print(f"Antwort des Servers: {e.response.text}")
return [] return []
def get_page_title(page: Dict) -> str: def get_page_title(page: Dict) -> str:
@@ -97,6 +101,22 @@ def get_page_property(page: Dict, prop_name: str, prop_type: str = "rich_text")
# Hier könnten weitere Typen wie 'select', 'number' etc. behandelt werden # Hier könnten weitere Typen wie 'select', 'number' etc. behandelt werden
return None return None
def get_page_number_property(page: Dict, prop_name: str) -> Optional[float]:
"""Extrahiert den Wert einer 'number'-Eigenschaft von einer Seite."""
prop = page.get("properties", {}).get(prop_name)
if not prop or prop.get("type") != "number":
return None
return prop.get("number")
def decimal_hours_to_hhmm(decimal_hours: float) -> str:
"""Wandelt Dezimalstunden in das Format 'HH:MM' um."""
if decimal_hours is None:
return "00:00"
hours = int(decimal_hours)
minutes = int((decimal_hours * 60) % 60)
return f"{hours:02d}:{minutes:02d}"
def get_page_content(token: str, page_id: str) -> str: def get_page_content(token: str, page_id: str) -> str:
"""Ruft den gesamten Textinhalt einer Notion-Seite ab, indem es die Blöcke zusammenfügt, mit Paginierung.""" """Ruft den gesamten Textinhalt einer Notion-Seite ab, indem es die Blöcke zusammenfügt, mit Paginierung."""
url = f"https://api.notion.com/v1/blocks/{page_id}/children" url = f"https://api.notion.com/v1/blocks/{page_id}/children"
@@ -165,12 +185,13 @@ def get_page_content(token: str, page_id: str) -> str:
return "\n".join(full_text) return "\n".join(full_text)
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
print(f"Fehler beim Abrufen des Seiteninhalts für Page-ID {page_id}: {e}") print(f"Fehler beim Abrufen des Seiteninhalts für Page-ID {page_id}: {e}")
try: if e.response is not None:
if e.response: # Wenn eine Antwort vorhanden ist try:
error_details = e.response.json() if e.response: # Wenn eine Antwort vorhanden ist
print(f"Notion API Fehlerdetails: {json.dumps(error_details, indent=2)}") error_details = e.response.json()
except json.JSONDecodeError: print(f"Notion API Fehlerdetails: {json.dumps(error_details, indent=2)}")
print(f"Notion API Rohantwort: {e.response.text}") except json.JSONDecodeError:
print(f"Notion API Rohantwort: {e.response.text}")
return "" return ""
@@ -193,36 +214,30 @@ def get_database_status_options(token: str, db_id: str) -> List[str]:
print(f"Fehler beim Abrufen der Datenbank-Eigenschaften: {e}") print(f"Fehler beim Abrufen der Datenbank-Eigenschaften: {e}")
return [] return []
def update_notion_task_status(token: str, task_id: str, status_value: str = "Doing") -> bool: def update_notion_task_property(token: str, task_id: str, payload: Dict) -> bool:
"""Aktualisiert den Status eines Notion-Tasks.""" """Aktualisiert eine beliebige Eigenschaft eines Notion-Tasks."""
print(f"\n--- Aktualisiere Status von Task '{task_id}' auf '{status_value}'... ---") print(f"\n--- Aktualisiere Eigenschaft von Task '{task_id}'... ---")
url = f"https://api.notion.com/v1/pages/{task_id}" url = f"https://api.notion.com/v1/pages/{task_id}"
headers = { headers = {
"Authorization": f"Bearer {token}", "Authorization": f"Bearer {token}",
"Content-Type": "application/json", "Content-Type": "application/json",
"Notion-Version": "2022-06-28" "Notion-Version": "2022-06-28"
} }
payload = {
"properties": { update_payload = {"properties": payload}
"Status": {
"status": {
"name": status_value
}
}
}
}
try: try:
response = requests.patch(url, headers=headers, json=payload) response = requests.patch(url, headers=headers, json=update_payload)
response.raise_for_status() response.raise_for_status()
print(f"✅ Task-Status erfolgreich auf '{status_value}' aktualisiert.") print(f"✅ Task-Eigenschaft erfolgreich aktualisiert.")
return True return True
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
print(f"❌ FEHLER beim Aktualisieren des Task-Status: {e}") print(f"❌ FEHLER beim Aktualisieren der Task-Eigenschaft: {e}")
try: if e.response is not None:
print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}") try:
except: print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}")
print(f"Antwort des Servers: {e.response.text}") except:
print(f"Antwort des Servers: {e.response.text}")
return False return False
def create_new_notion_task(token: str, project_id: str, tasks_db_id: str) -> Optional[Dict]: def create_new_notion_task(token: str, project_id: str, tasks_db_id: str) -> Optional[Dict]:
@@ -269,10 +284,11 @@ def create_new_notion_task(token: str, project_id: str, tasks_db_id: str) -> Opt
return new_task return new_task
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
print(f"❌ FEHLER beim Erstellen des Tasks: {e}") print(f"❌ FEHLER beim Erstellen des Tasks: {e}")
try: if e.response is not None:
print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}") try:
except: print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}")
print(f"Antwort des Servers: {e.response.text}") except:
print(f"Antwort des Servers: {e.response.text}")
return None return None
def add_comment_to_notion_task(token: str, task_id: str, comment: str) -> bool: def add_comment_to_notion_task(token: str, task_id: str, comment: str) -> bool:
@@ -298,10 +314,11 @@ def add_comment_to_notion_task(token: str, task_id: str, comment: str) -> bool:
return True return True
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
print(f"❌ FEHLER beim Hinzufügen des Kommentars zum Notion-Task: {e}") print(f"❌ FEHLER beim Hinzufügen des Kommentars zum Notion-Task: {e}")
try: if e.response is not None:
print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}") try:
except: print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}")
print(f"Antwort des Servers: {e.response.text}") except:
print(f"Antwort des Servers: {e.response.text}")
return False return False
def append_blocks_to_notion_page(token: str, page_id: str, blocks: List[Dict]) -> bool: def append_blocks_to_notion_page(token: str, page_id: str, blocks: List[Dict]) -> bool:
@@ -320,10 +337,11 @@ def append_blocks_to_notion_page(token: str, page_id: str, blocks: List[Dict]) -
return True return True
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
print(f"❌ FEHLER beim Anhängen des Statusberichts an die Notion-Seite: {e}") print(f"❌ FEHLER beim Anhängen des Statusberichts an die Notion-Seite: {e}")
try: if e.response is not None:
print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}") try:
except: print(f"Antwort des Servers: {json.dumps(e.response.json(), indent=2)}")
print(f"Antwort des Servers: {e.response.text}") except:
print(f"Antwort des Servers: {e.response.text}")
return False return False
# --- Session Management --- # --- Session Management ---
@@ -496,6 +514,44 @@ def report_status_to_notion(
print("❌ FEHLER: Session-Daten unvollständig. Kann keinen Statusbericht erstellen.") print("❌ FEHLER: Session-Daten unvollständig. Kann keinen Statusbericht erstellen.")
return return
# Time tracking logic
session_start_time_str = session_data.get("session_start_time")
if session_start_time_str:
session_start_time = datetime.fromisoformat(session_start_time_str)
elapsed_time = datetime.now() - session_start_time
elapsed_hours = elapsed_time.total_seconds() / 3600
# Get current task page to read existing duration
# Note: This is a simplified way. A more robust solution might query the DB
# to get the page object without a separate API call if we already have it.
# For now, a direct API call is clear and ensures we have the latest data.
task_page_url = f"https://api.notion.com/v1/pages/{task_id}"
headers = {
"Authorization": f"Bearer {token}",
"Notion-Version": "2022-06-28"
}
try:
page_response = requests.get(task_page_url, headers=headers)
page_response.raise_for_status()
task_page = page_response.json()
current_duration = get_page_number_property(task_page, "Total Duration (h)") or 0.0
new_total_duration = current_duration + elapsed_hours
duration_payload = {
"Total Duration (h)": {
"number": new_total_duration
}
}
update_notion_task_property(token, task_id, duration_payload)
print(f"✅ Zeiterfassung: {elapsed_hours:.2f} Stunden zum Task hinzugefügt. Neue Gesamtdauer: {new_total_duration:.2f} Stunden.")
# Reset session start time for the next interval
save_session_info(task_id, token)
except requests.exceptions.RequestException as e:
print(f"❌ FEHLER beim Abrufen der Task-Details für die Zeiterfassung: {e}")
print(f"--- Erstelle Statusbericht für Task {task_id} ---") print(f"--- Erstelle Statusbericht für Task {task_id} ---")
# Git-Zusammenfassung generieren (immer, wenn nicht explizit überschrieben) # Git-Zusammenfassung generieren (immer, wenn nicht explizit überschrieben)
@@ -558,6 +614,12 @@ def report_status_to_notion(
# Kommentar zusammenstellen # Kommentar zusammenstellen
report_lines = [] report_lines = []
# Diese Zeilen werden jetzt innerhalb des Code-Blocks formatiert # Diese Zeilen werden jetzt innerhalb des Code-Blocks formatiert
# Add invested time to the report if available
if 'elapsed_hours' in locals():
elapsed_hhmm = decimal_hours_to_hhmm(elapsed_hours)
report_lines.append(f"Investierte Zeit in dieser Session: {elapsed_hhmm}")
report_lines.append(f"Neuer Status: {actual_status}") report_lines.append(f"Neuer Status: {actual_status}")
if actual_summary: if actual_summary:
@@ -601,7 +663,8 @@ def report_status_to_notion(
# Notion aktualisieren # Notion aktualisieren
append_blocks_to_notion_page(token, task_id, notion_blocks) append_blocks_to_notion_page(token, task_id, notion_blocks)
update_notion_task_status(token, task_id, actual_status) status_payload = {"Status": {"status": {"name": actual_status}}}
update_notion_task_property(token, task_id, status_payload)
except (FileNotFoundError, json.JSONDecodeError) as e: except (FileNotFoundError, json.JSONDecodeError) as e:
print(f"❌ FEHLER beim Lesen der Session-Informationen für Statusbericht: {e}") print(f"❌ FEHLER beim Lesen der Session-Informationen für Statusbericht: {e}")
@@ -612,7 +675,7 @@ def report_status_to_notion(
# --- Context Generation --- # --- Context Generation ---
def generate_cli_context(project_title: str, task_title: str, task_id: str, readme_path: Optional[str], task_description: Optional[str]) -> str: def generate_cli_context(project_title: str, task_title: str, task_id: str, readme_path: Optional[str], task_description: Optional[str], total_duration: float) -> str:
"""Erstellt den reinen Kontext-String für die Gemini CLI.""" """Erstellt den reinen Kontext-String für die Gemini CLI."""
# Fallback, falls kein Pfad in Notion gesetzt ist # Fallback, falls kein Pfad in Notion gesetzt ist
@@ -625,10 +688,14 @@ def generate_cli_context(project_title: str, task_title: str, task_id: str, read
f"\n**Aufgabenbeschreibung:**\n" f"\n**Aufgabenbeschreibung:**\n"
f"```\n{task_description}\n```\n" f"```\n{task_description}\n```\n"
) )
duration_hhmm = decimal_hours_to_hhmm(total_duration)
duration_part = f"Bisher erfasster Zeitaufwand (Notion): {duration_hhmm} Stunden.\n"
context = ( context = (
f"Ich arbeite jetzt am Projekt '{project_title}'. Der Fokus liegt auf dem Task '{task_title}'.\n" f"Ich arbeite jetzt am Projekt '{project_title}'. Der Fokus liegt auf dem Task '{task_title}'.\n"
f"{description_part}\n" f"{description_part}\n"
f"{duration_part}"
"Die relevanten Dateien für dieses Projekt sind wahrscheinlich:\n" "Die relevanten Dateien für dieses Projekt sind wahrscheinlich:\n"
"- Die primäre Projektdokumentation: @readme.md\n" "- Die primäre Projektdokumentation: @readme.md\n"
f"- Die spezifische Dokumentation für dieses Modul: @{readme_path}\n\n" f"- Die spezifische Dokumentation für dieses Modul: @{readme_path}\n\n"
@@ -641,20 +708,13 @@ def generate_cli_context(project_title: str, task_title: str, task_id: str, read
# Die start_gemini_cli Funktion wird entfernt, da das aufrufende Skript jetzt die Gemini CLI startet. # Die start_gemini_cli Funktion wird entfernt, da das aufrufende Skript jetzt die Gemini CLI startet.
import shutil
import argparse
# --- Session Management ---
SESSION_DIR = ".dev_session"
SESSION_FILE_PATH = os.path.join(SESSION_DIR, "SESSION_INFO")
def save_session_info(task_id: str, token: str): def save_session_info(task_id: str, token: str):
"""Speichert die Task-ID und den Token für den Git-Hook.""" """Speichert die Task-ID, den Token und den Startzeitpunkt für den Git-Hook."""
os.makedirs(SESSION_DIR, exist_ok=True) os.makedirs(SESSION_DIR, exist_ok=True)
session_data = { session_data = {
"task_id": task_id, "task_id": task_id,
"token": token "token": token,
"session_start_time": datetime.now().isoformat()
} }
with open(SESSION_FILE_PATH, "w") as f: with open(SESSION_FILE_PATH, "w") as f:
json.dump(session_data, f) json.dump(session_data, f)
@@ -713,7 +773,8 @@ def complete_session():
status_options = get_database_status_options(token, tasks_db_id) status_options = get_database_status_options(token, tasks_db_id)
if status_options: if status_options:
done_status = status_options[-1] done_status = status_options[-1]
update_notion_task_status(token, task_id, done_status) status_payload = {"Status": {"status": {"name": done_status}}}
update_notion_task_property(token, task_id, status_payload)
except (FileNotFoundError, json.JSONDecodeError): except (FileNotFoundError, json.JSONDecodeError):
print("Fehler beim Lesen der Session-Informationen.") print("Fehler beim Lesen der Session-Informationen.")
@@ -767,8 +828,11 @@ def start_interactive_session():
task_id = selected_task["id"] task_id = selected_task["id"]
print(f"\nTask '{task_title}' ausgewählt.") print(f"\nTask '{task_title}' ausgewählt.")
# NEU: Lade die Task-Beschreibung # NEU: Lade die Task-Beschreibung und die bisherige Dauer
task_description = get_page_content(token, task_id) task_description = get_page_content(token, task_id)
total_duration_decimal = get_page_number_property(selected_task, "Total Duration (h)") or 0.0
total_duration_hhmm = decimal_hours_to_hhmm(total_duration_decimal)
print(f"> Bisher für diesen Task erfasst: {total_duration_hhmm} Stunden.")
# Session-Informationen für den Git-Hook speichern # Session-Informationen für den Git-Hook speichern
save_session_info(task_id, token) save_session_info(task_id, token)
@@ -782,7 +846,8 @@ def start_interactive_session():
suggested_branch_name = f"feature/task-{task_id.split('-')[0]}-{title_slug}" suggested_branch_name = f"feature/task-{task_id.split('-')[0]}-{title_slug}"
status_updated = update_notion_task_status(token, task_id, "Doing") status_payload = {"Status": {"status": {"name": "Doing"}}}
status_updated = update_notion_task_property(token, task_id, status_payload)
if not status_updated: if not status_updated:
print("Warnung: Notion-Task-Status konnte nicht aktualisiert werden.") print("Warnung: Notion-Task-Status konnte nicht aktualisiert werden.")
@@ -795,7 +860,7 @@ def start_interactive_session():
print("------------------------------------------------------------------") print("------------------------------------------------------------------")
# CLI-Kontext generieren und an stdout ausgeben, damit das Startskript ihn aufgreifen kann # CLI-Kontext generieren und an stdout ausgeben, damit das Startskript ihn aufgreifen kann
cli_context = generate_cli_context(project_title, task_title, task_id, readme_path, task_description) cli_context = generate_cli_context(project_title, task_title, task_id, readme_path, task_description, total_duration_decimal)
print("\n---GEMINI_CLI_CONTEXT_START---") print("\n---GEMINI_CLI_CONTEXT_START---")
print(cli_context) print(cli_context)
print("---GEMINI_CLI_CONTEXT_END---") print("---GEMINI_CLI_CONTEXT_END---")