diff --git a/.travis.yml b/.travis.yml
index 723aeb89..88eb6adf 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,9 +1,11 @@
language: python
python:
- - 3.5
+ - 3.6
install:
+ - pip install mypy
- pip install .
script:
+ - mypy --ignore-missing-imports oj
- python setup.py test
branches:
only:
diff --git a/onlinejudge/anarchygolf.py b/onlinejudge/anarchygolf.py
index 8810165f..12c00eae 100644
--- a/onlinejudge/anarchygolf.py
+++ b/onlinejudge/anarchygolf.py
@@ -8,31 +8,33 @@
import posixpath
import bs4
import requests
+from typing import *
@utils.singleton
class AnarchyGolfService(onlinejudge.service.Service):
- def get_url(self):
+ def get_url(self) -> str:
return 'http://golf.shinh.org/'
- def get_name(self):
+ def get_name(self) -> str:
return 'anarchygolf'
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['AnarchyGolfService']:
# example: http://golf.shinh.org/
result = urllib.parse.urlparse(s)
if result.scheme in ('', 'http', 'https') \
and result.netloc == 'golf.shinh.org':
return cls()
+ return None
class AnarchyGolfProblem(onlinejudge.problem.Problem):
- def __init__(self, problem_id):
+ def __init__(self, problem_id: str):
self.problem_id = problem_id
- def download(self, session=None):
+ def download(self, session: Optional[requests.Session] = None) -> List[onlinejudge.problem.TestCase]:
session = session or utils.new_default_session()
# get
resp = utils.request('GET', self.get_url(), session=session)
@@ -46,7 +48,7 @@ def download(self, session=None):
samples.add(s, name)
return samples.get()
- def _parse_sample_tag(self, tag):
+ def _parse_sample_tag(self, tag: bs4.Tag) -> Optional[Tuple[str, str]]:
assert isinstance(tag, bs4.Tag)
assert tag.name == 'h2'
name = tag.contents[0]
@@ -61,15 +63,16 @@ def _parse_sample_tag(self, tag):
else:
s = ''
return s, name
+ return None
- def get_url(self):
+ def get_url(self) -> str:
return 'http://golf.shinh.org/p.rb?{}'.format(self.problem_id)
- def get_service(self):
+ def get_service(self) -> AnarchyGolfService:
return AnarchyGolfService()
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['AnarchyGolfProblem']:
# example: http://golf.shinh.org/p.rb?The+B+Programming+Language
result = urllib.parse.urlparse(s)
if result.scheme in ('', 'http', 'https') \
@@ -77,6 +80,7 @@ def from_url(cls, s):
and utils.normpath(result.path) == '/p.rb' \
and result.query:
return cls(result.query)
+ return None
onlinejudge.dispatch.services += [ AnarchyGolfService ]
diff --git a/onlinejudge/aoj.py b/onlinejudge/aoj.py
index 6ce2dc03..b793715b 100644
--- a/onlinejudge/aoj.py
+++ b/onlinejudge/aoj.py
@@ -2,6 +2,7 @@
# -*- coding: utf-8 -*-
import onlinejudge.service
import onlinejudge.problem
+from onlinejudge.problem import LabeledString, TestCase
import onlinejudge.dispatch
import onlinejudge.implementation.utils as utils
import onlinejudge.implementation.logging as log
@@ -13,6 +14,7 @@
import zipfile
import collections
import itertools
+from typing import *
@utils.singleton
@@ -25,24 +27,25 @@ def get_name(self):
return 'aoj'
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['AOJService']:
# example: http://judge.u-aizu.ac.jp/onlinejudge/
result = urllib.parse.urlparse(s)
if result.scheme in ('', 'http', 'https') \
and result.netloc == 'judge.u-aizu.ac.jp':
return cls()
+ return None
class AOJProblem(onlinejudge.problem.Problem):
def __init__(self, problem_id):
self.problem_id = problem_id
- def download(self, session=None, is_system=False):
+ def download(self, session: Optional[requests.Session] = None, is_system: bool = False) -> List[TestCase]:
if is_system:
return self.download_system(session=session)
else:
return self.download_samples(session=session)
- def download_samples(self, session=None):
+ def download_samples(self, session: Optional[requests.Session] = None) -> List[TestCase]:
session = session or utils.new_default_session()
# get
resp = utils.request('GET', self.get_url(), session=session)
@@ -65,10 +68,10 @@ def download_samples(self, session=None):
name = hn.string
samples.add(s, name)
return samples.get()
- def download_system(self, session=None):
+ def download_system(self, session: Optional[requests.Session] = None) -> List[TestCase]:
session = session or utils.new_default_session()
get_url = lambda case, type: 'http://analytic.u-aizu.ac.jp:8080/aoj/testcase.jsp?id={}&case={}&type={}'.format(self.problem_id, case, type)
- testcases = []
+ testcases: List[TestCase] = []
for case in itertools.count(1):
# input
# get
@@ -76,23 +79,23 @@ def download_system(self, session=None):
if resp.status_code != 200:
break
in_txt = resp.text
- if case == 2 and testcases[0]['input']['data'] == in_txt:
+ if case == 2 and testcases[0].input.data == in_txt:
break # if the querystring case=??? is ignored
# output
# get
resp = utils.request('GET', get_url(case, 'out'), session=session)
out_txt = resp.text
- testcases += [ {
- 'input': { 'data': in_txt, 'name': 'in%d.txt' % case },
- 'output': { 'data': out_txt, 'name': 'out%d.txt' % case },
- } ]
+ testcases += [ TestCase(
+ LabeledString('in%d.txt' % case, in_txt),
+ LabeledString('out%d.txt' % case, out_txt),
+ ) ]
return testcases
- def get_url(self):
+ def get_url(self) -> str:
return 'http://judge.u-aizu.ac.jp/onlinejudge/description.jsp?id={}'.format(self.problem_id)
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['AOJProblem']:
# example: http://judge.u-aizu.ac.jp/onlinejudge/description.jsp?id=1169
# example: http://judge.u-aizu.ac.jp/onlinejudge/description.jsp?id=DSL_1_A&lang=jp
result = urllib.parse.urlparse(s)
@@ -104,8 +107,9 @@ def from_url(cls, s):
and len(querystring['id']) == 1:
n, = querystring['id']
return cls(n)
+ return None
- def get_service(self):
+ def get_service(self) -> AOJService:
return AOJService()
diff --git a/onlinejudge/atcoder.py b/onlinejudge/atcoder.py
index bc812c83..bf470c50 100644
--- a/onlinejudge/atcoder.py
+++ b/onlinejudge/atcoder.py
@@ -2,6 +2,7 @@
# -*- coding: utf-8 -*-
import onlinejudge.service
import onlinejudge.problem
+from onlinejudge.problem import SubmissionError
import onlinejudge.submission
import onlinejudge.dispatch
import onlinejudge.implementation.utils as utils
@@ -12,12 +13,13 @@
import urllib.parse
import posixpath
import json
+from typing import *
@utils.singleton
class AtCoderService(onlinejudge.service.Service):
- def login(self, get_credentials, session=None):
+ def login(self, get_credentials: onlinejudge.service.CredentialsProvider, session: Optional[requests.Session] = None) -> bool:
session = session or utils.new_default_session()
url = 'https://practice.contest.atcoder.jp/login'
# get
@@ -34,31 +36,32 @@ def login(self, get_credentials, session=None):
AtCoderService._report_messages(msgs)
return 'login' not in resp.url # AtCoder redirects to the top page if success
- def get_url(self):
+ def get_url(self) -> str:
return 'https://atcoder.jp/'
- def get_name(self):
+ def get_name(self) -> str:
return 'atcoder'
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['AtCoderService']:
# example: https://atcoder.jp/
# example: http://agc012.contest.atcoder.jp/
result = urllib.parse.urlparse(s)
if result.scheme in ('', 'http', 'https') \
and (result.netloc in ( 'atcoder.jp', 'beta.atcoder.jp' ) or result.netloc.endswith('.contest.atcoder.jp')):
return cls()
+ return None
@classmethod
- def _get_messages_from_cookie(cls, cookies):
- msgtags = []
+ def _get_messages_from_cookie(cls, cookies) -> List[str]:
+ msgtags: List[str] = []
for cookie in cookies:
log.debug('cookie: %s', str(cookie))
if cookie.name.startswith('__message_'):
msg = json.loads(urllib.parse.unquote_plus(cookie.value))
msgtags += [ msg['c'] ]
log.debug('message: %s: %s', cookie.name, str(msg))
- msgs = []
+ msgs: List[str] = []
for msgtag in msgtags:
soup = bs4.BeautifulSoup(msgtag, utils.html_parser)
msg = None
@@ -73,7 +76,7 @@ def _get_messages_from_cookie(cls, cookies):
return msgs
@classmethod
- def _report_messages(cls, msgs, unexpected=False):
+ def _report_messages(cls, msgs: List[str], unexpected: bool = False) -> bool:
for msg in msgs:
log.status('message: %s', msg)
if msgs and unexpected:
@@ -82,12 +85,12 @@ def _report_messages(cls, msgs, unexpected=False):
class AtCoderProblem(onlinejudge.problem.Problem):
- def __init__(self, contest_id, problem_id):
+ def __init__(self, contest_id: str, problem_id: str):
self.contest_id = contest_id
self.problem_id = problem_id
- self._task_id = None
+ self._task_id: Optional[int] = None
- def download(self, session=None):
+ def download(self, session: Optional[requests.Session] = None) -> List[onlinejudge.problem.TestCase]:
session = session or utils.new_default_session()
# get
resp = utils.request('GET', self.get_url(), session=session)
@@ -135,14 +138,14 @@ def _find_sample_tags(self, soup):
result += [( pre, prv )]
return result
- def get_url(self):
+ def get_url(self) -> str:
return 'http://{}.contest.atcoder.jp/tasks/{}'.format(self.contest_id, self.problem_id)
- def get_service(self):
+ def get_service(self) -> AtCoderService:
return AtCoderService()
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['AtCoderProblem']:
# example: http://agc012.contest.atcoder.jp/tasks/agc012_d
result = urllib.parse.urlparse(s)
dirname, basename = posixpath.split(utils.normpath(result.path))
@@ -165,7 +168,9 @@ def from_url(cls, s):
problem_id = m.group(2)
return cls(contest_id, problem_id)
- def get_input_format(self, session=None):
+ return None
+
+ def get_input_format(self, session: Optional[requests.Session] = None) -> str:
session = session or utils.new_default_session()
# get
resp = utils.request('GET', self.get_url(), session=session)
@@ -186,8 +191,9 @@ def get_input_format(self, session=None):
for it in tag:
s += it.string or it # AtCoder uses ... for math symbols
return s
+ return ''
- def get_language_dict(self, session=None):
+ def get_language_dict(self, session: Optional[requests.Session] = None) -> Dict[str, Any]:
session = session or utils.new_default_session()
# get
url = 'http://{}.contest.atcoder.jp/submit'.format(self.contest_id)
@@ -208,7 +214,7 @@ def get_language_dict(self, session=None):
language_dict[option.attrs['value']] = { 'description': option.string }
return language_dict
- def submit(self, code, language, session=None):
+ def submit(self, code: str, language: str, session: Optional[requests.Session] = None) -> 'AtCoderSubmission':
assert language in self.get_language_dict(session=session)
session = session or utils.new_default_session()
# get
@@ -216,18 +222,18 @@ def submit(self, code, language, session=None):
resp = utils.request('GET', url, session=session)
msgs = AtCoderService._get_messages_from_cookie(resp.cookies)
if AtCoderService._report_messages(msgs, unexpected=True):
- return None
+ raise SubmissionError
# check whether logged in
path = utils.normpath(urllib.parse.urlparse(resp.url).path)
if path.startswith('/login'):
log.error('not logged in')
- return None
+ raise SubmissionError
# parse
soup = bs4.BeautifulSoup(resp.content.decode(resp.encoding), utils.html_parser)
form = soup.find('form', action=re.compile(r'^/submit\?task_id='))
if not form:
log.error('form not found')
- return None
+ raise SubmissionError
log.debug('form: %s', str(form))
# post
task_id = self._get_task_id(session=session)
@@ -246,38 +252,43 @@ def submit(self, code, language, session=None):
log.success('success: result: %s', resp.url)
# NOTE: ignore the returned legacy URL and use beta.atcoder.jp's one
url = 'https://beta.atcoder.jp/contests/{}/submissions/me'.format(self.contest_id)
- return onlinejudge.submission.CompatibilitySubmission(url)
+ submission = AtCoderSubmission.from_url(url, problem_id=self.problem_id)
+ if not submission:
+ raise SubmissionError
+ return submission
else:
log.failure('failure')
- return None
+ raise SubmissionError
- def _get_task_id(self, session=None):
+ def _get_task_id(self, session: Optional[requests.Session] = None) -> int:
if self._task_id is None:
session = session or utils.new_default_session()
# get
resp = utils.request('GET', self.get_url(), session=session)
msgs = AtCoderService._get_messages_from_cookie(resp.cookies)
if AtCoderService._report_messages(msgs, unexpected=True):
- return {}
+ raise SubmissionError
# parse
soup = bs4.BeautifulSoup(resp.content.decode(resp.encoding), utils.html_parser)
submit = soup.find('a', href=re.compile(r'^/submit\?task_id='))
if not submit:
log.error('link to submit not found')
- return False
+ raise SubmissionError
m = re.match(r'^/submit\?task_id=([0-9]+)$', submit.attrs['href'])
assert m
self._task_id = int(m.group(1))
return self._task_id
class AtCoderSubmission(onlinejudge.submission.Submission):
- def __init__(self, contest_id, submission_id, problem_id=None):
+ def __init__(self, contest_id: str, submission_id: int, problem_id: Optional[str] = None):
self.contest_id = contest_id
self.submission_id = submission_id
self.problem_id = problem_id
@classmethod
- def from_url(cls, s, problem_id=None):
+ def from_url(cls, s: str, problem_id: Optional[str] = None) -> Optional['AtCoderSubmission']:
+ submission_id: Optional[int] = None
+
# example: http://agc001.contest.atcoder.jp/submissions/1246803
result = urllib.parse.urlparse(s)
dirname, basename = posixpath.split(utils.normpath(result.path))
@@ -290,6 +301,7 @@ def from_url(cls, s, problem_id=None):
try:
submission_id = int(basename)
except ValueError:
+ pass
submission_id = None
if submission_id is not None:
return cls(contest_id, submission_id, problem_id=problem_id)
@@ -307,23 +319,26 @@ def from_url(cls, s, problem_id=None):
if submission_id is not None:
return cls(contest_id, submission_id, problem_id=problem_id)
- def get_url(self):
+ return None
+
+ def get_url(self) -> str:
return 'http://{}.contest.atcoder.jp/submissions/{}'.format(self.contest_id, self.submission_id)
- def get_problem(self):
- if self.problem_id is not None:
- return AtCoderProblem(self.contest_id, self.problem_id)
+ def get_problem(self) -> AtCoderProblem:
+ if self.problem_id is None:
+ raise ValueError
+ return AtCoderProblem(self.contest_id, self.problem_id)
- def get_service(self):
+ def get_service(self) -> AtCoderService:
return AtCoderService()
- def download(self, session=None):
+ def download(self, session: Optional[requests.Session] = None) -> str:
session = session or utils.new_default_session()
# get
resp = utils.request('GET', self.get_url(), session=session)
msgs = AtCoderService._get_messages_from_cookie(resp.cookies)
if AtCoderService._report_messages(msgs, unexpected=True):
- return []
+ raise RuntimeError
# parse
soup = bs4.BeautifulSoup(resp.content.decode(resp.encoding), utils.html_parser)
code = None
@@ -335,6 +350,7 @@ def download(self, session=None):
code = pre.string
if code is None:
log.error('source code not found')
+ raise RuntimeError
return code
onlinejudge.dispatch.services += [ AtCoderService ]
diff --git a/onlinejudge/codeforces.py b/onlinejudge/codeforces.py
index 184866a8..2fabcbc7 100644
--- a/onlinejudge/codeforces.py
+++ b/onlinejudge/codeforces.py
@@ -4,17 +4,19 @@
import onlinejudge.dispatch
import onlinejudge.implementation.utils as utils
import onlinejudge.implementation.logging as log
+import requests
import re
import urllib.parse
import posixpath
import bs4
import string
+from typing import *
@utils.singleton
class CodeforcesService(onlinejudge.service.Service):
- def login(self, get_credentials, session=None):
+ def login(self, get_credentials: onlinejudge.service.CredentialsProvider, session: Optional[requests.Session] = None) -> bool:
session = session or utils.new_default_session()
url = 'http://codeforces.com/enter'
# get
@@ -41,36 +43,38 @@ def login(self, get_credentials, session=None):
log.failure('Invalid handle or password.')
return False
- def get_url(self):
+ def get_url(self) -> str:
return 'http://codeforces.com/'
- def get_name(self):
+ def get_name(self) -> str:
return 'codeforces'
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['CodeforcesService']:
# example: http://codeforces.com/
result = urllib.parse.urlparse(s)
if result.scheme in ('', 'http', 'https') \
and result.netloc == 'codeforces.com':
return cls()
+ return None
# NOTE: Codeforces has its API: http://codeforces.com/api/help
class CodeforcesProblem(onlinejudge.problem.Problem):
- def __init__(self, contest_id, index, kind=None):
+ def __init__(self, contest_id: int, index: str, kind: Optional[str] = None):
assert isinstance(contest_id, int)
assert index in string.ascii_uppercase
+ assert kind in ( None, 'contest', 'gym', 'problemset' )
self.contest_id = contest_id
self.index = index
- self.kind = kind # It seems 'gym' is specialized, 'contest' and 'problemset' are the same thing
if kind is None:
if self.contest_id < 100000:
- self.kind = 'contest'
+ kind = 'contest'
else:
- self.kind = 'gym'
+ kind = 'gym'
+ self.kind = kind # It seems 'gym' is specialized, 'contest' and 'problemset' are the same thing
- def download(self, session=None):
+ def download(self, session: Optional[requests.Session] = None) -> List[onlinejudge.problem.TestCase]:
session = session or utils.new_default_session()
# get
resp = utils.request('GET', self.get_url(), session=session)
@@ -92,18 +96,18 @@ def download(self, session=None):
samples.add(s, title.string)
return samples.get()
- def get_url(self):
+ def get_url(self) -> str:
table = {}
table['contest'] = 'http://codeforces.com/contest/{}/problem/{}'
table['problemset'] = 'http://codeforces.com/problemset/problem/{}/{}'
table['gym'] = 'http://codeforces.com/gym/{}/problem/{}'
return table[self.kind].format(self.contest_id, self.index)
- def get_service(self):
+ def get_service(self) -> CodeforcesService:
return CodeforcesService()
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['CodeforcesProblem']:
result = urllib.parse.urlparse(s)
if result.scheme in ('', 'http', 'https') \
and result.netloc == 'codeforces.com':
@@ -116,6 +120,7 @@ def from_url(cls, s):
m = re.match(expr, utils.normpath(result.path))
if m:
return cls(int(m.group(1)), normalize(m.group(2)), kind=kind)
+ return None
onlinejudge.dispatch.services += [ CodeforcesService ]
diff --git a/onlinejudge/csacademy.py b/onlinejudge/csacademy.py
index 2a44059e..2f2d3d03 100644
--- a/onlinejudge/csacademy.py
+++ b/onlinejudge/csacademy.py
@@ -1,6 +1,7 @@
# Python Version: 3.x
import onlinejudge.service
import onlinejudge.problem
+from onlinejudge.problem import LabeledString, TestCase
import onlinejudge.dispatch
import onlinejudge.implementation.utils as utils
import onlinejudge.implementation.logging as log
@@ -9,32 +10,34 @@
import urllib.parse
import posixpath
import json
+from typing import *
@utils.singleton
class CSAcademyService(onlinejudge.service.Service):
- def get_url(self):
+ def get_url(self) -> str:
return 'https://csacademy.com/'
- def get_name(self):
+ def get_name(self) -> str:
return 'csacademy'
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['CSAcademyService']:
# example: https://csacademy.com/
result = urllib.parse.urlparse(s)
if result.scheme in ('', 'http', 'https') \
and result.netloc in ('csacademy.com', 'www.csacademy.com'):
return cls()
+ return None
class CSAcademyProblem(onlinejudge.problem.Problem):
- def __init__(self, contest_name, task_name):
+ def __init__(self, contest_name: str, task_name: str):
self.contest_name = contest_name
self.task_name = task_name
- def download(self, session=None):
+ def download(self, session: Optional[requests.Session] = None) -> List[TestCase]:
session = session or utils.new_default_session()
base_url = self.get_url()
@@ -42,8 +45,8 @@ def download(self, session=None):
resp = utils.request('GET', base_url, session=session)
csrftoken = None
for cookie in session.cookies:
- if cookie.name == 'csrftoken' and cookie.domain == 'csacademy.com':
- csrftoken = cookie.value
+ if cookie.name == 'csrftoken' and cookie.domain == 'csacademy.com': # type: ignore
+ csrftoken = cookie.value # type: ignore
if csrftoken is None:
log.error('csrftoken is not found')
return []
@@ -85,21 +88,21 @@ def download(self, session=None):
for test_number, example_test in enumerate(contest_task['state']['EvalTask'][0]['exampleTests']):
inname = 'Input {}'.format(test_number)
outname = 'Output {}'.format(test_number)
- samples += [ {
- 'input': { 'data': example_test['input'], 'name': inname },
- 'output': { 'data': example_test['output'], 'name': outname },
- } ]
+ samples += [ TestCase(
+ LabeledString( inname, example_test[ 'input']),
+ LabeledString(outname, example_test['output']),
+ ) ]
return samples
- def get_url(self):
+ def get_url(self) -> str:
return 'https://csacademy.com/content/{}/task/{}/'.format(self.contest_name, self.task_name)
- def get_service(self):
+ def get_service(self) -> CSAcademyService:
return CSAcademyService()
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['CSAcademyProblem']:
# example: https://csacademy.com/contest/round-38/task/path-union/
# example: https://csacademy.com/contest/round-38/task/path-union/discussion/
# example: https://csacademy.com/contest/archive/task/swap_permutation/
@@ -110,6 +113,7 @@ def from_url(cls, s):
m = re.match(r'^/contest/([0-9A-Za-z_-]+)/task/([0-9A-Za-z_-]+)(|/statement|/solution|/discussion|/statistics|/submissions)/?$', utils.normpath(result.path))
if m:
return cls(m.group(1), m.group(2))
+ return None
onlinejudge.dispatch.services += [ CSAcademyService ]
onlinejudge.dispatch.problems += [ CSAcademyProblem ]
diff --git a/onlinejudge/dispatch.py b/onlinejudge/dispatch.py
index 7f7bf808..da6792a2 100644
--- a/onlinejudge/dispatch.py
+++ b/onlinejudge/dispatch.py
@@ -1,38 +1,46 @@
# Python Version: 3.x
import onlinejudge.implementation.logging as log
+from typing import List, Optional, Type, TYPE_CHECKING
+if TYPE_CHECKING:
+ from onlinejudge.submission import Submission
+ from onlinejudge.problem import Problem
+ from onlinejudge.service import Service
-submissions = []
-def submission_from_url(s):
+submissions: List[Type['Submission']] = []
+def submission_from_url(s: str) -> Optional['Submission']:
for cls in submissions:
- it = cls.from_url(s)
- if it is not None:
- log.status('submission recognized: %s: %s', str(it), s)
- return it
+ submission = cls.from_url(s)
+ if submission is not None:
+ log.status('submission recognized: %s: %s', str(submission), s)
+ return submission
log.failure('unknown submission: %s', s)
+ return None
-problems = []
-def problem_from_url(s):
+problems: List[Type['Problem']] = []
+def problem_from_url(s: str) -> Optional['Problem']:
for cls in problems:
- it = cls.from_url(s)
- if it is not None:
- log.status('problem recognized: %s: %s', str(it), s)
- return it
- it = submission_from_url(s)
- if it is not None:
- return it.get_problem()
+ problem = cls.from_url(s)
+ if problem is not None:
+ log.status('problem recognized: %s: %s', str(problem), s)
+ return problem
+ submission = submission_from_url(s)
+ if submission is not None:
+ return submission.get_problem()
log.failure('unknown problem: %s', s)
+ return None
-services = []
-def service_from_url(s):
+services: List[Type['Service']] = []
+def service_from_url(s: str) -> Optional['Service']:
for cls in services:
- it = cls.from_url(s)
- if it is not None:
- log.status('service recognized: %s: %s', str(it), s)
- return it
- it = submission_from_url(s)
- if it is not None:
- return it.get_service()
- it = problem_from_url(s)
- if it is not None:
- return it.get_service()
+ service = cls.from_url(s)
+ if service is not None:
+ log.status('service recognized: %s: %s', str(service), s)
+ return service
+ submission = submission_from_url(s)
+ if submission is not None:
+ return submission.get_service()
+ problem = problem_from_url(s)
+ if problem is not None:
+ return problem.get_service()
log.failure('unknown service: %s', s)
+ return None
diff --git a/onlinejudge/hackerrank.py b/onlinejudge/hackerrank.py
index 387872ed..95f653a5 100644
--- a/onlinejudge/hackerrank.py
+++ b/onlinejudge/hackerrank.py
@@ -1,6 +1,7 @@
# Python Version: 3.x
import onlinejudge.service
import onlinejudge.problem
+from onlinejudge.problem import LabeledString, TestCase
import onlinejudge.submission
import onlinejudge.dispatch
import onlinejudge.implementation.utils as utils
@@ -13,12 +14,13 @@
import json
import datetime
import time
+from typing import *
@utils.singleton
class HackerRankService(onlinejudge.service.Service):
- def login(self, get_credentials, session=None):
+ def login(self, get_credentials: onlinejudge.service.CredentialsProvider, session: Optional[requests.Session] = None) -> bool:
session = session or utils.new_default_session()
url = 'https://www.hackerrank.com/login'
# get
@@ -50,35 +52,36 @@ def login(self, get_credentials, session=None):
log.failure('You failed to sign in. Wrong user ID or password.')
return False
- def get_url(self):
+ def get_url(self) -> str:
return 'https://www.hackerrank.com/'
- def get_name(self):
+ def get_name(self) -> str:
return 'hackerrank'
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['HackerRankService']:
# example: https://www.hackerrank.com/dashboard
result = urllib.parse.urlparse(s)
if result.scheme in ('', 'http', 'https') \
and result.netloc in ('hackerrank.com', 'www.hackerrank.com'):
return cls()
+ return None
class HackerRankProblem(onlinejudge.problem.Problem):
- def __init__(self, contest_slug, challenge_slug):
+ def __init__(self, contest_slug: str, challenge_slug: str):
self.contest_slug = contest_slug
self.challenge_slug = challenge_slug
- def download(self, session=None, method='run_code'):
+ def download(self, session: Optional[requests.Session] = None, method: str = 'run_code') -> List[TestCase]:
if method == 'run_code':
return self.download_with_running_code(session=session)
elif method == 'parse_html':
return self.download_with_parsing_html(session=session)
else:
- assert False
+ raise ValueError
- def download_with_running_code(self, session=None):
+ def download_with_running_code(self, session: Optional[requests.Session] = None) -> List[TestCase]:
session = session or utils.new_default_session()
# get
resp = utils.request('GET', self.get_url(), session=session)
@@ -111,33 +114,33 @@ def download_with_running_code(self, session=None):
if not it['status']:
log.error('Run Code: failed')
return []
- samples = []
+ samples: List[TestCase] = []
for i, (inf, outf) in enumerate(zip(it['model']['stdin'], it['model']['expected_output'])):
inname = 'Testcase {} Input'.format(i)
outname = 'Testcase {} Expected Output'.format(i)
- samples += [ {
- 'input': { 'data': utils.textfile(inf), 'name': inname },
- 'output': { 'data': utils.textfile(outf), 'name': outname },
- } ]
+ samples += [ TestCase(
+ LabeledString( inname, utils.textfile(inf)),
+ LabeledString(outname, utils.textfile(outf)),
+ ) ]
return samples
- def download_with_parsing_html(self, session=None):
+ def download_with_parsing_html(self, session: Optional[requests.Session] = None) -> List[TestCase]:
session = session or utils.new_default_session()
url = 'https://www.hackerrank.com/rest/contests/{}/challenges/{}'.format(self.contest_slug, self.challenge_slug)
raise NotImplementedError
- def get_url(self):
+ def get_url(self) -> str:
if self.contest_slug == 'master':
return 'https://www.hackerrank.com/challenges/{}'.format(self.challenge_slug)
else:
return 'https://www.hackerrank.com/contests/{}/challenges/{}'.format(self.contest_slug, self.challenge_slug)
- def get_service(self):
+ def get_service(self) -> HackerRankService:
return HackerRankService()
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['HackerRankProblem']:
# example: https://www.hackerrank.com/contests/university-codesprint-2/challenges/the-story-of-a-tree
# example: https://www.hackerrank.com/challenges/fp-hello-world
result = urllib.parse.urlparse(s)
@@ -149,8 +152,9 @@ def from_url(cls, s):
m = re.match(r'^/challenges/([0-9A-Za-z-]+)$', utils.normpath(result.path))
if m:
return cls('master', m.group(1))
+ return None
- def _get_model(self, session=None):
+ def _get_model(self, session: Optional[requests.Session] = None) -> Dict[str, Any]:
session = session or utils.new_default_session()
# get
url = 'https://www.hackerrank.com/rest/contests/{}/challenges/{}'.format(self.contest_slug, self.challenge_slug)
@@ -160,10 +164,10 @@ def _get_model(self, session=None):
log.debug('json: %s', it)
if not it['status']:
log.error('get model: failed')
- return None
+ raise onlinejudge.problem.SubmissionError
return it['model']
- def get_language_dict(self, session=None):
+ def get_language_dict(self, session: Optional[requests.Session] = None) -> Dict[str, Dict[str, str]]:
session = session or utils.new_default_session()
info = self._get_model(session=session)
# lang_display_mapping from https://hrcdn.net/hackerrank/assets/codeshell/dist/codeshell-449bb296b091277fedc42b23f7c9c447.js, Sun Feb 19 02:25:36 JST 2017
@@ -177,7 +181,7 @@ def get_language_dict(self, session=None):
result[lang] = { 'description': descr }
return result
- def submit(self, code, language, session=None):
+ def submit(self, code: str, language: str, session: Optional[requests.Session] = None) -> onlinejudge.submission.Submission:
session = session or utils.new_default_session()
# get
resp = utils.request('GET', self.get_url(), session=session)
@@ -194,7 +198,7 @@ def submit(self, code, language, session=None):
log.debug('json: %s', it)
if not it['status']:
log.failure('Submit Code: failed')
- return None
+ raise onlinejudge.problem.SubmissionError
model_id = it['model']['id']
url = self.get_url().rstrip('/') + '/submissions/code/{}'.format(model_id)
log.success('success: result: %s', url)
diff --git a/onlinejudge/implementation/command/download.py b/onlinejudge/implementation/command/download.py
index 202b67d8..60202c02 100644
--- a/onlinejudge/implementation/command/download.py
+++ b/onlinejudge/implementation/command/download.py
@@ -5,8 +5,11 @@
import os
import colorama
import sys
+from typing import *
+if TYPE_CHECKING:
+ import argparse
-def download(args):
+def download(args: 'argparse.Namespace') -> None:
# prepare values
problem = onlinejudge.dispatch.problem_from_url(args.url)
if problem is None:
@@ -29,7 +32,7 @@ def download(args):
# get samples from the server
with utils.with_cookiejar(utils.new_default_session(), path=args.cookie) as sess:
- samples = problem.download(session=sess, **kwargs)
+ samples = problem.download(session=sess, **kwargs) # type: ignore
# write samples to files
for i, sample in enumerate(samples):
@@ -37,8 +40,8 @@ def download(args):
log.info('sample %d', i)
for kind in [ 'input', 'output' ]:
ext = kind[: -3]
- data = sample[kind]['data']
- name = sample[kind]['name']
+ data = getattr(sample, kind).data
+ name = getattr(sample, kind).name
table = {}
table['i'] = str(i+1)
table['e'] = ext
diff --git a/onlinejudge/implementation/command/utils.py b/onlinejudge/implementation/command/utils.py
index d2ceccc8..f4378202 100644
--- a/onlinejudge/implementation/command/utils.py
+++ b/onlinejudge/implementation/command/utils.py
@@ -8,8 +8,9 @@
import re
import glob
import collections
+from typing import Dict, List, Match, Optional
-def glob_with_format(directory, format):
+def glob_with_format(directory: str, format: str) -> List[str]:
table = {}
table['s'] = '*'
table['e'] = '*'
@@ -19,7 +20,7 @@ def glob_with_format(directory, format):
log.debug('testcase globbed: %s', path)
return paths
-def match_with_format(directory, format, path):
+def match_with_format(directory: str, format: str, path: str) -> Optional[Match[str]]:
table = {}
table['s'] = '(?P.+)'
table['e'] = '(?Pin|out)'
@@ -27,18 +28,18 @@ def match_with_format(directory, format, path):
path = os.path.normpath(os.path.relpath(path, directory))
return pattern.match(path)
-def path_from_format(directory, format, name, ext):
+def path_from_format(directory: str, format: str, name: str, ext: str) -> str:
table = {}
table['s'] = name
table['e'] = ext
return os.path.join(directory, utils.parcentformat(format, table))
-def is_backup_or_hidden_file(path):
+def is_backup_or_hidden_file(path: str) -> bool:
basename = os.path.basename(path)
return basename.endswith('~') or (basename.startswith('#') and basename.endswith('#')) or basename.startswith('.')
-def drop_backup_or_hidden_files(paths):
- result = []
+def drop_backup_or_hidden_files(paths: List[str]) -> List[str]:
+ result: List[str] = []
for path in paths:
if is_backup_or_hidden_file(path):
log.warning('ignore a backup file: %s', path)
@@ -46,8 +47,8 @@ def drop_backup_or_hidden_files(paths):
result += [ path ]
return result
-def construct_relationship_of_files(paths, directory, format):
- tests = collections.defaultdict(dict)
+def construct_relationship_of_files(paths: List[str], directory: str, format: str) -> Dict[str, Dict[str, str]]:
+ tests: Dict[str, Dict[str, str]] = collections.defaultdict(dict)
for path in paths:
m = match_with_format(directory, format, os.path.normpath(path))
if not m:
diff --git a/onlinejudge/implementation/logging.py b/onlinejudge/implementation/logging.py
index 94aa73c3..c24e9855 100644
--- a/onlinejudge/implementation/logging.py
+++ b/onlinejudge/implementation/logging.py
@@ -25,25 +25,25 @@ def removeHandler(handler):
'critical' : '[' + Fore.RED + 'CRITICAL' + Style.RESET_ALL + '] ',
}
-def emit(s, *args):
+def emit(s: str, *args) -> None:
logger.info(str(s), *args)
-def status(s, *args):
+def status(s: str, *args) -> None:
logger.info(prefix['status'] + str(s), *args)
-def success(s, *args):
+def success(s: str, *args) -> None:
logger.info(prefix['success'] + str(s), *args)
-def failure(s, *args):
+def failure(s: str, *args) -> None:
logger.info(prefix['failure'] + str(s), *args)
-def debug(s, *args):
+def debug(s: str, *args) -> None:
logger.debug(prefix['debug'] + str(s), *args)
-def info(s, *args):
+def info(s: str, *args) -> None:
logger.info(prefix['info'] + str(s), *args)
-def warning(s, *args):
+def warning(s: str, *args) -> None:
logger.warning(prefix['warning'] + str(s), *args)
-def error(s, *args):
+def error(s: str, *args) -> None:
logger.error(prefix['error'] + str(s), *args)
-def exception(s, *args):
+def exception(s: str, *args) -> None:
logger.error(prefix['exception'] + str(s), *args)
-def critical(s, *args):
+def critical(s: str, *args) -> None:
logger.critical(prefix['critical'] + str(s), *args)
bold = lambda s: colorama.Style.BRIGHT + s + colorama.Style.RESET_ALL
diff --git a/onlinejudge/implementation/main.py b/onlinejudge/implementation/main.py
index 67878f3b..bb9fcee1 100644
--- a/onlinejudge/implementation/main.py
+++ b/onlinejudge/implementation/main.py
@@ -17,10 +17,9 @@
import sys
import os
import os.path
+from typing import List, Optional
-def main(args=None):
-
- # argparse
+def get_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description='Tools for online judge services')
parser.add_argument('-v', '--verbose', action='store_true')
parser.add_argument('-c', '--cookie', help='path to cookie. (default: {})'.format(utils.default_cookie_path))
@@ -289,8 +288,10 @@ def main(args=None):
subparser.add_argument('url')
subparser.add_argument('-f', '--format', choices=[ 'csv', 'tsv', 'json' ], default='tsv', help='default: tsv')
- args = parser.parse_args(args=args)
+ return parser
+
+def run_program(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
# logging
log_level = log.logging.INFO
if args.verbose:
@@ -327,3 +328,7 @@ def main(args=None):
sys.exit(1)
+def main(args: Optional[List[str]] = None) -> None:
+ parser = get_parser()
+ namespace = parser.parse_args(args=args)
+ run_program(namespace, parser=parser)
diff --git a/onlinejudge/implementation/utils.py b/onlinejudge/implementation/utils.py
index efcc8da5..ccbaa978 100644
--- a/onlinejudge/implementation/utils.py
+++ b/onlinejudge/implementation/utils.py
@@ -2,6 +2,7 @@
# -*- coding: utf-8 -*-
import onlinejudge.implementation.logging as log
import onlinejudge.implementation.version as version
+from onlinejudge.problem import LabeledString, TestCase
import re
import os
import os.path
@@ -16,11 +17,12 @@
import sys
import ast
import time
+from typing import *
default_data_dir = os.path.join(os.environ.get('XDG_DATA_HOME') or os.path.expanduser('~/.local/share'), 'onlinejudge')
html_parser = 'lxml'
-def parcentformat(s, table):
+def parcentformat(s: str, table: Dict[str, str]) -> str:
assert '%' not in table or table['%'] == '%'
table['%'] = '%'
result = ''
@@ -32,7 +34,7 @@ def parcentformat(s, table):
result += m.group(0)
return result
-def describe_status_code(status_code):
+def describe_status_code(status_code: int) -> str:
return '{} {}'.format(status_code, http.client.responses[status_code])
def previous_sibling_tag(tag):
@@ -47,7 +49,7 @@ def next_sibling_tag(tag):
tag = tag.next_sibling
return tag
-def new_default_session(): # without setting cookiejar
+def new_default_session() -> requests.Session: # without setting cookiejar
session = requests.Session()
session.headers['User-Agent'] += ' (+{})'.format(version.__url__)
return session
@@ -55,17 +57,17 @@ def new_default_session(): # without setting cookiejar
default_cookie_path = os.path.join(default_data_dir, 'cookie.jar')
@contextlib.contextmanager
-def with_cookiejar(session, path):
+def with_cookiejar(session: requests.Session, path: str) -> Generator[requests.Session, None, None]:
path = path or default_cookie_path
- session.cookies = http.cookiejar.LWPCookieJar(path)
+ session.cookies = http.cookiejar.LWPCookieJar(path) # type: ignore
if os.path.exists(path):
log.info('load cookie from: %s', path)
- session.cookies.load()
+ session.cookies.load() # type: ignore
yield session
log.info('save cookie to: %s', path)
if os.path.dirname(path):
os.makedirs(os.path.dirname(path), exist_ok=True)
- session.cookies.save()
+ session.cookies.save() # type: ignore
os.chmod(path, 0o600) # NOTE: to make secure a little bit
@@ -74,22 +76,19 @@ def __init__(self):
self.data = []
self.dangling = None
- def add(self, s, name=''):
+ def add(self, s: str, name: str = '') -> None:
if self.dangling is None:
if re.search('output', name, re.IGNORECASE) or re.search('出力', name):
log.warning('strange name for input string: %s', name)
- self.dangling = { 'data': s, 'name': name }
+ self.dangling = LabeledString(name, s)
else:
if re.search('input', name, re.IGNORECASE) or re.search('入力', name):
if not (re.search('output', name, re.IGNORECASE) or re.search('出力', name)): # to ignore titles like "Output for Sample Input 1"
log.warning('strange name for output string: %s', name)
- self.data += [ {
- 'input': self.dangling,
- 'output': { 'data': s, 'name': name },
- } ]
+ self.data += [ TestCase(self.dangling, LabeledString(name, s)) ]
self.dangling = None
- def get(self):
+ def get(self) -> List[TestCase]:
if self.dangling is not None:
log.error('dangling sample string: %s', self.dangling[1])
return self.data
@@ -128,9 +127,9 @@ def request(self, session, action=None, **kwargs):
log.status(describe_status_code(resp.status_code))
return resp
-def dos2unix(s):
+def dos2unix(s: str) -> str:
return s.replace('\r\n', '\n')
-def textfile(s): # should have trailing newline
+def textfile(s: str) -> str: # should have trailing newline
if s.endswith('\n'):
return s
elif '\r\n' in s:
@@ -150,7 +149,7 @@ def singleton(cls):
pass
return cls
-def exec_command(command, timeout=None, **kwargs):
+def exec_command(command: List[str], timeout: float = None, **kwargs) -> Tuple[str, subprocess.Popen]:
try:
proc = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=sys.stderr, **kwargs)
except FileNotFoundError:
@@ -167,14 +166,14 @@ def exec_command(command, timeout=None, **kwargs):
# We should use this instead of posixpath.normpath
# posixpath.normpath doesn't collapse a leading duplicated slashes. see: https://stackoverflow.com/questions/7816818/why-doesnt-os-normpath-collapse-a-leading-double-slash
-def normpath(path):
+def normpath(path: str) -> str:
path = posixpath.normpath(path)
if path.startswith('//'):
path = '/' + path.lstrip('/')
return path
-def request(method, url, session=None, raise_for_status=True, **kwargs):
+def request(method: str, url: str, session: requests.Session, raise_for_status: bool = True, **kwargs) -> requests.Response:
assert method in [ 'GET', 'POST' ]
kwargs.setdefault('allow_redirects', True)
log.status('%s: %s', method, url)
diff --git a/onlinejudge/problem.py b/onlinejudge/problem.py
index b4762932..c52b334e 100644
--- a/onlinejudge/problem.py
+++ b/onlinejudge/problem.py
@@ -1,18 +1,32 @@
# Python Version: 3.x
+from typing import *
+if TYPE_CHECKING:
+ import requests
+ from onlinejudge.service import Service
+ from onlinejudge.submission import Submission
+
+LabeledString = NamedTuple('LabeledString', [ ('name', str), ('data', str) ])
+TestCase = NamedTuple('TestCase', [ ('input', LabeledString), ('output', LabeledString) ])
+# Language = NamedTuple('Language', [ ('id', str), ('name', str), ('description': str) ])
+Language = Dict[str, Any]
+Standings = Tuple[List[str], List[Dict[str, Any]]] # ( [ 'column1', 'column2', ... ], [ { 'column1': data1, ... } ... ] )
+
+class SubmissionError(RuntimeError):
+ pass
class Problem(object):
- def download(self, session=None): # => [ { 'input': { 'data': str, 'name': str }, 'output': { ... } } ]
+ def download(self, session: Optional['requests.Session'] = None) -> List[TestCase]:
raise NotImplementedError
- def submit(self, code, language, session=None):
+ def submit(self, code: str, language: str, session: Optional['requests.Session'] = None) -> 'Submission': # or SubmissionError
raise NotImplementedError
- def get_language_dict(self, session=None): # => { language_id: { 'description': str } }
+ def get_language_dict(self, session: Optional['requests.Session'] = None) -> Dict[str, Language]:
raise NotImplementedError
- def get_url(self):
+ def get_url(self) -> str:
raise NotImplementedError
- def get_service(self):
+ def get_service(self) -> 'Service':
raise NotImplementedError
- def get_standings(self, session=None): # => ( [ 'column1', 'column2', ... ], [ { 'column1': data1, ... } ... ] )
+ def get_standings(self, session: Optional['requests.Session'] = None) -> Standings:
raise NotImplementedError
@classmethod
- def from_url(self, s):
+ def from_url(self, s: str) -> Optional['Problem']:
pass
diff --git a/onlinejudge/service.py b/onlinejudge/service.py
index 7ff1dd6d..7e456aa1 100644
--- a/onlinejudge/service.py
+++ b/onlinejudge/service.py
@@ -1,12 +1,17 @@
# Python Version: 3.x
+from typing import Callable, Optional, Tuple, TYPE_CHECKING
+if TYPE_CHECKING:
+ import requests
+
+CredentialsProvider = Callable[[], Tuple[str, str]]
class Service(object):
- def login(self, get_credentials, session=None):
+ def login(self, get_credentials: CredentialsProvider, session: Optional['requests.Session'] = None) -> bool:
raise NotImplementedError
- def get_url(self):
+ def get_url(self) -> str:
raise NotImplementedError
- def get_name(self):
+ def get_name(self) -> str:
raise NotImplementedError
@classmethod
- def from_url(self, s):
+ def from_url(self, s: str) -> Optional['Service']:
pass
diff --git a/onlinejudge/submission.py b/onlinejudge/submission.py
index 067ef22b..7cd10a21 100644
--- a/onlinejudge/submission.py
+++ b/onlinejudge/submission.py
@@ -1,23 +1,28 @@
# Python Version: 3.x
+from typing import NamedTuple, Optional, TYPE_CHECKING
+if TYPE_CHECKING:
+ import requests
+ from onlinejudge.problem import Problem
+ from onlinejudge.service import Service
class Submission(object):
- def download(self, session=None):
+ def download(self, session: Optional['requests.Session'] = None) -> str:
raise NotImplementedError
- def get_url(self):
+ def get_url(self) -> str:
raise NotImplementedError
- def get_problem(self):
+ def get_problem(self) -> 'Problem':
raise NotImplementedError
- def get_service(self):
- raise self.get_problem().get_service()
+ def get_service(self) -> 'Service':
+ return self.get_problem().get_service()
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['Submission']:
pass
class CompatibilitySubmission(Submission):
- def __init__(self, url, problem=None):
+ def __init__(self, url: str, problem: 'Problem'):
self.url = url
self.problem = problem
- def get_url(self):
+ def get_url(self) -> str:
return self.url
- def get_problem(self):
+ def get_problem(self) -> 'Problem':
return self.problem
diff --git a/onlinejudge/topcoder.py b/onlinejudge/topcoder.py
index 13c45ddc..067354de 100644
--- a/onlinejudge/topcoder.py
+++ b/onlinejudge/topcoder.py
@@ -14,12 +14,13 @@
import time
import itertools
import collections
+from typing import *
@utils.singleton
class TopCoderService(onlinejudge.service.Service):
- def login(self, get_credentials, session=None):
+ def login(self, get_credentials: onlinejudge.service.CredentialsProvider, session: Optional[requests.Session] = None) -> bool:
session = session or utils.new_default_session()
# NOTE: you can see this login page with https://community.topcoder.com/longcontest/?module=Submit
@@ -41,19 +42,20 @@ def login(self, get_credentials, session=None):
log.failure('Failure')
return False
- def get_url(self):
+ def get_url(self) -> str:
return 'https://www.topcoder.com/'
- def get_name(self):
+ def get_name(self) -> str:
return 'topcoder'
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['TopCoderService']:
# example: https://www.topcoder.com/
result = urllib.parse.urlparse(s)
if result.scheme in ('', 'http', 'https') \
and result.netloc in [ 'www.topcoder.com', 'community.topcoder.com' ]:
return cls()
+ return None
class TopCoderLongContestProblem(onlinejudge.problem.Problem):
@@ -63,14 +65,14 @@ def __init__(self, rd, cd=None, compid=None, pm=None):
self.compid = compid
self.pm = pm
- def get_url(self):
+ def get_url(self) -> str:
return 'https://community.topcoder.com/tc?module=MatchDetails&rd=' + str(self.rd)
- def get_service(self):
+ def get_service(self) -> TopCoderService:
return TopCoderService()
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['TopCoderLongContestProblem']:
# example: https://community.topcoder.com/longcontest/?module=ViewProblemStatement&rd=16997&pm=14690
# example: https://community.topcoder.com/longcontest/?module=ViewProblemStatement&rd=16997&compid=57374
# example: https://community.topcoder.com/longcontest/?module=ViewStandings&rd=16997
@@ -86,8 +88,9 @@ def from_url(cls, s):
if name in querystring:
kwargs[name] = int(querystring[name])
return cls(**kwargs)
+ return None
- def get_language_dict(self, session=None):
+ def get_language_dict(self, session: Optional[requests.Session] = None) -> Dict[str, Dict[str, str]]:
session = session or utils.new_default_session()
# at 2017/09/21
@@ -99,7 +102,7 @@ def get_language_dict(self, session=None):
'Python': { 'value': '6', 'description': 'Pyhton 2' },
}
- def submit(self, code, language, kind='example', session=None):
+ def submit(self, code: str, language: str, session: Optional[requests.Session] = None, kind: str = 'example') -> onlinejudge.submission.Submission:
assert kind in [ 'example', 'full' ]
session = session or utils.new_default_session()
@@ -117,7 +120,7 @@ def submit(self, code, language, kind='example', session=None):
path = [ tag.attrs['href'] for tag in soup.find_all('a', text='Submit') if ('rd=%d' % self.rd) in tag.attrs['href'] ]
if len(path) == 0:
log.error('link to submit not found: Are you logged in? Are you registered? Is the contest running?')
- return None
+ raise onlinejudge.problem.SubmissionError
assert len(path) == 1
path = path[0]
assert path.startswith('/') and 'module=Submit' in path
@@ -149,20 +152,20 @@ def submit(self, code, language, kind='example', session=None):
if 'module=SubmitSuccess' in resp.content.decode(resp.encoding):
url = 'http://community.topcoder.com/longcontest/?module=SubmitSuccess&rd={}&cd={}&compid={}'.format(self.rd, self.cd, self.compid)
log.success('success: result: %s', url)
- return onlinejudge.submission.CompatibilitySubmission(url)
+ return onlinejudge.submission.CompatibilitySubmission(url, self)
else:
# module=Submit to get error messages
resp = utils.request('GET', submit_url, session=session)
soup = bs4.BeautifulSoup(resp.content.decode(resp.encoding), utils.html_parser)
messages = soup.find('textarea', { 'name': 'messages' }).text
log.failure('%s', messages)
- return None
+ raise onlinejudge.problem.SubmissionError
- def get_standings(self, session=None):
+ def get_standings(self, session: Optional[requests.Session] = None) -> onlinejudge.problem.Standings:
session = session or utils.new_default_session()
- header = None
- rows = []
+ header: Optional[List[str]] = None
+ rows: List[Dict[str, str]] = []
for start in itertools.count(1, 100):
# get
url = 'https://community.topcoder.com/longcontest/?sc=&sd=&nr=100&sr={}&rd={}&module=ViewStandings'.format(start, self.rd)
@@ -176,7 +179,7 @@ def get_standings(self, session=None):
tr = trs[1]
header = [ td.text.strip() for td in tr.find_all('td') ]
for tr in trs[2 :]:
- row = collections.OrderedDict()
+ row: Dict[str, str] = collections.OrderedDict()
for key, td in zip(header, tr.find_all('td')):
value = td.text.strip()
if not value:
@@ -191,6 +194,7 @@ def get_standings(self, session=None):
if link is None:
break
+ assert header is not None
return header, rows
onlinejudge.dispatch.services += [ TopCoderService ]
diff --git a/onlinejudge/yukicoder.py b/onlinejudge/yukicoder.py
index 2892906d..ff3fb33a 100644
--- a/onlinejudge/yukicoder.py
+++ b/onlinejudge/yukicoder.py
@@ -2,6 +2,7 @@
# -*- coding: utf-8 -*-
import onlinejudge.service
import onlinejudge.problem
+from onlinejudge.problem import LabeledString, TestCase
import onlinejudge.submission
import onlinejudge.dispatch
import onlinejudge.implementation.utils as utils
@@ -16,12 +17,13 @@
import urllib.parse
import zipfile
import collections
+from typing import *
@utils.singleton
class YukicoderService(onlinejudge.service.Service):
- def login(self, get_credentials, session=None, method=None):
+ def login(self, get_credentials: onlinejudge.service.CredentialsProvider, session: Optional[requests.Session] = None, method: Optional[str] = None) -> bool:
if method == 'github':
return self.login_with_github(get_credentials, session=session)
elif method == 'twitter':
@@ -29,7 +31,7 @@ def login(self, get_credentials, session=None, method=None):
else:
assert False
- def login_with_github(self, get_credentials, session=None):
+ def login_with_github(self, get_credentials: onlinejudge.service.CredentialsProvider, session: Optional[requests.Session] = None) -> bool:
session = session or utils.new_default_session()
url = 'https://yukicoder.me/auth/github'
# get
@@ -60,32 +62,34 @@ def login_with_github(self, get_credentials, session=None):
log.failure('You failed to sign in. Wrong user ID or password.')
return False
- def login_with_twitter(self, get_credentials, session=None):
+ def login_with_twitter(self, get_credentials: onlinejudge.service.CredentialsProvider, session: Optional[requests.Session] = None) -> bool:
session = session or utils.new_default_session()
url = 'https://yukicoder.me/auth/twitter'
raise NotImplementedError
- def get_url(self):
+ def get_url(self) -> str:
return 'https://yukicoder.me/'
- def get_name(self):
+ def get_name(self) -> str:
return 'yukicoder'
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['YukicoderService']:
# example: http://yukicoder.me/
result = urllib.parse.urlparse(s)
if result.scheme in ('', 'http', 'https') \
and result.netloc == 'yukicoder.me':
return cls()
+ return None
- def _issue_official_api(self, api, id=None, name=None, session=None):
+ def _issue_official_api(self, api: str, id: Optional[int] = None, name: Optional[str] = None, session: Optional[requests.Session] = None) -> Any:
assert (id is not None) != (name is not None)
if id is not None:
assert isinstance(id, int)
sometihng = { 'user': '', 'solved': 'id/' }[api]
url = 'https://yukicoder.me/api/v1/{}/{}{}'.format(api, sometihng, id)
else:
+ assert name is not None
url = 'https://yukicoder.me/api/v1/{}/name/{}'.format(api, urllib.parse.quote(name))
session = session or utils.new_default_session()
try:
@@ -96,16 +100,16 @@ def _issue_official_api(self, api, id=None, name=None, session=None):
return json.loads(resp.content.decode(resp.encoding))
# example: {"Id":10,"Name":"yuki2006","Solved":280,"Level":34,"Rank":59,"Score":52550,"Points":7105,"Notice":"匿名ユーザーの情報は取れません。ユーザー名が重複している場合は最初に作られたIDが優先されます(その場合は運営にご報告いただければマージします)。このAPIはベータ版です。予告なく変更される場合があります。404を返したら廃止です。"}
- def get_user(self, *args, **kwargs):
+ def get_user(self, *args, **kwargs) -> Dict[str, Any]:
return self._issue_official_api('user', *args, **kwargs)
# https://twitter.com/yukicoder/status/935943170210258944
# example: [{"No":46,"ProblemId":43,"Title":"はじめのn歩","AuthorId":25,"TesterId":0,"Level":1,"ProblemType":0,"Tags":"実装"}]
- def get_solved(self, *args, **kwargs):
+ def get_solved(self, *args, **kwargs) -> List[Dict[str, Any]]:
return self._issue_official_api('solved', *args, **kwargs)
# example: https://yukicoder.me/users/237/favorite
- def get_user_favorite(self, id, session=None):
+ def get_user_favorite(self, id: int, session: Optional[requests.Session] = None) -> List[Any]:
url = 'https://yukicoder.me/users/%d/favorite' % id
columns, rows = self._get_and_parse_the_table(url, session=session)
assert columns == [ '#', '提出時間', '提出者', '問題', '言語', '結果', '実行時間', 'コード長' ]
@@ -120,7 +124,7 @@ def get_user_favorite(self, id, session=None):
return rows
# example: https://yukicoder.me/users/504/favoriteProblem
- def get_user_favorite_problem(self, id, session=None):
+ def get_user_favorite_problem(self, id, session: Optional[requests.Session] = None) -> List[Any]:
url = 'https://yukicoder.me/users/%d/favoriteProblem' % id
columns, rows = self._get_and_parse_the_table(url, session=session)
assert columns == [ 'ナンバー', '問題名', 'レベル', 'タグ', '時間制限', 'メモリ制限', '作問者' ]
@@ -142,7 +146,7 @@ def get_user_favorite_problem(self, id, session=None):
return rows
# example: https://yukicoder.me/users/1786/favoriteWiki
- def get_user_favorite_wiki(self, id, session=None):
+ def get_user_favorite_wiki(self, id: int, session: Optional[requests.Session] = None) -> List[Any]:
url = 'https://yukicoder.me/users/%d/favoriteWiki' % id
columns, rows = self._get_and_parse_the_table(url, session=session)
assert columns == [ 'Wikiページ' ]
@@ -155,7 +159,7 @@ def get_user_favorite_wiki(self, id, session=None):
# example: https://yukicoder.me/submissions?page=4220
# example: https://yukicoder.me/submissions?page=2192&status=AC
# NOTE: 1ページしか読まない 全部欲しい場合は呼び出し側で頑張る
- def get_submissions(self, page, status=None, session=None):
+ def get_submissions(self, page: int, status: Optional[str] = None, session: Optional[requests.Session] = None) -> List[Any]:
assert isinstance(page, int) and page >= 1
url = 'https://yukicoder.me/submissions?page=%d' % page
if status is not None:
@@ -177,7 +181,7 @@ def get_submissions(self, page, status=None, session=None):
# example: https://yukicoder.me/problems?page=2
# NOTE: loginしてると
- def get_problems(self, page, comp_problem=True, other=False, sort=None, session=None):
+ def get_problems(self, page: int, comp_problem: bool = True, other: bool = False, sort: Optional[str] = None, session: Optional[requests.Session] = None) -> List[Any]:
assert isinstance(page, int) and page >= 1
url = 'https://yukicoder.me/problems'
if other:
@@ -206,7 +210,7 @@ def get_problems(self, page, comp_problem=True, other=False, sort=None, session=
row[column] = row[column].text.strip()
return rows
- def _get_and_parse_the_table(self, url, session=None):
+ def _get_and_parse_the_table(self, url: str, session: Optional[requests.Session] = None) -> Tuple[List[Any], List[Dict[str, bs4.Tag]]]:
# get
session = session or utils.new_default_session()
resp = utils.request('GET', url, session=session)
@@ -215,18 +219,18 @@ def _get_and_parse_the_table(self, url, session=None):
assert len(soup.find_all('table')) == 1
table = soup.find('table')
columns = [ th.text.strip() for th in table.find('thead').find('tr') if th.name == 'th' ]
- data = []
+ data: List[Dict[str, List[str]]] = []
for row in table.find('tbody').find_all('tr'):
values = [ td for td in row if td.name == 'td' ]
assert len(columns) == len(values)
data += [ dict(zip(columns, values)) ]
return columns, data
- def _parse_star(self, tag):
+ def _parse_star(self, tag: bs4.Tag) -> str:
star = str(len(tag.find_all(class_='fa-star')))
if tag.find_all(class_='fa-star-half-full'):
star += '.5'
- return star # str
+ return star
class YukicoderProblem(onlinejudge.problem.Problem):
def __init__(self, problem_no=None, problem_id=None):
@@ -236,12 +240,12 @@ def __init__(self, problem_no=None, problem_id=None):
self.problem_no = problem_no
self.problem_id = problem_id
- def download(self, session=None, is_system=False):
+ def download(self, session: Optional[requests.Session] = None, is_system: bool = False) -> List[TestCase]:
if is_system:
return self.download_system(session=session)
else:
return self.download_samples(session=session)
- def download_samples(self, session=None):
+ def download_samples(self, session: Optional[requests.Session] = None) -> List[TestCase]:
session = session or utils.new_default_session()
# get
resp = utils.request('GET', self.get_url(), session=session)
@@ -255,30 +259,34 @@ def download_samples(self, session=None):
data, name = it
samples.add(data, name)
return samples.get()
- def download_system(self, session=None):
+ def download_system(self, session: Optional[requests.Session] = None) -> List[TestCase]:
session = session or utils.new_default_session()
# get
url = 'https://yukicoder.me/problems/no/{}/testcase.zip'.format(self.problem_no)
resp = utils.request('GET', url, session=session)
# parse
- samples = collections.defaultdict(dict)
+ basenames: Dict[str, Dict[str, LabeledString]] = collections.defaultdict(dict)
with zipfile.ZipFile(io.BytesIO(resp.content)) as fh:
for filename in sorted(fh.namelist()): # "test_in" < "test_out"
dirname = os.path.dirname(filename)
basename = os.path.basename(filename)
kind = { 'test_in': 'input', 'test_out': 'output' }[dirname]
- data = fh.read(filename).decode()
+ content = fh.read(filename).decode()
name = basename
if os.path.splitext(name)[1] == '.in': # ".in" extension is confusing
name = os.path.splitext(name)[0]
- print(filename, name)
- samples[basename][kind] = { 'data': data, 'name': name }
- for sample in samples.values():
- if 'input' not in sample or 'output' not in sample:
- log.error('dangling sample found: %s', str(sample))
- return list(map(lambda it: it[1], sorted(samples.items())))
+ print(filename, name) # TODO: what is this?
+ basenames[basename][kind] = LabeledString(name, content)
+ samples: List[TestCase] = []
+ for basename in sorted(basenames.keys()):
+ data = basenames[basename]
+ if 'input' not in data or 'output' not in data or len(data) != 2:
+ log.error('dangling sample found: %s', str(data))
+ else:
+ samples += [ TestCase(data['input'], data['output']) ]
+ return samples
- def _parse_sample_tag(self, tag):
+ def _parse_sample_tag(self, tag: bs4.Tag) -> Optional[Tuple[str, str]]:
assert isinstance(tag, bs4.Tag)
assert tag.name == 'pre'
prv = utils.previous_sibling_tag(tag)
@@ -288,38 +296,41 @@ def _parse_sample_tag(self, tag):
log.debug('name.encode(): %s', prv.string.encode())
s = tag.string or '' # tag.string for the tag "" returns None
return utils.textfile(s.lstrip()), pprv.string + ' ' + prv.string
+ return None
- def get_url(self):
+ def get_url(self) -> str:
if self.problem_no:
return 'https://yukicoder.me/problems/no/{}'.format(self.problem_no)
elif self.problem_id:
return 'https://yukicoder.me/problems/{}'.format(self.problem_id)
else:
- assert False
+ raise ValueError
@classmethod
- def from_url(cls, s):
+ def from_url(cls, s: str) -> Optional['YukicoderProblem']:
# example: https://yukicoder.me/problems/no/499
# example: http://yukicoder.me/problems/1476
result = urllib.parse.urlparse(s)
dirname, basename = posixpath.split(utils.normpath(result.path))
if result.scheme in ('', 'http', 'https') \
and result.netloc == 'yukicoder.me':
+ n: Optional[int] = None
try:
n = int(basename)
except ValueError:
- n = None
+ pass
if n is not None:
if dirname == '/problems/no':
return cls(problem_no=int(n))
if dirname == '/problems':
return cls(problem_id=int(n))
return cls()
+ return None
- def get_service(self):
+ def get_service(self) -> YukicoderService:
return YukicoderService()
- def get_input_format(self, session=None):
+ def get_input_format(self, session: Optional[requests.Session] = None) -> Optional[str]:
session = session or utils.new_default_session()
# get
resp = utils.request('GET', self.get_url(), session=session)
@@ -328,6 +339,7 @@ def get_input_format(self, session=None):
for h4 in soup.find_all('h4'):
if h4.string == '入力':
return h4.parent.find('pre').string
+ return None
onlinejudge.dispatch.services += [ YukicoderService ]