Split ctf_status.py into a tracker package, config from .env only
ctf_status.py is now just the command line; the work lives in tracker/. The rules that decide whether a challenge is believed are in tracker/checks.py, one function per rule, listed in README_CHECKS and REPO_CHECKS. Output is unchanged. Read the config from .env only (ctf_status.env.example -> .env.example), let single_run_lock() hold its own lock files, and rewrite the README as a short overview of what the repo does and how to run it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,13 @@
|
||||
"""The pieces behind ctf_status.py, in the order the data flows through them:
|
||||
|
||||
config.py read .env and the environment
|
||||
gitlab.py list the group's repos, download README.md and description.yml
|
||||
challenge.py turn a README header into a Challenge (status, category, ...)
|
||||
checks.py decide what is wrong with a challenge, and whether to believe
|
||||
a Testing Ready+ status <- the rules live here
|
||||
overview.py run all of the above over every repo, group and sort the result
|
||||
terminal.py the terminal summary
|
||||
markdown.py the HedgeDoc note
|
||||
challs.py challs.txt for the deployment repo
|
||||
hedgedoc.py write the note into HedgeDoc
|
||||
"""
|
||||
@@ -0,0 +1,159 @@
|
||||
"""Reading a challenge's README header.
|
||||
|
||||
Only the template's header block is parsed:
|
||||
|
||||
### Challenge Name
|
||||
- Category: crypto
|
||||
- Difficulty: medium
|
||||
- Status: Testing Ready
|
||||
...
|
||||
|
||||
People fill it in loosely, so the values are normalised here. Whether they are
|
||||
good enough is decided in checks.py.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from .gitlab import Repo
|
||||
|
||||
FIELDS = ("Category", "Difficulty", "Event", "Author", "Tester", "Status")
|
||||
HEADING_RE = re.compile(r"^\s{0,3}#{1,4}\s+(.+?)\s*$", re.M)
|
||||
FIELD_RE = re.compile(r"^\s*[-*]\s*\**(?P<key>[A-Za-z ]+?)\**\s*:\s*(?P<val>.*?)\s*$", re.M)
|
||||
# a value the author never touched: "[crypto, pwn, ...]" or "[Author Name]"
|
||||
PLACEHOLDER_RE = re.compile(r"^\[.*\]$")
|
||||
NO_TESTER_RE = re.compile(r"^(\?+|none|tester name|not found\b.*|tbd)(\s.*)?$", re.I)
|
||||
|
||||
STATUS_ORDER = ["Planning", "PoC Draft", "Testing Ready", "Tested", "Ready"]
|
||||
STATUS_RE = re.compile(r"testing\s*ready|tested|ready|poc(?:\s*draft)?|planning", re.I)
|
||||
STATUS_ICON = {"Planning": "⚪", "PoC Draft": "🟠", "Testing Ready": "🟡",
|
||||
"Tested": "🔵", "Ready": "🟢", None: "❔"}
|
||||
BLOCKED_ICON = "⛔"
|
||||
CATEGORIES = ["crypto", "pwn", "web", "rev", "misc", "osint", "forensic", "hardware"]
|
||||
DIFFICULTIES = ["intro", "easy", "medium", "hard"]
|
||||
DIFF_RE = re.compile(r"intro|easy|medium|hard", re.I)
|
||||
DIFF_RANGE_RE = re.compile(r"(intro|easy|medium|hard)\s*(?:-|–|/|\bto\b)\s*(intro|easy|medium|hard)", re.I)
|
||||
CATEGORY_SPLIT_RE = re.compile(r"[,/&+]| and ")
|
||||
|
||||
|
||||
def clean(v: str) -> str:
|
||||
v = re.sub(r"\\+([\[\]_*])", r"\1", v) # markdown escapes, sometimes doubled
|
||||
return v.replace("**", "").strip()
|
||||
|
||||
|
||||
def is_placeholder(v: str) -> bool:
|
||||
v = clean(v)
|
||||
return not v or bool(PLACEHOLDER_RE.match(v))
|
||||
|
||||
|
||||
def stage_of(raw: str) -> tuple[str | None, bool]:
|
||||
"""Map a free-text status onto STATUS_ORDER -> (stage, clear).
|
||||
|
||||
"Tested with h4tum-chall-format" is clearly Tested. "Planning to PoC Draft"
|
||||
or a half-deleted option list is not; we take the least advanced stage it
|
||||
names, so nothing looks further along than it is, and call it unclear."""
|
||||
if is_placeholder(raw):
|
||||
return None, False
|
||||
found = set()
|
||||
for m in STATUS_RE.finditer(raw):
|
||||
w = m.group(0).lower()
|
||||
found.add("Testing Ready" if w.startswith("testing") else
|
||||
"PoC Draft" if w.startswith("poc") else w.title())
|
||||
if not found:
|
||||
return None, False
|
||||
return min(found, key=STATUS_ORDER.index), len(found) == 1
|
||||
|
||||
|
||||
def difficulty_of(raw: str) -> str | None:
|
||||
""""medium/hard", "medium-hard", "easy to medium" -> a range "medium-hard"."""
|
||||
if is_placeholder(raw):
|
||||
return None
|
||||
m = DIFF_RANGE_RE.search(raw)
|
||||
if m:
|
||||
ends = sorted({m.group(1).lower(), m.group(2).lower()}, key=DIFFICULTIES.index)
|
||||
return "-".join(ends) if len(ends) > 1 else ends[0]
|
||||
# prose like "currently medium, maybe also an easy and a hard version": the first one
|
||||
m = DIFF_RE.search(raw)
|
||||
return m.group(0).lower() if m else None
|
||||
|
||||
|
||||
def difficulty_rank(d: str | None) -> float:
|
||||
if not d:
|
||||
return 99
|
||||
ends = d.split("-")
|
||||
return sum(DIFFICULTIES.index(e) for e in ends) / len(ends)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Challenge:
|
||||
repo: Repo
|
||||
name: str = ""
|
||||
fields: dict = field(default_factory=dict)
|
||||
# filled in by overview.collect(), see checks.py
|
||||
problems: list = field(default_factory=list) # readme_problems()
|
||||
blockers: list = field(default_factory=list) # repo_blockers()
|
||||
|
||||
def get(self, key: str, dash: str = "—") -> str:
|
||||
v = clean(self.fields.get(key, ""))
|
||||
return dash if is_placeholder(v) else v
|
||||
|
||||
# -- derived -----------------------------------------------------------
|
||||
@property
|
||||
def named(self) -> bool:
|
||||
return bool(self.name) and not is_placeholder(self.name)
|
||||
|
||||
@property
|
||||
def title(self) -> str:
|
||||
return self.name if self.named else self.repo.path
|
||||
|
||||
@property
|
||||
def touched(self) -> bool:
|
||||
"""Did anyone put *any* real content in the README template?"""
|
||||
return self.named or any(not is_placeholder(self.fields.get(f, "")) for f in FIELDS)
|
||||
|
||||
@property
|
||||
def status_raw(self) -> str:
|
||||
return self.get("Status")
|
||||
|
||||
@property
|
||||
def stage(self) -> str | None:
|
||||
return stage_of(self.fields.get("Status", ""))[0]
|
||||
|
||||
@property
|
||||
def difficulty(self) -> str | None:
|
||||
return difficulty_of(self.fields.get("Difficulty", ""))
|
||||
|
||||
@property
|
||||
def category(self) -> str | None:
|
||||
"""The first category named; "rev, misc" counts as rev."""
|
||||
raw = self.get("Category", "")
|
||||
first = CATEGORY_SPLIT_RE.split(raw)[0].strip().lower()
|
||||
return first or None
|
||||
|
||||
@property
|
||||
def claims_testable(self) -> bool:
|
||||
"""The README says Testing Ready or beyond; checks.is_testable() also asks the repo."""
|
||||
return bool(self.stage) and STATUS_ORDER.index(self.stage) >= STATUS_ORDER.index("Testing Ready")
|
||||
|
||||
@property
|
||||
def icon(self) -> str:
|
||||
return BLOCKED_ICON if self.blockers else STATUS_ICON[self.stage]
|
||||
|
||||
@property
|
||||
def has_tester(self) -> bool:
|
||||
t = self.get("Tester", "")
|
||||
return bool(t) and not NO_TESTER_RE.match(t)
|
||||
|
||||
|
||||
def parse_challenge(repo: Repo) -> Challenge:
|
||||
c = Challenge(repo=repo)
|
||||
text = repo.readme or ""
|
||||
m = HEADING_RE.search(text)
|
||||
if m:
|
||||
c.name = clean(m.group(1))
|
||||
for fm in FIELD_RE.finditer(text):
|
||||
key = fm.group("key").strip().title()
|
||||
if key in FIELDS and key not in c.fields:
|
||||
c.fields[key] = fm.group("val")
|
||||
return c
|
||||
@@ -0,0 +1,17 @@
|
||||
"""challs.txt for the deployment repo: what infra deploys."""
|
||||
from __future__ import annotations
|
||||
|
||||
from .overview import Overview
|
||||
|
||||
|
||||
def render_challs(ov: Overview) -> str:
|
||||
"""One clone URL per line, for everything that is Testing Ready or beyond
|
||||
and not blocked."""
|
||||
return "".join(f"{c.repo.ssh_url}\n" for c in ov.testable)
|
||||
|
||||
|
||||
def challs_message(old: str, new: str, path: str) -> str:
|
||||
name = lambda u: u.rsplit("/", 1)[-1].removesuffix(".git")
|
||||
was, now = set(old.split()), set(new.split())
|
||||
parts = [f"+{name(u)}" for u in sorted(now - was)] + [f"-{name(u)}" for u in sorted(was - now)]
|
||||
return f"Update {path} from challenge status" + (f": {' '.join(parts)}" if parts else "")
|
||||
@@ -0,0 +1,198 @@
|
||||
"""What is wrong with a challenge, and whether its status is believed.
|
||||
|
||||
There are two kinds of finding:
|
||||
|
||||
* problems (readme_problems): the README header is incomplete or unclear.
|
||||
They are listed under "needs attention"; nothing else happens.
|
||||
|
||||
* blockers (repo_blockers): the README claims Testing Ready or beyond, but the
|
||||
repo shows the challenge is not finished. A blocked challenge keeps its
|
||||
status in the counts, but is left out of "ready for infra testing" and out of
|
||||
challs.txt, so it does not get deployed.
|
||||
|
||||
Every rule is one small function that returns what it found wrong, as a list
|
||||
of messages (empty: nothing wrong). README_CHECKS and REPO_CHECKS list the rules
|
||||
in the order their messages are shown. To add a rule, write a function and
|
||||
put it in the right list.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
import yaml
|
||||
|
||||
from .challenge import CATEGORY_SPLIT_RE, Challenge, stage_of
|
||||
from .gitlab import Repo
|
||||
|
||||
|
||||
def is_testable(c: Challenge) -> bool:
|
||||
"""Can infra deploy and test it now? The README says Testing Ready or beyond,
|
||||
and the repo has no blockers."""
|
||||
return c.claims_testable and not c.blockers
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- README ----
|
||||
# Run on every challenge whose README somebody filled in.
|
||||
|
||||
def status_is_clear(c: Challenge) -> list[str]:
|
||||
stage, clear = stage_of(c.fields.get("Status", ""))
|
||||
if stage is None:
|
||||
return [f"status unknown ({c.status_raw!r})" if c.status_raw != "—" else "no status"]
|
||||
if not clear:
|
||||
return [f"status unclear ({c.status_raw!r}, counted as {stage})"]
|
||||
return []
|
||||
|
||||
|
||||
def has_category(c: Challenge) -> list[str]:
|
||||
return [] if c.category else ["no category"]
|
||||
|
||||
|
||||
def difficulty_is_clear(c: Challenge) -> list[str]:
|
||||
if c.difficulty:
|
||||
return []
|
||||
raw = c.get("Difficulty")
|
||||
return [f"difficulty unclear ({raw!r})" if raw != "—" else "no difficulty"]
|
||||
|
||||
|
||||
def testable_has_tester(c: Challenge) -> list[str]:
|
||||
"""Once a challenge is Testing Ready, somebody has to test it."""
|
||||
return ["no tester"] if c.claims_testable and not c.has_tester else []
|
||||
|
||||
|
||||
def has_name(c: Challenge) -> list[str]:
|
||||
return [] if c.named else ["no challenge name in README"]
|
||||
|
||||
|
||||
README_CHECKS = (
|
||||
status_is_clear,
|
||||
has_category,
|
||||
difficulty_is_clear,
|
||||
testable_has_tester,
|
||||
has_name,
|
||||
)
|
||||
|
||||
|
||||
def readme_problems(c: Challenge) -> list[str]:
|
||||
return [msg for check in README_CHECKS for msg in check(c)]
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ repo ----
|
||||
# Run only on challenges whose README claims Testing Ready or beyond. These are
|
||||
# cheap signs that the challenge was never finished: passing them is no proof
|
||||
# that it builds or solves.
|
||||
|
||||
FLAG_RE = re.compile(r"^h4tum\{[^{}]+\}$")
|
||||
# what every repo starts with: the challenge-format example "netcat-check-multi-container"
|
||||
TEMPLATE_ID = "netcat-check-multi-container"
|
||||
TEMPLATE_NAME = "Netcat Check Multi Container"
|
||||
TEMPLATE_AUTHOR = "jh"
|
||||
TEMPLATE_FLAG = "h4tum{n3tc4t_1s_c0ol}"
|
||||
TEMPLATE_CONTAINERS = {"cont1", "cont2"}
|
||||
TEMPLATE_DIRS = {"container-1", "container-2"}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Description:
|
||||
"""A repo's description.yml: a single `<id>: {...}` block."""
|
||||
id: str
|
||||
data: dict # everything under <id>
|
||||
flag: dict # data["flag"], {} if missing
|
||||
containers: list[dict] # data["containers"], {} entries dropped
|
||||
deploy_dirs: list[str] # each container's deploy_dir, normalised; "." for the repo root
|
||||
files: set[str] # every file and dir in the repo
|
||||
|
||||
|
||||
class DescriptionError(Exception):
|
||||
"""description.yml is missing or unusable; nothing else can be checked."""
|
||||
|
||||
|
||||
def read_description(repo: Repo) -> Description:
|
||||
if repo.desc is None:
|
||||
raise DescriptionError("no description.yml")
|
||||
try:
|
||||
doc = yaml.safe_load(repo.desc)
|
||||
except yaml.YAMLError as e:
|
||||
raise DescriptionError(f"description.yml does not parse ({str(e).splitlines()[0]})") from None
|
||||
if not isinstance(doc, dict) or len(doc) != 1 or not isinstance(next(iter(doc.values())), dict):
|
||||
raise DescriptionError("description.yml is not a single `<id>: {...}` block")
|
||||
cid, data = next(iter(doc.items()))
|
||||
containers = [x for x in data.get("containers") or [] if isinstance(x, dict)]
|
||||
return Description(
|
||||
id=cid, data=data,
|
||||
flag=data.get("flag") if isinstance(data.get("flag"), dict) else {},
|
||||
containers=containers,
|
||||
deploy_dirs=[os.path.normpath(str(x.get("deploy_dir") or ".")) for x in containers],
|
||||
files=repo.files or set(),
|
||||
)
|
||||
|
||||
|
||||
def no_template_leftovers(c: Challenge, d: Description) -> list[str]:
|
||||
"""description.yml was copied from the template; anything still from it means
|
||||
nobody filled it in properly."""
|
||||
leftovers = [what for what, hit in (
|
||||
(f"id {d.id!r}", d.id == TEMPLATE_ID),
|
||||
(f"name {d.data.get('name')!r}", d.data.get("name") == TEMPLATE_NAME),
|
||||
(f"author {d.data.get('author')!r}", d.data.get("author") == TEMPLATE_AUTHOR),
|
||||
("flag", d.flag.get("content") == TEMPLATE_FLAG),
|
||||
("containers", any(x.get("name") in TEMPLATE_CONTAINERS for x in d.containers)
|
||||
or bool(TEMPLATE_DIRS & set(d.deploy_dirs))),
|
||||
) if hit]
|
||||
return ["description.yml still has the template's " + ", ".join(leftovers)] if leftovers else []
|
||||
|
||||
|
||||
def category_matches_readme(c: Challenge, d: Description) -> list[str]:
|
||||
"""Its category must be one of the README's ("rev, misc" allows either)."""
|
||||
cat = str(d.data.get("category") or "").strip().lower()
|
||||
readme_cats = {x.strip().lower() for x in CATEGORY_SPLIT_RE.split(c.get("Category", ""))}
|
||||
if c.category and cat not in readme_cats:
|
||||
return [f"category {cat or 'missing'} in description.yml, README says {c.get('Category')}"]
|
||||
return []
|
||||
|
||||
|
||||
def flag_is_valid(c: Challenge, d: Description) -> list[str]:
|
||||
"""The flag is h4tum{...}; a dynamic one needs a container to hand it to."""
|
||||
b = []
|
||||
content = str(d.flag.get("content") or "")
|
||||
if not FLAG_RE.match(content):
|
||||
b.append(f"flag {content} is not h4tum{{...}}" if content else "no flag in description.yml")
|
||||
if d.flag.get("dynamic") and not any(isinstance(x.get("flag"), dict) for x in d.containers):
|
||||
b.append("dynamic flag, but no container is handed one")
|
||||
return b
|
||||
|
||||
|
||||
def deploy_dirs_have_dockerfiles(c: Challenge, d: Description) -> list[str]:
|
||||
b = []
|
||||
for dd in d.deploy_dirs:
|
||||
if dd != "." and dd not in d.files:
|
||||
b.append(f"deploy_dir {dd} does not exist")
|
||||
elif ("Dockerfile" if dd == "." else f"{dd}/Dockerfile") not in d.files:
|
||||
b.append(f"no Dockerfile in {dd}")
|
||||
return b
|
||||
|
||||
|
||||
def has_solve_script(c: Challenge, d: Description) -> list[str]:
|
||||
"""A solve.py anywhere in the repo."""
|
||||
if any(p == "solve.py" or p.endswith("/solve.py") for p in d.files):
|
||||
return []
|
||||
return ["no solve.py"]
|
||||
|
||||
|
||||
REPO_CHECKS = (
|
||||
no_template_leftovers,
|
||||
category_matches_readme,
|
||||
flag_is_valid,
|
||||
deploy_dirs_have_dockerfiles,
|
||||
has_solve_script,
|
||||
)
|
||||
|
||||
|
||||
def repo_blockers(c: Challenge) -> list[str]:
|
||||
"""Reasons not to believe the README's Testing Ready+ status. Needs
|
||||
GitLab.load_files(c.repo) first."""
|
||||
try:
|
||||
d = read_description(c.repo)
|
||||
except DescriptionError as e:
|
||||
return [str(e)]
|
||||
return [msg for check in REPO_CHECKS for msg in check(c, d)]
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Config from .env next to ctf_status.py; environment variables override it."""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
CFG_KEYS = ("GITLAB_GROUP_URL", "GITLAB_TOKEN")
|
||||
OPT_KEYS = ("HEDGEDOC_NOTE", "HEDGEDOC_TAGS", "TITLE",
|
||||
"CHALLS_REPO_URL", "CHALLS_PATH", "CHALLS_BRANCH")
|
||||
# the names a plain `.env` tends to use for the same thing
|
||||
ALIASES = {"GL_PAT": "GITLAB_TOKEN", "GITLAB_URL": "GITLAB_GROUP_URL"}
|
||||
DEFAULT_TAGS = "Logistics, h4tum CTF 2026"
|
||||
DEFAULT_TITLE = "CTF challenge tracking"
|
||||
DEFAULT_CHALLS_PATH = "2026/challs.txt"
|
||||
|
||||
|
||||
def load_env() -> dict:
|
||||
cfg = {}
|
||||
env_file = Path(__file__).parent.parent / ".env" # the repo, next to ctf_status.py
|
||||
if env_file.exists():
|
||||
for line in env_file.read_text().splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
k, v = line.split("=", 1)
|
||||
k = k.strip().upper()
|
||||
cfg.setdefault(ALIASES.get(k, k), v.strip().strip("'\""))
|
||||
for k in CFG_KEYS + OPT_KEYS + tuple(ALIASES):
|
||||
if os.environ.get(k):
|
||||
cfg[ALIASES.get(k, k)] = os.environ[k]
|
||||
missing = [k for k in CFG_KEYS if not cfg.get(k)]
|
||||
if missing:
|
||||
sys.exit(f"missing config: {', '.join(missing)} (see .env.example)")
|
||||
return cfg
|
||||
@@ -0,0 +1,121 @@
|
||||
"""GitLab REST access: the challenge repos we read, and the one file we write."""
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from dataclasses import dataclass
|
||||
from urllib.parse import quote, urlsplit
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
@dataclass
|
||||
class Repo:
|
||||
path: str
|
||||
web_url: str
|
||||
ssh_url: str
|
||||
last_activity: str # YYYY-MM-DD
|
||||
readme: str | None # None: the repo has no README at all
|
||||
id: int = 0
|
||||
branch: str | None = None
|
||||
# only fetched for challenges whose README claims Testing Ready or beyond
|
||||
files: set[str] | None = None # every file and dir in the repo
|
||||
desc: str | None = None # description.yml, None if there is none
|
||||
|
||||
|
||||
class GitLab:
|
||||
"""Access to one group, plus the one file we write elsewhere. The token needs
|
||||
Project: Read (to list the group) and Repository: Read (to download files);
|
||||
--challs additionally needs read + write on the deployment repo."""
|
||||
|
||||
def __init__(self, group_url: str, token: str, timeout: int = 20):
|
||||
parts = urlsplit(group_url)
|
||||
self.api = f"{parts.scheme}://{parts.netloc}/api/v4"
|
||||
self.group = parts.path.strip("/")
|
||||
self.timeout = timeout
|
||||
self.s = requests.Session()
|
||||
self.s.headers["PRIVATE-TOKEN"] = token
|
||||
|
||||
def _get(self, url: str, **params) -> requests.Response:
|
||||
return self._req("GET", url, params=params)
|
||||
|
||||
def _req(self, method: str, url: str, ok: tuple = (), **kw) -> requests.Response:
|
||||
r = self.s.request(method, url, timeout=self.timeout, **kw)
|
||||
if r.status_code in ok:
|
||||
return r
|
||||
if r.status_code == 403:
|
||||
# fine-grained tokens say exactly which permission is missing
|
||||
try:
|
||||
why = r.json().get("error_description") or r.text
|
||||
except ValueError:
|
||||
why = r.text
|
||||
raise RuntimeError(f"gitlab refused {r.url}: {why}")
|
||||
r.raise_for_status()
|
||||
return r
|
||||
|
||||
def projects(self) -> list[dict]:
|
||||
out, page = [], "1"
|
||||
while page:
|
||||
r = self._get(f"{self.api}/groups/{quote(self.group, safe='')}/projects",
|
||||
include_subgroups="true", per_page="100", page=page,
|
||||
order_by="path", sort="asc")
|
||||
out += r.json()
|
||||
page = r.headers.get("X-Next-Page", "")
|
||||
# repos GitLab is about to delete are renamed "...-deletion_scheduled-<id>"
|
||||
return [p for p in out if not p.get("archived") and not p.get("marked_for_deletion_on")]
|
||||
|
||||
def readme(self, p: dict) -> str | None:
|
||||
branch, url = p.get("default_branch"), p.get("readme_url")
|
||||
if not branch or not url or f"/-/blob/{branch}/" not in url:
|
||||
return None
|
||||
return self._raw(p["id"], url.split(f"/-/blob/{branch}/", 1)[1], branch)
|
||||
|
||||
def _raw(self, project_id: int, path: str, branch: str) -> str:
|
||||
r = self._get(f"{self.api}/projects/{project_id}/repository/files/"
|
||||
f"{quote(path, safe='')}/raw", ref=branch)
|
||||
return r.content.decode("utf-8", "replace")
|
||||
|
||||
def repos(self) -> list[Repo]:
|
||||
projects = self.projects()
|
||||
|
||||
def fetch(p: dict) -> Repo:
|
||||
return Repo(path=p["path"], web_url=p["web_url"],
|
||||
ssh_url=p["ssh_url_to_repo"],
|
||||
last_activity=(p.get("last_activity_at") or "")[:10],
|
||||
readme=self.readme(p), id=p["id"], branch=p.get("default_branch"))
|
||||
|
||||
with ThreadPoolExecutor(max_workers=8) as pool:
|
||||
return list(pool.map(fetch, projects))
|
||||
|
||||
def load_files(self, repo: Repo) -> None:
|
||||
"""Fill in repo.files and repo.desc, for checks.repo_blockers()."""
|
||||
repo.files = set()
|
||||
if not repo.branch:
|
||||
return
|
||||
page = "1"
|
||||
while page:
|
||||
r = self._get(f"{self.api}/projects/{repo.id}/repository/tree", ref=repo.branch,
|
||||
recursive="true", per_page="100", page=page)
|
||||
repo.files |= {t["path"] for t in r.json()}
|
||||
page = r.headers.get("X-Next-Page", "")
|
||||
if "description.yml" in repo.files:
|
||||
repo.desc = self._raw(repo.id, "description.yml", repo.branch)
|
||||
|
||||
def write_file(self, repo_url: str, path: str, content: str, branch: str | None,
|
||||
message) -> bool:
|
||||
"""Commit `content` to `path` in the repo at `repo_url` (a web URL), creating
|
||||
the file if needed; `message(old_content)` makes the commit message.
|
||||
False if the file already said exactly this."""
|
||||
proj = quote(urlsplit(repo_url).path.strip("/").split("/-/", 1)[0], safe="")
|
||||
base = f"{self.api}/projects/{proj}"
|
||||
if not branch:
|
||||
branch = self._get(base).json()["default_branch"]
|
||||
url = f"{base}/repository/files/{quote(path, safe='')}"
|
||||
r = self._req("GET", url, ok=(404,), params={"ref": branch})
|
||||
exists = r.status_code != 404
|
||||
old = base64.b64decode(r.json()["content"]).decode("utf-8", "replace") if exists else ""
|
||||
if exists and old == content:
|
||||
return False
|
||||
self._req("PUT" if exists else "POST", url,
|
||||
json={"branch": branch, "content": content, "commit_message": message(old)})
|
||||
return True
|
||||
@@ -0,0 +1,180 @@
|
||||
"""Writing the note into HedgeDoc 1.x.
|
||||
|
||||
HedgeDoc 1.x has no REST endpoint that edits a note's body, so this pretends to
|
||||
be the web editor. Nothing in here knows about challenges.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import random
|
||||
import sys
|
||||
import time
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
import requests
|
||||
|
||||
# engine.io v3 packet types
|
||||
EIO_OPEN, EIO_CLOSE, EIO_PING, EIO_PONG, EIO_MSG = "0", "1", "2", "3", "4"
|
||||
# socket.io packet types, as the second character of an EIO_MSG packet
|
||||
SIO_CONNECT, SIO_DISCONNECT, SIO_EVENT, SIO_ERROR = "0", "1", "2", "4"
|
||||
|
||||
|
||||
def u16len(s: str) -> int:
|
||||
"""Length in UTF-16 code units — what JavaScript's `String.length` counts, and
|
||||
therefore what engine.io payload framing and OT operations count. Matters as
|
||||
soon as the markdown holds an emoji (🔴 is one Python char but two units)."""
|
||||
return len(s.encode("utf-16-le")) // 2
|
||||
|
||||
|
||||
def encode_payload(packets: tuple) -> bytes:
|
||||
return "".join(f"{u16len(p)}:{p}" for p in packets).encode("utf-8")
|
||||
|
||||
|
||||
def decode_payload(text: str) -> list[str]:
|
||||
buf = text.encode("utf-16-le")
|
||||
out, i = [], 0
|
||||
while i < len(buf):
|
||||
j = i
|
||||
while buf[j:j + 2] != b":\x00":
|
||||
j += 2
|
||||
if j >= len(buf):
|
||||
raise RuntimeError("malformed engine.io payload")
|
||||
n = int(buf[i:j].decode("utf-16-le"))
|
||||
j += 2
|
||||
out.append(buf[j:j + 2 * n].decode("utf-16-le"))
|
||||
i = j + 2 * n
|
||||
return out
|
||||
|
||||
|
||||
class HedgeDocNote:
|
||||
"""Overwrite an existing HedgeDoc 1.x note in place.
|
||||
|
||||
Speaks the same engine.io/socket.io protocol the web editor uses, over XHR
|
||||
long-polling (no extra dependencies beyond `requests`):
|
||||
|
||||
GET /<note> -> session cookie; the socket rejects us without one
|
||||
GET /socket.io/ -> engine.io handshake, yields a session id
|
||||
poll for 'doc' -> the note's current text and OT revision
|
||||
emit 'operation' -> [-len(old), new]: delete everything, insert ours
|
||||
poll for 'ack' -> the server applied it and told the other editors
|
||||
|
||||
The note's permission must allow us to write ("freely" for an anonymous run).
|
||||
"""
|
||||
|
||||
def __init__(self, url: str, timeout: int = 20):
|
||||
parts = urlsplit(url)
|
||||
path = parts.path.rstrip("/")
|
||||
if not parts.scheme or not parts.netloc or "/" not in path:
|
||||
raise ValueError(f"expected a note URL like https://host/<noteId>, got {url!r}")
|
||||
self.note = path.rsplit("/", 1)[-1]
|
||||
prefix = path[: -len(self.note)].rstrip("/")
|
||||
self.base = f"{parts.scheme}://{parts.netloc}{prefix}"
|
||||
self.timeout = timeout
|
||||
self.sid: str | None = None
|
||||
self.s = requests.Session()
|
||||
self.s.headers["Origin"] = f"{parts.scheme}://{parts.netloc}"
|
||||
|
||||
# -- transport ---------------------------------------------------------
|
||||
def _params(self) -> dict:
|
||||
p = {"EIO": "3", "transport": "polling", "noteId": self.note,
|
||||
"t": format(random.getrandbits(40), "x")}
|
||||
if self.sid:
|
||||
p["sid"] = self.sid
|
||||
return p
|
||||
|
||||
def _open(self) -> None:
|
||||
self.sid = None
|
||||
self.s.get(f"{self.base}/{self.note}", timeout=self.timeout).raise_for_status()
|
||||
r = self.s.get(f"{self.base}/socket.io/", params=self._params(), timeout=self.timeout)
|
||||
r.raise_for_status()
|
||||
pkts = decode_payload(r.text)
|
||||
if not pkts or pkts[0][:1] != EIO_OPEN:
|
||||
raise RuntimeError(f"unexpected engine.io handshake: {pkts[:1]}")
|
||||
self.sid = json.loads(pkts[0][1:])["sid"]
|
||||
|
||||
def _send(self, *packets: str) -> None:
|
||||
r = self.s.post(f"{self.base}/socket.io/", params=self._params(),
|
||||
data=encode_payload(packets),
|
||||
headers={"Content-Type": "text/plain;charset=UTF-8"},
|
||||
timeout=self.timeout)
|
||||
r.raise_for_status()
|
||||
|
||||
def _events(self):
|
||||
"""Poll once, answer keepalives, yield decoded socket.io events."""
|
||||
r = self.s.get(f"{self.base}/socket.io/", params=self._params(), timeout=self.timeout)
|
||||
r.raise_for_status()
|
||||
for p in decode_payload(r.text):
|
||||
if p == EIO_PING:
|
||||
self._send(EIO_PONG)
|
||||
elif p == EIO_CLOSE:
|
||||
raise RuntimeError("hedgedoc closed the connection")
|
||||
elif p[:1] == EIO_MSG:
|
||||
kind, data = p[1:2], p[2:]
|
||||
if kind == SIO_ERROR:
|
||||
raise RuntimeError(f"hedgedoc refused the socket: {data}")
|
||||
if kind == SIO_EVENT:
|
||||
yield json.loads(data)
|
||||
|
||||
def _wait(self, name: str, polls: int = 8):
|
||||
for _ in range(polls):
|
||||
for ev in self._events():
|
||||
if ev and ev[0] == name:
|
||||
return ev[1] if len(ev) > 1 else None
|
||||
raise RuntimeError(f"timed out waiting for {name!r} from hedgedoc")
|
||||
|
||||
def _close(self) -> None:
|
||||
try:
|
||||
self._send(EIO_MSG + SIO_DISCONNECT, EIO_CLOSE)
|
||||
except requests.RequestException:
|
||||
pass # we already got our ack; a tidy goodbye is a nicety
|
||||
|
||||
# -- api ---------------------------------------------------------------
|
||||
def download(self) -> str:
|
||||
r = self.s.get(f"{self.base}/{self.note}/download", timeout=self.timeout)
|
||||
r.raise_for_status()
|
||||
return r.content.decode("utf-8", "replace")
|
||||
|
||||
def replace(self, markdown: str) -> bool:
|
||||
"""Replace the note body. False if it already said exactly this."""
|
||||
self._open()
|
||||
doc = self._wait("doc")
|
||||
old, revision = doc["str"], doc["revision"]
|
||||
if old == markdown:
|
||||
self._close()
|
||||
return False
|
||||
# an OT operation is a list of: n>0 keep n chars, n<0 drop n, str insert
|
||||
ops = ([-u16len(old)] if old else []) + ([markdown] if markdown else [])
|
||||
self._send(EIO_MSG + SIO_EVENT
|
||||
+ json.dumps(["operation", revision, ops, None], ensure_ascii=False))
|
||||
self._wait("ack")
|
||||
time.sleep(2) # give the realtime server a moment to flush the note to its DB
|
||||
self._close()
|
||||
return True
|
||||
|
||||
|
||||
def update_hedgedoc(url: str, md: str, retries: int = 3, quiet: bool = False) -> bool:
|
||||
"""Write `md` into the existing note at `url`, then read it back to be sure."""
|
||||
for attempt in range(1, retries + 1):
|
||||
try:
|
||||
note = HedgeDocNote(url)
|
||||
changed = note.replace(md)
|
||||
if note.download() != md:
|
||||
raise RuntimeError("note does not match what we wrote")
|
||||
return changed
|
||||
except (requests.RequestException, RuntimeError, ValueError, KeyError) as e:
|
||||
if attempt == retries:
|
||||
raise
|
||||
if not quiet:
|
||||
print(f"hedgedoc update failed ({e}) — retry {attempt + 1}/{retries}",
|
||||
file=sys.stderr)
|
||||
time.sleep(3 * attempt)
|
||||
return False
|
||||
|
||||
|
||||
def push_hedgedoc(base: str, md: str) -> str:
|
||||
"""POST `md` as a *new* note; returns its URL."""
|
||||
r = requests.post(base.rstrip("/") + "/new", data=md.encode(),
|
||||
headers={"Content-Type": "text/markdown"}, allow_redirects=False)
|
||||
r.raise_for_status()
|
||||
loc = r.headers.get("Location", "")
|
||||
return loc if loc.startswith("http") else base.rstrip("/") + loc
|
||||
@@ -0,0 +1,114 @@
|
||||
"""The HedgeDoc note, as markdown."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import Counter
|
||||
from datetime import datetime
|
||||
|
||||
from .challenge import BLOCKED_ICON, STATUS_ICON, STATUS_ORDER
|
||||
from .overview import Overview, icons
|
||||
|
||||
|
||||
def md_cell(v: str) -> str:
|
||||
return v.replace("|", "\\|")
|
||||
|
||||
|
||||
def render_hedgedoc(ov: Overview, title: str, tags: str = "") -> str:
|
||||
stamp = datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S %Z")
|
||||
stages = ov.stages
|
||||
groups = ov.by_category()
|
||||
total = Counter(ch.stage for ch in ov.challenges)
|
||||
L = []
|
||||
if tags:
|
||||
# HedgeDoc reads this YAML block off the top of the note
|
||||
L += ["---", f"tags: {tags}", "---", ""]
|
||||
L += [f"# {title}", "",
|
||||
f"**Last updated:** {stamp}", "",
|
||||
f"*generated by `ctf_status.py` from the README.md of every repo in the "
|
||||
f"GitLab challenges group — {len(ov.challenges)} challenges*", "",
|
||||
" · ".join(f"{STATUS_ICON[s]} **{s or 'unknown'}** {total[s]}"
|
||||
for s in stages if total[s] or s), ""]
|
||||
|
||||
ready = ov.testable
|
||||
L += [f"## ✅ Ready for infra testing ({len(ready)} of {len(ov.challenges)})", "",
|
||||
"*every challenge at Testing Ready or beyond whose repo backs that up — "
|
||||
"these can be deployed and tested against the infrastructure now*", ""]
|
||||
if ready:
|
||||
L += ["| | Challenge | Category | Difficulty | Status | Event | Author | Tester |",
|
||||
"| --- | --- | --- | --- | --- | --- | --- | --- |"]
|
||||
for ch in ready:
|
||||
L.append("| " + " | ".join(md_cell(x) for x in (
|
||||
STATUS_ICON[ch.stage], f"[{ch.title}]({ch.repo.web_url})",
|
||||
ch.category or "?", ch.difficulty or "?", ch.stage, ch.get("Event"),
|
||||
ch.get("Author"), ch.get("Tester") if ch.has_tester else "—")) + " |")
|
||||
else:
|
||||
L.append("*nothing yet*")
|
||||
L.append("")
|
||||
|
||||
if ov.blocked:
|
||||
L += [f"## {BLOCKED_ICON} Claims Testing Ready+, but the repo disagrees "
|
||||
f"({len(ov.blocked)})", "",
|
||||
"*the README says so, but description.yml or the repo shows it is not "
|
||||
"finished — kept out of challs.txt until this is fixed*", ""]
|
||||
for ch in ov.blocked:
|
||||
L.append(f"- {STATUS_ICON[ch.stage]} [{md_cell(ch.title)}]({ch.repo.web_url}) "
|
||||
f"({ch.stage}): " + "; ".join(ch.blockers))
|
||||
L.append("")
|
||||
|
||||
L += ["## Status by category", "",
|
||||
"| Category | " + " | ".join(f"{STATUS_ICON[s]} {s or '?'}" for s in stages)
|
||||
+ " | total |",
|
||||
"| --- |" + " ---: |" * (len(stages) + 1)]
|
||||
for cat in ov.categories:
|
||||
chs = groups.get(cat, [])
|
||||
cnt = Counter(ch.stage for ch in chs)
|
||||
L.append(f"| {cat} | " + " | ".join(str(cnt[s] or "·") for s in stages)
|
||||
+ f" | **{len(chs)}** |")
|
||||
L.append("| **total** | " + " | ".join(f"**{total[s]}**" for s in stages)
|
||||
+ f" | **{len(ov.challenges)}** |")
|
||||
|
||||
diffs = ov.difficulties
|
||||
L += ["", "## Category × difficulty", "",
|
||||
"*one dot per challenge, coloured by status: "
|
||||
+ " ".join(f"{STATUS_ICON[s]} {s}" for s in STATUS_ORDER)
|
||||
+ f" {BLOCKED_ICON} blocked*", "",
|
||||
"| Category | " + " | ".join(diffs) + " | total |",
|
||||
"| --- |" + " --- |" * len(diffs) + " ---: |"]
|
||||
for cat in ov.categories:
|
||||
chs = groups.get(cat, [])
|
||||
cells = [icons([ch for ch in chs if (ch.difficulty or "?") == d]) or "·" for d in diffs]
|
||||
L.append(f"| {cat} | " + " | ".join(cells) + f" | **{len(chs)}** |")
|
||||
dtot = Counter(ch.difficulty or "?" for ch in ov.challenges)
|
||||
L.append("| **total** | " + " | ".join(f"**{dtot[d]}**" for d in diffs)
|
||||
+ f" | **{len(ov.challenges)}** |")
|
||||
|
||||
L += ["", "## Challenges"]
|
||||
for cat in ov.categories:
|
||||
chs = groups.get(cat)
|
||||
if not chs:
|
||||
continue
|
||||
L += ["", f"### {cat} ({len(chs)})", "",
|
||||
"| | Challenge | Difficulty | Status | Event | Author | Tester | Last activity |",
|
||||
"| --- | --- | --- | --- | --- | --- | --- | --- |"]
|
||||
for ch in chs:
|
||||
status = ch.stage or "?"
|
||||
if ch.status_raw.lower() != status.lower():
|
||||
status += f" *({ch.status_raw})*"
|
||||
cat_raw = ch.get("Category", "")
|
||||
name = f"[{md_cell(ch.title)}]({ch.repo.web_url})"
|
||||
if cat_raw.lower() != cat:
|
||||
name += f" *({cat_raw})*"
|
||||
L.append("| " + " | ".join(md_cell(x) for x in (
|
||||
ch.icon, name, ch.difficulty or ch.get("Difficulty"), status,
|
||||
ch.get("Event"), ch.get("Author"),
|
||||
ch.get("Tester") if ch.has_tester else "—", ch.repo.last_activity)) + " |")
|
||||
|
||||
L += ["", "## Needs attention", ""]
|
||||
issues = [ch for ch in sorted(ov.challenges, key=lambda c: c.repo.path.lower()) if ch.problems]
|
||||
for ch in issues:
|
||||
L.append(f"- [{md_cell(ch.title)}]({ch.repo.web_url}): " + "; ".join(ch.problems))
|
||||
for ch in ov.blank:
|
||||
why = "no README" if ch.repo.readme is None else "README is still the blank template"
|
||||
L.append(f"- [{ch.repo.path}]({ch.repo.web_url}): {why}")
|
||||
if not issues and not ov.blank:
|
||||
L.append("*nothing — every README is filled in. suspicious.*")
|
||||
return "\n".join(L) + "\n"
|
||||
@@ -0,0 +1,78 @@
|
||||
"""Every challenge in the group, checked, grouped and sorted for the reports."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from dataclasses import dataclass
|
||||
|
||||
from .challenge import CATEGORIES, DIFFICULTIES, STATUS_ORDER, Challenge, difficulty_rank, parse_challenge
|
||||
from .checks import is_testable, readme_problems, repo_blockers
|
||||
from .gitlab import GitLab
|
||||
|
||||
|
||||
@dataclass
|
||||
class Overview:
|
||||
challenges: list # repos with a README somebody filled in
|
||||
blank: list # repos with no README, or an untouched template
|
||||
|
||||
@property
|
||||
def categories(self) -> list[str]:
|
||||
extra = sorted({c.category or "?" for c in self.challenges} - set(CATEGORIES))
|
||||
return CATEGORIES + extra
|
||||
|
||||
@property
|
||||
def difficulties(self) -> list[str]:
|
||||
seen = {c.difficulty for c in self.challenges if c.difficulty}
|
||||
unknown = ["?"] if any(not c.difficulty for c in self.challenges) else []
|
||||
return sorted(set(DIFFICULTIES) | seen, key=difficulty_rank) + unknown
|
||||
|
||||
@property
|
||||
def stages(self) -> list:
|
||||
"""The five stages, plus None ("?") only if some status is unreadable."""
|
||||
return STATUS_ORDER + ([None] if any(not c.stage for c in self.challenges) else [])
|
||||
|
||||
@property
|
||||
def testable(self) -> list:
|
||||
"""Everything infra can already deploy and test: Testing Ready or beyond."""
|
||||
return sorted((c for c in self.challenges if is_testable(c)), key=self._ready_order)
|
||||
|
||||
@property
|
||||
def blocked(self) -> list:
|
||||
"""Testing Ready or beyond according to the README, but not the repo."""
|
||||
return sorted((c for c in self.challenges if c.claims_testable and c.blockers),
|
||||
key=self._ready_order)
|
||||
|
||||
def _ready_order(self, c: Challenge) -> tuple:
|
||||
return (self.categories.index(c.category or "?"),
|
||||
difficulty_rank(c.difficulty), c.title.lower())
|
||||
|
||||
def by_category(self) -> dict[str, list]:
|
||||
out = defaultdict(list)
|
||||
for c in self.challenges:
|
||||
out[c.category or "?"].append(c)
|
||||
for chs in out.values():
|
||||
chs.sort(key=lambda c: (-STATUS_ORDER.index(c.stage) if c.stage else 1,
|
||||
difficulty_rank(c.difficulty), c.title.lower()))
|
||||
return out
|
||||
|
||||
|
||||
def collect(gl: GitLab) -> Overview:
|
||||
chs, blank = [], []
|
||||
for repo in gl.repos():
|
||||
c = parse_challenge(repo)
|
||||
(chs if c.touched else blank).append(c)
|
||||
for c in chs:
|
||||
c.problems = readme_problems(c)
|
||||
|
||||
# only a Testing Ready+ claim is worth the extra downloads to check the repo
|
||||
def check(c: Challenge) -> None:
|
||||
gl.load_files(c.repo)
|
||||
c.blockers = repo_blockers(c)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=8) as pool:
|
||||
list(pool.map(check, [c for c in chs if c.claims_testable]))
|
||||
return Overview(chs, sorted(blank, key=lambda c: c.repo.path.lower()))
|
||||
|
||||
|
||||
def icons(chs: list) -> str:
|
||||
return "".join(c.icon for c in chs)
|
||||
@@ -0,0 +1,63 @@
|
||||
"""The summary printed when ctf_status.py runs in a terminal."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import Counter
|
||||
|
||||
from .challenge import BLOCKED_ICON, STATUS_ICON, STATUS_ORDER
|
||||
from .overview import Overview, icons
|
||||
|
||||
BOLD, DIM, RED, YEL, GRN, CYA, RST = (
|
||||
"\033[1m", "\033[2m", "\033[31m", "\033[33m", "\033[32m", "\033[36m", "\033[0m")
|
||||
|
||||
|
||||
def print_terminal(ov: Overview, c: bool) -> None:
|
||||
def col(x): return x if c else ""
|
||||
stages = ov.stages
|
||||
groups = ov.by_category()
|
||||
|
||||
total = Counter(ch.stage for ch in ov.challenges)
|
||||
print(f"\n{col(BOLD)}=== {len(ov.challenges)} challenges ==={col(RST)} "
|
||||
+ " ".join(f"{STATUS_ICON[s]} {s or '?'}:{total[s]}" for s in stages if total[s]))
|
||||
|
||||
ready = ov.testable
|
||||
print(f"\n{col(BOLD)}{col(GRN)}=== ready for infra testing ({len(ready)}) ==={col(RST)}")
|
||||
for ch in ready:
|
||||
print(f" {STATUS_ICON[ch.stage]} {ch.title[:34]:<35}{ch.category or '?':<10}"
|
||||
f"{ch.difficulty or '?':<13}{col(DIM)}{ch.repo.web_url}{col(RST)}")
|
||||
|
||||
if ov.blocked:
|
||||
print(f"\n{col(BOLD)}{col(RED)}=== {BLOCKED_ICON} claims {STATUS_ORDER[2]}+, "
|
||||
f"but the repo disagrees ({len(ov.blocked)}) ==={col(RST)}")
|
||||
for ch in ov.blocked:
|
||||
print(f" {STATUS_ICON[ch.stage]} {ch.title[:34]:<35}{col(DIM)}"
|
||||
+ "; ".join(ch.blockers) + col(RST))
|
||||
|
||||
print(f"\n{col(BOLD)}=== status by category ==={col(RST)}")
|
||||
print(f" {'':<10}" + "".join(f"{s or '?':>14}" for s in stages) + f"{'total':>7}")
|
||||
for cat in ov.categories:
|
||||
chs = groups.get(cat, [])
|
||||
cnt = Counter(ch.stage for ch in chs)
|
||||
tint = "" if chs else DIM
|
||||
print(f" {col(tint)}{cat:<10}" + "".join(f"{cnt[s] or '·':>14}" for s in stages)
|
||||
+ f"{len(chs):>7}{col(RST)}")
|
||||
|
||||
print(f"\n{col(BOLD)}=== category × difficulty ==={col(RST)} "
|
||||
+ " ".join(f"{STATUS_ICON[s]} {s}" for s in STATUS_ORDER) + f" {BLOCKED_ICON} blocked")
|
||||
diffs = ov.difficulties
|
||||
print(f" {'':<10}" + "".join(f"{d:<13}" for d in diffs))
|
||||
for cat in ov.categories:
|
||||
chs = groups.get(cat, [])
|
||||
cells = [icons([ch for ch in chs if (ch.difficulty or "?") == d]) for d in diffs]
|
||||
# emoji are two columns wide in a terminal
|
||||
print(f" {cat:<10}" + "".join(f"{x or '·'}{' ' * (13 - (2 * len(x) if x else 1))}"
|
||||
for x in cells))
|
||||
|
||||
print(f"\n{col(BOLD)}=== needs attention ==={col(RST)}")
|
||||
for ch in sorted(ov.challenges, key=lambda c: c.repo.path.lower()):
|
||||
if ch.problems:
|
||||
print(f" {col(YEL)}{ch.repo.path:<32}{col(RST)} {col(DIM)}"
|
||||
+ "; ".join(ch.problems) + col(RST))
|
||||
for ch in ov.blank:
|
||||
why = "no README" if ch.repo.readme is None else "README is still the blank template"
|
||||
print(f" {col(RED)}{ch.repo.path:<32}{col(RST)} {col(DIM)}{why}{col(RST)}")
|
||||
print()
|
||||
Reference in New Issue
Block a user