maintainers: run ruff

additionally, the issue_import has been removed, as our mirror has been
disconued

Change-Id: I8cb4c807db6c415b0fd340ead58be341a05e6a34
This commit is contained in:
Commentator2.0
2025-12-29 17:55:16 +00:00
committed by Rutile
parent fae54ba5a8
commit ac1b8e91d6
3 changed files with 89 additions and 212 deletions
+88 -63
View File
@@ -1,35 +1,39 @@
#!@python@
# ruff: noqa: T201 # Our generated MD file is expected on stdout
# This is the only file and is executable
from collections import defaultdict
import frontmatter
import pathlib
import textwrap
from typing import Any, Tuple
from typing import Any
import dataclasses
import yaml
import argparse
GH_ROOT = "https://github.com/"
GH_REPO_BASE = "https://github.com/NixOS/nix"
FORGEJO_REPO_BASE = "https://git.lix.systems/lix-project/lix"
FORGEJO_ROOT = "https://git.lix.systems/"
GERRIT_BASE = "https://gerrit.lix.systems/c/lix/+"
KNOWN_KEYS = ('synopsis', 'cls', 'issues', 'prs', 'significance', 'category', 'credits')
KNOWN_KEYS = ("synopsis", "cls", "issues", "prs", "significance", "category", "credits")
SIGNIFICANCECES = {
None: 0,
'significant': 10,
}
SIGNIFICANCECES = {None: 0, "significant": 10}
# This is just hardcoded for better validation. If you think there should be
# more of them, feel free to add more.
#
# Please update doc/manual/src/contributing/hacking.md if you do. Thanks~
CATEGORIES = [
'Breaking Changes',
'Features',
'Improvements',
'Fixes',
'Packaging',
'Development',
'Miscellany',
"Breaking Changes",
"Features",
"Improvements",
"Fixes",
"Packaging",
"Development",
"Miscellany",
]
@@ -45,11 +49,10 @@ class AuthorInfo:
def __str__(self) -> str:
if self.forgejo:
return f'[{self.show_name()}]({FORGEJO_ROOT}{self.forgejo})'
elif self.github:
return f'[{self.show_name()}]({GH_ROOT}{self.github})'
else:
return self.show_name()
return f"[{self.show_name()}]({FORGEJO_ROOT}{self.forgejo})"
if self.github:
return f"[{self.show_name()}]({GH_ROOT}{self.github})"
return self.show_name()
class AuthorInfoDB:
@@ -57,14 +60,13 @@ class AuthorInfoDB:
self.author_info = {name: AuthorInfo(name=name, **d) for (name, d) in author_info.items()}
self.throw_on_missing = throw_on_missing
def __getitem__(self, name) -> str:
def __getitem__(self, name: str) -> str:
if name in self.author_info:
return str(self.author_info[name])
else:
if self.throw_on_missing:
raise Exception(f'Missing author info for author {name}')
else:
return name
if self.throw_on_missing:
msg = f"Missing author info for author {name}"
raise Exception(msg)
return name
def format_link(ident: str, gh_part: str, fj_part: str) -> str:
@@ -80,37 +82,46 @@ def format_link(ident: str, gh_part: str, fj_part: str) -> str:
elif ident.startswith("lix#"):
num, link, base = int(ident[4:]), ident, f"{FORGEJO_REPO_BASE}/{fj_part}"
else:
raise Exception("unrecognized reference format", ident)
msg = f"unrecognized reference format: {ident}"
raise Exception(msg)
return f"[{link}]({base}/{num})"
def format_issue(issue: str) -> str:
return format_link(issue, "issues", "issues")
def format_pr(pr: str) -> str:
return format_link(pr, "pull", "pulls")
def format_cl(clid: int) -> str:
return f"[cl/{clid}]({GERRIT_BASE}/{clid})"
def plural_list(strs: list[str]) -> str:
if len(strs) <= 1:
return ''.join(strs)
else:
comma = ',' if len(strs) >= 3 else ''
return '{}{} and {}'.format(', '.join(strs[:-1]), comma, strs[-1])
return "".join(strs)
comma = "," if len(strs) >= 3 else ""
return "{}{} and {}".format(", ".join(strs[:-1]), comma, strs[-1])
def listify(l: list | int) -> list:
if not isinstance(l, list):
return [l]
else:
return l
def do_category(author_info: AuthorInfoDB, entries: list[Tuple[pathlib.Path, Any]]):
for p, entry in sorted(entries, key=lambda e: (-SIGNIFICANCECES[e[1].metadata.get('significance')], e[0])):
def listify(li: list | int) -> list:
if not isinstance(li, list):
return [li]
return li
def do_category(author_info: AuthorInfoDB, entries: list[tuple[pathlib.Path, Any]]):
for p, entry in sorted(
entries, key=lambda e: (-SIGNIFICANCECES[e[1].metadata.get("significance")], e[0])
):
try:
header = entry.metadata['synopsis']
header = entry.metadata["synopsis"]
links = []
links += [format_issue(str(s)) for s in listify(entry.metadata.get('issues', []))]
links += [format_pr(str(s)) for s in listify(entry.metadata.get('prs', []))]
links += [format_cl(int(cl)) for cl in listify(entry.metadata.get('cls', []))]
links += [format_issue(str(s)) for s in listify(entry.metadata.get("issues", []))]
links += [format_pr(str(s)) for s in listify(entry.metadata.get("prs", []))]
links += [format_cl(int(cl)) for cl in listify(entry.metadata.get("cls", []))]
if links != []:
header += " " + " ".join(links)
@@ -118,12 +129,17 @@ def do_category(author_info: AuthorInfoDB, entries: list[Tuple[pathlib.Path, Any
print(f"- {header}")
print()
else:
print("- ", end='')
print("- ", end="")
print(textwrap.indent(entry.content, ' '))
if credits := listify(entry.metadata.get('credits', [])):
print(textwrap.indent(entry.content, " "))
if credits_authors := listify(entry.metadata.get("credits", [])):
print()
print(textwrap.indent('Many thanks to {} for this.'.format(plural_list(list(author_info[c] for c in credits))), ' '))
print(
textwrap.indent(
f"Many thanks to {plural_list([author_info[c] for c in credits_authors])} for this.",
" ",
)
)
# Blank line after each entry.
print()
@@ -132,23 +148,27 @@ def do_category(author_info: AuthorInfoDB, entries: list[Tuple[pathlib.Path, Any
raise
def run_on_dir(author_info: AuthorInfoDB, d):
def run_on_dir(author_info: AuthorInfoDB, d: str):
d = pathlib.Path(d)
if not d.is_dir():
raise ValueError(f'provided path {d} is not a directory')
paths = pathlib.Path(d).glob('[!.]*.md')
msg = f"provided path {d} is not a directory"
raise ValueError(msg)
paths = d.glob("[!.]*.md")
entries = defaultdict(list)
for p in paths:
try:
e = frontmatter.load(p) # type: ignore
if 'synopsis' not in e.metadata:
raise Exception('missing synopsis')
unknownKeys = set(e.metadata.keys()) - set(KNOWN_KEYS)
if unknownKeys:
raise Exception('unknown keys', unknownKeys)
category = e.metadata.get('category', 'Miscellany')
e = frontmatter.load(p) # type: ignore
if "synopsis" not in e.metadata:
msg = "missing synopsis"
raise ValueError(msg)
unknown_keys = set(e.metadata.keys()) - set(KNOWN_KEYS)
if unknown_keys:
msg = f"unknown keys: {unknown_keys}"
raise ValueError(msg)
category = e.metadata.get("category", "Miscellany")
if category not in CATEGORIES:
raise Exception('unknown category', category)
msg = f"unknown category: {category}"
raise ValueError(msg)
entries[category].append((p, e))
except Exception as e:
e.add_note(f"in {p}")
@@ -156,7 +176,7 @@ def run_on_dir(author_info: AuthorInfoDB, d):
for category in CATEGORIES:
if entries[category]:
print('##', category)
print("##", category)
# Blank line after each heading.
print()
do_category(author_info, entries[category])
@@ -165,22 +185,27 @@ def run_on_dir(author_info: AuthorInfoDB, d):
# after each entry.
print()
def main():
import argparse
def main():
ap = argparse.ArgumentParser()
ap.add_argument('--change-authors', help='File name of the change authors metadata YAML file', type=argparse.FileType('r'))
ap.add_argument('dirs', help='Directories to run on', nargs='+')
ap.add_argument(
"--change-authors",
help="File name of the change authors metadata YAML file",
type=argparse.FileType("r"),
)
ap.add_argument("dirs", help="Directories to run on", nargs="+")
args = ap.parse_args()
author_info = AuthorInfoDB(yaml.safe_load(args.change_authors), throw_on_missing=True) \
if args.change_authors \
author_info = (
AuthorInfoDB(yaml.safe_load(args.change_authors), throw_on_missing=True)
if args.change_authors
else AuthorInfoDB({}, throw_on_missing=False)
)
for d in args.dirs:
run_on_dir(author_info, d)
if __name__ == '__main__':
if __name__ == "__main__":
main()
-148
View File
@@ -1,148 +0,0 @@
import requests
import textwrap
import dataclasses
import logging
import re
import os
API_BASE = 'https://git.lix.systems/api/v1'
API_KEY = os.environ['FORGEJO_API_KEY']
log = logging.getLogger(__name__)
log.setLevel(logging.INFO)
fmt = logging.Formatter('{asctime} {levelname} {name}: {message}',
datefmt='%b %d %H:%M:%S',
style='{')
if not any(isinstance(h, logging.StreamHandler) for h in log.handlers):
hand = logging.StreamHandler()
hand.setFormatter(fmt)
log.addHandler(hand)
# These are erring in the direction of re-triage, rather than necessarily
# mapping all metadata of the issue
LABEL_MAPPING = {
'lix-import': 153, # 'imported',
'contributor-experience': 148, # 'devx',
'bug': 150, # 'bug',
'UX': 149, # 'ux',
'error-messages': 149, # 'ux',
'lix-stability': 146, # 'stability',
'performance': 147, # 'performance',
'tests': 121, # 'tests',
}
def api(method, endpoint: str, resp_json=True, **kwargs):
log.info('http %s %s', method, endpoint)
if not endpoint.startswith('https'):
endpoint = API_BASE + endpoint
resp = requests.request(method,
endpoint,
headers={'Authorization': f'Bearer {API_KEY}'},
**kwargs)
resp.raise_for_status()
if resp_json:
return resp.json()
else:
return resp
def paginate(method: str, url: str):
while True:
resp = api(method, url, resp_json=False)
yield from resp.json()
next_one = resp.links.get('next')
if not next_one:
return
url = next_one.get('url')
if not url:
return
class DataClassUnpack:
"""Taken from: https://stackoverflow.com/a/72164665"""
classFieldCache = {}
@classmethod
def instantiate(cls, classToInstantiate, argDict):
if classToInstantiate not in cls.classFieldCache:
cls.classFieldCache[classToInstantiate] = {
f.name
for f in getattr(classToInstantiate, dataclasses._FIELDS).values() if f._field_type is not dataclasses._FIELD_CLASSVAR # type: ignore
}
fieldSet = cls.classFieldCache[classToInstantiate]
filteredArgDict = {k: v for k, v in argDict.items() if k in fieldSet}
return classToInstantiate(**filteredArgDict)
@dataclasses.dataclass
class Label:
name: str
description: str
@dataclasses.dataclass
class Issue:
number: int
url: str
html_url: str
title: str
body: str
labels: dataclasses.InitVar[list[dict]]
labels_clean: list[Label] = dataclasses.field(init=False)
def __post_init__(self, labels):
self.labels_clean = [DataClassUnpack.instantiate(Label, l) for l in labels]
def issues_to_import():
yield from paginate('GET', '/repos/nixos/nix/issues?state=open&labels=lix-import')
def issues_already_imported():
yield from paginate('GET', '/repos/lix-project/lix/issues?state=all&labels=imported')
UPSTREAM_ISSUE_RE = re.compile(r'^Upstream-Issue: https://git\.lix\.systems/NixOS/nix/issues/(\d+)$', re.MULTILINE)
def make_already_imported():
d = {}
for issue in issues_already_imported():
iss = DataClassUnpack.instantiate(Issue, issue)
print(iss)
match = UPSTREAM_ISSUE_RE.search(iss.body)
if match:
d[int(match.group(1))] = iss
return d
def new_issue(title, body, labels):
api('POST', '/repos/lix-project/lix/issues', resp_json=True, json={
'labels': labels,
'body': body,
'title': title,
'dont_notify': True,
})
already_imported = make_already_imported()
def import_issue(iss: Issue):
if iss.number in already_imported:
log.info('Skipping already imported %d', iss.number)
return
new_body = textwrap.dedent('''
Upstream-Issue: {iss}
{original_body}
''').format(iss=iss.html_url, original_body=iss.body)
new_labels = [LABEL_MAPPING[l.name] for l in iss.labels_clean if l.name in LABEL_MAPPING]
new_title = '[Nix#{num}] {title}'.format(num=iss.number, title=iss.title)
log.info('%s', f'create issue with: {new_labels} {new_title} {new_body}')
new_issue(new_title, new_body, new_labels)
def go():
log.info('Importing issues!')
for issue in issues_to_import():
import_issue(DataClassUnpack.instantiate(Issue, issue))
if __name__ == '__main__':
go()
+1 -1
View File
@@ -12,5 +12,5 @@ priority = 0
[formatter.ruff]
command = "ruff"
options = ["check"]
includes = ["tests/functional2/**/*.py", "meson/clang-tidy/*.py"]
includes = ["tests/functional2/**/*.py", "meson/clang-tidy/*.py", "maintainers/*.py"]
priority = 1