personal_data/csv_import.py

@@ -36,15 +36,6 @@ 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,
 ) -> (
@@ -69,8 +60,6 @@ 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':
@@ -114,21 +103,13 @@ 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 is_duration_key(k,v)
+        if isinstance(v, Decimal) and 'duration_seconds' in k
     ]
     name_keys = [k for k, v in event_data.items() if isinstance(v, str)]
     image_keys = [
@@ -171,16 +152,12 @@ def start_end(
 
     if keys.time_start and keys.duration:
         start = sample[keys.time_start[0]]
-        duration = sample[keys.duration[0]]
-        if not isinstance(duration, datetime.timedelta):
-            duration = datetime.timedelta(seconds=float(duration))
+        duration = datetime.timedelta(seconds=float(sample[keys.duration[0]]))
         return (start, start + duration)
 
     if keys.time_end and keys.duration:
         end = sample[keys.time_end[0]]
-        duration = sample[keys.duration[0]]
-        if not isinstance(duration, datetime.timedelta):
-            duration = datetime.timedelta(seconds=float(duration))
+        duration = datetime.timedelta(seconds=float(sample[keys.duration[0]]))
         return (end - duration, end)
 
     if keys.time_start:

personal_data/fetchers/stepmania.py

@@ -1,65 +0,0 @@
-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()),
-            }