Improved type parsing

personal_data/csv_import.py

@@ -36,6 +36,15 @@ def try_value(fn: Callable[[str], T], s: str) -> T | None:
         return None
 
 
+def parse_timedelta(text: str) -> datetime.timedelta:
+    if t := try_value(lambda t: datetime.datetime.strptime(t, '%H:%M:%S.%f'), text):
+        return datetime.timedelta(hours=t.hour, minutes=t.minute, seconds=t.second)
+    elif t := try_value(lambda t: datetime.datetime.strptime(t, '%H:%M:%S'), text):
+        return datetime.timedelta(hours=t.hour, minutes=t.minute, seconds=t.second)
+    else:
+        return None
+
+
 def csv_str_to_value(
     s: str,
 ) -> (
@@ -60,6 +69,8 @@ def csv_str_to_value(
         return v_date
     if (v_datetime := try_value(datetime.datetime.fromisoformat, s)) is not None:
         return v_datetime
+    if (v_timedelta := parse_timedelta(s)) is not None:
+        return v_timedelta
     if s.startswith(('http://', 'https://')):
         return urllib.parse.urlparse(s)
     if s.lower() == 'false':
@@ -103,13 +114,21 @@ class PossibleKeys:
     misc: list[str]
 
 
+def is_duration_key(k,v):
+    if isinstance(v, Decimal) and 'duration_seconds' in k:
+        return True
+    if isinstance(v, datetime.timedelta) and 'duration' in k:
+        return True
+    return False
+
+
 def determine_possible_keys(event_data: dict[str, Any]) -> PossibleKeys:
     # Select data
     time_keys = [k for k, v in event_data.items() if isinstance(v, datetime.date)]
     duration_keys = [
         k
         for k, v in event_data.items()
-        if isinstance(v, Decimal) and 'duration_seconds' in k
+        if is_duration_key(k,v)
     ]
     name_keys = [k for k, v in event_data.items() if isinstance(v, str)]
     image_keys = [
@@ -152,12 +171,16 @@ def start_end(
 
     if keys.time_start and keys.duration:
         start = sample[keys.time_start[0]]
-        duration = datetime.timedelta(seconds=float(sample[keys.duration[0]]))
+        duration = sample[keys.duration[0]]
+        if not isinstance(duration, datetime.timedelta):
+            duration = datetime.timedelta(seconds=float(duration))
         return (start, start + duration)
 
     if keys.time_end and keys.duration:
         end = sample[keys.time_end[0]]
-        duration = datetime.timedelta(seconds=float(sample[keys.duration[0]]))
+        duration = sample[keys.duration[0]]
+        if not isinstance(duration, datetime.timedelta):
+            duration = datetime.timedelta(seconds=float(duration))
         return (end - duration, end)
 
     if keys.time_start:

personal_data/fetchers/stepmania.py

@@ -0,0 +1,65 @@
+import dataclasses
+import datetime
+import logging
+import datetime
+from collections.abc import Iterator, Mapping
+from decimal import Decimal
+from pathlib import Path
+import bs4
+import zoneinfo
+
+from personal_data.data import DeduplicateMode, Scraper
+
+from .. import secrets
+
+logger = logging.getLogger(__name__)
+
+STATS_FILE_PATH: Path = Path('/home/jmaa/.itgmania/Save/LocalProfiles/00000000/Stats.xml')
+
+@dataclasses.dataclass(frozen=True)
+class Stepmania(Scraper):
+    dataset_name = 'stepmania'
+    deduplicate_mode = DeduplicateMode.BY_ALL_COLUMNS
+    deduplicate_ignore_columns = []
+
+    def scrape(self) -> Iterator[Mapping[str, object]]:
+        timezone = zoneinfo.ZoneInfo(
+            'Europe/Copenhagen',
+        )  # TODO: Parameterize in an intelligent manner
+
+        print(STATS_FILE_PATH)
+
+        with open(STATS_FILE_PATH) as f:
+            soup = bs4.BeautifulSoup(f.read(), 'lxml-xml')
+
+        # Derp
+        for score in soup.select('SongScores Song HighScoreList HighScore'):
+            print(score.parent.parent)
+            song = score.parent.parent.parent
+            song_path = Path(song ['Dir'].removesuffix('/'))
+
+            disqualified = score.select_one('Disqualified').get_text().strip() != '0'
+            if disqualified:
+                logger.warning('Ignored disqualified')
+                continue
+
+            play_start = datetime.datetime.fromisoformat(score.select_one('DateTime').get_text())
+            play_start = play_start.replace(tzinfo=timezone).astimezone(datetime.UTC)
+
+            play_seconds = float(score.select_one('SurviveSeconds').get_text())
+
+            yield {
+                    'song.name': song_path.stem,
+                    'song.pack': song_path.parent.stem,
+                    'song.difficulty': score.parent.parent['Difficulty'],
+                    'song.grade': score.select_one('Grade').get_text(),
+                    'play.start': play_start,
+                    'play.duration': datetime.timedelta(seconds=play_seconds),
+                    'score.score': float(score.select_one('PercentDP').get_text()),
+                    'score.w1': int(score.select_one('W1').get_text()),
+                    'score.w2': int(score.select_one('W2').get_text()),
+                    'score.w3': int(score.select_one('W3').get_text()),
+                    'score.w4': int(score.select_one('W4').get_text()),
+                    'score.w5': int(score.select_one('W5').get_text()),
+                    'score.miss': int(score.select_one('Miss').get_text()),
+            }