Merge pull request #250 from kmyk/issue/245

#240 #247 #245

onlinejudge/implementation/command/download.py

@@ -3,14 +3,17 @@
 import onlinejudge.type
 import onlinejudge.implementation.utils as utils
 import onlinejudge.implementation.logging as log
+import onlinejudge.implementation.download_history
 import colorama
+import datetime
 import json
 import os
+import pathlib
+import random
 import sys
 from typing import *
 if TYPE_CHECKING:
     import argparse
-    import pathlib
 
 def convert_sample_to_dict(sample: onlinejudge.type.TestCase) -> dict:
     data = {}
@@ -25,6 +28,9 @@ def download(args: 'argparse.Namespace') -> None:
     problem = onlinejudge.dispatch.problem_from_url(args.url)
     if problem is None:
         sys.exit(1)
+    is_default_format = args.format is None and args.directory is None  # must be here since args.directory and args.format are overwritten
+    if args.directory is None:
+        args.directory = pathlib.Path('test')
     if args.format is None:
         if args.system:
             if problem.get_service().get_name() == 'yukicoder':
@@ -41,6 +47,11 @@ def download(args: 'argparse.Namespace') -> None:
         else:
             samples = problem.download_sample_cases(session=sess)  # type: ignore
 
+    # append the history for submit command
+    if not args.dry_run and is_default_format:
+        history = onlinejudge.implementation.download_history.DownloadHistory()
+        history.add(problem)
+
     # write samples to files
     for i, sample in enumerate(samples):
         log.emit('')

onlinejudge/implementation/command/submit.py

@@ -2,6 +2,7 @@
 import onlinejudge
 import onlinejudge.implementation.utils as utils
 import onlinejudge.implementation.logging as log
+import onlinejudge.implementation.download_history
 import pathlib
 import re
 import shutil
@@ -15,6 +16,18 @@
 default_url_opener = [ 'sensible-browser', 'xdg-open', 'open' ]
 
 def submit(args: 'argparse.Namespace') -> None:
+    # guess url
+    history = onlinejudge.implementation.download_history.DownloadHistory()
+    guessed_urls = history.get()
+    if args.url is None:
+        if len(guessed_urls) == 1:
+            args.url = guessed_urls[0]
+            log.info('guessed problem: %s', args.url)
+        else:
+            log.error('failed to guess the URL to submit')
+            log.info('please manually specify URL as: $ oj submit URL FILE')
+            sys.exit(1)
+
     # parse url
     problem = onlinejudge.dispatch.problem_from_url(args.url)
     if problem is None:
@@ -36,7 +49,13 @@ def submit(args: 'argparse.Namespace') -> None:
         log.failure('%s: %s', e.__class__.__name__, str(e))
         s = repr(code)[ 1 : ]
     log.info('code (%d byte):', len(code))
-    log.emit(log.bold(s))
+    lines = s.splitlines(keepends=True)
+    if len(lines) < 30:
+        log.emit(log.bold(s))
+    else:
+        log.emit(log.bold(''.join(lines[: 10])))
+        log.emit('... (%s lines) ...', len(lines[10 : -10]))
+        log.emit(log.bold(''.join(lines[-10 :])))
 
 
     with utils.with_cookiejar(utils.new_default_session(), path=args.cookie) as sess:
@@ -70,7 +89,7 @@ def submit(args: 'argparse.Namespace') -> None:
         # report selected language ids
         if matched_lang_ids is not None and len(matched_lang_ids) == 1:
             args.language = matched_lang_ids[0]
-            log.info('choosed language: %s (%s)', args.language, langs[args.language]['description'])
+            log.info('chosen language: %s (%s)', args.language, langs[args.language]['description'])
         else:
             if matched_lang_ids is None:
                 log.error('language is unknown')
@@ -86,16 +105,29 @@ def submit(args: 'argparse.Namespace') -> None:
             sys.exit(1)
 
         # confirm
+        guessed_unmatch = ([ problem.get_url() ] != guessed_urls)
+        if guessed_unmatch:
+            log.warning('the problem "%s" is specified to submit, but samples of "%s" were downloaded in this directory. this may be mis-operation', problem.get_url(), '", "'.join(guessed_urls))
         if args.wait:
             log.status('sleep(%.2f)', args.wait)
             time.sleep(args.wait)
         if not args.yes:
-            sys.stdout.write('Are you sure? [y/N] ')
-            sys.stdout.flush()
-            c = sys.stdin.read(1)
-            if c != 'y':
-                log.info('terminated.')
-                return
+            if guessed_unmatch:
+                problem_id = problem.get_url().rstrip('/').split('/')[-1].split('?')[-1]  # this is too ad-hoc
+                key = problem_id[: 3] + (problem_id[-1] if len(problem_id) >= 4 else '')
+                sys.stdout.write('Are you sure? Please type "{}" '.format(key))
+                sys.stdout.flush()
+                c = sys.stdin.readline().rstrip()
+                if c != key:
+                    log.info('terminated.')
+                    return
+            else:
+                sys.stdout.write('Are you sure? [y/N] ')
+                sys.stdout.flush()
+                c = sys.stdin.read(1)
+                if c.lower() != 'y':
+                    log.info('terminated.')
+                    return
 
         # submit
         kwargs = {}
@@ -108,7 +140,7 @@ def submit(args: 'argparse.Namespace') -> None:
             submission = problem.submit_code(code, language=args.language, session=sess, **kwargs)  # type: ignore
         except onlinejudge.type.SubmissionError:
             log.failure('submission failed')
-            return
+            sys.exit(1)
 
         # show result
         if args.open:
@@ -159,12 +191,12 @@ def guess_lang_ids_of_file(filename: pathlib.Path, code: bytes, language_dict, c
 
         # compiler
         if select('gcc', lang_ids) and select('clang', lang_ids):
-            log.info('both GCC and Clang are available for C++ compiler')
+            log.status('both GCC and Clang are available for C++ compiler')
             if cxx_compiler.lower() == 'gcc':
-                log.info('use: GCC')
+                log.status('use: GCC')
                 lang_ids = select('gcc', lang_ids)
             elif cxx_compiler.lower() == 'clang':
-                log.info('use: Clang')
+                log.status('use: Clang')
                 lang_ids = select('clang', lang_ids)
             else:
                 assert cxx_compiler.lower() == 'all'
@@ -192,7 +224,7 @@ def guess_lang_ids_of_file(filename: pathlib.Path, code: bytes, language_dict, c
     elif ext == 'py':
         log.debug('language guessing: Python')
         if select('pypy', language_dict.keys()):
-            log.info('PyPy is available for Python interpreter')
+            log.status('PyPy is available for Python interpreter')
 
         # interpreter
         lang_ids = []
@@ -203,7 +235,7 @@ def guess_lang_ids_of_file(filename: pathlib.Path, code: bytes, language_dict, c
 
         # version
         if select('python2', lang_ids) and select('python3', lang_ids):
-            log.info('both Python2 and Python3 are available for version of Python')
+            log.status('both Python2 and Python3 are available for version of Python')
             if python_version in ( '2', '3' ):
                 versions = [ int(python_version) ]
             elif python_version == 'all':
@@ -214,14 +246,15 @@ def guess_lang_ids_of_file(filename: pathlib.Path, code: bytes, language_dict, c
                 if code.startswith(b'#!'):
                     s = lines[0]  # use shebang
                 else:
-                    s = b'\n'.join(lines[: 5] + lines[-5 :])  # use modelines
+                    s = b'\n'.join(lines[: 10] + lines[-5 :])  # use modelines
                 versions = []
                 for version in ( 2, 3 ):
-                    if re.search(r'python ?%d'.encode() % version, s.lower()):
+                    if re.search(r'python *(version:? *)?%d'.encode() % version, s.lower()):
                         versions += [ version ]
                 if not versions:
+                    log.status('no version info in code')
                     versions = [ 2, 3 ]
-            log.info('use: %s', ', '.join(map(str, versions)))
+            log.status('use: %s', ', '.join(map(str, versions)))
 
             saved_ids = lang_ids
             lang_ids = []

onlinejudge/implementation/download_history.py

@@ -0,0 +1,55 @@
+# Python Version: 3.x
+import onlinejudge
+import onlinejudge.type
+import onlinejudge.implementation.utils as utils
+import onlinejudge.implementation.logging as log
+import datetime
+import json
+import pathlib
+import time
+import traceback
+from typing import *
+
+class DownloadHistory(object):
+    def __init__(self, path: pathlib.Path = utils.cache_dir / 'download-history.jsonl'):
+        self.path = path
+
+    def add(self, problem: onlinejudge.type.Problem, directory: pathlib.Path = pathlib.Path.cwd()) -> None:
+        now = datetime.datetime.now(datetime.timezone.utc).astimezone()
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with open(str(self.path), 'a') as fh:
+            fh.write(json.dumps({
+                'timestamp': int(time.time()),  # this should not be int, but Python's strptime is too weak and datetime.fromisoformat is from 3.7
+                'directory': str(directory),
+                'url': problem.get_url(),
+            }) + '\n')
+        log.status('append history to: %s', self.path)
+        self._flush()
+
+    def _flush(self) -> None:
+        # halve the size if it is more than 1MiB
+        if self.path.stat().st_size >= 1024 * 1024:
+            with open(str(self.path)) as fh:
+                history_lines = fh.readlines()
+            with open(str(self.path), 'w') as fh:
+                fh.write(''.join(history_lines[: - len(history_lines) // 2]))
+            log.status('halve history at: %s', self.path)
+
+    def get(self, directory: pathlib.Path = pathlib.Path.cwd()) -> List[str]:
+        if not self.path.exists():
+            return []
+
+        log.status('read history from: %s', self.path)
+        found = set()
+        with open(str(self.path)) as fh:
+            for line in fh:
+                try:
+                    data = json.loads(line)
+                except json.decoder.JSONDecodeError as e:
+                    log.warning('corrupted line found in: %s', self.path)
+                    log.debug('%s', traceback.format_exc())
+                    continue
+                if pathlib.Path(data['directory']) == directory:
+                    found.add(data['url'])
+        log.status('found urls in history:\n%s', '\n'.join(found))
+        return list(found)

onlinejudge/implementation/main.py

@@ -67,7 +67,7 @@ def get_parser() -> argparse.ArgumentParser:
 ''')
     subparser.add_argument('url')
     subparser.add_argument('-f', '--format', help='a format string to specify paths of cases (defaut: "sample-%%i.%%e" if not --system)')  # default must be None for --system
-    subparser.add_argument('-d', '--directory', type=pathlib.Path, default=pathlib.Path('test'), help='a directory name for test cases (default: test/)')
+    subparser.add_argument('-d', '--directory', type=pathlib.Path, help='a directory name for test cases (default: test/)')  # default must be None for guessing in submit command
     subparser.add_argument('--overwrite', action='store_true')
     subparser.add_argument('-n', '--dry-run', action='store_true', help='don\'t write to files')
     subparser.add_argument('-a', '--system', action='store_true', help='download system testcases')
@@ -107,7 +107,7 @@ def get_parser() -> argparse.ArgumentParser:
   Yukicoder
   TopCoder (Marathon Match)
 ''')
-    subparser.add_argument('url')
+    subparser.add_argument('url', nargs='?', help='the URL of the problem to submit. if not given, guessed from history of download command.')
     subparser.add_argument('file', type=pathlib.Path)
     subparser.add_argument('-l', '--language', help='narrow down language choices if ambiguous')
     subparser.add_argument('--no-guess', action='store_false', dest='guess')

tests/command_download.py

@@ -1,12 +1,11 @@
 import unittest
 
+import tests.utils
+
 import hashlib
 import os
-import os.path
-import shutil
 import subprocess
 import sys
-import tempfile
 
 def get_files_from_json(samples):
     files = {}
@@ -24,13 +23,8 @@ def snippet_call_download(self, url, files, is_system=False, type='files'):
     if type == 'json':
         files = get_files_from_json(files)
 
-    cwd = os.getcwd()
-    ojtools = os.path.join( cwd, 'oj' )
-    try:
-        tempdir = tempfile.mkdtemp()
-        os.chdir(tempdir)
-        if os.path.exists('test'):
-            shutil.rmtree('test')
+    ojtools = os.path.abspath('oj')
+    with tests.utils.sandbox([]):
         cmd = [ ojtools, 'download', url ]
         if is_system:
             cmd += [ '--system' ]
@@ -41,6 +35,3 @@ def snippet_call_download(self, url, files, is_system=False, type='files'):
                 with open(os.path.join('test', name)) as fh:
                     result[name] = hashlib.md5(fh.buffer.read()).hexdigest()
         self.assertEqual(files, result)
-    finally:
-        os.chdir(cwd)
-        shutil.rmtree(tempdir)

tests/command_download_yukicoder.py

@@ -1,4 +1,5 @@
 import unittest
+
 import tests.command_download
 
 import os

tests/command_generate_output.py

@@ -1,47 +1,24 @@
 import unittest
 
-import contextlib
-import glob
+import tests.utils
+
 import os
 import subprocess
 import sys
-import tempfile
-
-
-@contextlib.contextmanager
-def chdir(path):
-    cwd = os.getcwd()
-    try:
-        os.chdir(path)
-        yield
-    finally:
-        for file in glob.glob('*/*.out'):
-            os.remove(file)
-        os.chdir(cwd)
 
-def prepare_files(input_files):
-    for f in input_files:
-        if os.path.dirname(f['path']):
-            os.makedirs(os.path.dirname(f['path']), exist_ok=True)
-        with open(f['path'], 'w') as fh:
-            fh.write(f['data'])
-        if f.get('executable', False):
-            os.chmod(f['path'], 0o755)
 
 class GenerateOutputTest(unittest.TestCase):
 
     def snippet_call_generate_output(self, args, input_files, expected_values, disallowed_files=None):
         ojtools = os.path.abspath('oj')
-        with tempfile.TemporaryDirectory() as tempdir:
-            with chdir(tempdir):
-                prepare_files(input_files)
-                _ = subprocess.check_output([ojtools, 'generate-output'] + args, stderr=sys.stderr)
-                for expect in expected_values:
-                    with open(expect['path']) as f:
-                        self.assertEqual(''.join(f.readlines()), expect['data'])
-                if disallowed_files is not None:
-                    for file in disallowed_files:
-                        self.assertFalse(os.path.exists(file))
+        with tests.utils.sandbox(input_files) as tempdir:
+            _ = subprocess.check_output([ojtools, 'generate-output'] + args, stderr=sys.stderr)
+            for expect in expected_values:
+                with open(expect['path']) as f:
+                    self.assertEqual(''.join(f.readlines()), expect['data'])
+            if disallowed_files is not None:
+                for file in disallowed_files:
+                    self.assertFalse(os.path.exists(file))
 
     def test_call_generate_output_simple(self):
         self.snippet_call_generate_output(