feat(codeforces): implement submit; cache CSES token (#300)

## Problem

Codeforces submit was a stub. CSES submit re-ran the full login flow on
every invocation (~1.5s overhead).

## Solution

**Codeforces**: headless browser submit via StealthySession (same
pattern as AtCoder). Solves Cloudflare Turnstile on login, uploads
source via file input, caches cookies at
`~/.cache/cp-nvim/codeforces-cookies.json` so repeat submits skip login.

**CSES**: persist the API token in credentials via a `credentials`
ndjson event. Subsequent submits validate the cached token with a single
GET before falling back to full login.

Also includes a vimdoc table of contents.
This commit is contained in:
Barrett Ruth 2026-03-05 10:37:39 -05:00 committed by GitHub
parent e9f72dfbbc
commit 6fcb5d1bbc
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 307 additions and 96 deletions

View file

@ -2,7 +2,9 @@
import asyncio
import json
import os
import re
import tempfile
from typing import Any
import requests
@ -18,10 +20,16 @@ from .models import (
SubmitResult,
TestCase,
)
from .timeouts import (
BROWSER_ELEMENT_WAIT,
BROWSER_NAV_TIMEOUT,
BROWSER_SESSION_TIMEOUT,
BROWSER_SETTLE_DELAY,
HTTP_TIMEOUT,
)
BASE_URL = "https://codeforces.com"
API_CONTEST_LIST_URL = f"{BASE_URL}/api/contest.list"
TIMEOUT_SECONDS = 30
HEADERS = {
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
}
@ -136,7 +144,7 @@ def _is_interactive(block: Tag) -> bool:
def _fetch_problems_html(contest_id: str) -> str:
url = f"{BASE_URL}/contest/{contest_id}/problems"
response = curl_requests.get(url, impersonate="chrome", timeout=TIMEOUT_SECONDS)
response = curl_requests.get(url, impersonate="chrome", timeout=HTTP_TIMEOUT)
response.raise_for_status()
return response.text
@ -223,7 +231,7 @@ class CodeforcesScraper(BaseScraper):
async def scrape_contest_list(self) -> ContestListResult:
try:
r = requests.get(API_CONTEST_LIST_URL, timeout=TIMEOUT_SECONDS)
r = requests.get(API_CONTEST_LIST_URL, timeout=HTTP_TIMEOUT)
r.raise_for_status()
data = r.json()
if data.get("status") != "OK":
@ -289,13 +297,187 @@ class CodeforcesScraper(BaseScraper):
language_id: str,
credentials: dict[str, str],
) -> SubmitResult:
return await asyncio.to_thread(
_submit_headless,
contest_id,
problem_id,
source_code,
language_id,
credentials,
)
def _wait_for_gate_reload(page, wait_selector: str) -> None:
from .atcoder import _solve_turnstile
if "Verification" not in page.title():
return
_solve_turnstile(page)
page.wait_for_function(
f"() => !!document.querySelector('{wait_selector}')",
timeout=BROWSER_ELEMENT_WAIT,
)
def _submit_headless(
contest_id: str,
problem_id: str,
source_code: str,
language_id: str,
credentials: dict[str, str],
) -> SubmitResult:
from pathlib import Path
try:
from scrapling.fetchers import StealthySession # type: ignore[import-untyped,unresolved-import]
except ImportError:
return SubmitResult(
success=False,
error="Codeforces submit not yet implemented",
submission_id="",
verdict="",
error="scrapling is required for Codeforces submit",
)
from .atcoder import _ensure_browser, _solve_turnstile
_ensure_browser()
cookie_cache = Path.home() / ".cache" / "cp-nvim" / "codeforces-cookies.json"
cookie_cache.parent.mkdir(parents=True, exist_ok=True)
saved_cookies: list[dict[str, Any]] = []
if cookie_cache.exists():
try:
saved_cookies = json.loads(cookie_cache.read_text())
except Exception:
pass
logged_in = False
login_error: str | None = None
submit_error: str | None = None
def check_login(page):
nonlocal logged_in
logged_in = page.evaluate(
"() => Array.from(document.querySelectorAll('a'))"
".some(a => a.textContent.includes('Logout'))"
)
def login_action(page):
nonlocal login_error
try:
_wait_for_gate_reload(page, "#enterForm")
except Exception:
pass
try:
page.fill(
'input[name="handleOrEmail"]',
credentials.get("username", ""),
)
page.fill(
'input[name="password"]',
credentials.get("password", ""),
)
page.locator('#enterForm input[type="submit"]').click()
page.wait_for_url(
lambda url: "/enter" not in url, timeout=BROWSER_NAV_TIMEOUT
)
except Exception as e:
login_error = str(e)
def submit_action(page):
nonlocal submit_error
try:
_solve_turnstile(page)
except Exception:
pass
tmp_path: str | None = None
try:
page.select_option(
'select[name="submittedProblemIndex"]',
problem_id.upper(),
)
page.select_option('select[name="programTypeId"]', language_id)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".cpp", delete=False, prefix="cf_"
) as tf:
tf.write(source_code)
tmp_path = tf.name
try:
page.set_input_files('input[name="sourceFile"]', tmp_path)
page.wait_for_timeout(BROWSER_SETTLE_DELAY)
except Exception:
page.fill('textarea[name="source"]', source_code)
page.locator("form.submit-form input.submit").click(no_wait_after=True)
try:
page.wait_for_url(
lambda url: "/my" in url or "/status" in url,
timeout=BROWSER_NAV_TIMEOUT * 2,
)
except Exception:
err_el = page.query_selector("span.error")
if err_el:
submit_error = err_el.inner_text().strip()
else:
submit_error = "Submit failed: page did not navigate"
except Exception as e:
submit_error = str(e)
finally:
if tmp_path:
try:
os.unlink(tmp_path)
except OSError:
pass
try:
with StealthySession(
headless=True,
timeout=BROWSER_SESSION_TIMEOUT,
google_search=False,
cookies=saved_cookies,
) as session:
print(json.dumps({"status": "checking_login"}), flush=True)
session.fetch(
f"{BASE_URL}/",
page_action=check_login,
network_idle=True,
)
if not logged_in:
print(json.dumps({"status": "logging_in"}), flush=True)
session.fetch(
f"{BASE_URL}/enter",
page_action=login_action,
solve_cloudflare=True,
)
if login_error:
return SubmitResult(
success=False, error=f"Login failed: {login_error}"
)
print(json.dumps({"status": "submitting"}), flush=True)
session.fetch(
f"{BASE_URL}/contest/{contest_id}/submit",
page_action=submit_action,
solve_cloudflare=True,
)
try:
browser_cookies = session.context.cookies()
if any(c["name"] == "JSESSIONID" for c in browser_cookies):
cookie_cache.write_text(json.dumps(browser_cookies))
except Exception:
pass
if submit_error:
return SubmitResult(success=False, error=submit_error)
return SubmitResult(
success=True,
error="",
submission_id="",
verdict="submitted",
)
except Exception as e:
return SubmitResult(success=False, error=str(e))
if __name__ == "__main__":
CodeforcesScraper().run_cli()