wuyang
87 lines
3.2 KiB
Python
Executable File
87 lines
3.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Check source URLs stored in collection item frontmatter."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
from build_index import parse_frontmatter
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
URL_KEYS = ("url", "source_url", "code_url")
|
|
|
|
|
|
def check_url(url: str, timeout: int) -> dict:
|
|
request = urllib.request.Request(url, method="HEAD", headers={"User-Agent": "agent-kb-url-check/1.0"})
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
return {"url": url, "ok": True, "status": response.status}
|
|
except urllib.error.HTTPError as exc:
|
|
if exc.code in (403, 405):
|
|
return check_url_get(url, timeout, exc.code)
|
|
return {"url": url, "ok": False, "status": exc.code, "error": str(exc)}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"url": url, "ok": False, "status": None, "error": str(exc)}
|
|
|
|
|
|
def check_url_get(url: str, timeout: int, head_status: int) -> dict:
|
|
request = urllib.request.Request(url, method="GET", headers={"User-Agent": "agent-kb-url-check/1.0"})
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
return {"url": url, "ok": True, "status": response.status, "head_status": head_status}
|
|
except urllib.error.HTTPError as exc:
|
|
if exc.code == 403:
|
|
return {
|
|
"url": url,
|
|
"ok": True,
|
|
"status": exc.code,
|
|
"head_status": head_status,
|
|
"blocked": True,
|
|
"error": "blocked automated request; verify in browser",
|
|
}
|
|
return {"url": url, "ok": False, "status": exc.code, "head_status": head_status, "error": str(exc)}
|
|
except Exception as exc: # noqa: BLE001
|
|
return {"url": url, "ok": False, "status": head_status, "error": str(exc)}
|
|
|
|
|
|
def iter_urls():
|
|
for collection in ("jobs", "papers", "industry"):
|
|
for path in sorted((ROOT / collection / "items").glob("*.md")):
|
|
if path.name == "README.md":
|
|
continue
|
|
meta = parse_frontmatter(path.read_text(encoding="utf-8"))
|
|
for key in URL_KEYS:
|
|
value = meta.get(key)
|
|
if value:
|
|
yield path.relative_to(ROOT).as_posix(), key, str(value)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--json", action="store_true")
|
|
parser.add_argument("--timeout", type=int, default=12)
|
|
args = parser.parse_args()
|
|
|
|
results = []
|
|
for path, key, url in iter_urls():
|
|
result = check_url(url, args.timeout)
|
|
result.update({"path": path, "key": key})
|
|
results.append(result)
|
|
|
|
if args.json:
|
|
print(json.dumps(results, ensure_ascii=False, indent=2))
|
|
else:
|
|
for result in results:
|
|
marker = "BLOCKED" if result.get("blocked") else ("OK" if result["ok"] else "FAIL")
|
|
print(f"{marker} {result.get('status')} {result['path']} {result['key']} {result['url']}")
|
|
return 1 if any(not result["ok"] for result in results) else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|