audit_mhtml.py 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166
  1. #!/usr/bin/env python3
  2. """Audit saved VJudge MHTML files without copying problem statements.
  3. The script reports the metadata needed by the course and verifies that each
  4. snapshot contains an embedded statement, rather than only the VJudge shell.
  5. It intentionally stores no statement text.
  6. """
  7. from __future__ import annotations
  8. import argparse
  9. import html
  10. import json
  11. import re
  12. import sys
  13. from email import policy
  14. from email.parser import BytesParser
  15. from html.parser import HTMLParser
  16. from pathlib import Path
  17. class TextExtractor(HTMLParser):
  18. def __init__(self) -> None:
  19. super().__init__()
  20. self.parts: list[str] = []
  21. def handle_data(self, data: str) -> None:
  22. if data.strip():
  23. self.parts.append(data.strip())
  24. def text(self) -> str:
  25. return " ".join(self.parts)
  26. def plain_text(fragment: str) -> str:
  27. parser = TextExtractor()
  28. parser.feed(fragment)
  29. return html.unescape(parser.text())
  30. def decode_part(part) -> str:
  31. payload = part.get_payload(decode=True) or b""
  32. charset = part.get_content_charset() or "utf-8"
  33. return payload.decode(charset, errors="replace")
  34. def parse_snapshot(path: Path) -> dict[str, object]:
  35. message = BytesParser(policy=policy.default).parsebytes(path.read_bytes())
  36. snapshot_url = str(message.get("Snapshot-Content-Location", ""))
  37. subject = str(message.get("Subject", path.stem))
  38. html_parts = [
  39. part
  40. for part in message.walk()
  41. if not part.is_multipart() and part.get_content_type() == "text/html"
  42. ]
  43. main_part = next(
  44. (
  45. part
  46. for part in html_parts
  47. if str(part.get("Content-Location", "")) == snapshot_url
  48. ),
  49. None,
  50. )
  51. if main_part is None:
  52. raise ValueError("missing VJudge main HTML part")
  53. main_html = decode_part(main_part)
  54. match = re.search(
  55. r'<textarea[^>]*name=["\']dataJson["\'][^>]*>(.*?)</textarea>',
  56. main_html,
  57. flags=re.DOTALL,
  58. )
  59. if match is None:
  60. raise ValueError("missing dataJson metadata")
  61. metadata = json.loads(html.unescape(match.group(1)))
  62. official_keys = {
  63. str(item["key"])
  64. for item in metadata.get("descBriefs", [])
  65. if item.get("mainOfficial") or (item.get("official") and item.get("type") == "main")
  66. }
  67. statement_candidates: list[tuple[bool, str]] = []
  68. for part in html_parts:
  69. location = str(part.get("Content-Location", ""))
  70. if "/problem/description/" not in location:
  71. continue
  72. body = decode_part(part)
  73. key_match = re.search(r"/problem/description/(\d+)", location)
  74. is_official = bool(key_match and key_match.group(1) in official_keys)
  75. statement_candidates.append((is_official, body))
  76. if not statement_candidates:
  77. raise ValueError("missing embedded statement HTML")
  78. statement_candidates.sort(key=lambda item: (item[0], len(item[1])), reverse=True)
  79. statement_html = statement_candidates[0][1]
  80. statement_text = plain_text(statement_html)
  81. properties = {
  82. str(item.get("title", "")): plain_text(str(item.get("content", "")))
  83. for item in metadata.get("properties", [])
  84. }
  85. title = re.sub(r"\s+-\s+Virtual Judge$", "", subject)
  86. title = re.sub(r"\s+-\s+[^-]+\s+[^-]+$", "", title)
  87. has_input = bool(re.search(r"\bInput\b|输入", statement_text, re.IGNORECASE))
  88. has_output = bool(re.search(r"\bOutput\b|输出", statement_text, re.IGNORECASE))
  89. sample_count = len(re.findall(r'class=["\']vjudge_sample["\']', statement_html))
  90. return {
  91. "file": path.as_posix(),
  92. "query": f"{metadata.get('oj', '?')}-{metadata.get('prob', '?')}",
  93. "title": title,
  94. "source": properties.get("source", ""),
  95. "difficulty": properties.get("difficulty", ""),
  96. "time_limit": properties.get("time_limit", ""),
  97. "memory_limit": properties.get("mem_limit", ""),
  98. "statement_chars": len(statement_text),
  99. "samples": sample_count,
  100. "has_input": has_input,
  101. "has_output": has_output,
  102. "official_statement": statement_candidates[0][0],
  103. "url": snapshot_url,
  104. }
  105. def main() -> int:
  106. parser = argparse.ArgumentParser()
  107. parser.add_argument("root", nargs="?", default="problems")
  108. parser.add_argument("--json", action="store_true", dest="as_json")
  109. args = parser.parse_args()
  110. paths = sorted(Path(args.root).rglob("*.mhtml"))
  111. if not paths:
  112. print(f"no .mhtml files found under {args.root}", file=sys.stderr)
  113. return 2
  114. rows: list[dict[str, object]] = []
  115. errors: list[str] = []
  116. for path in paths:
  117. try:
  118. rows.append(parse_snapshot(path))
  119. except Exception as exc: # keep auditing the remaining snapshots
  120. errors.append(f"{path}: {exc}")
  121. if args.as_json:
  122. print(json.dumps(rows, ensure_ascii=False, indent=2))
  123. else:
  124. print("query\ttitle\tdifficulty\tsource\tstatement\tsamples\tofficial")
  125. for row in rows:
  126. statement = "ok" if row["has_input"] and row["has_output"] else "incomplete"
  127. official = "yes" if row["official_statement"] else "fallback"
  128. print(
  129. f"{row['query']}\t{row['title']}\t{row['difficulty'] or '-'}\t"
  130. f"{row['source'] or '-'}\t{statement}\t{row['samples']}\t{official}"
  131. )
  132. for error in errors:
  133. print(error, file=sys.stderr)
  134. incomplete = [
  135. row for row in rows if not row["has_input"] or not row["has_output"] or row["statement_chars"] < 200
  136. ]
  137. return 1 if errors or incomplete else 0
  138. if __name__ == "__main__":
  139. raise SystemExit(main())