diff --git a/coding/codeforces.py b/coding/codeforces.py new file mode 100644 index 0000000..c6b9378 --- /dev/null +++ b/coding/codeforces.py @@ -0,0 +1,115 @@ +#!/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 kython import cproperty, fget +from kython.konsume import zoom, ignore +from kython.kerror import Res, ytry, unwrap +from kython.kdatetime import as_utc + + +_BDIR = Path('/L/zzz_syncthing/data/codeforces') + + +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: + last = max(_BDIR.glob('allcontests*.json')) + j = json.loads(last.read_text()) + d = {} + for c in j['result']: + cc = Contest.make(c) + d[cc.cid] = cc + return d + + +def get_latest(): + last = max(_BDIR.glob('codeforces*.json')) + return json.loads(last.read_text()) + + +class Competition(NamedTuple): + contest_id: str + contest: str + cmap: Cmap + + @cproperty + def uid(self) -> str: + 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') + +from kython.konsume import wrap +def iter_data() -> Iterator[Res[Competition]]: + cmap = get_contests() + + with wrap(get_latest()) 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() diff --git a/coding/topcoder.py b/coding/topcoder.py new file mode 100644 index 0000000..d41c2ae --- /dev/null +++ b/coding/topcoder.py @@ -0,0 +1,96 @@ +#!/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 kython import cproperty, fget +from kython.konsume import zoom, wrap, ignore +from kython.kerror import Res, ytry, unwrap + + +def get_latest(): + last = max(Path('/L/zzz_syncthing/data/topcoder').glob('*.json')) + return json.loads(last.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()