version 0.33

src/plugin.py

@@ -7,7 +7,8 @@
 from collections import defaultdict
 
 from galaxy.api.plugin import Plugin, create_and_run_plugin
-from galaxy.api.types import Authentication, Game, NextStep, Achievement, UserPresence, PresenceState, SubscriptionGame, Subscription
+from galaxy.api.types import Authentication, Game, NextStep, Achievement, UserPresence, PresenceState, SubscriptionGame, \
+    Subscription, GameTime
 from galaxy.api.consts import Platform
 from galaxy.api.errors import ApplicationError, InvalidCredentials, UnknownError
 from galaxy.api.jsonrpc import InvalidParams
@@ -17,7 +18,7 @@
 from http_client import AuthenticatedHttpClient
 from psn_client import (
     CommunicationId, TitleId, TrophyTitles, UnixTimestamp, TrophyTitleInfo,
-    PSNClient, MAX_TITLE_IDS_PER_REQUEST
+    PSNClient, MAX_TITLE_IDS_PER_REQUEST, parse_timestamp, parse_play_duration
 )
 from typing import Dict, List, Set, Iterable, Tuple, Optional, Any, AsyncGenerator
 from version import __version__
@@ -43,6 +44,9 @@
 TROPHY_TITLE_INFO_INVALIDATION_PERIOD_SEC = 3600 * 24 * 7
 
 
+logger = logging.getLogger(__name__)
+
+
 class PSNPlugin(Plugin):
     def __init__(self, reader, writer, token):
         super().__init__(Platform.Psn, __version__, reader, writer, token)
@@ -258,6 +262,19 @@ def handle_error(error_):
             logging.exception("Unhandled exception. Please report it to the plugin developers")
             handle_error(UnknownError())
 
+    async def prepare_game_times_context(self, game_ids: List[str]) -> Any:
+        return {game['titleId']: game for game in await self._psn_client.async_get_played_games()}
+
+    async def get_game_time(self, game_id: str, context: Any) -> GameTime:
+        time_played, last_played_game = None, None
+        try:
+            game = context[game_id]
+            last_played_game = parse_timestamp(game['lastPlayedDateTime'])
+            time_played = parse_play_duration(game['playDuration'])
+        except KeyError as e:
+            logger.debug(e)
+        return GameTime(game_id, time_played, last_played_game)
+
     async def prepare_user_presence_context(self, user_ids: List[str]) -> Any:
         return await self._psn_client.async_get_friends_presences()
 

src/psn_client.py

@@ -1,15 +1,18 @@
 import asyncio
 import logging
-from datetime import datetime, timezone
+import re
+import math
+from datetime import datetime, timezone, timedelta
 from functools import partial
-from typing import Dict, List, NewType, Tuple, NamedTuple
+from typing import Dict, List, NewType, Optional, Tuple, NamedTuple
 
 from galaxy.api.errors import UnknownBackendResponse
 from galaxy.api.types import Achievement, Game, LicenseInfo, UserInfo, UserPresence, PresenceState, SubscriptionGame
 from galaxy.api.consts import LicenseType
 from http_client import paginate_url
 from parsers import PSNGamesParser
 
+
 # game_id_list is limited to 5 IDs per request
 GAME_DETAILS_URL = "https://pl-tpy.np.community.playstation.net/trophy/v1/apps/trophyTitles" \
     "?npTitleIds={game_id_list}" \
@@ -24,6 +27,14 @@
     "&ih=240"\
     "&fields=@default"
 
+PLAYED_GAME_LIST_URL = "https://gamelist.api.playstation.com/v2/users/{user_id}/titles" \
+    "?type=owned,played" \
+    "&app=richProfile" \
+    "&sort=-lastPlayedDate" \
+    "&iw=240"\
+    "&ih=240"\
+    "&fields=@default"
+
 TROPHY_TITLES_URL = "https://pl-tpy.np.community.playstation.net/trophy/v1/trophyTitles" \
     "?fields=@default" \
     "&platform=PS4" \
@@ -68,11 +79,32 @@ class TrophyTitleInfo(NamedTuple):
 
 
 def parse_timestamp(earned_date) -> UnixTimestamp:
-    dt = datetime.strptime(earned_date, "%Y-%m-%dT%H:%M:%SZ")
+    date_format = "%Y-%m-%dT%H:%M:%S.%fZ" if '.' in earned_date else "%Y-%m-%dT%H:%M:%SZ"
+    dt = datetime.strptime(earned_date, date_format)
     dt = datetime.combine(dt.date(), dt.time(), timezone.utc)
     return UnixTimestamp(int(dt.timestamp()))
 
 
+def parse_play_duration(duration: Optional[str]) -> int:
+    """Returns time of played game in minutes from PSN API format `PT{HOURS}H{MINUTES}M{SECONDS}S`. Example: `PT2H33M3S`"""
+    if not duration:
+        raise UnknownBackendResponse(f'nullable playtime duration: {type(duration)}')
+    try:
+        result = re.match(
+            r'(?:PT)?'
+            r'(?:(?P<hours>\d*)H)?'
+            r'(?:(?P<minutes>\d*)M)?'
+            r'(?:(?P<seconds>\d*)S)?$',
+            duration
+        )
+        mapped_result = {k: float(v) for k, v in result.groupdict(0).items()}
+        time = timedelta(**mapped_result)
+    except (ValueError, AttributeError, TypeError):
+        raise UnknownBackendResponse(f'Unmatchable gametime: {duration}')
+
+    total_minutes = math.ceil(time.seconds / 60 + time.days * 24 * 60)
+    return total_minutes
+
 def date_today():
     return datetime.today()
 
@@ -293,3 +325,9 @@ def friends_with_presence_parser(response):
 
     async def get_subscription_games(self) -> List[SubscriptionGame]:
         return await self.fetch_data(PSNGamesParser().parse, PSN_PLUS_SUBSCRIPTIONS_URL, get_json=False, silent=True)
+
+    async def async_get_played_games(self):
+        def get_games_parser(response):
+            return response.get('titles', [])
+
+        return await self.fetch_paginated_data(get_games_parser, PLAYED_GAME_LIST_URL.format(user_id="me"), 'totalItemCount')

src/version.py

@@ -1,6 +1,9 @@
-__version__ = "0.32"
+__version__ = "0.33"
 
 __changelog__ = {
+    "0.33": """
+        - Add game time feature 
+    """,
     "0.32": """
         - Fix showing correct PS Plus monthly games in Galaxy Subscription tab
     """,

tests/integration_test.py

@@ -49,7 +49,8 @@ def test_integration():
                 'ImportUserPresence',
                 'ImportSubscriptions',
                 'ImportSubscriptionGames',
-                'ImportFriends'
+                'ImportFriends',
+                'ImportGameTime'
             ])
     assert response["result"]["token"] == token
 

tests/test_game_time.py

@@ -0,0 +1,100 @@
+from unittest.mock import Mock
+from galaxy.api.errors import UnknownBackendResponse
+
+import pytest
+from galaxy.api.types import GameTime
+
+from psn_client import parse_play_duration
+
+
+@pytest.fixture()
+def played_games_backend_response():
+    return {"titles":
+        [
+            {
+                "playDuration": "PT2H33M3S",
+                "firstPlayedDateTime": "2021-02-27T13:17:31.460Z",
+                "lastPlayedDateTime": "2021-03-06T16:29:22.490Z",
+                "playCount": 5,
+                "category": "unknown",
+                "name": "Call of Duty®: Modern Warfare®",
+                "titleId": "GAME_ID_1",
+            },
+            {
+                "playDuration": "PT00H1M0S",
+                "firstPlayedDateTime": "2021-02-27T13:17:31.460Z",
+                "lastPlayedDateTime": "1970-01-01T00:00:01.000Z",
+                "playCount": 5,
+                "category": "unknown",
+                "name": "Call of Duty®: Modern Warfare®",
+                "titleId": "GAME_ID_2",
+            },
+        ]
+    }
+
+
+@pytest.mark.asyncio
+async def test_prepare_game_times_context(
+    http_get,
+    authenticated_plugin,
+    played_games_backend_response
+):
+    http_get.return_value = played_games_backend_response
+
+    result = await authenticated_plugin.prepare_game_times_context(Mock(list))
+    for title in played_games_backend_response['titles']:
+        assert title in result.values()
+
+
+@pytest.mark.asyncio
+async def test_getting_game_time(
+    http_get,
+    authenticated_plugin,
+    played_games_backend_response
+):
+    http_get.return_value = played_games_backend_response
+    ctx = await authenticated_plugin.prepare_game_times_context(Mock(list))
+    assert GameTime('GAME_ID_1', 154, 1615048162) == await authenticated_plugin.get_game_time('GAME_ID_1', ctx)
+    assert GameTime('GAME_ID_2', 1, 1) == await authenticated_plugin.get_game_time('GAME_ID_2', ctx)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("input_time, expected_result", [
+        ("PT2H33M3S", 154),
+        ("PT0H0M60S", 1),
+        ("PT0H0M61S", 2),
+        ("PT0H0M30S", 1),
+        ("PT0H0M01S", 1),
+        ("PT1H0M0S", 60),
+        ("PT1H01M0S", 61),
+        ("PT1H0M01S", 61),
+        ("PT33H60M3S", 2041),
+        ("PT1H4M", 64),
+        ("PT1H1S", 61),
+        ("PT30M0S", 30),
+        ("PT0M1S", 1),
+        ("PT1H", 60),
+        ("PT1M", 1),
+        ("PT1S", 1),
+        ("1S", 1),
+        ("PT0H0M0S", 0),
+        ("PT", 0),
+    ])
+async def test_play_duration_parser(input_time, expected_result):
+    minutes = parse_play_duration(input_time)
+    assert minutes == expected_result
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("input_time", [
+        None,
+        "",
+        "bad_value",
+        "PTXH33M3S",
+        "PT0HM3S",
+        "PTHMS",
+        "HMS",
+    ])
+async def test_unknown_foramat_of_play_duration_parser(input_time):
+    with pytest.raises(UnknownBackendResponse):
+        parse_play_duration(input_time)

tests/test_parsers.py

@@ -2,7 +2,7 @@
 from galaxy.api.errors import UnknownBackendResponse
 from galaxy.api.types import SubscriptionGame
 
-from src.parsers import PSNGamesParser
+from parsers import PSNGamesParser
 
 
 @pytest.mark.asyncio