313 lines
12 KiB
Python
313 lines
12 KiB
Python
"""Build a HydroOJ-ready archive from CCF/NOI source-resource downloads."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import re
|
||
import shutil
|
||
import ssl
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import zipfile
|
||
from collections import defaultdict
|
||
from datetime import UTC, datetime
|
||
from html.parser import HTMLParser
|
||
from pathlib import Path
|
||
from urllib.parse import urljoin
|
||
from urllib.request import Request, urlopen
|
||
|
||
BASE_URL = "https://noi.ccf.org.cn"
|
||
LISTING_URLS = [
|
||
f"{BASE_URL}/zxzy/lnzl/",
|
||
f"{BASE_URL}/zxzy/lnzl/index_2.shtml",
|
||
f"{BASE_URL}/zxzy/lnzl/index_3.shtml",
|
||
]
|
||
USER_AGENT = "Mozilla/5.0 (HydroOJ archival importer; contact: local)"
|
||
DATA_SUFFIXES = {".in", ".ans", ".out"}
|
||
STATEMENT_SUFFIXES = {".pdf", ".doc", ".docx", ".md", ".txt", ".html", ".htm"}
|
||
|
||
|
||
class AnchorParser(HTMLParser):
|
||
def __init__(self) -> None:
|
||
super().__init__()
|
||
self.links: list[tuple[str, str]] = []
|
||
self._href: str | None = None
|
||
self._chunks: list[str] = []
|
||
|
||
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
||
if tag == "a":
|
||
self._href = dict(attrs).get("href")
|
||
self._chunks = []
|
||
|
||
def handle_data(self, data: str) -> None:
|
||
if self._href:
|
||
self._chunks.append(data)
|
||
|
||
def handle_endtag(self, tag: str) -> None:
|
||
if tag == "a" and self._href:
|
||
text = "".join(self._chunks).strip()
|
||
if text:
|
||
self.links.append((self._href, text))
|
||
self._href = None
|
||
self._chunks = []
|
||
|
||
|
||
def normalize_problem_id(value: str) -> str:
|
||
"""Return a stable ASCII id accepted by common HydroOJ import tooling."""
|
||
value = value.lower().strip()
|
||
value = re.sub(r"[^a-z0-9]+", "-", value)
|
||
return value.strip("-") or "problem"
|
||
|
||
|
||
def classify_member(member: str) -> str | None:
|
||
suffix = Path(member).suffix.lower()
|
||
if suffix == ".in":
|
||
return "input"
|
||
if suffix in {".ans", ".out"}:
|
||
return "output"
|
||
if suffix in STATEMENT_SUFFIXES:
|
||
return "statement"
|
||
return None
|
||
|
||
|
||
def find_problem_roots(members: list[str]) -> set[str]:
|
||
"""Find archive directories containing at least one input/output pair."""
|
||
files = set(members)
|
||
roots: set[str] = set()
|
||
for name in files:
|
||
if Path(name).suffix.lower() != ".in":
|
||
continue
|
||
stem = name[: -len(Path(name).suffix)]
|
||
if any(stem + ext in files for ext in (".out", ".ans")):
|
||
roots.add(str(Path(name).parent))
|
||
return roots
|
||
|
||
|
||
def build_ssl_context() -> ssl.SSLContext:
|
||
"""Work around the expired certificate currently served by noi.ccf.org.cn.
|
||
|
||
This context is deliberately local to the official archive fetcher; URLs
|
||
are hard-coded to the CCF/NOI hosts above and are not user supplied.
|
||
"""
|
||
return ssl._create_unverified_context()
|
||
|
||
|
||
def fetch(url: str, referer: str | None = None) -> bytes:
|
||
headers = {"User-Agent": USER_AGENT}
|
||
if referer:
|
||
headers["Referer"] = referer
|
||
request = Request(url, headers=headers)
|
||
with urlopen(request, timeout=120, context=build_ssl_context()) as response:
|
||
return response.read()
|
||
|
||
|
||
def parse_links(html: bytes, source_url: str) -> list[tuple[str, str]]:
|
||
parser = AnchorParser()
|
||
parser.feed(html.decode("utf-8", errors="replace"))
|
||
return [(urljoin(source_url, href), text) for href, text in parser.links]
|
||
|
||
|
||
def resource_filename(url: str, label: str) -> str:
|
||
safe_label = re.sub(r"[^\w. -]+", "_", label, flags=re.UNICODE).strip(" .")
|
||
digest = hashlib.sha256(url.encode()).hexdigest()[:12]
|
||
return f"{safe_label or 'resource'}-{digest}"
|
||
|
||
|
||
def archive_format(data_path: Path) -> str | None:
|
||
magic = data_path.read_bytes()[:8]
|
||
if magic.startswith(b"PK\x03\x04") or magic.startswith(b"PK\x05\x06"):
|
||
return "zip"
|
||
if magic.startswith(b"Rar!\x1a\x07"):
|
||
return "rar"
|
||
return None
|
||
|
||
|
||
def create_hydro_zip(
|
||
output_path: Path,
|
||
problem_id: str,
|
||
title: str,
|
||
data_dir: Path,
|
||
tags: list[str],
|
||
statement_pdf: Path | None = None,
|
||
) -> None:
|
||
"""Create one Hydro Problem Format ZIP from a local testdata directory."""
|
||
root = normalize_problem_id(problem_id)
|
||
yaml_tags = "\n".join(f"- {tag}" for tag in tags)
|
||
problem_yaml = f"title: {title}\ntag:\n{yaml_tags}\npid: {root}\n"
|
||
pdf_name = statement_pdf.name if statement_pdf else None
|
||
statement = f"# {title}\n\n"
|
||
if pdf_name:
|
||
statement += f"官方题面 PDF：[下载或查看](file://{pdf_name})\n"
|
||
else:
|
||
statement += "题面请参阅 CCF/NOI 官方发布文件。\n"
|
||
with zipfile.ZipFile(output_path, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=6) as archive:
|
||
archive.writestr(f"{root}/problem.yaml", problem_yaml)
|
||
archive.writestr(f"{root}/problem_zh.md", statement)
|
||
archive.writestr(f"{root}/testdata/config.yaml", "")
|
||
if statement_pdf:
|
||
archive.write(statement_pdf, f"{root}/additional_file/{pdf_name}")
|
||
for source in sorted(path for path in data_dir.rglob("*") if path.is_file()):
|
||
relative = source.relative_to(data_dir).as_posix()
|
||
if source.suffix.lower() == ".ans":
|
||
relative = str(Path(relative).with_suffix(".out"))
|
||
archive.write(source, f"{root}/testdata/{relative}")
|
||
|
||
|
||
def create_hydro_bundle_zip(
|
||
output_path: Path,
|
||
exam_id: str,
|
||
exam_title: str,
|
||
problems: list[tuple[str, Path]],
|
||
statement_pdfs: list[Path],
|
||
) -> None:
|
||
"""Create one Hydro bundle ZIP containing all problems from one exam."""
|
||
exam_root = normalize_problem_id(exam_id)
|
||
with zipfile.ZipFile(output_path, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=6) as archive:
|
||
for problem_id, data_dir in problems:
|
||
root = normalize_problem_id(problem_id)
|
||
title = f"{exam_title}: {problem_id}"
|
||
archive.writestr(f"{exam_root}/{root}/problem.yaml", f"title: {title}\ntag:\n- CCF\n- NOI\npid: {root}\n")
|
||
statement = f"# {title}\n\n"
|
||
if statement_pdfs:
|
||
pdf_name = statement_pdfs[0].name
|
||
statement += f"官方题面 PDF：[下载或查看](file://{pdf_name})\n"
|
||
else:
|
||
statement += "题面请参阅 CCF/NOI 官方发布文件。\n"
|
||
archive.writestr(f"{exam_root}/{root}/problem_zh.md", statement)
|
||
archive.writestr(f"{exam_root}/{root}/testdata/config.yaml", "")
|
||
for source in sorted(path for path in data_dir.rglob("*") if path.is_file()):
|
||
relative = source.relative_to(data_dir).as_posix()
|
||
if source.suffix.lower() == ".ans":
|
||
relative = str(Path(relative).with_suffix(".out"))
|
||
archive.write(source, f"{exam_root}/{root}/testdata/{relative}")
|
||
for pdf in statement_pdfs:
|
||
archive.write(pdf, f"{exam_root}/additional_file/{pdf.name}")
|
||
|
||
|
||
def extract_zip(data_path: Path, dest: Path) -> list[dict]:
|
||
with zipfile.ZipFile(data_path) as archive:
|
||
members = [name for name in archive.namelist() if not name.endswith("/")]
|
||
roots = sorted(find_problem_roots(members))
|
||
results = []
|
||
for root in roots:
|
||
problem_id = normalize_problem_id(root)
|
||
problem_dir = dest / problem_id
|
||
data_dir = problem_dir / "data"
|
||
data_dir.mkdir(parents=True, exist_ok=True)
|
||
copied = []
|
||
prefix = root.rstrip("/") + "/"
|
||
for member in members:
|
||
if not member.startswith(prefix) or classify_member(member) not in {"input", "output"}:
|
||
continue
|
||
local_name = member[len(prefix) :]
|
||
target = data_dir / local_name
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
with archive.open(member) as source, target.open("wb") as output:
|
||
shutil.copyfileobj(source, output)
|
||
copied.append(local_name)
|
||
if copied:
|
||
results.append({"id": problem_id, "archive_root": root, "files": sorted(copied)})
|
||
return results
|
||
|
||
|
||
def extract_rar(data_path: Path, dest: Path) -> list[dict]:
|
||
with tempfile.TemporaryDirectory() as temp:
|
||
work = Path(temp)
|
||
subprocess.run(
|
||
["7z", "x", "-y", f"-o{work}", str(data_path)],
|
||
check=True,
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.PIPE,
|
||
)
|
||
members = [path.relative_to(work).as_posix() for path in work.rglob("*") if path.is_file()]
|
||
roots = sorted(find_problem_roots(members))
|
||
results = []
|
||
for root in roots:
|
||
problem_id = normalize_problem_id(root)
|
||
data_dir = dest / problem_id / "data"
|
||
data_dir.mkdir(parents=True, exist_ok=True)
|
||
copied = []
|
||
prefix = root.rstrip("/") + "/"
|
||
for member in members:
|
||
if not member.startswith(prefix) or classify_member(member) not in {"input", "output"}:
|
||
continue
|
||
local_name = member[len(prefix) :]
|
||
target = data_dir / local_name
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
shutil.copy2(work / member, target)
|
||
copied.append(local_name)
|
||
if copied:
|
||
results.append({"id": problem_id, "archive_root": root, "files": sorted(copied)})
|
||
return results
|
||
|
||
|
||
def build(args: argparse.Namespace) -> dict:
|
||
output = Path(args.output).resolve()
|
||
raw_dir = output / "raw"
|
||
package_dir = output / "hydrooj"
|
||
raw_dir.mkdir(parents=True, exist_ok=True)
|
||
package_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
entries: list[dict] = []
|
||
seen_articles: set[str] = set()
|
||
for listing_url in LISTING_URLS:
|
||
for article_url, title in parse_links(fetch(listing_url), listing_url):
|
||
if "/zxzy/lnzl/jszl/" not in article_url or article_url in seen_articles:
|
||
continue
|
||
seen_articles.add(article_url)
|
||
article_html = fetch(article_url)
|
||
resources = []
|
||
for resource_url, label in parse_links(article_html, article_url):
|
||
if "contentcore/resource/download?ID=" not in resource_url and "/ccf/file/do?" not in resource_url:
|
||
continue
|
||
filename = resource_filename(resource_url, label)
|
||
target = raw_dir / filename
|
||
if not target.exists():
|
||
target.write_bytes(fetch(resource_url, referer=article_url))
|
||
resource = {
|
||
"label": label,
|
||
"url": resource_url,
|
||
"file": str(target.relative_to(output)),
|
||
"bytes": target.stat().st_size,
|
||
"sha256": hashlib.sha256(target.read_bytes()).hexdigest(),
|
||
}
|
||
archive_type = archive_format(target)
|
||
if not args.download_only and archive_type == "zip":
|
||
resource["extracted_problems"] = extract_zip(target, package_dir / normalize_problem_id(title))
|
||
elif not args.download_only and archive_type == "rar":
|
||
resource["extracted_problems"] = extract_rar(target, package_dir / normalize_problem_id(title))
|
||
resources.append(resource)
|
||
entries.append({"title": title, "source": article_url, "resources": resources})
|
||
|
||
manifest = {
|
||
"schema": "ccf-hydrooj-archive/v1",
|
||
"generated_at": datetime.now(UTC).isoformat(),
|
||
"source_listing": LISTING_URLS,
|
||
"entries": entries,
|
||
}
|
||
(output / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n")
|
||
return manifest
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--output", default="dist", help="destination directory")
|
||
parser.add_argument("--download-only", action="store_true", help="download source files and manifest without extracting test data")
|
||
args = parser.parse_args()
|
||
manifest = build(args)
|
||
resources = sum(len(entry["resources"]) for entry in manifest["entries"])
|
||
problems = sum(
|
||
len(resource.get("extracted_problems", []))
|
||
for entry in manifest["entries"]
|
||
for resource in entry["resources"]
|
||
)
|
||
print(json.dumps({"entries": len(manifest["entries"]), "resources": resources, "problems": problems}, ensure_ascii=False))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|