lix/maintainers/issue_import.py
Jade Lovelace 81be5eb7c6 issue importer: list issues that are *not* closed when finding existing issues
Turns out also, you cannot set the queue to 0 with any success. So we
really should just like, prevent notifications in forgejo itself.

Filed a bug for that:
https://git.lix.systems/lix-project/web-services/issues/38

Change-Id: Ib96749f3159659182904963cab7b2ef88fc64442
2024-03-18 18:14:31 -07:00

152 lines
4.5 KiB
Python

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,
})
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():
print('Have you turned off the forgejo mailer? Enter "We have" if so:')
answer = input('> ')
if answer != 'We have':
return
log.info('Importing issues!')
for issue in issues_to_import():
import_issue(DataClassUnpack.instantiate(Issue, issue))
if __name__ == '__main__':
go()