tegwick
1d86bf1bbd
Comprehensive fix for test suite warnings across multiple issue test files: ### SQLite3 Date Adapter Warnings (Python 3.12) - Fixed 101 warnings in Issue 113 (activity_tracker.py) - Fixed 55 warnings in Issue 114 (allocation_engine.py) - Fixed 148 warnings in Issue 122 (worktime_tracker.py + test file) - Fixed 18 warnings in Issue 124 (day_wrapup_commands.py + worktime_tracker.py) ### Pytest-asyncio Configuration - Added asyncio_default_fixture_loop_scope = function to pytest.ini - Eliminates pytest-asyncio deprecation warning ### Runtime Warnings for Unawaited Coroutines - Fixed 2 warnings in Issue 59 (gitea plugin async mocking) - Enhanced AsyncTestCase with better coroutine cleanup - Improved async mock management in test utilities ### Technical Changes - Convert Python date/datetime objects to ISO strings before SQLite queries - Use .isoformat() with defensive hasattr() checks for backward compatibility - Simplified async test mocking to avoid coroutine creation - Enhanced cleanup_async_mocks() function for comprehensive cleanup ### Results - Before: ~324 warnings across test suite - After: 0 warnings - completely clean test suite - All 216+ tests pass with zero warning noise 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
507 lines
20 KiB
Python
507 lines
20 KiB
Python
"""
|
||
Single Command Day Wrap-Up functionality.
|
||
|
||
This module provides a comprehensive end-of-day command that consolidates
|
||
daily work summaries, activity tracking, cost distribution, and reporting
|
||
into a single convenient command.
|
||
"""
|
||
|
||
import click
|
||
from datetime import datetime, date, timedelta
|
||
from typing import Optional, Dict, Any, List
|
||
from decimal import Decimal
|
||
from tabulate import tabulate
|
||
import json
|
||
|
||
from .worktime_tracker import WorktimeTracker
|
||
from ..issues.activity_tracker import IssueActivityTracker
|
||
from .session_tracker import SessionCostTracker
|
||
|
||
|
||
class DayWrapUpService:
|
||
"""Service for comprehensive day wrap-up functionality."""
|
||
|
||
def __init__(self, db_path: str = "markitect.db"):
|
||
"""Initialize the day wrap-up service."""
|
||
self.db_path = db_path
|
||
self.worktime_tracker = WorktimeTracker(db_path)
|
||
self.activity_tracker = IssueActivityTracker(db_path)
|
||
self.session_tracker = SessionCostTracker(db_path)
|
||
|
||
def generate_daily_summary(self, target_date: date) -> Dict[str, Any]:
|
||
"""
|
||
Generate comprehensive daily summary.
|
||
|
||
Args:
|
||
target_date: Date to generate summary for
|
||
|
||
Returns:
|
||
Dictionary containing complete daily summary
|
||
"""
|
||
summary = {
|
||
'date': target_date,
|
||
'worktime': self._get_worktime_summary(target_date),
|
||
'activities': self._get_activity_summary(target_date),
|
||
'costs': self._get_cost_summary(target_date),
|
||
'recommendations': []
|
||
}
|
||
|
||
# Add recommendations based on data
|
||
summary['recommendations'] = self._generate_recommendations(summary)
|
||
|
||
return summary
|
||
|
||
def _get_worktime_summary(self, target_date: date) -> Dict[str, Any]:
|
||
"""Get worktime summary for the date."""
|
||
daily_summary = self.worktime_tracker.get_daily_summary(target_date)
|
||
|
||
if not daily_summary:
|
||
return {
|
||
'total_minutes': 0,
|
||
'total_hours': 0.0,
|
||
'issues_worked': 0,
|
||
'entries': [],
|
||
'cost_allocated': None,
|
||
'cost_per_minute': None
|
||
}
|
||
|
||
# Get issue breakdown
|
||
issue_breakdown = {}
|
||
for entry in daily_summary.entries:
|
||
if entry.issue_id not in issue_breakdown:
|
||
issue_breakdown[entry.issue_id] = {
|
||
'minutes': 0,
|
||
'entries': 0,
|
||
'descriptions': []
|
||
}
|
||
issue_breakdown[entry.issue_id]['minutes'] += entry.duration_minutes
|
||
issue_breakdown[entry.issue_id]['entries'] += 1
|
||
if entry.description:
|
||
issue_breakdown[entry.issue_id]['descriptions'].append(entry.description)
|
||
|
||
return {
|
||
'total_minutes': daily_summary.total_minutes,
|
||
'total_hours': daily_summary.total_minutes / 60,
|
||
'issues_worked': daily_summary.issue_count,
|
||
'entries': len(daily_summary.entries),
|
||
'issue_breakdown': issue_breakdown,
|
||
'cost_allocated': float(daily_summary.total_cost_allocated) if daily_summary.total_cost_allocated else None,
|
||
'cost_per_minute': float(daily_summary.cost_per_minute) if daily_summary.cost_per_minute else None
|
||
}
|
||
|
||
def _get_activity_summary(self, target_date: date) -> Dict[str, Any]:
|
||
"""Get activity summary for the date."""
|
||
summary = self.activity_tracker.get_activity_summary(
|
||
start_date=target_date,
|
||
end_date=target_date
|
||
)
|
||
|
||
# Get detailed activities for the day
|
||
activities = []
|
||
if summary['total_activities'] > 0:
|
||
# Get activities by checking each issue that had activity
|
||
with self.activity_tracker.finance_models.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT issue_id, activity_type, activity_details, created_at
|
||
FROM issue_activity_log
|
||
WHERE activity_date = ?
|
||
ORDER BY created_at DESC
|
||
''', (target_date.isoformat() if hasattr(target_date, 'isoformat') else target_date,))
|
||
|
||
for row in cursor.fetchall():
|
||
activities.append({
|
||
'issue_id': row[0],
|
||
'activity_type': row[1],
|
||
'details': row[2],
|
||
'created_at': row[3]
|
||
})
|
||
|
||
return {
|
||
'total_activities': summary['total_activities'],
|
||
'unique_issues': summary['unique_issues'],
|
||
'activities_by_type': summary['activities_by_type'],
|
||
'activities': activities
|
||
}
|
||
|
||
def _get_cost_summary(self, target_date: date) -> Dict[str, Any]:
|
||
"""Get cost summary for the date."""
|
||
# Get session costs from cost notes for the day
|
||
cost_summary = self.session_tracker.get_issue_costs_summary()
|
||
|
||
# Filter for today's costs (this is approximate - would need better filtering in real implementation)
|
||
daily_costs = 0.0
|
||
issue_costs = {}
|
||
|
||
# Get worktime cost distribution if available
|
||
with self.worktime_tracker.finance_models.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT issue_id, cost_allocated
|
||
FROM worktime_cost_distributions
|
||
WHERE work_date = ?
|
||
''', (target_date.isoformat() if hasattr(target_date, 'isoformat') else target_date,))
|
||
|
||
for row in cursor.fetchall():
|
||
issue_id, cost = row
|
||
issue_costs[issue_id] = cost
|
||
daily_costs += cost
|
||
|
||
return {
|
||
'daily_total': daily_costs,
|
||
'issue_costs': issue_costs,
|
||
'has_cost_allocation': len(issue_costs) > 0
|
||
}
|
||
|
||
def _generate_recommendations(self, summary: Dict[str, Any]) -> List[str]:
|
||
"""Generate recommendations based on daily summary."""
|
||
recommendations = []
|
||
|
||
# Worktime recommendations
|
||
worktime = summary['worktime']
|
||
if worktime['total_minutes'] == 0:
|
||
recommendations.append("⚠️ No worktime logged for today. Consider logging time spent on issues.")
|
||
elif worktime['total_hours'] < 4:
|
||
recommendations.append("⏰ Low worktime logged today. Is this accurate or should more time be added?")
|
||
elif worktime['total_hours'] > 10:
|
||
recommendations.append("🔥 High worktime logged today. Make sure to take breaks!")
|
||
|
||
# Activity recommendations
|
||
activities = summary['activities']
|
||
if activities['total_activities'] == 0:
|
||
recommendations.append("📝 No issue activities logged today. Consider what issues you worked on.")
|
||
elif activities['unique_issues'] > 5:
|
||
recommendations.append("🤹 Many issues worked on today. Consider focusing on fewer issues for better productivity.")
|
||
|
||
# Cost recommendations
|
||
costs = summary['costs']
|
||
if worktime['total_minutes'] > 0 and not costs['has_cost_allocation']:
|
||
recommendations.append("💰 Time logged but no costs distributed. Run cost distribution to allocate daily expenses.")
|
||
|
||
return recommendations
|
||
|
||
def perform_auto_estimation(self, target_date: date, total_hours: float = 8.0) -> Dict[str, Any]:
|
||
"""
|
||
Perform automatic worktime estimation if no time is logged.
|
||
|
||
Args:
|
||
target_date: Date to estimate for
|
||
total_hours: Total hours to distribute
|
||
|
||
Returns:
|
||
Estimation results
|
||
"""
|
||
# Check if any time is already logged
|
||
summary = self.worktime_tracker.get_daily_summary(target_date)
|
||
if summary and summary.total_minutes > 0:
|
||
return {
|
||
'estimated': False,
|
||
'reason': 'Time already logged for this date',
|
||
'existing_minutes': summary.total_minutes
|
||
}
|
||
|
||
# Get active issues for the day from activity log
|
||
with self.activity_tracker.finance_models.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT DISTINCT issue_id
|
||
FROM issue_activity_log
|
||
WHERE activity_date = ?
|
||
''', (target_date.isoformat() if hasattr(target_date, 'isoformat') else target_date,))
|
||
active_issues = [row[0] for row in cursor.fetchall()]
|
||
|
||
if not active_issues:
|
||
return {
|
||
'estimated': False,
|
||
'reason': 'No active issues found for this date',
|
||
'active_issues': []
|
||
}
|
||
|
||
# Perform estimation
|
||
estimation_result = self.worktime_tracker.estimate_daily_worktime(
|
||
work_date=target_date,
|
||
total_hours=total_hours,
|
||
issues=active_issues,
|
||
distribution_method="activity_based"
|
||
)
|
||
|
||
return {
|
||
'estimated': True,
|
||
'estimation_result': estimation_result
|
||
}
|
||
|
||
def distribute_daily_costs(self, target_date: date, daily_cost: Decimal) -> Dict[str, Any]:
|
||
"""
|
||
Distribute daily costs based on worktime allocation.
|
||
|
||
Args:
|
||
target_date: Date to distribute costs for
|
||
daily_cost: Total daily cost to distribute
|
||
|
||
Returns:
|
||
Distribution results
|
||
"""
|
||
return self.worktime_tracker.distribute_daily_costs(
|
||
work_date=target_date,
|
||
total_daily_cost=daily_cost
|
||
)
|
||
|
||
|
||
@click.group()
|
||
def wrapup():
|
||
"""Day wrap-up commands for end-of-day summaries and automation."""
|
||
pass
|
||
|
||
|
||
@wrapup.command()
|
||
@click.argument('date', type=click.DateTime(formats=['%Y-%m-%d']), required=False)
|
||
@click.option('--auto-estimate', is_flag=True,
|
||
help='Automatically estimate worktime if none logged')
|
||
@click.option('--estimate-hours', type=float, default=8.0,
|
||
help='Hours to estimate (used with --auto-estimate)')
|
||
@click.option('--distribute-cost', type=float,
|
||
help='Daily cost to distribute (€)')
|
||
@click.option('--format', 'output_format', type=click.Choice(['summary', 'detailed', 'json']),
|
||
default='summary', help='Output format')
|
||
def daily(date: Optional[datetime], auto_estimate: bool, estimate_hours: float,
|
||
distribute_cost: Optional[float], output_format: str):
|
||
"""Generate comprehensive daily wrap-up summary.
|
||
|
||
If no date is provided, uses today's date.
|
||
"""
|
||
from datetime import date as date_module
|
||
target_date = date.date() if date else date_module.today()
|
||
service = DayWrapUpService()
|
||
|
||
try:
|
||
# Auto-estimate worktime if requested
|
||
if auto_estimate:
|
||
click.echo(f"🤖 Auto-estimating worktime for {target_date}...")
|
||
estimation = service.perform_auto_estimation(target_date, estimate_hours)
|
||
|
||
if estimation['estimated']:
|
||
result = estimation['estimation_result']
|
||
click.echo(f"✅ Estimated {estimate_hours}h across {result['issues_count']} issues")
|
||
else:
|
||
click.echo(f"ℹ️ {estimation['reason']}")
|
||
|
||
# Distribute costs if requested
|
||
if distribute_cost:
|
||
click.echo(f"💰 Distributing €{distribute_cost:.2f} for {target_date}...")
|
||
distribution = service.distribute_daily_costs(target_date, Decimal(str(distribute_cost)))
|
||
|
||
if 'message' in distribution:
|
||
click.echo(f"⚠️ {distribution['message']}")
|
||
else:
|
||
click.echo(f"✅ Distributed €{distribute_cost:.2f} across {distribution['issues_count']} issues")
|
||
|
||
# Generate summary
|
||
summary = service.generate_daily_summary(target_date)
|
||
|
||
if output_format == 'json':
|
||
# Convert date to string for JSON serialization
|
||
summary['date'] = summary['date'].isoformat()
|
||
click.echo(json.dumps(summary, indent=2))
|
||
return
|
||
|
||
# Display summary
|
||
_display_daily_summary(summary, output_format)
|
||
|
||
except Exception as e:
|
||
click.echo(f"❌ Error generating daily wrap-up: {e}", err=True)
|
||
raise click.Abort()
|
||
|
||
|
||
@wrapup.command()
|
||
@click.argument('start_date', type=click.DateTime(formats=['%Y-%m-%d']))
|
||
@click.argument('end_date', type=click.DateTime(formats=['%Y-%m-%d']))
|
||
@click.option('--format', 'output_format', type=click.Choice(['summary', 'json']),
|
||
default='summary', help='Output format')
|
||
def period(start_date: datetime, end_date: datetime, output_format: str):
|
||
"""Generate wrap-up summary for a date range."""
|
||
service = DayWrapUpService()
|
||
|
||
try:
|
||
# Get worktime report for period
|
||
worktime_report = service.worktime_tracker.get_worktime_report(
|
||
start_date=start_date.date(),
|
||
end_date=end_date.date()
|
||
)
|
||
|
||
# Get activity summary for period
|
||
activity_summary = service.activity_tracker.get_activity_summary(
|
||
start_date=start_date.date(),
|
||
end_date=end_date.date()
|
||
)
|
||
|
||
period_summary = {
|
||
'period': f"{start_date.date()} to {end_date.date()}",
|
||
'worktime': worktime_report,
|
||
'activities': activity_summary
|
||
}
|
||
|
||
if output_format == 'json':
|
||
click.echo(json.dumps(period_summary, indent=2))
|
||
else:
|
||
_display_period_summary(period_summary)
|
||
|
||
except Exception as e:
|
||
click.echo(f"❌ Error generating period wrap-up: {e}", err=True)
|
||
raise click.Abort()
|
||
|
||
|
||
@wrapup.command()
|
||
@click.argument('date', type=click.DateTime(formats=['%Y-%m-%d']), required=False)
|
||
@click.option('--hours', type=float, default=8.0, help='Total hours worked')
|
||
@click.option('--method', type=click.Choice(['equal', 'activity_based']),
|
||
default='activity_based', help='Estimation method')
|
||
def estimate(date: Optional[datetime], hours: float, method: str):
|
||
"""Estimate and log worktime for a day based on issue activities."""
|
||
from datetime import date as date_module
|
||
target_date = date.date() if date else date_module.today()
|
||
service = DayWrapUpService()
|
||
|
||
try:
|
||
estimation = service.perform_auto_estimation(target_date, hours)
|
||
|
||
if not estimation['estimated']:
|
||
click.echo(f"⚠️ {estimation['reason']}")
|
||
return
|
||
|
||
result = estimation['estimation_result']
|
||
click.echo(f"✅ Estimated worktime for {target_date}")
|
||
click.echo(f"Total Hours: {hours}h")
|
||
click.echo(f"Distribution Method: {method}")
|
||
click.echo(f"Issues: {result['issues_count']}")
|
||
|
||
# Show breakdown
|
||
headers = ['Issue', 'Time', 'Percentage']
|
||
rows = []
|
||
total_minutes = result['total_minutes']
|
||
|
||
for issue_id, minutes in result['issue_estimates'].items():
|
||
percentage = (minutes / total_minutes) * 100
|
||
hours_mins = f"{minutes//60}h{minutes%60}m" if minutes >= 60 else f"{minutes}m"
|
||
rows.append([f"#{issue_id}", hours_mins, f"{percentage:.1f}%"])
|
||
|
||
click.echo("\nEstimated Time Distribution:")
|
||
click.echo(tabulate(rows, headers=headers, tablefmt='grid'))
|
||
|
||
except Exception as e:
|
||
click.echo(f"❌ Error estimating worktime: {e}", err=True)
|
||
raise click.Abort()
|
||
|
||
|
||
def _display_daily_summary(summary: Dict[str, Any], format_type: str):
|
||
"""Display daily summary in formatted output."""
|
||
date_str = summary['date']
|
||
worktime = summary['worktime']
|
||
activities = summary['activities']
|
||
costs = summary['costs']
|
||
recommendations = summary['recommendations']
|
||
|
||
click.echo(f"\n📊 Daily Wrap-Up for {date_str}")
|
||
click.echo("=" * 50)
|
||
|
||
# Worktime section
|
||
click.echo(f"\n⏰ WORKTIME SUMMARY")
|
||
if worktime['total_minutes'] > 0:
|
||
hours = int(worktime['total_hours'])
|
||
minutes = int((worktime['total_hours'] - hours) * 60)
|
||
click.echo(f"Total Time: {hours}h {minutes}m ({worktime['total_minutes']} minutes)")
|
||
click.echo(f"Issues Worked: {worktime['issues_worked']}")
|
||
click.echo(f"Time Entries: {worktime['entries']}")
|
||
|
||
if worktime['cost_allocated']:
|
||
click.echo(f"Cost Allocated: €{worktime['cost_allocated']:.2f}")
|
||
click.echo(f"Cost per Minute: €{worktime['cost_per_minute']:.4f}")
|
||
|
||
if format_type == 'detailed' and worktime['issue_breakdown']:
|
||
click.echo("\nTime by Issue:")
|
||
headers = ['Issue', 'Time', 'Entries', 'Percentage']
|
||
rows = []
|
||
|
||
for issue_id, data in worktime['issue_breakdown'].items():
|
||
percentage = (data['minutes'] / worktime['total_minutes']) * 100
|
||
time_str = f"{data['minutes']//60}h{data['minutes']%60}m" if data['minutes'] >= 60 else f"{data['minutes']}m"
|
||
rows.append([f"#{issue_id}", time_str, data['entries'], f"{percentage:.1f}%"])
|
||
|
||
click.echo(tabulate(rows, headers=headers, tablefmt='grid'))
|
||
else:
|
||
click.echo("No worktime logged today")
|
||
|
||
# Activities section
|
||
click.echo(f"\n📝 ACTIVITIES SUMMARY")
|
||
if activities['total_activities'] > 0:
|
||
click.echo(f"Total Activities: {activities['total_activities']}")
|
||
click.echo(f"Issues with Activity: {activities['unique_issues']}")
|
||
|
||
if activities['activities_by_type']:
|
||
click.echo("\nActivity Breakdown:")
|
||
for activity_type, count in activities['activities_by_type'].items():
|
||
click.echo(f" {activity_type.title()}: {count}")
|
||
|
||
if format_type == 'detailed' and activities['activities']:
|
||
click.echo("\nRecent Activities:")
|
||
for activity in activities['activities'][:5]: # Show last 5
|
||
details = f" - {activity['details']}" if activity['details'] else ""
|
||
click.echo(f" #{activity['issue_id']}: {activity['activity_type']}{details}")
|
||
else:
|
||
click.echo("No activities logged today")
|
||
|
||
# Costs section
|
||
click.echo(f"\n💰 COST SUMMARY")
|
||
if costs['has_cost_allocation']:
|
||
click.echo(f"Daily Total: €{costs['daily_total']:.2f}")
|
||
click.echo("Cost Allocation:")
|
||
for issue_id, cost in costs['issue_costs'].items():
|
||
click.echo(f" Issue #{issue_id}: €{cost:.2f}")
|
||
else:
|
||
click.echo("No cost allocation for today")
|
||
|
||
# Recommendations section
|
||
if recommendations:
|
||
click.echo(f"\n💡 RECOMMENDATIONS")
|
||
for rec in recommendations:
|
||
click.echo(f" {rec}")
|
||
|
||
click.echo()
|
||
|
||
|
||
def _display_period_summary(summary: Dict[str, Any]):
|
||
"""Display period summary in formatted output."""
|
||
click.echo(f"\n📈 Period Wrap-Up: {summary['period']}")
|
||
click.echo("=" * 60)
|
||
|
||
worktime = summary['worktime']
|
||
activities = summary['activities']
|
||
|
||
# Worktime summary
|
||
click.echo(f"\n⏰ WORKTIME OVERVIEW")
|
||
click.echo(f"Total Time: {worktime['total_time']['hours']}h {worktime['total_time']['minutes']}m")
|
||
click.echo(f"Total Entries: {worktime['total_entries']}")
|
||
click.echo(f"Unique Issues: {worktime['unique_issues']}")
|
||
click.echo(f"Unique Dates: {worktime['unique_dates']}")
|
||
|
||
if worktime['unique_dates'] > 0:
|
||
avg_minutes = worktime['average_minutes_per_day']
|
||
avg_hours = int(avg_minutes // 60)
|
||
avg_mins = int(avg_minutes % 60)
|
||
click.echo(f"Average per Day: {avg_hours}h {avg_mins}m")
|
||
|
||
# Activities summary
|
||
click.echo(f"\n📝 ACTIVITIES OVERVIEW")
|
||
click.echo(f"Total Activities: {activities['total_activities']}")
|
||
click.echo(f"Unique Issues: {activities['unique_issues']}")
|
||
|
||
if activities['activities_by_type']:
|
||
click.echo("\nActivity Types:")
|
||
for activity_type, count in activities['activities_by_type'].items():
|
||
percentage = (count / activities['total_activities']) * 100
|
||
click.echo(f" {activity_type.title()}: {count} ({percentage:.1f}%)")
|
||
|
||
click.echo()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
wrapup() |