barrett/cp.nvim
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 13

ci: format

Browse source
This commit is contained in:
Barrett Ruth 2026-03-03 14:58:57 -05:00 • committed by Barrett Ruth
parent ad90d564ca
commit bad219e578
9 changed files with 134 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

14
lua/cp/constants.lua
View file

@ -1,7 +1,19 @@
local M = {}
M.PLATFORMS = { 'atcoder', 'codechef', 'codeforces', 'cses', 'kattis', 'usaco' }
M.ACTIONS = { 'run', 'panel', 'next', 'prev', 'pick', 'cache', 'interact', 'edit', 'race', 'stress', 'submit' }
M.ACTIONS = {
'run',
'panel',
'next',
'prev',
'pick',
'cache',
'interact',
'edit',
'race',
'stress',
'submit',
}
M.PLATFORM_DISPLAY_NAMES = {
atcoder = 'AtCoder',

25
scrapers/atcoder.py
View file

@ -369,15 +369,26 @@ class AtcoderScraper(BaseScraper):
await asyncio.gather(*(emit(r) for r in rows))
async def submit(self, contest_id: str, problem_id: str, source_code: str, language_id: str, credentials: dict[str, str]) -> SubmitResult:
async def submit(
self,
contest_id: str,
problem_id: str,
source_code: str,
language_id: str,
credentials: dict[str, str],
) -> SubmitResult:
def _submit_sync() -> SubmitResult:
try:
login_page = _session.get(f"{BASE_URL}/login", headers=HEADERS, timeout=TIMEOUT_SECONDS)
login_page = _session.get(
f"{BASE_URL}/login", headers=HEADERS, timeout=TIMEOUT_SECONDS
)
login_page.raise_for_status()
soup = BeautifulSoup(login_page.text, "html.parser")
csrf_input = soup.find("input", {"name": "csrf_token"})
if not csrf_input:
return SubmitResult(success=False, error="Could not find CSRF token on login page")
return SubmitResult(
success=False, error="Could not find CSRF token on login page"
)
csrf_token = csrf_input.get("value", "")
login_resp = _session.post(
@ -401,7 +412,9 @@ class AtcoderScraper(BaseScraper):
soup = BeautifulSoup(submit_page.text, "html.parser")
csrf_input = soup.find("input", {"name": "csrf_token"})
if not csrf_input:
return SubmitResult(success=False, error="Could not find CSRF token on submit page")
return SubmitResult(
success=False, error="Could not find CSRF token on submit page"
)
csrf_token = csrf_input.get("value", "")
task_screen_name = f"{contest_id}_{problem_id}"
@ -418,7 +431,9 @@ class AtcoderScraper(BaseScraper):
)
submit_resp.raise_for_status()
return SubmitResult(success=True, error="", submission_id="", verdict="submitted")
return SubmitResult(
success=True, error="", submission_id="", verdict="submitted"
)
except Exception as e:
return SubmitResult(success=False, error=str(e))

27
scrapers/base.py
View file

@ -6,7 +6,13 @@ import sys
from abc import ABC, abstractmethod
from .language_ids import get_language_id
from .models import CombinedTest, ContestListResult, MetadataResult, SubmitResult, TestsResult
from .models import (
CombinedTest,
ContestListResult,
MetadataResult,
SubmitResult,
TestsResult,
)
_PRECISION_ABS_REL_RE = re.compile(
r"(?:absolute|relative)\s+error[^.]*?10\s*[\^{]\s*\{?\s*[-\u2212]\s*(\d+)\s*\}?",
@ -43,7 +49,14 @@ class BaseScraper(ABC):
async def stream_tests_for_category_async(self, category_id: str) -> None: ...
@abstractmethod
async def submit(self, contest_id: str, problem_id: str, source_code: str, language_id: str, credentials: dict[str, str]) -> SubmitResult: ...
async def submit(
self,
contest_id: str,
problem_id: str,
source_code: str,
language_id: str,
credentials: dict[str, str],
) -> SubmitResult: ...
def _usage(self) -> str:
name = self.platform_name
@ -102,7 +115,11 @@ class BaseScraper(ABC):
case "submit":
if len(args) != 5:
print(self._submit_error("Usage: <platform> submit <contest_id> <problem_id> <language_id>").model_dump_json())
print(
self._submit_error(
"Usage: <platform> submit <contest_id> <problem_id> <language_id>"
).model_dump_json()
)
return 1
source_code = sys.stdin.read()
creds_raw = os.environ.get("CP_CREDENTIALS", "{}")
@ -111,7 +128,9 @@ class BaseScraper(ABC):
except json.JSONDecodeError:
credentials = {}
language_id = get_language_id(self.platform_name, args[4]) or args[4]
result = await self.submit(args[2], args[3], source_code, language_id, credentials)
result = await self.submit(
args[2], args[3], source_code, language_id, credentials
)
print(result.model_dump_json())
return 0 if result.success else 1

16
scrapers/codechef.py
View file

@ -252,8 +252,20 @@ class CodeChefScraper(BaseScraper):
payload = await coro
print(json.dumps(payload), flush=True)
async def submit(self, contest_id: str, problem_id: str, source_code: str, language_id: str, credentials: dict[str, str]) -> SubmitResult:
return SubmitResult(success=False, error="CodeChef submit not yet implemented", submission_id="", verdict="")
async def submit(
self,
contest_id: str,
problem_id: str,
source_code: str,
language_id: str,
credentials: dict[str, str],
) -> SubmitResult:
return SubmitResult(
success=False,
error="CodeChef submit not yet implemented",
submission_id="",
verdict="",
)
if __name__ == "__main__":

16
scrapers/codeforces.py
View file

@ -281,8 +281,20 @@ class CodeforcesScraper(BaseScraper):
flush=True,
)
async def submit(self, contest_id: str, problem_id: str, source_code: str, language_id: str, credentials: dict[str, str]) -> SubmitResult:
return SubmitResult(success=False, error="Codeforces submit not yet implemented", submission_id="", verdict="")
async def submit(
self,
contest_id: str,
problem_id: str,
source_code: str,
language_id: str,
credentials: dict[str, str],
) -> SubmitResult:
return SubmitResult(
success=False,
error="Codeforces submit not yet implemented",
submission_id="",
verdict="",
)
if __name__ == "__main__":

16
scrapers/cses.py
View file

@ -262,8 +262,20 @@ class CSESScraper(BaseScraper):
payload = await coro
print(json.dumps(payload), flush=True)
async def submit(self, contest_id: str, problem_id: str, source_code: str, language_id: str, credentials: dict[str, str]) -> SubmitResult:
return SubmitResult(success=False, error="CSES submit not yet implemented", submission_id="", verdict="")
async def submit(
self,
contest_id: str,
problem_id: str,
source_code: str,
language_id: str,
credentials: dict[str, str],
) -> SubmitResult:
return SubmitResult(
success=False,
error="CSES submit not yet implemented",
submission_id="",
verdict="",
)
if __name__ == "__main__":

17
scrapers/kattis.py
View file

@ -222,9 +222,20 @@ class KattisScraper(BaseScraper):
flush=True,
)
async def submit(self, contest_id: str, problem_id: str, source_code: str, language_id: str, credentials: dict[str, str]) -> SubmitResult:
return SubmitResult(success=False, error="Kattis submit not yet implemented", submission_id="", verdict="")
async def submit(
self,
contest_id: str,
problem_id: str,
source_code: str,
language_id: str,
credentials: dict[str, str],
) -> SubmitResult:
return SubmitResult(
success=False,
error="Kattis submit not yet implemented",
submission_id="",
verdict="",
)
if __name__ == "__main__":

17
scrapers/usaco.py
View file

@ -279,9 +279,20 @@ class USACOScraper(BaseScraper):
payload = await coro
print(json.dumps(payload), flush=True)
async def submit(self, contest_id: str, problem_id: str, source_code: str, language_id: str, credentials: dict[str, str]) -> SubmitResult:
return SubmitResult(success=False, error="USACO submit not yet implemented", submission_id="", verdict="")
async def submit(
self,
contest_id: str,
problem_id: str,
source_code: str,
language_id: str,
credentials: dict[str, str],
) -> SubmitResult:
return SubmitResult(
success=False,
error="USACO submit not yet implemented",
submission_id="",
verdict="",
)
if __name__ == "__main__":

10
scripts/stress.py
View file

@ -41,7 +41,10 @@ def main() -> None:
timeout=timeout,
)
except subprocess.TimeoutExpired:
print(f"[stress] generator timed out on iteration {iteration}", file=sys.stderr)
print(
f"[stress] generator timed out on iteration {iteration}",
file=sys.stderr,
)
sys.exit(1)
if gen_result.returncode != 0:
@ -80,7 +83,10 @@ def main() -> None:
timeout=timeout,
)
except subprocess.TimeoutExpired:
print(f"[stress] candidate timed out on iteration {iteration}", file=sys.stderr)
print(
f"[stress] candidate timed out on iteration {iteration}",
file=sys.stderr,
)
print(f"\n--- input ---\n{test_input}", end="")
sys.exit(1)