批量 block 用中文污染 issue 的人

github_batch_block.py

# -*- coding: utf-8 -*-
# Created by johnj at 2019/12/19

"""
批量 block 用中文污染 issue 的人

需要设置 USERNAME 和 TOKEN 两个环境变量
USERNAME 为 GitHub 用户名
TOKEN 获取参考这里：# https://help.github.com/en/github/authenticating-to-github/creating-a-personal-access-token-for-the-command-line
"""
import asyncio
import json
import os
import sys

import httpx

client = httpx.Client()
client.headers.update({
    'Accept': 'application/vnd.github.v3+json,application/vnd.github.giant-sentry-fist-preview+json'
})
personal_token = os.environ['TOKEN']
username = os.environ['USERNAME']

client.auth = (username, personal_token)
client.timeout = None


def has_chinese(sentense):
    for ch in sentense:
        if '\u4e00' <= ch <= '\u9fff':
            return True

    return False


async def block(username):
    resp = await client.put('https://api.github.com/user/blocks/' + username)
    return resp.status_code == httpx.codes.NO_CONTENT


async def has_blocked(username):
    resp = await client.get('https://api.github.com/user/blocks/' + username)
    if resp.status_code == httpx.codes.NO_CONTENT:
        return True
    if resp.status_code == httpx.codes.NOT_FOUND:
        return False
    raise ValueError(f'Invalid response: {resp.status_code}')


async def list_issues(repo):
    url = f'https://api.github.com/repos/{repo}/issues?state=all&per_page=100'
    while True:
        resp = await client.get(url)
        links = resp.links

        issues = resp.json()
        for issue in issues:
            yield issue
        if 'next' in links:
            next_url = links['next']['url']
            url = next_url
        else:
            break


cache = {}


def _dump_cache(repo):
    c = {}
    try:
        with open('./issue_caches.json', 'rt') as f:
            c = json.load(f)
    except Exception:
        pass
    c[repo] = cache
    with open('./issue_caches.json', 'wt') as f:
        json.dump(c, f, ensure_ascii=False, indent=2)


async def dump_cache(repo):
    while True:
        await asyncio.sleep(10)
        _dump_cache(repo)


def load_cache(repo):
    try:
        with open('./issue_caches.json', 'rt') as f:
            return json.load(f)[repo]
    except Exception:
        return {}


async def main(repo):
    global cache
    cache = load_cache(repo)
    asyncio.create_task(dump_cache(repo))

    async for issue in list_issues(repo):
        issue_id = str(issue['number'])
        if issue_id in cache:
            continue

        title = issue['title'].strip()
        body = issue['body']
        body = body.strip().replace('\r\n', ' ').replace('\n', ' ')
        creator = issue['user']['login']
        # if await has_blocked(creator):
        #     cache[issue_id] = False
        #     continue
        if not has_chinese(title):
            cache[issue_id] = False
            continue

        print('Title:', title)
        print('Body: ', body)
        y = input(f'Block "{creator}"? (press y to confirm, enter to skip)').strip()
        if y.lower().startswith('y'):
            cache[issue_id] = True
            success = await block(creator)
            if success:
                print(f'Blocked {creator}')
            else:
                print(f'Block {creator} failed')
        else:
            cache[issue_id] = False

    _dump_cache(repo)


if __name__ == '__main__':
    if len(sys.argv) > 1:
        repo = sys.argv[1]
        asyncio.run(main(repo))
    else:
        print('Usage:   python github_batch_block.py <repo_name>')
        print('Example: python github_batch_block.py LingDong-/wenyan-lang')