Merge branch 'develop'

.github/workflows/autotest.yaml

@@ -40,7 +40,7 @@ jobs:
         run: poetry install
 
       - name: Run the automated tests
-        run: poetry run pytest --cov --cov-report=xml -n 6 -v
+        run: poetry run pytest --cov --cov-report=xml -v
 
       - uses: codecov/codecov-action@v4
         with:

configs.py

@@ -0,0 +1,3 @@
+# 配置LRUN用户和组（不要将其设置为有root权限的用户！）
+LRUN_UID = 1001
+LRUN_GID = 1001

judge/runtime.py

@@ -1,3 +1,4 @@
+import os
 import pathlib
 import subprocess
 from typing import Literal, Union
@@ -7,60 +8,110 @@
 from .status import Status
 
 
-def simple_runtime(executeable_file: pathlib.Path, input_content: str, input_type: Literal['STDIN', 'FILE'], file_input_path: pathlib.Path = None, timeout: float = 1.0) -> Union[str, Status]:
-    """返回 STDOUT（STDOUT 无输出时返回第一个后缀为 .out 的文件的内容）"""
-    if file_input_path is not None:
-        if file_input_path.is_absolute():
-            judge_logger.warning(
-                f'执行评测时的输入文件路径 {file_input_path} 是绝对路径，正确的格式应当为 `Path("hello.in")` 一类。\n运行失败，返回状态码 RE。')
+class SimpleRuntime(object):
+    def __init__(self) -> None:
+        pass
+
+    def calling_precheck(self, executeable_file: pathlib.Path, input_content: str, input_type: Literal['STDIN', 'FILE'], file_input_path: pathlib.Path = None, timeout: float = 1.0) -> bool:
+        if file_input_path is not None:
+            if file_input_path.is_absolute():
+                judge_logger.warning(
+                    f'执行评测时的输入文件路径 {file_input_path} 是绝对路径，正确的格式应当为 `Path("hello.in")` 一类。\n运行失败，返回状态码 RE。')
+                return False
+
+        executeable_file.chmod(0o700)
+
+    def __call__(self, executeable_file: pathlib.Path, input_content: str, input_type: Literal['STDIN', 'FILE'], file_input_path: pathlib.Path = None, timeout: float = 1.0) -> Union[str, Status]:
+        """返回 STDOUT（STDOUT 无输出时返回第一个后缀为 .out 的文件的内容）"""
+        if self.calling_precheck(executeable_file, input_content, input_type, file_input_path, timeout) is False:
             return Status.RuntimeError
 
-    executeable_file.chmod(0o700)
+        if input_type == 'STDIN':
+            process = self.stdin_executor(executeable_file)
+            try:
+                stdout, stderr = process.communicate(
+                    input_content.encode('utf-8'), timeout=timeout)
+
+                if process.returncode != 0:
+                    return Status.RuntimeError
+
+                return stdout.decode('utf-8')
+            except subprocess.TimeoutExpired:
+                return Status.TimeLimitExceeded
+        else:
+            process = self.file_input_executor(
+                executeable_file, file_input_path, input_content)
+
+            try:
+                process.wait(timeout)
+
+                if process.returncode != 0:
+                    return Status.RuntimeError
+
+                try:
+                    return executeable_file.with_name(file_input_path.name).with_suffix('.out').read_text(encoding='utf-8')
+                except:
+                    return Status.WrongAnswer
+            except subprocess.TimeoutExpired:
+                return Status.TimeLimitExceeded
+
+    def file_input_executor(self, executeable_file: pathlib.Path, file_input_path: pathlib.Path, input_content: str) -> subprocess.Popen[bytes]:
+        executeable_file.parent.joinpath(file_input_path).write_text(
+            input_content, encoding='utf-8')
 
-    if input_type == 'STDIN':
         process = subprocess.Popen(
             executeable_file.absolute().__str__(),
             bufsize=-1,
-            stdin=subprocess.PIPE,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
             cwd=executeable_file.parent
         )
 
-        try:
-            stdout, stderr = process.communicate(
-                input_content.encode('utf-8'), timeout=timeout)
-            process.wait(timeout=timeout)
-        except subprocess.TimeoutExpired:
-            return Status.TimeLimitExceeded
-
-        if process.returncode != 0:
-            return Status.RuntimeError
-
-        return stdout.decode('utf-8')
-    else:
-        executeable_file.parent.joinpath(file_input_path).write_text(
-            input_content, encoding='utf-8')
+        return process
 
+    def stdin_executor(self, executeable_file: pathlib.Path) -> subprocess.Popen[bytes]:
         process = subprocess.Popen(
             executeable_file.absolute().__str__(),
             bufsize=-1,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
             cwd=executeable_file.parent
         )
 
-        try:
-            process.wait(timeout=timeout)
-        except subprocess.TimeoutExpired:
-            return Status.TimeLimitExceeded
+        return process
 
-        if process.returncode != 0:
-            return Status.RuntimeError
 
-        output_file = None
-        for file in executeable_file.parent.iterdir():
-            if file.suffix == '.out':
-                output_file = file
+# TODO: 暂缓开发安全运行时
+# class SafetyRuntimeWithLrun(SimpleRuntime):
+#     def __init__(self) -> None:
+#         super().__init__()
+
+#         if os.getuid() != 0 or os.getgid() != 0:  # pragma: no cover
+#             raise RuntimeError('必须为 Root 用户才能使用 SafetyRuntimeWithLrun。')
+
+#     def __call__(self, executeable_file: pathlib.Path, input_content: str, input_type: Literal['STDIN'] | Literal['FILE'], file_input_path: pathlib.Path = None, timeout: float = 1) -> str | Status:
+#         if self.calling_precheck(executeable_file, input_content, input_type, file_input_path, timeout) is False:
+#             return Status.RuntimeError
+
+#         if input_type == 'STDIN':
+#             self.stdin_executor(executeable_file, network=False, timeout=timeout, max_memory=0, uid=0, gid=0)
+
+#     def stdin_executor(self, executeable_file: pathlib.Path, network: bool, timeout: float, max_memory: int, uid: int, gid: int) -> subprocess.Popen[bytes]:
+#         # 注：max_memory 的单位为byte
+
+#         process = subprocess.Popen(
+#             [
+#                 'sudo',
+#                 'lrun',
+#                 '--network', 'true' if network else 'false',
+#                 '--max-cpu-time', str(timeout),
+#                 '--max-memory', str(max_memory),
+#                 '--isolate-process', 'true',
+#                 '--uid', str(uid),
+#                 '--gid', str(gid)
+#             ]
+#         )
+
+#         return process
+
 
-        if output_file is None:
-            return Status.RuntimeError  # pragma: no cover
-        return output_file.read_text(encoding='utf-8')
+simple_runtime = SimpleRuntime()

judge/status.py

@@ -9,3 +9,4 @@ class Status(Enum):
     TimeLimitExceeded = 'TLE'
     MemoryLimitExceeded = 'MLE'
     OutputLimitExceeded = 'OLE'
+    JudgeInternalError = 'JIE'

main.py

@@ -2,7 +2,8 @@
 import pathlib
 
 from ccf_parser import CCF
-from manager import start_server_background
+from manager import cli_app, start_server_background
 
 if '__main__' == __name__:
-    start_server_background()
+    cli_app()
+    # start_server_background()

manager/__init__.py

@@ -1 +1,2 @@
 from .base import start_server_background
+from .cli import cli_app

manager/cli/__init__.py

@@ -0,0 +1 @@
+from .base import app as cli_app

manager/cli/base.py

@@ -0,0 +1,26 @@
+from typing import Optional
+
+import requests
+import typer
+from rich import print
+from rich.markdown import Markdown
+from rich.panel import Panel
+from rich.text import Text
+
+from .judge import app as judge_typer
+
+app = typer.Typer()
+app.add_typer(judge_typer)
+
+
+@app.command(name='intro', help='查看ItsWA介绍')
+def intro():
+    md = Markdown(
+        """
+## 欢迎使用 ItsWA 评测系统
+ItsWA是一个基于Python搭建，使用`Lrun`提供安全运行时的Linux下的竞赛代码评测系统。
+查看`docs/guide.pdf`获取使用教程
+"""
+    )
+
+    print(md)

manager/cli/judge.py

@@ -0,0 +1,26 @@
+import json
+from pathlib import Path
+from typing import Annotated, Optional, Union
+
+import typer
+
+from ccf_parser import CCF
+from judge import start_judging
+from utils import manager_logger
+
+app = typer.Typer(name='judge', help='评测相关')
+
+
+@app.command('start', help='开始评测')
+def start_judging_command(path: Annotated[Path, typer.Argument(help='比赛目录或CCF文件')] = Path('.')):
+    ccf: Optional[CCF] = None
+
+    if path.name == 'ccf.json':
+        ccf = CCF(**json.loads(path.read_text('utf-8')))
+    elif path.is_dir() and path.joinpath('ccf.json').exists():
+        ccf = CCF(**json.loads(path.joinpath('ccf.json').read_text('utf-8')))
+    else:
+        raise FileNotFoundError('评测目录不正确，可能是不存在CCF文件')
+
+    start_judging(ccf)
+    return 0