adapt codeforces/topcoder data from private package

my/coding/codeforces.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+from datetime import datetime
+from typing import NamedTuple
+from pathlib import Path
+import json
+from typing import Dict, Iterator, Any
+
+from ..common import cproperty, get_files
+from ..error import Res, unwrap
+
+from kython import fget
+from kython.konsume import zoom, ignore, wrap
+# TODO remove
+from kython.kdatetime import as_utc
+
+
+Cid = int
+
+class Contest(NamedTuple):
+    cid: Cid
+    when: datetime
+
+    @classmethod
+    def make(cls, j) -> 'Contest':
+        return cls(
+            cid=j['id'],
+            when=as_utc(j['startTimeSeconds']),
+        )
+
+Cmap = Dict[Cid, Contest]
+
+
+def get_contests() -> Cmap:
+    from mycfg import paths
+    last = max(get_files(paths.codeforces.export_path, 'allcontests*.json'))
+    j = json.loads(last.read_text())
+    d = {}
+    for c in j['result']:
+        cc = Contest.make(c)
+        d[cc.cid] = cc
+    return d
+
+
+class Competition(NamedTuple):
+    contest_id: Cid
+    contest: str
+    cmap: Cmap
+
+    @cproperty
+    def uid(self) -> Cid:
+        return self.contest_id
+
+    def __hash__(self):
+        return hash(self.contest_id)
+
+    @cproperty
+    def when(self) -> datetime:
+        return self.cmap[self.uid].when
+
+    @cproperty
+    def summary(self) -> str:
+        return f'participated in {self.contest}' # TODO 
+
+    @classmethod
+    def make(cls, cmap, json) -> Iterator[Res['Competition']]:
+        # TODO try here??
+        contest_id = json['contestId'].zoom().value
+        contest = json['contestName'].zoom().value
+        yield cls(
+            contest_id=contest_id,
+            contest=contest,
+            cmap=cmap,
+        )
+        # TODO ytry???
+        ignore(json, 'rank', 'oldRating', 'newRating')
+
+
+def iter_data() -> Iterator[Res[Competition]]:
+    cmap = get_contests()
+    from mycfg import paths
+    last = max(get_files(paths.codeforces.export_path, 'codeforces*.json'))
+
+    with wrap(json.loads(last.read_text())) as j:
+        j['status'].ignore()
+        res = j['result'].zoom()
+
+        for c in list(res): # TODO maybe we want 'iter' method??
+            ignore(c, 'handle', 'ratingUpdateTimeSeconds')
+            yield from Competition.make(cmap=cmap, json=c)
+            c.consume()
+            # TODO maybe if they are all empty, no need to consume??
+
+
+def get_data():
+    return list(sorted(iter_data(), key=fget(Competition.when)))
+
+
+def test():
+    assert len(get_data()) > 10
+
+
+def main():
+    for d in iter_data():
+        try:
+            d = unwrap(d)
+        except Exception as e:
+            print(f'ERROR! {d}')
+        else:
+            print(f'{d.when}: {d.summary}')
+
+
+
+if __name__ == '__main__':
+    main()

my/coding/topcoder.py

@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+from datetime import datetime
+from typing import NamedTuple
+from pathlib import Path
+import json
+from typing import Dict, Iterator, Any
+
+from ..common import cproperty, get_files
+from ..error import Res, unwrap
+
+# TODO get rid of fget?
+from kython import fget
+from kython.konsume import zoom, wrap, ignore
+
+
+# TODO json type??
+def _get_latest() -> Dict:
+    from mycfg import paths
+    pp = max(get_files(paths.topcoder.export_path, glob='*.json'))
+    return json.loads(pp.read_text())
+
+
+class Competition(NamedTuple):
+    contest_id: str
+    contest: str
+    percentile: float
+    dates: str
+
+    @cproperty
+    def uid(self) -> str:
+        return self.contest_id
+
+    def __hash__(self):
+        return hash(self.contest_id)
+
+    @cproperty
+    def when(self) -> datetime:
+        return datetime.strptime(self.dates, '%Y-%m-%dT%H:%M:%S.%fZ')
+
+    @cproperty
+    def summary(self) -> str:
+        return f'participated in {self.contest}: {self.percentile:.0f}'
+
+    @classmethod
+    def make(cls, json) -> Iterator[Res['Competition']]:
+        ignore(json, 'rating', 'placement')
+        cid = json['challengeId'].zoom().value
+        cname = json['challengeName'].zoom().value
+        percentile = json['percentile'].zoom().value
+        dates = json['date'].zoom().value
+        yield cls(
+            contest_id=cid,
+            contest=cname,
+            percentile=percentile,
+            dates=dates,
+        )
+
+
+def iter_data() -> Iterator[Res[Competition]]:
+    with wrap(_get_latest()) as j:
+        ignore(j, 'id', 'version')
+
+        res = j['result'].zoom()
+        ignore(res, 'success', 'status', 'metadata')
+
+        cont = res['content'].zoom()
+        ignore(cont, 'handle', 'handleLower', 'userId', 'createdAt', 'updatedAt', 'createdBy', 'updatedBy')
+
+        cont['DEVELOP'].ignore() # TODO FIXME handle it??
+        ds = cont['DATA_SCIENCE'].zoom()
+
+        mar, srm = zoom(ds, 'MARATHON_MATCH', 'SRM')
+
+        mar = mar['history'].zoom()
+        srm = srm['history'].zoom()
+    # TODO right, I guess I could rely on pylint for unused variables??
+
+        for c in mar + srm:
+            yield from Competition.make(json=c)
+            c.consume()
+
+
+def get_data():
+    return list(sorted(iter_data(), key=fget(Competition.when)))
+
+
+def test():
+    assert len(get_data()) > 10
+
+def main():
+    for d in iter_data():
+        try:
+            d = unwrap(d)
+        except Exception as e:
+            print(f'ERROR! {d}')
+        else:
+            print(d.summary)
+
+
+if __name__ == '__main__':
+    main()