Coverage for src / competitive_verifier / oj / problem.py: 63%

1import glob 

2import json 

3import os 

4import pathlib 

5import posixpath 

6import re 

7import shutil 

8import subprocess 

9import sys 

10import tempfile 

11import time 

12import urllib.parse 

13import zipfile 

14from abc import abstractmethod 

15from collections.abc import Iterable, Iterator 

16from dataclasses import dataclass 

17from logging import getLogger 

18from typing import ClassVar, Optional, TypeVar 

19 

20import requests 

21 

22from competitive_verifier import config 

23from competitive_verifier.log import GitHubMessageParams 

24from competitive_verifier.models import ( 

25 Problem, 

26 TestCaseData, 

27 TestCaseFile, 

28 TestCaseProvider, 

29) 

30 

31logger = getLogger(__name__) 

32 

33_ASCII_SPACE = 0x20 

34_ASCII_DELETE = 0x7F 

35 

36 

37class NotLoggedInError(RuntimeError): 

38 pass 

39 

40 

41class _BaseProblem(Problem): 

42 def iter_system_cases(self) -> Iterator[TestCaseFile]: 

43 return iter_testcases(directory=self.test_directory) 

44 

45 def download_system_cases(self) -> Iterable[TestCaseData] | bool: 

46 test_directory = self.test_directory 

47 

48 if test_directory.exists() and any(test_directory.iterdir()): 

49 logger.info("download:already exists: %s", self.url) 

50 return True 

51 

52 self.problem_directory.mkdir(parents=True, exist_ok=True) 

53 

54 samples = list(self._download_cases()) 

55 

56 # Check samples 

57 if not samples: 57 ↛ 58line 57 didn't jump to line 58 because the condition on line 57 was never true

58 logger.error( 

59 "Sample not found", 

60 extra={"github": GitHubMessageParams()}, 

61 ) 

62 return False 

63 

64 # write samples to files 

65 save_testcases(samples, directory=test_directory) 

66 return samples 

67 

68 @abstractmethod 

69 def _download_cases(self) -> Iterable[TestCaseData]: ... 

70 

71 

72class LibraryCheckerProblem(Problem): 

73 checker_exe_name: ClassVar[str] = ( 

74 "checker.exe" if sys.platform == "win32" else "checker" 

75 ) 

76 

77 def __init__(self, *, problem_id: str): 

78 self.problem_id = problem_id 

79 self._source_directory = None 

80 

81 def __hash__(self) -> int: 

82 return hash((self.problem_id, self.repo_path)) 

83 

84 def __eq__(self, value: object) -> bool: 

85 if not isinstance(value, LibraryCheckerProblem): 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true

86 return False 

87 return self.problem_id == value.problem_id and self.repo_path == value.repo_path 

88 

89 @property 

90 def repo_path(self): 

91 return config.get_cache_dir() / "library-checker-problems" 

92 

93 def iter_system_cases(self) -> Iterator[TestCaseFile]: 

94 inputs: dict[str, pathlib.Path] = {} 

95 outputs: dict[str, pathlib.Path] = {} 

96 for path in self.source_directory.glob("in/*.in"): 

97 inputs[path.stem] = path 

98 for path in self.source_directory.glob("out/*.out"): 

99 outputs[path.stem] = path 

100 return merge_testcase_files(inputs, outputs) 

101 

102 def download_system_cases(self) -> bool: 

103 self.problem_directory.mkdir(parents=True, exist_ok=True) 

104 self.generate_test_cases() 

105 return True 

106 

107 @property 

108 def checker(self) -> pathlib.Path | None: 

109 return self.source_directory / self.checker_exe_name 

110 

111 def generate_test_cases(self) -> None: 

112 self.update_cloned_repository() 

113 path = self.repo_path 

114 

115 spec = str(self.source_directory / "info.toml") 

116 command = [sys.executable, str(path / "generate.py"), spec] 

117 logger.info("$ %s", " ".join(command)) 

118 try: 

119 subprocess.check_call(command, stdout=sys.stderr, stderr=sys.stderr) 

120 except subprocess.CalledProcessError: 

121 logger.exception( 

122 "the generate.py failed: check https://github.com/yosupo06/library-checker-problems/issues", 

123 extra={"github": GitHubMessageParams()}, 

124 ) 

125 raise 

126 

127 @property 

128 def source_directory(self): 

129 if self._source_directory is None: 

130 problem_id = self.problem_id 

131 info_tomls = list( 

132 self.repo_path.glob(f"**/{glob.escape(problem_id)}/info.toml") 

133 ) 

134 if len(info_tomls) != 1: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true

135 raise RuntimeError(f"the problem {problem_id!r} not found or broken") 

136 self._source_directory = info_tomls[0].parent 

137 return self._source_directory 

138 

139 @property 

140 def url(self) -> str: 

141 return f"https://judge.yosupo.jp/problem/{self.problem_id}" 

142 

143 @classmethod 

144 def from_url(cls, url: str) -> Optional["LibraryCheckerProblem"]: 

145 # example: https://judge.yosupo.jp/problem/unionfind 

146 result = urllib.parse.urlparse(url) 

147 if result.scheme in ("", "http", "https") and result.netloc in ( 

148 "judge.yosupo.jp", 

149 "old.yosupo.jp", 

150 ): 

151 m = re.match(r"/problem/(\w+)/?", result.path) 

152 if m: 152 ↛ 154line 152 didn't jump to line 154 because the condition on line 152 was always true

153 return cls(problem_id=m.group(1)) 

154 return None 

155 

156 _is_repository_updated: ClassVar[set[pathlib.Path]] = set() 

157 

158 def update_cloned_repository(self) -> None: 

159 if self.repo_path in self._is_repository_updated: 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true

160 return 

161 

162 try: 

163 subprocess.check_call( 

164 ["git", "--version"], # noqa: S607 

165 stdout=sys.stderr, 

166 stderr=sys.stderr, 

167 ) 

168 except FileNotFoundError: 

169 logger.exception( 

170 "git command not found", 

171 exc_info=False, 

172 extra={"github": GitHubMessageParams()}, 

173 ) 

174 raise 

175 

176 path = self.repo_path 

177 if not path.exists(): 177 ↛ 188line 177 didn't jump to line 188 because the condition on line 177 was always true

178 # init the problem repository 

179 url = "https://github.com/yosupo06/library-checker-problems" 

180 logger.info("$ git clone %s %s", url, path) 

181 subprocess.check_call( 

182 ["git", "clone", url, str(path)], # noqa: S607 

183 stdout=sys.stderr, 

184 stderr=sys.stderr, 

185 ) 

186 else: 

187 # sync the problem repository 

188 logger.info("$ git -C %s pull", path) 

189 subprocess.check_call( 

190 ["git", "-C", str(path), "pull"], # noqa: S607 

191 stdout=sys.stderr, 

192 stderr=sys.stderr, 

193 ) 

194 

195 LibraryCheckerProblem._is_repository_updated.add(self.repo_path) 

196 

197 

198class _YukicoderProblemNo(int): 

199 def __new__(cls, value: int): 

200 return super().__new__(cls, value) 

201 

202 def __str__(self) -> str: 

203 return "no/" + super().__str__() 

204 

205 

206class _YukicoderProblemId(int): 

207 def __new__(cls, value: int): 

208 return super().__new__(cls, value) 

209 

210 

211class YukicoderProblem(_BaseProblem): 

212 problem: _YukicoderProblemNo | _YukicoderProblemId 

213 

214 def __init__(self, *, problem_no: int | None = None, problem_id: int | None = None): 

215 if problem_no is not None: 

216 self.problem = _YukicoderProblemNo(problem_no) 

217 elif problem_id is not None: 217 ↛ 220line 217 didn't jump to line 220 because the condition on line 217 was always true

218 self.problem = _YukicoderProblemId(problem_id) 

219 else: 

220 raise ValueError("Needs problem_no or problem_id") 

221 

222 @staticmethod 

223 def _env_float(name: str, default: float) -> float: 

224 value = os.environ.get(name) 

225 if value is None or value == "": 

226 return default 

227 try: 

228 return float(value) 

229 except ValueError as e: 

230 raise ValueError(f"{name} must be a float: {value!r}") from e 

231 

232 @staticmethod 

233 def _env_int(name: str, default: int) -> int: 

234 value = os.environ.get(name) 

235 if value is None or value == "": 

236 return default 

237 try: 

238 return int(value) 

239 except ValueError as e: 

240 raise ValueError(f"{name} must be an integer: {value!r}") from e 

241 

242 @staticmethod 

243 def _validate_yukicoder_token(token: str) -> str: 

244 if not token: 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true

245 raise NotLoggedInError("Required: $YUKICODER_TOKEN environment variable") 

246 

247 if token.startswith("YUKICODER_TOKEN="): 

248 raise NotLoggedInError( 

249 "YUKICODER_TOKEN must contain only the token value, " 

250 "not a 'YUKICODER_TOKEN=...' assignment." 

251 ) 

252 

253 if token.lower().startswith("bearer "): 253 ↛ 254line 253 didn't jump to line 254 because the condition on line 253 was never true

254 raise NotLoggedInError( 

255 "YUKICODER_TOKEN must contain only the token value, " 

256 "not a 'Bearer ...' authorization header value." 

257 ) 

258 

259 if token != token.strip(): 

260 raise NotLoggedInError( 

261 "YUKICODER_TOKEN contains leading or trailing whitespace." 

262 ) 

263 

264 bad_control_chars: list[str] = [] 

265 for index, ch in enumerate(token): 

266 code = ord(ch) 

267 if code < _ASCII_SPACE or code == _ASCII_DELETE: 

268 name = { 

269 0x09: "TAB", 

270 0x0A: "LF", 

271 0x0D: "CR", 

272 0x1B: "ESC", 

273 _ASCII_DELETE: "DEL", 

274 }.get(code, "control") 

275 bad_control_chars.append(f"offset {index}: 0x{code:02X} ({name})") 

276 

277 if bad_control_chars: 

278 extra = "" 

279 if "\x1b[" in token: 279 ↛ 285line 279 didn't jump to line 285 because the condition on line 279 was always true

280 extra = ( 

281 " It looks like an ANSI escape sequence was included, " 

282 "possibly by pressing an arrow key while entering the token." 

283 ) 

284 

285 raise NotLoggedInError( 

286 "YUKICODER_TOKEN contains control characters: " 

287 + ", ".join(bad_control_chars) 

288 + "." 

289 + extra 

290 ) 

291 

292 non_visible_ascii: list[str] = [] 

293 for index, ch in enumerate(token): 

294 code = ord(ch) 

295 if code <= _ASCII_SPACE or code >= _ASCII_DELETE: 295 ↛ 296line 295 didn't jump to line 296 because the condition on line 295 was never true

296 non_visible_ascii.append(f"offset {index}: U+{code:04X}") 

297 

298 if non_visible_ascii: 298 ↛ 299line 298 didn't jump to line 299 because the condition on line 298 was never true

299 raise NotLoggedInError( 

300 "YUKICODER_TOKEN contains characters that are not visible ASCII: " 

301 + ", ".join(non_visible_ascii) 

302 ) 

303 

304 return token 

305 

306 @classmethod 

307 def yukicoder_headers(cls) -> dict[str, str]: 

308 token = cls._validate_yukicoder_token(os.environ.get("YUKICODER_TOKEN", "")) 

309 return {"Authorization": f"Bearer {token}"} 

310 

311 def download_system_cases(self) -> Iterable[TestCaseData] | bool: 

312 test_directory = self.test_directory 

313 if test_directory.exists() and any(test_directory.iterdir()): 

314 logger.info("download:already exists: %s", self.url) 

315 return True 

316 

317 headers = self.yukicoder_headers() 

318 if not self._is_logged_in(headers=headers): 

319 raise NotLoggedInError("Required: $YUKICODER_TOKEN environment variable") 

320 

321 self.problem_directory.parent.mkdir(parents=True, exist_ok=True) 

322 

323 tmp_root = pathlib.Path( 

324 tempfile.mkdtemp( 

325 prefix=f"{self.hash_id}.", 

326 dir=self.problem_directory.parent, 

327 ) 

328 ) 

329 zip_path = tmp_root / "testcase.zip" 

330 staging_directory = tmp_root / "test" 

331 

332 try: 

333 self._download_testcase_zip(zip_path, headers=headers) 

334 case_count = self._extract_testcase_zip(zip_path, staging_directory) 

335 

336 if case_count == 0: 

337 logger.error( 

338 "Sample not found", 

339 extra={"github": GitHubMessageParams()}, 

340 ) 

341 return False 

342 

343 self.problem_directory.mkdir(parents=True, exist_ok=True) 

344 

345 if test_directory.exists(): 

346 shutil.rmtree(test_directory) 

347 

348 staging_directory.rename(test_directory) 

349 logger.info("download:saved: %s cases: %s", case_count, self.url) 

350 return True 

351 finally: 

352 shutil.rmtree(tmp_root, ignore_errors=True) 

353 

354 def _download_cases(self) -> list[TestCaseData]: 

355 headers = self.yukicoder_headers() 

356 if not self._is_logged_in(headers=headers): 

357 raise NotLoggedInError("Required: $YUKICODER_TOKEN environment variable") 

358 

359 self.problem_directory.parent.mkdir(parents=True, exist_ok=True) 

360 

361 tmp_root = pathlib.Path( 

362 tempfile.mkdtemp( 

363 prefix=f"{self.hash_id}.", 

364 dir=self.problem_directory.parent, 

365 ) 

366 ) 

367 zip_path = tmp_root / "testcase.zip" 

368 

369 try: 

370 self._download_testcase_zip(zip_path, headers=headers) 

371 with zipfile.ZipFile(zip_path) as fh: 

372 inputs: dict[str, bytes] = {} 

373 outputs: dict[str, bytes] = {} 

374 

375 for info in fh.infolist(): 

376 filename = info.filename 

377 if filename.endswith("/"): 

378 continue 

379 

380 path = pathlib.PurePosixPath(filename) 

381 self._validate_zip_member_path(path) 

382 

383 if filename.startswith("test_in/"): 

384 inputs[path.stem] = fh.read(info) 

385 elif filename.startswith("test_out/"): 

386 outputs[path.stem] = fh.read(info) 

387 

388 return [ 

389 TestCaseData(name=name, input_data=i, output_data=o) 

390 for name, i, o in enumerate_input_outputs(inputs, outputs) 

391 ] 

392 finally: 

393 shutil.rmtree(tmp_root, ignore_errors=True) 

394 

395 def _download_testcase_zip( 

396 self, 

397 destination: pathlib.Path, 

398 *, 

399 headers: dict[str, str] | None, 

400 ) -> None: 

401 url = f"{self.url}/testcase.zip" 

402 

403 connect_timeout = self._env_float( 

404 "COMPETITIVE_VERIFIER_YUKICODER_CONNECT_TIMEOUT", 

405 10.0, 

406 ) 

407 read_timeout = self._env_float( 

408 "COMPETITIVE_VERIFIER_YUKICODER_READ_TIMEOUT", 

409 30.0, 

410 ) 

411 download_timeout = self._env_float( 

412 "COMPETITIVE_VERIFIER_YUKICODER_DOWNLOAD_TIMEOUT", 

413 0.0, 

414 ) 

415 report_interval = self._env_float( 

416 "COMPETITIVE_VERIFIER_YUKICODER_REPORT_INTERVAL", 

417 5.0, 

418 ) 

419 chunk_size = self._env_int( 

420 "COMPETITIVE_VERIFIER_YUKICODER_CHUNK_SIZE", 

421 1024 * 1024, 

422 ) 

423 

424 logger.info("download:yukicoder testcase.zip: %s", url) 

425 

426 started_at = time.perf_counter() 

427 last_reported_at = started_at 

428 total = 0 

429 

430 with requests.get( 

431 url, 

432 headers=headers, 

433 allow_redirects=True, 

434 stream=True, 

435 timeout=(connect_timeout, read_timeout), 

436 ) as resp: 

437 resp.raise_for_status() 

438 

439 content_length_text = resp.headers.get("content-length") 

440 content_length = ( 

441 int(content_length_text) 

442 if content_length_text is not None and content_length_text.isdigit() 

443 else None 

444 ) 

445 

446 logger.info( 

447 "download:yukicoder response: status=%s, content-type=%s, content-length=%s", 

448 resp.status_code, 

449 resp.headers.get("content-type"), 

450 content_length_text, 

451 ) 

452 

453 with destination.open("wb") as out: 

454 for chunk in resp.iter_content(chunk_size=chunk_size): 

455 if not chunk: 

456 continue 

457 

458 out.write(chunk) 

459 total += len(chunk) 

460 

461 now = time.perf_counter() 

462 elapsed = now - started_at 

463 

464 if download_timeout > 0 and elapsed > download_timeout: 

465 raise TimeoutError( 

466 f"download timeout: {url}: " 

467 f"{total} bytes in {download_timeout:.0f} sec" 

468 ) 

469 

470 if ( 

471 report_interval > 0 

472 and now - last_reported_at >= report_interval 

473 ): 

474 mib = total / 1024 / 1024 

475 speed = mib / elapsed if elapsed > 0 else 0.0 

476 

477 if content_length: 

478 percent = total * 100.0 / content_length 

479 logger.info( 

480 "download:yukicoder progress: %.1f / %.1f MiB, %.1f%%, %.2f MiB/s", 

481 mib, 

482 content_length / 1024 / 1024, 

483 percent, 

484 speed, 

485 ) 

486 else: 

487 logger.info( 

488 "download:yukicoder progress: %.1f MiB, %.2f MiB/s", 

489 mib, 

490 speed, 

491 ) 

492 

493 last_reported_at = now 

494 

495 elapsed = time.perf_counter() - started_at 

496 mib = total / 1024 / 1024 

497 speed = mib / elapsed if elapsed > 0 else 0.0 

498 logger.info( 

499 "download:yukicoder done: %.1f MiB, %.2f MiB/s, %.1f sec", 

500 mib, 

501 speed, 

502 elapsed, 

503 ) 

504 

505 def _extract_testcase_zip( 

506 self, 

507 zip_path: pathlib.Path, 

508 destination: pathlib.Path, 

509 ) -> int: 

510 inputs: dict[str, zipfile.ZipInfo] = {} 

511 outputs: dict[str, zipfile.ZipInfo] = {} 

512 

513 with zipfile.ZipFile(zip_path) as fh: 

514 for info in fh.infolist(): 

515 filename = info.filename 

516 if filename.endswith("/"): 

517 continue 

518 

519 path = pathlib.PurePosixPath(filename) 

520 self._validate_zip_member_path(path) 

521 

522 if filename.startswith("test_in/"): 

523 inputs[path.stem] = info 

524 elif filename.startswith("test_out/"): 

525 outputs[path.stem] = info 

526 

527 common_names = sorted(inputs.keys() & outputs.keys()) 

528 

529 if len(inputs) != len(common_names) or len(outputs) != len(common_names): 

530 logger.warning("dangling output case") 

531 

532 if not common_names: 

533 logger.warning("no cases found") 

534 return 0 

535 

536 destination.mkdir(parents=True, exist_ok=False) 

537 

538 for name in common_names: 

539 input_path = destination / _name_to_filename(name, "in") 

540 output_path = destination / _name_to_filename(name, "out") 

541 

542 with fh.open(inputs[name]) as src, input_path.open("wb") as out: 

543 shutil.copyfileobj(src, out) 

544 

545 with fh.open(outputs[name]) as src, output_path.open("wb") as out: 

546 shutil.copyfileobj(src, out) 

547 

548 return len(common_names) 

549 

550 @staticmethod 

551 def _validate_zip_member_path(path: pathlib.PurePosixPath) -> None: 

552 if path.is_absolute() or ".." in path.parts: 

553 raise RuntimeError(f"unsafe path in testcase.zip: {path}") 

554 

555 @property 

556 def url(self) -> str: 

557 return f"https://yukicoder.me/problems/{self.problem}" 

558 

559 @classmethod 

560 def from_url(cls, url: str) -> Optional["YukicoderProblem"]: 

561 # example: https://yukicoder.me/problems/no/499 

562 # example: http://yukicoder.me/problems/1476 

563 result = urllib.parse.urlparse(url) 

564 dirname, basename = posixpath.split(normalize_url_path(result.path)) 

565 if result.scheme in ("", "http", "https") and result.netloc == "yukicoder.me": 

566 try: 

567 n = int(basename) 

568 except ValueError: 

569 pass 

570 else: 

571 if dirname == "/problems/no": 

572 return cls(problem_no=n) 

573 if dirname == "/problems": 

574 return cls(problem_id=n) 

575 return None 

576 

577 def _is_logged_in(self, *, headers: dict[str, str] | None = None) -> bool: 

578 url = "https://yukicoder.me" 

579 resp = requests.get(url, headers=headers, allow_redirects=True, timeout=10) 

580 resp.raise_for_status() 

581 return "login-btn" not in str(resp.content) 

582 

583 

584class AOJProblem(_BaseProblem): 

585 def __init__(self, *, problem_id: str): 

586 self.problem_id = problem_id 

587 

588 def _download_cases(self) -> Iterable[TestCaseData]: 

589 return AOJProblem.download_cases(self.problem_id) 

590 

591 @staticmethod 

592 def download_cases(problem_id: str) -> Iterable[TestCaseData]: 

593 # get header 

594 # reference: http://developers.u-aizu.ac.jp/api?key=judgedat%2Ftestcases%2F%7BproblemId%7D%2Fheader_GET 

595 url = f"https://judgedat.u-aizu.ac.jp/testcases/{problem_id}/header" 

596 resp = requests.get(url, allow_redirects=True, timeout=10) 

597 resp.raise_for_status() 

598 header_res = json.loads(resp.text) 

599 

600 # get testcases via the official API 

601 for header in header_res["headers"]: 

602 # NOTE: the endpoints are not same to http://developers.u-aizu.ac.jp/api?key=judgedat%2Ftestcases%2F%7BproblemId%7D%2F%7Bserial%7D_GET since the json API often says "..... (terminated because of the limitation)" 

603 # NOTE: even when using https://judgedat.u-aizu.ac.jp/testcases/PROBLEM_ID/SERIAL, there is the 1G limit (see https://twitter.com/beet_aizu/status/1194947611100188672) 

604 serial = header["serial"] 

605 url = f"https://judgedat.u-aizu.ac.jp/testcases/{problem_id}/{serial}" 

606 

607 resp_in = requests.get(url + "/in", allow_redirects=True, timeout=10) 

608 resp_in.raise_for_status() 

609 resp_out = requests.get(url + "/out", allow_redirects=True, timeout=10) 

610 resp_out.raise_for_status() 

611 

612 yield TestCaseData( 

613 header["name"], 

614 resp_in.content, 

615 resp_out.content, 

616 ) 

617 

618 @property 

619 def url(self) -> str: 

620 return f"http://judge.u-aizu.ac.jp/onlinejudge/description.jsp?id={self.problem_id}" 

621 

622 @classmethod 

623 def from_url(cls, url: str) -> Optional["AOJProblem"]: 

624 result = urllib.parse.urlparse(url) 

625 

626 # example: http://judge.u-aizu.ac.jp/onlinejudge/description.jsp?id=1169 

627 # example: http://judge.u-aizu.ac.jp/onlinejudge/description.jsp?id=DSL_1_A&lang=jp 

628 querystring = urllib.parse.parse_qs(result.query) 

629 if ( 

630 result.scheme in ("", "http", "https") 

631 and result.netloc == "judge.u-aizu.ac.jp" 

632 and normalize_url_path(result.path) == "/onlinejudge/description.jsp" 

633 and querystring.get("id") 

634 and len(querystring["id"]) == 1 

635 ): 

636 (n,) = querystring["id"] 

637 return cls(problem_id=n) 

638 

639 # example: https://onlinejudge.u-aizu.ac.jp/challenges/sources/JAG/Prelim/2881 

640 # example: https://onlinejudge.u-aizu.ac.jp/courses/library/4/CGL/3/CGL_3_B 

641 m = re.match( 

642 r"^/(challenges|courses)/(sources|library/\d+|lesson/\d+)/(\w+)/(\w+)/(\w+)$", 

643 normalize_url_path(result.path), 

644 ) 

645 if ( 

646 result.scheme in ("", "http", "https") 

647 and result.netloc == "onlinejudge.u-aizu.ac.jp" 

648 and m 

649 ): 

650 n = m.group(5) 

651 return cls(problem_id=n) 

652 

653 # example: https://onlinejudge.u-aizu.ac.jp/problems/0423 

654 # example: https://onlinejudge.u-aizu.ac.jp/problems/CGL_3_B 

655 m = re.match(r"^/problems/(\w+)$", normalize_url_path(result.path)) 

656 if ( 

657 result.scheme in ("", "http", "https") 

658 and result.netloc == "onlinejudge.u-aizu.ac.jp" 

659 and m 

660 ): 

661 n = m.group(1) 

662 return cls(problem_id=n) 

663 

664 return None 

665 

666 

667class AOJArenaProblem(_BaseProblem): 

668 def __init__(self, *, arena_id: str, alphabet: str): 

669 if len(alphabet) != 1 or not alphabet.isupper(): 669 ↛ 670line 669 didn't jump to line 670 because the condition on line 669 was never true

670 raise ValueError(arena_id, alphabet) 

671 self.arena_id = arena_id 

672 self.alphabet = alphabet 

673 

674 self._problem_id: str | None = None 

675 

676 def get_problem_id(self) -> str: 

677 if self._problem_id is None: 677 ↛ 689line 677 didn't jump to line 689 because the condition on line 677 was always true

678 url = f"https://judgeapi.u-aizu.ac.jp/arenas/{self.arena_id}/problems" 

679 resp = requests.get(url, allow_redirects=True, timeout=10) 

680 resp.raise_for_status() 

681 problems = json.loads(resp.text) 

682 for problem in problems: 682 ↛ 688line 682 didn't jump to line 688 because the loop on line 682 didn't complete

683 if problem["id"] == self.alphabet: 683 ↛ 682line 683 didn't jump to line 682 because the condition on line 683 was always true

684 p = problem["problemId"] 

685 logger.debug("problem: %s", p) 

686 self._problem_id = p 

687 return p 

688 raise ValueError("Problem is not found.") 

689 return self._problem_id 

690 

691 def _download_cases(self) -> Iterable[TestCaseData]: 

692 return AOJProblem.download_cases(self.get_problem_id()) 

693 

694 @property 

695 def url(self) -> str: 

696 return f"https://onlinejudge.u-aizu.ac.jp/services/room.html#{self.arena_id}/problems/{self.alphabet}" 

697 

698 @classmethod 

699 def from_url(cls, url: str) -> Optional["AOJArenaProblem"]: 

700 # example: https://onlinejudge.u-aizu.ac.jp/services/room.html#RitsCamp19Day2/problems/A 

701 result = urllib.parse.urlparse(url) 

702 if ( 

703 result.scheme in ("", "http", "https") 

704 and result.netloc == "onlinejudge.u-aizu.ac.jp" 

705 and normalize_url_path(result.path) == "/services/room.html" 

706 ): 

707 fragment = result.fragment.split("/") 

708 if len(fragment) == 3 and fragment[1] == "problems": # noqa: PLR2004 708 ↛ 710line 708 didn't jump to line 710 because the condition on line 708 was always true

709 return cls(arena_id=fragment[0], alphabet=fragment[2].upper()) 

710 return None 

711 

712 

713@dataclass 

714class LocalProblem(TestCaseProvider): 

715 path: pathlib.Path 

716 

717 def download_system_cases(self) -> Iterable[TestCaseData] | bool: 

718 return bool(any(self.iter_system_cases())) 

719 

720 def iter_system_cases(self) -> Iterable[TestCaseFile]: 

721 return iter_testcases(directory=self.path, recursive=True) 

722 

723 

724def normalize_url_path(path: str) -> str: 

725 """A wrapper of posixpath.normpath. 

726 

727 posixpath.normpath doesn't collapse a leading duplicated slashes. 

728 """ 

729 path = posixpath.normpath(path) 

730 if path.startswith("//"): 

731 path = "/" + path.lstrip("/") 

732 return path 

733 

734 

735def _subclasses_recursive(cls: type[object]) -> Iterable[type[Problem]]: 

736 for ch in cls.__subclasses__(): 

737 if issubclass(ch, Problem): 737 ↛ 736line 737 didn't jump to line 736 because the condition on line 737 was always true

738 yield ch 

739 yield from _subclasses_recursive(ch) 

740 

741 

742def problem_from_url(url: str) -> Problem | None: 

743 for ch in set(_subclasses_recursive(Problem)): 

744 if (problem := ch.from_url(url)) is not None: 

745 return problem 

746 return None 

747 

748 

749_InputOutput = TypeVar("_InputOutput") 

750 

751 

752def enumerate_input_outputs( 

753 inputs: dict[str, _InputOutput], 

754 outputs: dict[str, _InputOutput], 

755) -> Iterator[tuple[str, _InputOutput, _InputOutput]]: 

756 common_keys = inputs.keys() & outputs.keys() 

757 if len(inputs) != len(common_keys) or len(outputs) != len(common_keys): 

758 logger.warning("dangling output case") 

759 

760 if len(common_keys) == 0: 

761 logger.warning("no cases found") 

762 

763 for key in sorted(common_keys): 

764 yield (key, inputs[key], outputs[key]) 

765 

766 

767def merge_testcase_files( 

768 inputs: dict[str, pathlib.Path], 

769 outputs: dict[str, pathlib.Path], 

770) -> Iterator[TestCaseFile]: 

771 for name, i, o in enumerate_input_outputs(inputs, outputs): 

772 yield TestCaseFile(name=name, input_path=i, output_path=o) 

773 

774 

775def _casename(path: pathlib.Path, *, directory: pathlib.Path) -> str: 

776 return path.relative_to(directory).with_suffix("").as_posix()