From 056349cb77b692a6ac1dedb91a0e34a0d26b3238 Mon Sep 17 00:00:00 2001 From: AlexAndrewsAI Date: Sat, 20 Jun 2026 08:45:50 -0400 Subject: [PATCH] add standard utc timezone to db and cleanup multi app code --- README.md | 15 +- patch_serialise.py | 33 --- pyproject.toml | 2 +- tests/test_base_tracker.py | 74 ++++++- tests/test_cli.py | 191 +++++++++++----- tests/test_database.py | 271 ++++++++++++++++++++++- tests/test_simple_time_tracker.py | 94 ++++++-- timetracker_utils/base_tracker.py | 72 +++++- timetracker_utils/cli.py | 153 +++++-------- timetracker_utils/database.py | 36 +-- timetracker_utils/simple_time_tracker.py | 52 +++-- 11 files changed, 716 insertions(+), 277 deletions(-) delete mode 100644 patch_serialise.py diff --git a/README.md b/README.md index 5b0dd1a..910cd21 100644 --- a/README.md +++ b/README.md @@ -55,24 +55,23 @@ max_conflict_display: 100 ### CLI -The package provides a `timetracker` CLI with commands for both supported formats: +The package provides a `timetracker` CLI with unified commands for both supported formats: ```bash # Show version uv run timetracker --version # Import a TimeCop CSV file and display entries -uv run timetracker timecop --config tests/timetracker.yml --input tests/example_timecop.csv - -# Export the database back to TimeCop CSV -uv run timetracker timecop --config tests/timetracker.yml --output timecop_export.csv - +uv run timetracker add --config tests/timetracker.yml --format timecop tests/example_timecop.csv # Import a Simple Time Tracker CSV file and display entries -uv run timetracker stt --config tests/timetracker.yml --input tests/example_stt.csv +uv run timetracker add --config tests/timetracker.yml --format stt tests/example_stt.csv + +# Export the database back to TimeCop CSV +uv run timetracker export --config tests/timetracker.yml --format timecop timecop_export.csv # Export the database back to Simple Time Tracker CSV -uv run timetracker stt --config tests/timetracker.yml --output stt_export.csv +uv run timetracker export --config tests/timetracker.yml --format stt stt_export.csv ``` ### Python API diff --git a/patch_serialise.py b/patch_serialise.py deleted file mode 100644 index 99a033e..0000000 --- a/patch_serialise.py +++ /dev/null @@ -1,33 +0,0 @@ -"""Patch script to fix _serialise_lists in database.py. - -This script updates the _serialise_lists function to convert empty lists -to empty strings instead of "[]". -""" - -from pathlib import Path - -# Fix _serialise_lists: empty lists -> "" not "[]" -p = Path("/root/timetracker-utils/timetracker_utils/database.py") -src = p.read_text() -old = """ df[col] = df[col].apply( - lambda x: json.dumps(x) if isinstance(x, list) - else ("" if _is_blank(x) else str(x)) - ) - return df - - -def _deserialise_lists""" -new = """ df[col] = df[col].apply( - lambda x: json.dumps(x) if isinstance(x, list) and len(x) > 0 - else ("" if _is_blank(x) else str(x)) - ) - return df - - -def _deserialise_lists""" -if old in src: - src = src.replace(old, new, 1) - p.write_text(src) - print("patched _serialise_lists") -else: - print("old text not found") diff --git a/pyproject.toml b/pyproject.toml index 5812280..8e396f4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -97,7 +97,7 @@ warn_required_dynamic_aliases = true [tool.pytest.ini_options] testpaths = ["tests"] -addopts = "--cov=timetracker_utils --cov-report=term-missing --cov-fail-under=80" +addopts = "--cov=timetracker_utils --cov-report=term-missing --cov-fail-under=95" [tool.coverage.run] source = ["timetracker_utils"] diff --git a/tests/test_base_tracker.py b/tests/test_base_tracker.py index e22096b..9ae6d9b 100644 --- a/tests/test_base_tracker.py +++ b/tests/test_base_tracker.py @@ -77,6 +77,17 @@ def test_base_entry_parse_list_fields_string_fallback() -> None: assert entry.categories == ["42"] +def test_base_entry_parse_list_fields_edge_case_empty_list_string() -> None: + """Test parse_list_fields edge case where str(value).strip() == [] (line 177).""" + entry = BaseTimeEntry( + activity="Test", + start_time="2200-01-15T09:00:00.000Z", + end_time="2200-01-15T11:30:00.000Z", + categories=[], + ) + assert entry.categories == [] + + def test_base_entry_parse_list_fields_non_string_empty() -> None: """Test that empty categories/tags defaults to empty list.""" entry = BaseTimeEntry( @@ -199,26 +210,67 @@ def test_base_tracker_total_hours_by_activity_with_data() -> None: tracker.total_hours_by_activity() -def test_base_entry_parse_datetime_naive_datetime_object() -> None: - """Test parse_datetime with naive datetime object (tzinfo is None, line 120).""" +def test_base_entry_parse_datetime_naive_datetime_object_zone_none() -> None: + """Test parse_datetime with naive datetime object when resolve_tz returns None.""" from datetime import datetime + from unittest.mock import patch + + with patch("timetracker_utils.datetime_utils.resolve_tz", return_value=None): + entry = BaseTimeEntry.model_validate( + { + "activity": "Test", + "start_time": datetime(2200, 1, 15, 9, 0, 0), # naive datetime + "end_time": "2200-01-15T11:30:00.000Z", + }, + context={"default_timezone": "ET"}, + ) + # Should still work, falling back to UTC + assert entry.start_time.tzinfo is not None - entry = BaseTimeEntry( - activity="Test", - start_time=datetime(2200, 1, 15, 9, 0, 0), # naive datetime - end_time="2200-01-15T11:30:00.000Z", - ) - # Naive datetime should be assumed UTC + +def test_base_entry_parse_datetime_naive_datetime_object_zone_not_none() -> None: + """Test parse_datetime with naive datetime object when resolve_tz returns zone.""" + from datetime import datetime, timedelta, timezone + from unittest.mock import patch + + test_zone = timezone(timedelta(hours=-5)) + with patch("timetracker_utils.datetime_utils.resolve_tz", return_value=test_zone): + entry = BaseTimeEntry.model_validate( + { + "activity": "Test", + "start_time": datetime(2200, 1, 15, 9, 0, 0), # naive datetime + "end_time": "2200-01-15T11:30:00.000Z", + }, + context={"default_timezone": "ET"}, + ) + # Should convert to UTC + assert entry.start_time.tzinfo is not None + + +def test_base_entry_parse_datetime_naive_string_zone_none() -> None: + """Test parse_datetime with naive string when resolve_tz returns None (line 152).""" + from unittest.mock import patch + + with patch("timetracker_utils.datetime_utils.resolve_tz", return_value=None): + entry = BaseTimeEntry.model_validate( + { + "activity": "Test", + "start_time": "2200-01-15T09:00:00", # naive string + "end_time": "2200-01-15T11:30:00.000Z", + }, + context={"default_timezone": "ET"}, + ) + # Should still work, falling back to UTC assert entry.start_time.tzinfo is not None -def test_base_entry_hours_empty_string_returns_none() -> None: - """Test hours validator with empty string (line 157).""" +def test_base_entry_hours_whitespace_string() -> None: + """Test hours validator with whitespace-only string (line 187).""" entry = BaseTimeEntry( activity="Test", start_time="2200-01-15T09:00:00.000Z", end_time="2200-01-15T11:30:00.000Z", - hours="", + hours=" ", ) # hours should be computed from end_time - start_time assert entry.hours == 2.5 diff --git a/tests/test_cli.py b/tests/test_cli.py index af5d4ed..87ef569 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -90,12 +90,12 @@ def test_version_callback() -> None: def test_timecop_command(tmp_path: Path) -> None: - """Test the timecop CLI command loads a CSV and prints the DataFrame.""" + """Test the add CLI command with timecop format loads a CSV and prints the DataFrame.""" csv_path = tmp_path / "test.csv" csv_path.write_text(SAMPLE_CSV, encoding="utf-8") config_path = _write_config(tmp_path, timezone="ET") result = runner.invoke( - app, ["timecop", "--config", str(config_path), "--input", str(csv_path)] + app, ["add", "--config", str(config_path), "--format", "timecop", str(csv_path)] ) assert result.exit_code == 0 assert "Loaded DataFrame" in result.output @@ -103,17 +103,18 @@ def test_timecop_command(tmp_path: Path) -> None: def test_timecop_command_head(tmp_path: Path) -> None: - """Test the timecop CLI command with --head option.""" + """Test the add CLI command with timecop format and --head option.""" csv_path = tmp_path / "test.csv" csv_path.write_text(SAMPLE_CSV, encoding="utf-8") config_path = _write_config(tmp_path, timezone="ET") result = runner.invoke( app, [ - "timecop", + "add", "--config", str(config_path), - "--input", + "--format", + "timecop", str(csv_path), "--head", "1", @@ -124,18 +125,19 @@ def test_timecop_command_head(tmp_path: Path) -> None: def test_timecop_command_missing_config() -> None: - """Test that the timecop CLI command fails without required --config.""" - result = runner.invoke(app, ["timecop"]) + """Test that the add CLI command fails without required --config.""" + result = runner.invoke(app, ["add", "--format", "timecop", "test.csv"]) assert result.exit_code != 0 assert "Missing option" in result.stderr or "required" in result.stderr.lower() def test_timecop_command_no_input_or_output_exits_with_error(tmp_path: Path) -> None: - """Test that the timecop CLI command fails without --input or --output.""" + """Test that the add CLI command fails without input file argument.""" config_path = _write_config(tmp_path, timezone="ET") - result = runner.invoke(app, ["timecop", "--config", str(config_path)]) - assert result.exit_code == 1 - assert "input" in result.output or "output" in result.output + result = runner.invoke( + app, ["add", "--config", str(config_path), "--format", "timecop"] + ) + assert result.exit_code != 0 def test_timecop_command_timezone_from_config(tmp_path: Path) -> None: @@ -144,7 +146,7 @@ def test_timecop_command_timezone_from_config(tmp_path: Path) -> None: csv_path.write_text(SAMPLE_CSV, encoding="utf-8") config_path = _write_config(tmp_path, timezone="ET") result = runner.invoke( - app, ["timecop", "--config", str(config_path), "--input", str(csv_path)] + app, ["add", "--config", str(config_path), "--format", "timecop", str(csv_path)] ) assert result.exit_code == 0 assert "Loaded DataFrame" in result.output @@ -159,7 +161,7 @@ def test_timecop_command_different_timezone(tmp_path: Path) -> None: csv_path.write_text(SAMPLE_CSV, encoding="utf-8") config_path = _write_config(tmp_path, timezone="PT") result = runner.invoke( - app, ["timecop", "--config", str(config_path), "--input", str(csv_path)] + app, ["add", "--config", str(config_path), "--format", "timecop", str(csv_path)] ) assert result.exit_code == 0 assert "Loaded DataFrame" in result.output @@ -168,11 +170,19 @@ def test_timecop_command_different_timezone(tmp_path: Path) -> None: def test_timecop_output_empty_db(tmp_path: Path) -> None: - """Test --output with an empty database writes header-only CSV.""" + """Test export with timecop format and empty database writes header-only CSV.""" config_path = _write_config(tmp_path, timezone="ET") output_path = tmp_path / "output.csv" result = runner.invoke( - app, ["timecop", "--config", str(config_path), "--output", str(output_path)] + app, + [ + "export", + "--config", + str(config_path), + "--format", + "timecop", + str(output_path), + ], ) assert result.exit_code == 0 assert "Database is empty" in result.output @@ -186,19 +196,27 @@ def test_timecop_output_empty_db(tmp_path: Path) -> None: def test_timecop_output_with_data(tmp_path: Path) -> None: - """Test --output exports a previously imported database to CSV.""" + """Test export with timecop format exports a previously imported database to CSV.""" csv_path = tmp_path / "input.csv" csv_path.write_text(SAMPLE_CSV, encoding="utf-8") config_path = _write_config(tmp_path, timezone="ET") # First, import the CSV to populate the database result = runner.invoke( - app, ["timecop", "--config", str(config_path), "--input", str(csv_path)] + app, ["add", "--config", str(config_path), "--format", "timecop", str(csv_path)] ) assert result.exit_code == 0 # Now export to output CSV output_path = tmp_path / "output.csv" result = runner.invoke( - app, ["timecop", "--config", str(config_path), "--output", str(output_path)] + app, + [ + "export", + "--config", + str(config_path), + "--format", + "timecop", + str(output_path), + ], ) assert result.exit_code == 0 assert output_path.exists() @@ -220,25 +238,38 @@ def test_timecop_output_with_data(tmp_path: Path) -> None: def test_timecop_output_combined_with_input(tmp_path: Path) -> None: - """Test using --output together with --input.""" + """Test using add and export commands together.""" csv_path = tmp_path / "input.csv" csv_path.write_text(SAMPLE_CSV, encoding="utf-8") config_path = _write_config(tmp_path, timezone="ET") output_path = tmp_path / "output.csv" + # First add the data result = runner.invoke( app, [ - "timecop", + "add", "--config", str(config_path), - "--input", + "--format", + "timecop", str(csv_path), - "--output", - str(output_path), ], ) assert result.exit_code == 0 assert "Loaded DataFrame" in result.output + # Then export the data + result = runner.invoke( + app, + [ + "export", + "--config", + str(config_path), + "--format", + "timecop", + str(output_path), + ], + ) + assert result.exit_code == 0 assert "Exporting" in result.output assert output_path.exists() content = output_path.read_text(encoding="utf-8") @@ -246,7 +277,7 @@ def test_timecop_output_combined_with_input(tmp_path: Path) -> None: def test_stt_rejects_timecop_csv(tmp_path: Path) -> None: - """Test that the stt CLI command rejects a TimeCop-format CSV.""" + """Test that the add CLI command with stt format rejects a TimeCop-format CSV.""" timecop_csv = """\ "Date","Project","Description","Combined Project & Description","Start Time","End Time","Time (hours)","Notes" "1/15/2200","StellarCartography","nebula mapping","StellarCartography: nebula mapping","2200-01-15T09:00:00.000Z","2200-01-15T11:30:00.000Z","2.5","" @@ -255,14 +286,14 @@ def test_stt_rejects_timecop_csv(tmp_path: Path) -> None: csv_path.write_text(timecop_csv, encoding="utf-8") config_path = _write_config(tmp_path, timezone="ET") result = runner.invoke( - app, ["stt", "--config", str(config_path), "--input", str(csv_path)] + app, ["add", "--config", str(config_path), "--format", "stt", str(csv_path)] ) assert result.exit_code != 0 assert "Missing required STT columns" in (result.output or result.stderr or "") def test_timecop_rejects_stt_csv(tmp_path: Path) -> None: - """Test that the timecop CLI command rejects an STT-format CSV.""" + """Test that the add CLI command with timecop format rejects an STT-format CSV.""" stt_csv = """\ "activity name","time started","time ended","comment","categories","record tags","duration","duration minutes" "StellarCartography","2200-01-15T09:00:00.000Z","2200-01-15T11:30:00.000Z","nebula mapping","nebula mapping","","2:30:00","150" @@ -271,18 +302,26 @@ def test_timecop_rejects_stt_csv(tmp_path: Path) -> None: csv_path.write_text(stt_csv, encoding="utf-8") config_path = _write_config(tmp_path, timezone="ET") result = runner.invoke( - app, ["timecop", "--config", str(config_path), "--input", str(csv_path)] + app, ["add", "--config", str(config_path), "--format", "timecop", str(csv_path)] ) assert result.exit_code != 0 assert "Missing required TimeCop columns" in (result.output or result.stderr or "") def test_timecop_output_non_existent_db(tmp_path: Path) -> None: - """Test --output when the database file does not exist yet.""" + """Test export with timecop format when the database file does not exist yet.""" config_path = _write_config(tmp_path, timezone="ET") output_path = tmp_path / "output.csv" result = runner.invoke( - app, ["timecop", "--config", str(config_path), "--output", str(output_path)] + app, + [ + "export", + "--config", + str(config_path), + "--format", + "timecop", + str(output_path), + ], ) assert result.exit_code == 0 assert "Database is empty" in result.output @@ -528,7 +567,7 @@ def test_compute_simple_duration_datetime_objects() -> None: def test_stt_command(tmp_path: Path) -> None: - """Test the stt CLI command loads a CSV and prints the DataFrame.""" + """Test the add CLI command with stt format loads a CSV and prints the DataFrame.""" stt_csv = ( '"activity name","time started","time ended","comment","categories",' '"record tags","duration","duration minutes"\n' @@ -539,7 +578,7 @@ def test_stt_command(tmp_path: Path) -> None: csv_path.write_text(stt_csv, encoding="utf-8") config_path = _write_config(tmp_path, timezone="ET") result = runner.invoke( - app, ["stt", "--config", str(config_path), "--input", str(csv_path)] + app, ["add", "--config", str(config_path), "--format", "stt", str(csv_path)] ) assert result.exit_code == 0 assert "Loaded DataFrame" in result.output @@ -547,7 +586,7 @@ def test_stt_command(tmp_path: Path) -> None: def test_stt_command_head(tmp_path: Path) -> None: - """Test the stt CLI command with --head option.""" + """Test the add CLI command with stt format and --head option.""" stt_csv = ( '"activity name","time started","time ended","comment","categories",' '"record tags","duration","duration minutes"\n' @@ -560,10 +599,11 @@ def test_stt_command_head(tmp_path: Path) -> None: result = runner.invoke( app, [ - "stt", + "add", "--config", str(config_path), - "--input", + "--format", + "stt", str(csv_path), "--head", "1", @@ -574,18 +614,19 @@ def test_stt_command_head(tmp_path: Path) -> None: def test_stt_command_missing_config() -> None: - """Test that stt command fails without required --config.""" - result = runner.invoke(app, ["stt"]) + """Test that add command fails without required --config.""" + result = runner.invoke(app, ["add", "--format", "stt", "test.csv"]) assert result.exit_code != 0 assert "Missing option" in result.stderr or "required" in result.stderr.lower() def test_stt_command_no_input_or_output_exits_with_error(tmp_path: Path) -> None: - """Test that stt command fails without --input or --output.""" + """Test that add command fails without input file argument.""" config_path = _write_config(tmp_path, timezone="ET") - result = runner.invoke(app, ["stt", "--config", str(config_path)]) - assert result.exit_code == 1 - assert "input" in result.output or "output" in result.output + result = runner.invoke( + app, ["add", "--config", str(config_path), "--format", "stt"] + ) + assert result.exit_code != 0 def test_stt_command_timezone_conversion(tmp_path: Path) -> None: @@ -600,7 +641,7 @@ def test_stt_command_timezone_conversion(tmp_path: Path) -> None: csv_path.write_text(stt_csv, encoding="utf-8") config_path = _write_config(tmp_path, timezone="PT") result = runner.invoke( - app, ["stt", "--config", str(config_path), "--input", str(csv_path)] + app, ["add", "--config", str(config_path), "--format", "stt", str(csv_path)] ) assert result.exit_code == 0 # PT in January is UTC-8: 09:00Z -> 01:00 PT @@ -608,11 +649,12 @@ def test_stt_command_timezone_conversion(tmp_path: Path) -> None: def test_stt_output_empty_db(tmp_path: Path) -> None: - """Test stt --output with an empty database writes header-only CSV.""" + """Test export with stt format and empty database writes header-only CSV.""" config_path = _write_config(tmp_path, timezone="ET") output_path = tmp_path / "stt_output.csv" result = runner.invoke( - app, ["stt", "--config", str(config_path), "--output", str(output_path)] + app, + ["export", "--config", str(config_path), "--format", "stt", str(output_path)], ) assert result.exit_code == 0 assert "Database is empty" in result.output @@ -622,7 +664,7 @@ def test_stt_output_empty_db(tmp_path: Path) -> None: def test_stt_output_with_data(tmp_path: Path) -> None: - """Test stt --output exports a previously imported database.""" + """Test export with stt format exports a previously imported database.""" stt_csv = ( '"activity name","time started","time ended","comment","categories",' '"record tags","duration","duration minutes"\n' @@ -633,12 +675,13 @@ def test_stt_output_with_data(tmp_path: Path) -> None: csv_path.write_text(stt_csv, encoding="utf-8") config_path = _write_config(tmp_path, timezone="ET") result = runner.invoke( - app, ["stt", "--config", str(config_path), "--input", str(csv_path)] + app, ["add", "--config", str(config_path), "--format", "stt", str(csv_path)] ) assert result.exit_code == 0 output_path = tmp_path / "stt_output.csv" result = runner.invoke( - app, ["stt", "--config", str(config_path), "--output", str(output_path)] + app, + ["export", "--config", str(config_path), "--format", "stt", str(output_path)], ) assert result.exit_code == 0 assert output_path.exists() @@ -648,7 +691,7 @@ def test_stt_output_with_data(tmp_path: Path) -> None: def test_stt_output_combined_with_input(tmp_path: Path) -> None: - """Test using stt --output together with --input.""" + """Test using add and export commands together.""" stt_csv = ( '"activity name","time started","time ended","comment","categories",' '"record tags","duration","duration minutes"\n' @@ -659,20 +702,33 @@ def test_stt_output_combined_with_input(tmp_path: Path) -> None: csv_path.write_text(stt_csv, encoding="utf-8") config_path = _write_config(tmp_path, timezone="ET") output_path = tmp_path / "stt_output.csv" + # First add the data result = runner.invoke( app, [ - "stt", + "add", "--config", str(config_path), - "--input", + "--format", + "stt", str(csv_path), - "--output", - str(output_path), ], ) assert result.exit_code == 0 assert "Loaded DataFrame" in result.output + # Then export the data + result = runner.invoke( + app, + [ + "export", + "--config", + str(config_path), + "--format", + "stt", + str(output_path), + ], + ) + assert result.exit_code == 0 assert "Exporting" in result.output assert output_path.exists() content = output_path.read_text(encoding="utf-8") @@ -691,7 +747,7 @@ def test_stt_command_invalid_csv(tmp_path: Path) -> None: csv_path.write_text(bad_csv, encoding="utf-8") config_path = _write_config(tmp_path, timezone="ET") result = runner.invoke( - app, ["stt", "--config", str(config_path), "--input", str(csv_path)] + app, ["add", "--config", str(config_path), "--format", "stt", str(csv_path)] ) assert result.exit_code != 0 @@ -771,3 +827,34 @@ def test_format_simple_csv_non_list_categories(tmp_path: Path) -> None: content = output_path.read_text(encoding="utf-8") assert "raw_category" in content assert "raw_tag" in content + + +def test_add_command_unknown_format(tmp_path: Path) -> None: + """Test add command with unknown format (lines 297-298).""" + csv_path = tmp_path / "test.csv" + csv_path.write_text(SAMPLE_CSV, encoding="utf-8") + config_path = _write_config(tmp_path, timezone="ET") + result = runner.invoke( + app, ["add", "--config", str(config_path), "--format", "unknown", str(csv_path)] + ) + assert result.exit_code == 1 + assert "Unknown format" in result.stderr or "Unknown format" in result.output + + +def test_export_command_unknown_format(tmp_path: Path) -> None: + """Test export command with unknown format (lines 347-348).""" + config_path = _write_config(tmp_path, timezone="ET") + output_path = tmp_path / "output.csv" + result = runner.invoke( + app, + [ + "export", + "--config", + str(config_path), + "--format", + "unknown", + str(output_path), + ], + ) + assert result.exit_code == 1 + assert "Unknown format" in result.stderr or "Unknown format" in result.output diff --git a/tests/test_database.py b/tests/test_database.py index 4ce7ca7..9360a1a 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -404,9 +404,10 @@ def test_database_write_with_update(tmp_path: Path) -> None: def test_merge_conflict_error() -> None: - """Test MergeConflictError creation (lines 38-40).""" + """Test MergeConflictError class (kept for backward compatibility).""" from timetracker_utils.database import MergeConflictError + # Test that the class can still be instantiated (for backward compatibility) err = MergeConflictError( "test conflict", [{"date": "1/15/2200", "notes": "conflict"}], @@ -414,3 +415,271 @@ def test_merge_conflict_error() -> None: assert str(err) == "test conflict" assert len(err.conflicts) == 1 assert err.conflicts[0]["date"] == "1/15/2200" + + +def test_activity_entry_parse_datetime_none() -> None: + """Test ActivityEntry.parse_datetime with None returns None (line 45).""" + result = ActivityEntry.parse_datetime(None) + assert result is None + + +def test_activity_entry_parse_datetime_empty_string() -> None: + """Test ActivityEntry.parse_datetime with empty string returns None (line 45).""" + result = ActivityEntry.parse_datetime("") + assert result is None + + +def test_activity_entry_parse_datetime_datetime_object() -> None: + """Test ActivityEntry.parse_datetime with datetime object (line 46-47).""" + dt = datetime(2200, 1, 15, 9, 0, 0) + result = ActivityEntry.parse_datetime(dt) + assert result == dt + + +def test_activity_entry_parse_datetime_valid_string() -> None: + """Test ActivityEntry.parse_datetime with valid ISO string (lines 48-51).""" + result = ActivityEntry.parse_datetime("2200-01-15T09:00:00") + assert result == datetime(2200, 1, 15, 9, 0, 0) + + +def test_activity_entry_parse_datetime_z_string() -> None: + """Test ActivityEntry.parse_datetime with Z suffix (line 50).""" + from datetime import timezone + + result = ActivityEntry.parse_datetime("2200-01-15T09:00:00Z") + # Z suffix creates UTC datetime + assert result == datetime(2200, 1, 15, 9, 0, 0, tzinfo=timezone.utc) + + +def test_activity_entry_parse_datetime_invalid_string() -> None: + """Test ActivityEntry.parse_datetime with invalid string raises ValueError.""" + with pytest.raises(ValueError, match="Invalid datetime value"): + ActivityEntry.parse_datetime("not-a-datetime") + + +def test_activity_entry_parse_datetime_invalid_type() -> None: + """Test ActivityEntry.parse_datetime with invalid type raises TypeError.""" + with pytest.raises(TypeError, match="Invalid datetime type"): + ActivityEntry.parse_datetime(12345) + + +def test_try_json_loads_valid_json() -> None: + """Test _try_json_loads with valid JSON returns True.""" + from timetracker_utils.database import _try_json_loads + + assert _try_json_loads('["cat1", "cat2"]') is True + + +def test_try_json_loads_invalid_json() -> None: + """Test _try_json_loads with invalid JSON returns False (lines 110-111).""" + from timetracker_utils.database import _try_json_loads + + assert _try_json_loads("not-json") is False + + +def test_try_json_loads_value_error() -> None: + """Test _try_json_loads with value error returns False (line 110).""" + from timetracker_utils.database import _try_json_loads + + assert _try_json_loads("") is False + + +def test_try_json_loads_type_error() -> None: + """Test _try_json_loads with type error returns False (line 110).""" + from timetracker_utils.database import _try_json_loads + + assert _try_json_loads(None) is False + + +def test_format_row_for_display() -> None: + """Test _format_row_for_display formats row correctly (lines 115-127).""" + from timetracker_utils.database import _format_row_for_display + + row = { + "date": "1/15/2200", + "activity": "Test", + "start_time": "2200-01-15T09:00:00", + "end_time": "2200-01-15T10:00:00", + "notes": "note", + "categories": ["cat1"], + "tags": ["tag1"], + } + result = _format_row_for_display(row) + assert "date='1/15/2200'" in result + assert "activity='Test'" in result + + +def test_database_is_blank_fill_new_is_na() -> None: + """Test _is_blank_fill returns False when new value is NaN (line 296-297).""" + import numpy as np + + old = pd.Series( + {"date": "1/15/2200", "notes": "existing", "categories": [], "tags": []} + ) + new = pd.Series( + {"date": "1/15/2200", "notes": np.nan, "categories": [], "tags": []} + ) + assert not Database._is_blank_fill(old, new) + + +def test_database_merge_nan_key_value() -> None: + """Test _merge_dataframes handles NaN in key columns (line 327).""" + import numpy as np + + existing = pd.DataFrame( + { + "date": ["1/15/2200"], + "activity": ["A"], + "start_time": ["09:00"], + "end_time": ["10:00"], + "notes": [""], + "categories": [""], + "tags": [""], + } + ) + incoming = pd.DataFrame( + { + "date": ["1/16/2200"], + "activity": [np.nan], + "start_time": ["09:00"], + "end_time": ["10:00"], + "notes": [""], + "categories": [""], + "tags": [""], + } + ) + _result, new_count, _skipped, _updated = Database._merge_dataframes( + existing, incoming, 100 + ) + assert new_count == 1 + + +def test_database_merge_new_rows_added() -> None: + """Test _merge_dataframes adds new rows when no match (lines 348-352).""" + existing = pd.DataFrame( + { + "date": ["1/15/2200"], + "activity": ["A"], + "start_time": ["09:00"], + "end_time": ["10:00"], + "notes": [""], + "categories": [""], + "tags": [""], + } + ) + incoming = pd.DataFrame( + { + "date": ["1/16/2200"], + "activity": ["B"], + "start_time": ["09:00"], + "end_time": ["10:00"], + "notes": [""], + "categories": [""], + "tags": [""], + } + ) + result, new_count, _skipped, _updated = Database._merge_dataframes( + existing, incoming, 100 + ) + assert new_count == 1 + assert len(result) == 2 + + +def test_database_merge_unresolved_treats_as_new() -> None: + """Test _merge_dataframes treats unresolved as new (lines 370-373).""" + existing = pd.DataFrame( + { + "date": ["1/15/2200"], + "activity": ["A"], + "start_time": ["09:00"], + "end_time": ["10:00"], + "notes": ["old"], + "categories": ["cat1"], + "tags": [""], + } + ) + incoming = pd.DataFrame( + { + "date": [""], # blank - no non-blank mergeable values + "activity": ["A"], + "start_time": ["09:00"], # Same key + "end_time": ["10:00"], + "notes": [""], # blank + "categories": [""], # blank + "tags": [""], # blank + } + ) + # This will treat as new since incoming has no non-blank mergeable values + # and rows are not identical (existing has non-blank values) + result, new_count, skipped, updated = Database._merge_dataframes( + existing, incoming, 100 + ) + # Should treat as new since no non-blank values to update + assert new_count == 1 + assert updated == 0 + assert skipped == 0 + assert len(result) == 2 + + +def test_database_merge_conflict_updates() -> None: + """Test _merge_dataframes updates with new values on conflict.""" + existing = pd.DataFrame( + { + "date": ["1/15/2200"], + "activity": ["A"], + "start_time": ["09:00"], + "end_time": ["10:00"], + "notes": ["note1"], + "categories": ["cat1"], + "tags": ["tag1"], + } + ) + incoming = pd.DataFrame( + { + "date": ["1/15/2200"], + "activity": ["A"], + "start_time": ["09:00"], + "end_time": ["10:00"], + "notes": ["note2"], + "categories": ["cat2"], + "tags": ["tag2"], + } + ) + # The implementation updates with new values rather than raising conflicts + result, _new_count, _skipped, updated = Database._merge_dataframes( + existing, incoming, 100 + ) + # It updates with new non-blank values + assert updated == 1 + assert result.iloc[0]["notes"] == "note2" + + +def test_database_merge_concat_new_rows() -> None: + """Test _merge_dataframes concatenates new rows (lines 407-408).""" + existing = pd.DataFrame( + { + "date": ["1/15/2200"], + "activity": ["A"], + "start_time": ["09:00"], + "end_time": ["10:00"], + "notes": [""], + "categories": [""], + "tags": [""], + } + ) + incoming = pd.DataFrame( + { + "date": ["1/16/2200", "1/17/2200"], + "activity": ["B", "C"], + "start_time": ["09:00", "10:00"], + "end_time": ["10:00", "11:00"], + "notes": ["", ""], + "categories": ["", ""], + "tags": ["", ""], + } + ) + result, new_count, _skipped, _updated = Database._merge_dataframes( + existing, incoming, 100 + ) + assert new_count == 2 + assert len(result) == 3 diff --git a/tests/test_simple_time_tracker.py b/tests/test_simple_time_tracker.py index b75f5dc..f8269de 100644 --- a/tests/test_simple_time_tracker.py +++ b/tests/test_simple_time_tracker.py @@ -165,8 +165,66 @@ def test_parse_datetime_naive_no_tz() -> None: duration="1:00:00", duration_minutes="60", ) - # Naive STT timestamps should remain naive - assert entry.start_time.tzinfo is None + # Naive timestamps should be converted to UTC using the default timezone (UTC) + assert entry.start_time.tzinfo is not None + assert entry.start_time.tzinfo == timezone.utc + + +def test_parse_datetime_naive_zone_none() -> None: + """Test parse_datetime with naive datetime when resolve_tz returns None (line 99, 110).""" + from datetime import datetime + from unittest.mock import patch + + # Test with naive datetime string (line 99) + with patch("timetracker_utils.datetime_utils.resolve_tz", return_value=None): + entry = SimpleTimeEntry.model_validate( + { + "activity": "Test", + "start_time": "2200-01-15T09:00:00", + "end_time": "2200-01-15T10:00:00", + "duration": "1:00:00", + "duration_minutes": "60", + }, + context={"default_timezone": "ET"}, + ) + # Should still work, falling back to UTC + assert entry.start_time.tzinfo is not None + assert entry.start_time.tzinfo == timezone.utc + + # Test with naive datetime object (line 110) + with patch("timetracker_utils.datetime_utils.resolve_tz", return_value=None): + entry2 = SimpleTimeEntry.model_validate( + { + "activity": "Test", + "start_time": datetime(2200, 1, 15, 9, 0, 0), + "end_time": "2200-01-15T10:00:00.000Z", + "duration": "1:00:00", + "duration_minutes": "60", + }, + context={"default_timezone": "ET"}, + ) + assert entry2.start_time.tzinfo is not None + + +def test_parse_datetime_naive_zone_not_none() -> None: + """Test parse_datetime with naive datetime when resolve_tz returns zone (lines 94-99).""" + from datetime import datetime, timedelta, timezone + from unittest.mock import patch + + test_zone = timezone(timedelta(hours=-5)) + with patch("timetracker_utils.datetime_utils.resolve_tz", return_value=test_zone): + entry = SimpleTimeEntry.model_validate( + { + "activity": "Test", + "start_time": datetime(2200, 1, 15, 9, 0, 0), # naive datetime object + "end_time": "2200-01-15T10:00:00.000Z", + "duration": "1:00:00", + "duration_minutes": "60", + }, + context={"default_timezone": "ET"}, + ) + # Should convert to UTC + assert entry.start_time.tzinfo is not None def test_parse_datetime_explicit_timezone() -> None: @@ -182,7 +240,7 @@ def test_parse_datetime_explicit_timezone() -> None: def test_parse_datetime_invalid_raises() -> None: - """Test parse_datetime with invalid string raises ValueError.""" + """Test parse_datetime with invalid string raises ValueError (line 113).""" with pytest.raises(ValidationError, match="Invalid datetime"): SimpleTimeEntry( activity="Test", @@ -222,8 +280,8 @@ def test_coerce_duration_minutes_unexpected_type() -> None: assert result is None -def test_validate_duration_crosscheck_empty_duration_none_minutes() -> None: - """Test crosscheck with empty duration_str and None duration_minutes (line 99).""" +def test_parse_hms_to_minutes_empty_string() -> None: + """Test _parse_hms_to_minutes with empty string returns None.""" result = SimpleTimeEntry._parse_hms_to_minutes("") assert result is None @@ -385,20 +443,18 @@ def test_post_process_drops_validation_cols() -> None: assert "duration_minutes" not in tracker.entries.columns -def test_validate_duration_crosscheck_zero_duration_no_minutes() -> None: - """Test crosscheck returns self when no duration_str and no duration_minutes.""" - with pytest.raises(ValidationError, match="Field required"): - SimpleTimeEntry( - activity="Test", - start_time="2200-01-15T09:00:00.000Z", - end_time="2200-01-15T10:00:00.000Z", - ) - - -def test_parse_hms_to_minutes_empty() -> None: - """Test _parse_hms_to_minutes with empty string.""" - result = SimpleTimeEntry._parse_hms_to_minutes("") - assert result is None +def test_validate_duration_crosscheck_empty_duration_none_minutes() -> None: + """Test crosscheck with empty duration_str and None duration_minutes (line 122).""" + entry = SimpleTimeEntry( + activity="Test", + start_time="2200-01-15T09:00:00.000Z", + end_time="2200-01-15T10:00:00.000Z", + duration="", + duration_minutes=None, + ) + # Should return self without error + assert entry.duration_str == "" + assert entry.duration_minutes is None def test_parse_hms_to_minutes_invalid() -> None: diff --git a/timetracker_utils/base_tracker.py b/timetracker_utils/base_tracker.py index 78f9e90..2ea0b0e 100644 --- a/timetracker_utils/base_tracker.py +++ b/timetracker_utils/base_tracker.py @@ -13,7 +13,14 @@ from typing import Any, ClassVar import pandas as pd -from pydantic import AliasChoices, BaseModel, Field, field_validator, model_validator +from pydantic import ( + AliasChoices, + BaseModel, + Field, + ValidationInfo, + field_validator, + model_validator, +) warnings.filterwarnings( "ignore", @@ -114,24 +121,57 @@ def validate_end_time_and_hours(self) -> "BaseTimeEntry": @field_validator("start_time", "end_time", mode="before") @classmethod - def parse_datetime(cls, value: str | None) -> datetime | None: + def parse_datetime( + cls, value: str | None, info: ValidationInfo | None = None + ) -> datetime | None: """Parse a datetime string or return None for empty values. - Timezone policy: Naive datetime objects are treated as UTC by adding - timezone.utc. Timezone-aware datetimes are converted to UTC. This ensures - consistent UTC representation in the database and merge operations. + Timezone policy: If the datetime string has an explicit timezone + (Z, +00:00, -04:00, etc.), it is converted to UTC. If no timezone is + specified, the timezone from the config is assumed and then converted + to UTC. This ensures consistent UTC representation in the database. + + Args: + value: The datetime string or datetime object to parse. + info: Pydantic validation info containing context data. + + Returns: + A timezone-aware datetime in UTC, or None for empty values. + """ if value is None or value == "": return None if isinstance(value, datetime): if value.tzinfo is None: + # Naive datetime: assume config timezone and convert to UTC + default_tz = ( + info.context.get("default_timezone", "UTC") + if info and info.context + else "UTC" + ) + from timetracker_utils.datetime_utils import resolve_tz + + zone = resolve_tz(default_tz) + if zone is not None: + return value.replace(tzinfo=zone).astimezone(timezone.utc) return value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc) try: dt = datetime.fromisoformat(value.replace("Z", "+00:00")) if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) - return dt + # Naive datetime string: assume config timezone and convert to UTC + default_tz = ( + info.context.get("default_timezone", "UTC") + if info and info.context + else "UTC" + ) + from timetracker_utils.datetime_utils import resolve_tz + + zone = resolve_tz(default_tz) + if zone is not None: + return dt.replace(tzinfo=zone).astimezone(timezone.utc) + return dt.replace(tzinfo=timezone.utc) + return dt.astimezone(timezone.utc) except (ValueError, TypeError) as exc: msg = f"Invalid datetime value: {value!r}" raise ValueError(msg) from exc @@ -205,9 +245,16 @@ class BaseTimeTracker: _ENTRY_CLASS: ClassVar[type[BaseTimeEntry]] = BaseTimeEntry _GROUPBY_FIELD: ClassVar[str] = "activity" - def __init__(self) -> None: - """Initialize the tracker with an empty DataFrame.""" + def __init__(self, default_timezone: str = "UTC") -> None: + """Initialize the tracker with an empty DataFrame. + + Args: + default_timezone: The timezone to assume for naive datetime strings + (e.g., "ET", "PT", "UTC"). Defaults to "UTC". + + """ self.entries: pd.DataFrame = pd.DataFrame() + self.default_timezone: str = default_timezone def read_csv(self, path: str | Path) -> pd.DataFrame: """Read a CSV file and return a DataFrame of parsed time entries.""" @@ -241,7 +288,12 @@ def read_csv_string(self, csv_data: str) -> pd.DataFrame: "Extra columns in CSV that will be ignored: %s", sorted(extra_cols), ) - validated_entries = [self._ENTRY_CLASS.model_validate(row) for row in reader] + validated_entries = [ + self._ENTRY_CLASS.model_validate( + row, context={"default_timezone": self.default_timezone} + ) + for row in reader + ] if validated_entries: self.entries = pd.DataFrame( [entry.model_dump() for entry in validated_entries] diff --git a/timetracker_utils/cli.py b/timetracker_utils/cli.py index b202de3..b2a0150 100644 --- a/timetracker_utils/cli.py +++ b/timetracker_utils/cli.py @@ -253,28 +253,25 @@ def _compute_hours(start_time: object, end_time: object) -> str: @app.command() -def timecop( +def add( config: Path = typer.Option( ..., "--config", "-c", help="Path to the YAML configuration file." ), - input: Path = typer.Option( - None, "--input", "-i", help="Path to the CSV file to load." + format: str = typer.Option( + ..., "--format", "-f", help="Format of the input file (timecop or stt)." ), + input_file: Path = typer.Argument(..., help="Path to the CSV file to load."), head: int = typer.Option(100, "--head", "-h", help="Rows to display."), - output: Path = typer.Option( - None, "--output", "-o", help="Path to export database as TimeCop CSV." - ), ) -> None: - """Load, display, and export TimeCop CSV data.""" + """Load CSV data into the database.""" logging.basicConfig(level=logging.INFO, format="%(message)s") cfg = load_config(config) - if input is not None: + if format == "timecop": try: - cop = TimeCop() - cop.read_csv(input) + cop = TimeCop(default_timezone=cfg.timezone) + cop.read_csv(input_file) except ValueError as exc: - # Emit a clear error message for the user and exit with non-zero code. typer.echo(str(exc), err=True) raise typer.Exit(code=1) from exc db = Database() @@ -283,105 +280,71 @@ def timecop( ) typer.echo(f"Loaded DataFrame ({len(cop.entries)} rows total):") display_df = cop.entries.copy() - if not display_df.empty: - if "start_time" in display_df.columns: - display_df["start_time"] = convert_column_tz( - display_df["start_time"], cfg.timezone - ) - if "end_time" in display_df.columns: - display_df["end_time"] = convert_column_tz( - display_df["end_time"], cfg.timezone - ) - with pd.option_context( - "display.max_columns", - None, - "display.max_colwidth", - None, - "display.width", - None, - ): - typer.echo(str(display_df.head(head))) - - if output is not None: + elif format == "stt": + try: + tracker = SimpleTimeTracker(default_timezone=cfg.timezone) + tracker.read_csv(input_file) + except ValueError as exc: + typer.echo(str(exc), err=True) + raise typer.Exit(code=1) from exc db = Database() - db_entries = db.read(cfg.database) - if db_entries.empty: - typer.echo("Database is empty, writing header-only CSV.") - else: - typer.echo(f"Exporting {len(db_entries)} entries to {output}") - _format_timecop_csv(db_entries, output, cfg.timezone) - - if input is None and output is None: - typer.echo( - "No --input or --output specified.", - err=True, + db.write( + tracker.entries, cfg.database, max_conflict_display=cfg.max_conflict_display ) + typer.echo(f"Loaded DataFrame ({len(tracker.entries)} rows total):") + display_df = tracker.entries.copy() + else: + typer.echo(f"Unknown format: {format}. Use 'timecop' or 'stt'.", err=True) raise typer.Exit(code=1) + if not display_df.empty: + if "start_time" in display_df.columns: + display_df["start_time"] = convert_column_tz( + display_df["start_time"], cfg.timezone + ) + if "end_time" in display_df.columns: + display_df["end_time"] = convert_column_tz( + display_df["end_time"], cfg.timezone + ) + with pd.option_context( + "display.max_columns", + None, + "display.max_colwidth", + None, + "display.width", + None, + ): + typer.echo(str(display_df.head(head))) + @app.command() -def stt( +def export( config: Path = typer.Option( ..., "--config", "-c", help="Path to the YAML configuration file." ), - input: Path = typer.Option( - None, "--input", "-i", help="Path to the STT-format CSV file to load." - ), - head: int = typer.Option(100, "--head", "-h", help="Rows to display."), - output: Path = typer.Option( - None, "--output", "-o", help="Path to export database as STT-format CSV." + format: str = typer.Option( + ..., "--format", "-f", help="Format of the output file (timecop or stt)." ), + output_file: Path = typer.Argument(..., help="Path to export database as CSV."), ) -> None: - """Load, display, and export SimpleTimeTracker CSV data.""" + """Export database data to CSV.""" logging.basicConfig(level=logging.INFO, format="%(message)s") cfg = load_config(config) - if input is not None: - try: - tracker = SimpleTimeTracker() - tracker.read_csv(input) - except ValueError as exc: - typer.echo(str(exc), err=True) - raise typer.Exit(code=1) from exc - db = Database() - db.write( - tracker.entries, cfg.database, max_conflict_display=cfg.max_conflict_display - ) - typer.echo(f"Loaded DataFrame ({len(tracker.entries)} rows total):") - display_df = tracker.entries.copy() - if not display_df.empty: - if "start_time" in display_df.columns: - display_df["start_time"] = convert_column_tz( - display_df["start_time"], cfg.timezone - ) - if "end_time" in display_df.columns: - display_df["end_time"] = convert_column_tz( - display_df["end_time"], cfg.timezone - ) - with pd.option_context( - "display.max_columns", - None, - "display.max_colwidth", - None, - "display.width", - None, - ): - typer.echo(str(display_df.head(head))) - - if output is not None: - db = Database() - db_entries = db.read(cfg.database) - if db_entries.empty: - typer.echo("Database is empty, writing header-only CSV.") - else: - typer.echo(f"Exporting {len(db_entries)} entries to {output}") - _format_simple_csv(db_entries, output, cfg.timezone) + db = Database() + db_entries = db.read(cfg.database) - if input is None and output is None: - typer.echo( - "No --input or --output specified.", - err=True, - ) + if db_entries.empty: + typer.echo("Database is empty, writing header-only CSV.") + else: + typer.echo(f"Exporting {len(db_entries)} entries to {output_file}") + + if format == "timecop": + _format_timecop_csv(db_entries, output_file, cfg.timezone) + elif format == "stt": + _format_simple_csv(db_entries, output_file, cfg.timezone) + else: + typer.echo(f"Unknown format: {format}. Use 'timecop' or 'stt'.", err=True) raise typer.Exit(code=1) diff --git a/timetracker_utils/database.py b/timetracker_utils/database.py index 50a53a6..6c22064 100644 --- a/timetracker_utils/database.py +++ b/timetracker_utils/database.py @@ -102,8 +102,8 @@ def _deserialise_lists(df: pd.DataFrame) -> pd.DataFrame: return df -def _try_json_loads(value: str) -> bool: - """Safely test if a string can be parsed as JSON.""" +def _try_json_loads(value: Any) -> bool: + """Safely test if a value can be parsed as JSON.""" try: json.loads(value) return True @@ -314,7 +314,7 @@ def _is_conflict(old_row: pd.Series, new_row: pd.Series) -> bool: def _merge_dataframes( existing: pd.DataFrame, incoming: pd.DataFrame, - max_conflict_display: int, + _max_conflict_display: int, ) -> tuple[pd.DataFrame, int, int, int]: if incoming.empty: return existing, 0, 0, 0 @@ -335,7 +335,6 @@ def _make_key(row: pd.Series) -> str: incoming["_merge_key"] = incoming.apply(_make_key, axis=1) new_rows: list[pd.DataFrame] = [] - conflicts: list[dict[str, Any]] = [] new_count = 0 skipped_count = 0 updated_count = 0 @@ -374,34 +373,7 @@ def _make_key(row: pd.Series) -> str: new_rows.append( incoming.iloc[[inc_idx]].drop(columns=["_merge_key"]) # type: ignore[index] ) - if conflicts: - seen_keys: set[str] = set() - unique_conflicts: list[dict[str, Any]] = [] - for c in conflicts: - key = "|".join(str(c.get(col, "")) for col in _MERGE_KEY_COLUMNS) - if key not in seen_keys: - seen_keys.add(key) - unique_conflicts.append(c) - display_conflicts = ( - unique_conflicts[:max_conflict_display] - if max_conflict_display > 0 - else [] - ) - total_conflicts = len(unique_conflicts) - conflict_msgs = [_format_row_for_display(c) for c in display_conflicts] - conflict_detail = chr(10).join(conflict_msgs) - if total_conflicts > max_conflict_display > 0: - remaining = total_conflicts - max_conflict_display - conflict_detail += f"{chr(10)} ... and {remaining} more conflicts." - suffix = "y" if total_conflicts == 1 else "ies" - prefix = "y has" if total_conflicts == 1 else "ies have" - msg = ( - f"Merge conflict detected for {total_conflicts} entr{suffix}. " - f"The following entr{prefix} the same " - "start_time, end_time, and activity " - f"but conflicting non-blank values:{chr(10)}{conflict_detail}" - ) - raise MergeConflictError(msg, unique_conflicts) + new_count += 1 result = existing.drop(columns=["_merge_key"]) if new_rows: new_concat = pd.concat(new_rows, ignore_index=True) diff --git a/timetracker_utils/simple_time_tracker.py b/timetracker_utils/simple_time_tracker.py index 1649ea0..7fb875b 100644 --- a/timetracker_utils/simple_time_tracker.py +++ b/timetracker_utils/simple_time_tracker.py @@ -7,11 +7,11 @@ import csv import io import logging -from datetime import datetime +from datetime import datetime, timezone from typing import Any, ClassVar, cast import pandas as pd -from pydantic import Field, field_validator, model_validator +from pydantic import Field, ValidationInfo, field_validator, model_validator from timetracker_utils.base_tracker import BaseTimeEntry, BaseTimeTracker @@ -79,27 +79,49 @@ def parse_duration_hms(cls, value: Any) -> str: @field_validator("start_time", "end_time", mode="before") @classmethod - def parse_datetime(cls, value: str | None) -> datetime | None: + def parse_datetime( + cls, value: str | None, info: ValidationInfo | None = None + ) -> datetime | None: """Parse datetime from string or return existing datetime. - Timezone policy: For Simple Time Tracker format, if no explicit timezone - is specified in the input string, the datetime is kept as naive (treated - as local/config timezone). If an explicit timezone is present (e.g., 'Z', - '+00:00', or offset), it is preserved. This differs from BaseTimeEntry - which normalizes all datetimes to UTC. + Timezone policy: If the datetime string has an explicit timezone + (Z, +00:00, -04:00, etc.), it is converted to UTC. If no timezone is + specified, the timezone from the config is assumed and then converted + to UTC. This ensures consistent UTC representation in the database. """ if value is None or value == "": return None if isinstance(value, datetime): - return value + if value.tzinfo is None: + # Naive datetime: assume config timezone and convert to UTC + default_tz = ( + info.context.get("default_timezone", "UTC") + if info and info.context + else "UTC" + ) + from timetracker_utils.datetime_utils import resolve_tz + + zone = resolve_tz(default_tz) + if zone is not None: + return value.replace(tzinfo=zone).astimezone(timezone.utc) + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) try: dt = datetime.fromisoformat(value.replace("Z", "+00:00")) - # For STT format: if no timezone is specified, keep the value - # as naive (treat it as local / config timezone). - has_explicit_tz = "Z" in value or (value.count("-") > 2 or "+" in value) - if not has_explicit_tz: - return dt.replace(tzinfo=None) - return dt + if dt.tzinfo is None: + # Naive datetime string: assume config timezone and convert to UTC + default_tz = ( + info.context.get("default_timezone", "UTC") + if info and info.context + else "UTC" + ) + from timetracker_utils.datetime_utils import resolve_tz + + zone = resolve_tz(default_tz) + if zone is not None: + return dt.replace(tzinfo=zone).astimezone(timezone.utc) + return dt.replace(tzinfo=timezone.utc) + return dt.astimezone(timezone.utc) except (ValueError, TypeError) as exc: msg = f"Invalid datetime value: {value!r}" raise ValueError(msg) from exc