CCF推荐会议论文的获取和推送
from .feishu import *
from .wolai import *

bots/__init__.py
VulnTotal-Team/openccf
import time
import json
import requests
from pathlib import Path
from Crypto.Cipher import AES
from urllib.parse import urlparse, parse_qs
from playwright.sync_api import sync_playwright

from utils import *


class feishuBot:
    """飞书群机器人
    https://open.feishu.cn/document/ukTMukTMukTM/ucTM5YjL3ETO24yNxkjN
    https://open.feishu.cn/tool/cardbuilder
    """
    URL_API = 'https://open.feishu.cn/open-apis'

    def __init__(self, key) -> None:
        self.key = key

    def make_card(self, paper: dict):
        keywords = ','.join(paper['keywords'])

        if files := paper.get('files'):
            files = "\n".join(f'{name}：{url}' for name, url in files.items() if url)
        if slides := paper.get('slides'):
            slides = "\n".join(f'{name}：{url}' for name, url in slides.items() if url)

        # 优先使用中文
        abstract = paper['abstract_zh'] or paper['abstract']

        card = {
            'header': {
                'template': 'red',
                'title': {
                    'content': paper['title'],
                    'tag': 'plain_text'
                }
            },
            'elements': [
                {
                    'tag': 'div',
                    'fields': [
                        {
                            'is_short': True,
                            'text': {
                                'content': f'**命中依据：**{keywords}',
                                'tag': 'lark_md'
                            }
                        },
                        {
                            'is_short': True,
                            'text': {
                                'content': f'**发表年份：**{paper["year"]}',
                                'tag': 'lark_md'
                            }
                        }
                    ]
                },
                {
                    'tag': 'hr'
                },
                {
                    'tag': 'div',
                    'text': {
                        'content': f'**期刊/会议**\n{paper["conf_title"]}\n{paper["dblp_url"]}\n{paper["conf_url"]}',
                        'tag': 'lark_md'
                    }
                },
                {
                    'tag': 'div',
                    'text': {
                        'content': f'**标题**\n[{paper["title"]}]({paper["url"]})',
                        'tag': 'lark_md'
                    }
                },
                {
                    'tag': 'div',
                    'text': {
                        'content': f'**作者**\n{paper["authors"]}',
                        'tag': 'lark_md'
                    }
                },
                {
                    'tag': 'div',
                    'text': {
                        'content': f'**摘要**\n{abstract}',
                        'tag': 'lark_md'
                    }
                },
                {
                    'tag': 'div',
                    'text': {
                        'content': f'**文件**\n{files or ""}',
                        'tag': 'lark_md'
                    }
                },
                {
                    'tag': 'div',
                    'text': {
                        'content': f'**幻灯片**\n{slides or ""}',
                        'tag': 'lark_md'
                    }
                },
                {
                    'tag': 'div',
                    'text': {
                        'content': f'**视频**\n{paper.get("video") or ""}',
                        'tag': 'lark_md'
                    }
                },
            ]
        }
        bottom = [
            {
                'tag': 'hr'
            },
            {
                'tag': 'action',
                'actions': [
                    {
                        'tag': 'button',
                        'text': {
                            'tag': 'plain_text',
                            'content': '安全学术论文'
                        },
                        'url': '',
                        'type': 'primary'
                    },
                    {
                        'tag': 'button',
                        'text': {
                            'tag': 'plain_text',
                            'content': '安全研究资料'
                        },
                        'url': '',
                        'type': 'primary'
                    },
                    {
                        "tag": "button",
                        "text": {
                            "tag": "plain_text",
                            "content": "关于机器人"
                        },
                        "url": "",
                        "type": "default"
                    }
                ]
            }
        ]
        card['elements'] += bottom
        return card

    def send(self, paper: dict):
        card = self.make_card(paper)

        url = f'{self.URL_API}/bot/v2/hook/{self.key}'
        headers = {'Content-Type': 'application/json; charset=utf-8'}
        data = {'msg_type': 'interactive', 'card': card}

        r = requests.post(url=url, headers=headers, data=json.dumps(data))
        if r.status_code == 200:
            console.print(f'发送成功 {paper}', style='bold green')
            return True
        else:
            console.print(f'发送失败 {paper}', style='bold red')
            console.print_exception()
            return False


class feishuOper:
    """服务端程序
    https://open.feishu.cn/document/ukTMukTMukTM/uQjN3QjL0YzN04CN2cDN
    """
    TOKEN_TENANT = 'tenant'
    TOKEN_APP = 'app'
    TOKEN_USER = 'user'
    URL_API = 'https://open.feishu.cn/open-apis'

    def __init__(self, app_id, app_secret, mail=None, pwd=None):
        # 可在飞书开放平台->开发者后台中获取
        self.app_id = app_id
        self.app_secret = app_secret

        # tenant
        self.tenant_access_token = False
        self.tenant_expire = float()

        # app
        self.app_access_token = False
        self.app_expire = float()

        # user
        self.login_mail = mail
        self.login_pwd = pwd

        self.user_access_token = False
        self.user_refresh_token = False
        self.user_expire = float()
        self.user_refresh_expire = float()

        # 缓存文件
        cache_path = Path(__file__).parent.absolute().joinpath('cache')
        cache_path.mkdir(exist_ok=True)
        self.access_cache_file = cache_path.joinpath('feishu_access.bin')
        if self.access_cache_file.exists():
            self.load_access_cache()

    def load_access_cache(self):
        """加载缓存"""
        with open(self.access_cache_file, 'rb') as f:
            nonce, tag, ciphertext = [f.read(x) for x in (16, 16, -1)]
            cipher = AES.new(self.app_secret.encode(), AES.MODE_EAX, nonce)
            plaintext = cipher.decrypt_and_verify(ciphertext, tag)
            access_info = json.loads(plaintext)

        self.tenant_access_token = access_info['tenant_access_token']
        self.tenant_expire = access_info['tenant_expire']
        self.app_access_token = access_info['app_access_token']
        self.app_expire = access_info['app_expire']
        self.user_access_token = access_info['user_access_token']
        self.user_refresh_token = access_info['user_refresh_token']
        self.user_expire = access_info['user_expire']
        self.user_refresh_expire = access_info['user_refresh_expire']

    def update_access_cache(self):
        """更新缓存"""
        access_info = {
            'tenant_access_token': self.tenant_access_token,
            'tenant_expire': self.tenant_expire,
            'app_access_token': self.app_access_token,
            'app_expire': self.app_expire,
            'user_access_token': self.user_access_token,
            'user_refresh_token': self.user_refresh_token,
            'user_expire': self.user_expire,
            'user_refresh_expire': self.user_refresh_expire,
        }
        with open(self.access_cache_file, 'wb') as f:
            plaintext = json.dumps(access_info).encode()
            cipher = AES.new(self.app_secret.encode(), AES.MODE_EAX)
            ciphertext, tag = cipher.encrypt_and_digest(plaintext)
            [f.write(x) for x in (cipher.nonce, tag, ciphertext)]

    def check_access_valid(self, name: str) -> bool:
        """
        检查token是否有效
        name可选值：tenant、app、user
        """
        auth_permission = False

        # 缓存文件存在
        current_time = time.time()
        if name == self.TOKEN_TENANT and current_time < self.tenant_expire:
            auth_permission = True

        elif name == self.TOKEN_APP and current_time < self.app_expire:
            auth_permission = True

        elif name == self.TOKEN_USER and current_time < self.user_expire:
            auth_permission = True

        # 缓存文件不存在，或者缓存无效，则重新获取
        if not auth_permission:
            if name == self.TOKEN_TENANT:
                auth_permission = self.get_tenant_access()

            elif name == self.TOKEN_APP:
                auth_permission = self.get_app_access()

            elif name == self.TOKEN_USER:
                if current_time < self.user_refresh_expire:
                    auth_permission = self.get_user_access(self.user_refresh_token)
                else:
                    auth_permission = self.get_user_access()

        # 更新缓存文件
        if auth_permission:
            self.update_access_cache()
        return auth_permission

    def get_tenant_access(self) -> bool:
        """有效期2小时"""
        console.print('获取tenant_access_token ...', style='bold yellow')

        try:
            url = f'{self.URL_API}/auth/v3/tenant_access_token/internal'
            headers = {'Content-Type': 'application/json; charset=utf-8'}
            data = json.dumps({
                'app_id': self.app_id,
                'app_secret': self.app_secret
            })
            r = requests.post(url, headers=headers, data=data).json()

            self.tenant_access_token = r['tenant_access_token']
            self.tenant_expire = time.time() + r['expire'] - 60
            return True
        except Exception:
            console.print_exception()
            return False

    def get_app_access(self) -> bool:
        """有效期2小时"""
        console.print('获取app_access_token ...', style='bold yellow')

        try:
            url = f'{self.URL_API}/auth/v3/app_access_token/internal'
            headers = {'Content-Type': 'application/json; charset=utf-8'}
            data = json.dumps({
                'app_id': self.app_id,
                'app_secret': self.app_secret
            })
            r = requests.post(url, headers=headers, data=data).json()

            self.app_access_token = r['app_access_token']
            self.app_expire = time.time() + r['expire'] - 60
            return True
        except Exception:
            console.print_exception()
            return False

    def get_user_access(self, token='') -> bool:
        try:
            # 先获取app_access_token
            if self.get_app_access():
                data = self.refresh_user_access_token(token) if token else self.get_user_access_token()

                self.user_access_token = data['access_token']
                self.user_refresh_token = data['refresh_token']
                self.user_expire = time.time() + data['expire_in'] - 60
                self.user_refresh_expire = time.time() + data['refresh_expires_in'] - 60
                return True
            else:
                return False
        except Exception:
            console.print_exception()
            return False

    def get_login_code(self) -> str:
        """有效期5分钟，且只能使用一次
        不支持开启短信验证，需要手动输入验证码
        """
        console.print('获取预授权码 ...', style='bold yellow')

        url = f'{self.URL_API}/authen/v1/index?app_id={self.app_id}&redirect_uri='
        p = sync_playwright().start()
        browser = p.chromium.launch(headless=True)
        page = browser.new_page(locale='zh-CN')
        page.goto(url)

        with page.expect_navigation():
            page.locator('svg').nth(1).click()
            page.get_by_text('邮箱').click()
            page.locator('[data-test="login-mail-input"]').click()
            page.locator('[data-test="login-mail-input"]').fill(self.login_mail)
            page.get_by_label('我已阅读并同意 服务协议 和 隐私政策').check()
            page.locator('[data-test="login-phone-next-btn"]').click()
            page.locator('[data-test="login-pwd-input"]').click()
            page.locator('[data-test="login-pwd-input"]').fill(self.login_pwd)
            page.locator('[data-test="login-pwd-next-btn"]').click()

        code_url = page.url
        browser.close()
        return parse_qs(urlparse(code_url).query)['code'][0]

    def get_user_access_token(self) -> dict:
        """有效期6900秒"""
        console.print('获取user_access_token ...', style='bold yellow')

        url = f'{self.URL_API}/authen/v1/access_token'
        data = json.dumps({
            'code': self.get_login_code(),
            'grant_type': 'authorization_code'
        })
        headers = {
            'Content-Type': 'application/json; charset=utf-8',
            'Authorization': f'Bearer {self.app_access_token}',
        }

        return requests.post(url, headers=headers, data=data).json()['data']

    def refresh_user_access_token(self, token: str) -> dict:
        """有效期30天"""
        console.print('刷新user_access_token ...', style='bold yellow')

        url = f'{self.URL_API}/authen/v1/refresh_access_token'
        data = json.dumps({
            'refresh_token': token,
            'grant_type': 'refresh_token'
        })
        headers = {
            'Content-Type': 'application/json; charset=utf-8',
            'Authorization': f'Bearer {self.app_access_token}',
        }

        return requests.post(url, headers=headers, data=data).json()['data']

    def bitable_batch_create(self, table: str, data: dict) -> bool:
        """tenant_access_token 或 user_access_token"""
        app_token, table_id = table.split(':')
        url = f'{self.URL_API}/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_create'
        headers = {
            'Content-Type': 'application/json; charset=utf-8',
            'Authorization': f'Bearer {self.tenant_access_token}',
        }
        try:
            result = requests.post(url, headers=headers, data=json.dumps(data)).json()
            if result['msg'] == 'success':
                console.print('写入表格成功', style='bold green')
                return True
            else:
                console.print(f'写入表格失败:\n{data}\n{result}', style='bold red')
        except Exception:
            console.print('无法访问表格', style='bold red')
            console.print_exception()

        return False

bots/feishu.py
VulnTotal-Team/openccf
import requests

from utils import *


class wolaiOper:
    """我来应用
    https://www.wolai.com/wolai/7FB9PLeqZ1ni9FfD11WuUi
    """
    URL_API = 'https://openapi.wolai.com/v1'

    def __init__(self, app_id, app_secret) -> None:
        self.app_id = app_id
        self.app_secret = app_secret
        self.token = self.get_token()

    def get_token(self) -> bool:
        """创建或重置token"""
        console.print('获取token ...', style='bold yellow')

        url = f'{self.URL_API}/token'
        data = {
            'appId': self.app_id,
            'appSecret': self.app_secret
        }
        try:
            return requests.post(url, json=data).json()['data']['app_token']
        except Exception:
            console.print_exception()
            return False

    def database_get(self, id: str) -> bool:
        """获取数据表格数据"""
        url = f'{self.URL_API}/databases/{id}'
        headers = {'Authorization': self.token}
        try:
            return requests.get(url, headers=headers).json()['data']
        except Exception:
            console.print_exception()
            return False

    def database_post(self, id: str, data: dict) -> bool:
        """创建数据表格数据"""
        url = f'{self.URL_API}/databases/{id}/rows'
        headers = {'Authorization': self.token}
        try:
            r = requests.post(url, headers=headers, json=data)
            if r.status_code == 200:
                console.print('写入表格成功', style='bold green')
                return True
            else:
                console.print(f'写入表格失败:\n{data}\n{r.json()}', style='bold red')
        except Exception:
            console.print('无法访问表格', style='bold red')
            console.print_exception()

        return False

bots/wolai.py
VulnTotal-Team/openccf
from .ccf import *
from .dblp import *
from .scholar import *
from .libpaper import *

crawler/__init__.py
VulnTotal-Team/openccf
import contextlib
import json
import requests
from bs4 import BeautifulSoup

from utils import *

"""
ARCH_DCP_SS 计算机体系结构/并行与分布计算/存储系统
CN          计算机网络
NIS         网络与信息安全
TCSE_SS_PDL 软件工程/系统软件/程序设计语言
DM_CS       数据库/数据挖掘/内容检索
TCS         计算机科学理论
CGAndMT     计算机图形学与多媒体
AI          人工智能
HCIAndPC    人机交互与普适计算
Cross_Compre_Emerging   交叉/综合/新兴
"""

FIELDS = ['ARCH_DCP_SS', 'CN', 'NIS', 'TCSE_SS_PDL', 'DM_CS', 'TCS', 'CGAndMT', 'AI', 'HCIAndPC', 'Cross_Compre_Emerging']

def get_ccf_data(update: bool=False):
    ccf_fields = FIELDS
    ccf_file = data_path.joinpath('ccf.json')

    if update or not ccf_file.exists():
        ccf_data = get_ccf(ccf_fields)
        with open(ccf_file, 'w') as f:
            json.dump(ccf_data, f, indent=4, ensure_ascii=False)
    else:
        ccf_data = json.loads(ccf_file.read_text())

    num1 = sum(len(j) for i in ccf_data.values() for j in i['journals'].values())
    num2 = sum(len(j) for i in ccf_data.values() for j in i['conf'].values())
    print(f'ccf: {ccf_fields}\njournals: {num1}\tconf: {num2}\n')

    return ccf_data


def get_ccf(fields: list=FIELDS):

    def parse_ul(ul):
        temp = []
        for li in ul.find_all('li')[1:]:
            fields = li.find_all('div')
            index = fields[0].text.strip()
            name = fields[1].text.strip()
            full_name = fields[2].text.strip()
            publisher = fields[3].text.strip()
            address = fields[4].text.strip()

            if address in temp_addr:
                console.print(f'Duplicate address: {full_name}\n{address}\n', style='bold red')

            line = {'name': name, 'full_name': full_name, 'publisher': publisher, 'address': address}
            temp.append(line)
            temp_addr.append(address)
        return temp

    results = {}
    for field in fields:
        temp_addr = []  # 同一个领域去重
        base_url = 'https://www.ccf.org.cn/Academic_Evaluation'
        r = requests.get(f'{base_url}/{field}', timeout=30)
        soup = BeautifulSoup(r.content, 'html.parser')
        ul_tags = soup.select('ul.g-ul.x-list3')

        item = {
            'journals': {
                'A': parse_ul(ul_tags[0]),
                'B': parse_ul(ul_tags[1]),
                'C': parse_ul(ul_tags[2]),
            },
            'conf': {
                'A': parse_ul(ul_tags[3]),
                'B': parse_ul(ul_tags[4]),
                'C': parse_ul(ul_tags[5]),
            }
        }
        results[field] = item

    # 特殊情况处理
    with contextlib.suppress(Exception):
        for a, b in results.items():
            for c, d in b.items():
                for e, f in d.items():
                    for idx, item in enumerate(f):
                        if item['name'] == 'AsiaCCS':
                            # 修改网址
                            item['address'] = 'https://dblp.org/db/conf/asiaccs/'
                            results[a][c][e][idx] = item

                        if item['name'] == 'PETS':
                            # 2015年开始改为PoPETs
                            temp = {
                                'name': 'PoPETs',
                                'full_name': 'Proceedings on Privacy Enhancing Technologies',
                                'publisher': item['publisher'],
                                'address': 'https://dblp.uni-trier.de/db/journals/popets/'
                            }
                            results[a]['journals'][e].append(temp)

                        if item['name'] == 'FSE':
                            # 2017年开始改为ToSC
                            temp = {
                                'name': 'ToSC',
                                'full_name': 'IACR Transactions on Symmetric Cryptology',
                                'publisher': item['publisher'],
                                'address': 'https://dblp.uni-trier.de/db/journals/tosc/'
                            }
                            results[a]['journals'][e].append(temp)

                        if item['name'] == 'CHES':
                            # 2018年开始改为TCHES
                            temp = {
                                'name': 'TCHES',
                                'full_name': 'IACR Transactions on Cryptographic Hardware and Embedded Systems',
                                'publisher': item['publisher'],
                                'address': 'https://dblp.uni-trier.de/db/journals/tches/'
                            }
                            results[a]['journals'][e].append(temp)

                        if item['name'] == 'DFRWS':
                            # 2006年开始改为DI
                            temp = {
                                'name': 'DI',
                                'full_name': 'Forensic Science International: Digital Investigation',
                                'publisher': item['publisher'],
                                'address': 'https://dblp.uni-trier.de/db/journals/di/'
                            }
                            results[a]['journals'][e].append(temp)

    return results

crawler/ccf.py
VulnTotal-Team/openccf
import json
import time
import asyncio
import requests
from bs4 import BeautifulSoup

from utils import *
from crawler.scholar import *


def get_dblp_data(url: str, start_year: str, end_year: str):
    dblp_key = '/'.join(url.split('/')[-3:-1])
    dblp_file = dblp_path.joinpath(f'{dblp_key.replace("/", "_")}.json')

    old_data = json.loads(dblp_file.read_text()) if dblp_file.exists() else {}
    all_data, new_data, update_data = get_dblp(dblp_key, old_data, start_year, end_year)

    # 只更新start_year到end_year的数据
    for year, year_data in all_data.items():
        old_data[year] = year_data

    with open(dblp_file, 'w') as f:
        json.dump(old_data, f, indent=4, ensure_ascii=False)

    return url, all_data, new_data, update_data


def get_dblp(key: str, old_data: dict, start_year: str, end_year: str):
    # 期刊/会议主页
    url = f'https://dblp.uni-trier.de/db/{key}/index.html'
    console.print(url, style='bold yellow')
    while True:
        try:
            r = requests.get(url, timeout=30)
            break
        except Exception:
            time.sleep(1)
            print('Retrying...')

    soup = BeautifulSoup(r.content, 'html.parser')
    href_list = []
    tasks = []
    all_papers_dict = {}
    new_papers_dict = {}
    update_papers_dict = {}
    loop = asyncio.new_event_loop()

    old_dict = {}
    for year_data in old_data.values():
        for item in year_data:
            for paper in item['papers']:
                old_dict[paper['title']] = paper

    # 遍历所有年份的所有文章
    if '/journals/' in url:
        href_list = parse_journals_index(soup, start_year, end_year)

        print(href_list)
        for year, href in href_list:
            task = asyncio.ensure_future(parse_journals(year, href, old_dict), loop=loop)
            tasks.append(task)

        console.print(f'tasks: {len(tasks)}\n', style='bold yellow')

    elif '/conf/' in url:
        href_list = parse_conf_index(soup, start_year, end_year)

        print(href_list)
        for year, href in href_list:
            task = asyncio.ensure_future(parse_conf(year, href, old_dict), loop=loop)
            tasks.append(task)

        console.print(f'tasks: {len(tasks)}\n', style='bold yellow')

    if tasks:
        try:
            done, pending = loop.run_until_complete(asyncio.wait(tasks))
            print(f'done: {len(done)}\t pending: {len(pending)}\n')
            for task in done:
                ret = task.result()
                year, all_papers, new_papers, update_papers = ret

                all_papers_dict.setdefault(year, []).append(all_papers)
                new_papers_dict.setdefault(year, []).append(new_papers)
                update_papers_dict.setdefault(year, []).append(update_papers)
        except ValueError:
            console.print(f'ValueError: {ret}', style='bold red')
        except Exception:
            console.print_exception()
        finally:
            loop.close()

    # 按会议排序
    for year in all_papers_dict:
        all_papers_dict[year].sort(key=lambda x: x['dblp_url'])

    return all_papers_dict, new_papers_dict, update_papers_dict


async def parse_journals(year: str, url: str, old_dict: dict):
    """获取一年的所有文章"""
    # 获取网页数据
    data = await fetch(url)
    if not data:
        return url

    try:
        soup = BeautifulSoup(data, 'html.parser')
        # 期刊
        journals_title = soup.select_one('h1').text

        # 文章
        all_papers = []
        new_papers = []
        update_papers = []
        article = soup.select('li.entry.article')
        for a in article:
            new_flag = False
            update_flag = False

            paper = {'url': a.select_one('li.drop-down').select_one('a').attrs.get('href')}
            paper['title'] = a.select_one('span.title').text.strip('.')
            paper['authors'] = ', '.join([a.text for a in a.select('span[itemprop="name"]')[:-1]])

            # 用旧数据补充
            if paper['title'] in old_dict:
                paper = old_dict[paper['title']]
            else:
                new_flag = True

            # 获取新数据补充
            if not paper.get('abstract'):
                paper = await get_scholar(paper)
                if paper['abstract'] and not new_flag:
                    update_flag = True

            all_papers.append(paper)
            if new_flag:
                new_papers.append(paper)
            if update_flag:
                update_papers.append(paper)

            await asyncio.sleep(1)

        result_all = {'dblp_url': url, 'journals_title': journals_title, 'papers': sorted(all_papers, key=lambda x: x['url'])}
        result_new = {'dblp_url': url, 'journals_title': journals_title, 'papers': new_papers}
        result_update = {'dblp_url': url, 'journals_title': journals_title, 'papers': update_papers}

        print(f'{journals_title}\n{url}\nAll Papers: {len(all_papers)}\tNew Papers: {len(new_papers)}\tUpdate Papers: {len(update_papers)}\n')
        return year, result_all, result_new, result_update
    except Exception:
        console.print(url, style='bold red')
        console.print_exception()
        return url


async def parse_conf(year: str, url: str, old_dict: dict):
    """获取一年的所有文章"""
    # 获取网页数据
    data = await fetch(url)
    if not data:
        return url

    try:
        soup = BeautifulSoup(data, 'html.parser')
        # 会议
        conf_title = soup.select_one('h1').text
        if '404' in conf_title:
            print(f'{conf_title}\n{url}\npapers: 0\n')
            return url

        editor = soup.select_one('li.entry.editor')
        conf_url = editor.select_one('li.drop-down').select_one('a').attrs.get('href')
        # conf_title = editor.select_one('span.title').text

        # 文章
        all_papers = []
        new_papers = []
        update_papers = []
        inproceedings = soup.select('li.entry.inproceedings')
        for i in inproceedings:
            new_flag = False
            update_flag = False

            paper = {'url': i.select_one('li.drop-down').select_one('a').attrs.get('href')}
            paper['title'] = i.select_one('span.title').text.strip('.')
            paper['authors'] = ', '.join([a.text for a in i.select('span[itemprop="name"]')[:-1]])

            # 用旧数据补充
            if paper['title'] in old_dict:
                paper = old_dict[paper['title']]
            else:
                new_flag = True

            # 获取新数据补充
            if not paper.get('abstract'):
                paper = await get_scholar(paper)
                if paper['abstract'] and not new_flag:
                    update_flag = True

            all_papers.append(paper)
            if new_flag:
                new_papers.append(paper)
            if update_flag:
                update_papers.append(paper)

            await asyncio.sleep(1)

        result_all = {'dblp_url': url, 'conf_title': conf_title, 'conf_url': conf_url, 'papers': sorted(all_papers, key=lambda x: x['url'])}
        result_new = {'dblp_url': url, 'conf_title': conf_title, 'conf_url': conf_url, 'papers': new_papers}
        result_update = {'dblp_url': url, 'conf_title': conf_title, 'conf_url': conf_url, 'papers': update_papers}

        print(f'{conf_title}\n{url}\nAll Papers: {len(all_papers)}\tNew Papers: {len(new_papers)}\tUpdate Papers: {len(update_papers)}\n')
        return year, result_all, result_new, result_update
    except Exception:
        console.print(url, style='bold red')
        console.print_exception()
        return url


def parse_journals_index(soup, start_year: str, end_year: str):
    """解析期刊首页"""
    results = []
    section = soup.select_one('div#info-section')
    ul = section.find_next_siblings('ul')
    for u in ul:
        for li in u.select('li'):
            aa = li.select('a')
            for a in aa:
                a_text = a.text.strip()
                if len(li.contents) == 1:
                    if ',' in a_text or ':' in a_text:
                        v, year = [i.strip() for i in a_text.replace(',', ':').split(':')]
                    elif ' ' not in a_text:
                        year = a_text
                    else:
                        v, year = a_text.split(' ')
                    # title = f'{v}: {year}'
                else:
                    year, v = [i.strip() for i in li.contents[0].split(':')]
                    # title = f'{v} {a_text}: {year}'

                href = a.attrs.get('href')
                year = year.split('/')[-1] if '/' in year else year
                if year and end_year >= year >= start_year:    # 筛选年份
                    results.append((year, href))
    return results


def parse_conf_index(soup, start_year: str, end_year: str):
    """解析会议首页"""
    results = []

    # conf/ccs
    if workshops := soup.select_one('p:-soup-contains("Workshops:")'):
        ul = workshops.find_next('ul')
        for li in ul.select('li'):
            try:
                title = li.contents[0].strip()
                aa = li.select('a')
            except Exception:
                title = li.find_next('a').text.strip() + li.contents[1]
                aa = li.select('a')[1:]
            title = title.strip().strip(':').replace('\n', ' ')

            for a in aa:
                year = a.text.strip()
                href = a.attrs.get('href')
                if year and end_year >= year >= start_year:
                    results.append((year, href))

    for h2 in soup.select('header.h2'):
        year = h2.select_one('h2').attrs.get('id')
        # 筛选年份，示例：'2016' < '2016a' < '2017'
        if year and end_year >= year >= start_year:
            if n := h2.next_sibling:
                if n.name == 'ul':
                    toc = n.select('li.entry.editor.toc')
                    for t in toc:
                        href = t.select_one('li.drop-down').select_one('a').attrs.get('href')
                        results.append((year, href))

                    workshop = n.next_sibling
                    if 'workshop' in workshop.text.lower():
                        for a in workshop.select('a'):
                            title = a.text.strip()
                            href = a.attrs.get('href')
                            results.append((year, href))
                # elif n.name == 'p':
                #     a = n.select_one('a')
                #     title = a.text.strip()
                #     href = a.attrs.get('href')
                #     results.append((year, href))

    return results

crawler/dblp.py
VulnTotal-Team/openccf
from .sp import *
from .ccs import *
from .ndss import *
from .usenix import *

crawler/libpaper/__init__.py
VulnTotal-Team/openccf
import json
import asyncio

from utils import *
from .common import *


def get_ccs_data(data: dict, update: bool=False):
    data_file = paper_path.joinpath('conf_ccs.json')

    if update or not data_file.exists():
        ret_data = get_ccs(data)
        with open(data_file, 'w') as f:
            json.dump(ret_data, f, indent=4, ensure_ascii=False)
    else:
        ret_data = json.loads(data_file.read_text())

    num = sum(len(j['papers']) for i in ret_data.values() for j in i)
    console.print(f'CCS: {num}\n', style='bold green')

    return ret_data


def get_ccs(data: dict):
    results = {}

    # 遍历年份
    for year, v in data.items():
        results[year] = []

        # 遍历会议
        for i in v:
            tasks = []
            loop = asyncio.new_event_loop()

            # 遍历文章
            for idx, paper in enumerate(i['papers']):
                url = paper['url']
                if 'doi.org' in url:
                    task = asyncio.ensure_future(parse_doi_org(paper), loop=loop)
                    tasks.append(task)
                else:
                    console.print(f'Unknow: {url}', style='bold red')
                    i['papers'][idx]['url'] = url

            if tasks:
                i['papers'] = []    # 清空原数据
                try:
                    done, pending = loop.run_until_complete(asyncio.wait(tasks))
                    print(f'done: {len(done)}\t pending: {len(pending)}\n')
                    for task in done:
                        i['papers'].append(task.result())
                    results[year].append(i)
                except Exception:
                    console.print_exception()
                finally:
                    loop.close()
            else:
                results[year].append(i)

    return results

crawler/libpaper/ccs.py
VulnTotal-Team/openccf
from bs4 import BeautifulSoup

from utils import *


async def parse_doi_org(paper: dict):
    data = await fetch(paper['url'])
    if not data:
        paper['status'] = 'error'
        return paper

    try:
        soup = BeautifulSoup(data, 'html.parser')

        # author = soup.select('a.author-name')
        # authors = ''
        # for a in author:
        #     name = a.find_next('div', class_='author-data').text
        #     inst = a.find_next('span', class_='loa_author_inst').text
        #     authors += f'{name}, '
        # paper['authors'] = authors[:-2]

        abstract = soup.select_one('div.abstractSection.abstractInFull').text
        paper['abstract'] = abstract.strip() if abstract else ''

        files = {}
        file = soup.select('li.pdf-file')
        for f in file:
            href = f.find_next('a').attrs.get('href')
            files['Paper'] = f'https://dl.acm.org{href}'
        paper['files'].update(files)

    except Exception:
        paper['status'] = 'error'
        console.print(paper['url'], style='bold red')
        console.print_exception()
    finally:
        return paper

crawler/libpaper/common.py
VulnTotal-Team/openccf
import re
import json
import asyncio
from bs4 import BeautifulSoup

from utils import *
from .common import *


def get_ndss_data(data: dict, update: bool=False):
    data_file = paper_path.joinpath('conf_ndss.json')

    if update or not data_file.exists():
        ret_data = data_ndss(data)
        with open(data_file, 'w') as f:
            json.dump(ret_data, f, indent=4, ensure_ascii=False)
    else:
        ret_data = json.loads(data_file.read_text())

    num = sum(len(j['papers']) for i in ret_data.values() for j in i)
    console.print(f'NDSS: {num}\n', style='bold green')

    return ret_data


def data_ndss(data: dict):
    results = {}

    # 遍历年份
    for year, v in data.items():
        results[year] = []

        # 遍历会议
        for i in v:
            tasks = []
            loop = asyncio.new_event_loop()

            # 遍历文章
            for idx, paper in enumerate(i['papers']):
                url = paper['url']
                if 'ndss-symposium.org' in url:
                    task = asyncio.ensure_future(parse_ndss_symposium_org(paper), loop=loop)
                    tasks.append(task)
                elif 'wp.internetsociety.org' in url:
                    # 转换此类链接
                    base_url = 'https://www.ndss-symposium.org/wp-content/uploads'
                    url = f'{base_url}/{"/".join(url.split("/")[-3:])}'
                    i['papers'][idx]['url'] = url
                else:
                    console.print(f'Unknow: {url}', style='bold red')
                    i['papers'][idx]['url'] = url

            if tasks:
                i['papers'] = []    # 清空原数据
                try:
                    done, pending = loop.run_until_complete(asyncio.wait(tasks))
                    print(f'done: {len(done)}\t pending: {len(pending)}\n')
                    for task in done:
                        i['papers'].append(task.result())
                    results[year].append(i)
                except Exception:
                    console.print_exception()
                finally:
                    loop.close()
            else:
                results[year].append(i)

    return results


async def parse_ndss_symposium_org(paper: dict):
    data = await fetch(paper['url'])
    if not data:
        paper['status'] = 'error'
        return paper

    try:
        soup = BeautifulSoup(data, 'html.parser')
        data = soup.select_one('div.paper-data')

        # authors = data.contents[1].text.strip()
        # paper['authors'] = authors.strip() if authors else ''

        abstract = data.contents[3].text.strip()
        paper['abstract'] = abstract.strip() if abstract else ''

        files = {}
        slides = {}
        video = ''
        buttons = soup.select_one('div.paper-buttons')
        for a in buttons.select('a'):
            if 'paper' in a.text.lower():
                files['Paper'] = a.attrs.get('href')
            if 'slides' in a.text.lower():
                slides['Slides'] = a.attrs.get('href')
            if 'video' in a.text.lower():
                href = a.attrs.get('href')
                try:
                    video_id = re.search(r"v=([A-Za-z0-9_-]{11})", href)[1]
                    video = f'https://www.youtube.com/watch?v={video_id}'
                except Exception:
                    video = href
        paper['files'].update(files)
        paper['slides'] = slides
        paper['video'] = video

    except Exception:
        paper['status'] = 'error'
        console.print(paper['url'], style='bold red')
        console.print_exception()
    finally:
        return paper

crawler/libpaper/ndss.py
VulnTotal-Team/openccf
import json
import asyncio

from utils import *
from .common import *


def get_sp_data(data: dict, update: bool=False):
    data_file = paper_path.joinpath('conf_sp.json')

    if update or not data_file.exists():
        ret_data = get_sp(data)
        with open(data_file, 'w') as f:
            json.dump(ret_data, f, indent=4, ensure_ascii=False)
    else:
        ret_data = json.loads(data_file.read_text())

    num = sum(len(j['papers']) for i in ret_data.values() for j in i)
    console.print(f'S&P: {num}\n', style='bold green')

    return ret_data


def get_sp(data: dict):
    results = {}

    # 遍历年份
    for year, v in data.items():
        results[year] = []

        # 遍历会议
        for i in v:
            tasks = []
            loop = asyncio.new_event_loop()

            # 遍历文章
            for idx, paper in enumerate(i['papers']):
                url = paper['url']
                if 'doi.org' in url:
                    task = asyncio.ensure_future(parse_doi_org(paper), loop=loop)
                    tasks.append(task)
                else:
                    console.print(f'Unknow: {url}', style='bold red')
                    i['papers'][idx]['url'] = url

            if tasks:
                i['papers'] = []    # 清空原数据
                try:
                    done, pending = loop.run_until_complete(asyncio.wait(tasks))
                    print(f'done: {len(done)}\t pending: {len(pending)}\n')
                    for task in done:
                        i['papers'].append(task.result())
                    results[year].append(i)
                except Exception:
                    console.print_exception()
                finally:
                    loop.close()
            else:
                results[year].append(i)

    return results

crawler/libpaper/sp.py
VulnTotal-Team/openccf
import re
import json
import asyncio
import bibtexparser
from bs4 import BeautifulSoup

from utils import *
from .common import *


def get_usenix_data(data: dict, update: bool=False):
    data_file = paper_path.joinpath('conf_uss.json')

    if update or not data_file.exists():
        ret_data = get_usenix(data)
        with open(data_file, 'w') as f:
            json.dump(ret_data, f, indent=4, ensure_ascii=False)
    else:
        ret_data = json.loads(data_file.read_text())

    num = sum(len(j['papers']) for i in ret_data.values() for j in i)
    console.print(f'USENIX: {num}\n', style='bold green')

    return ret_data


def get_usenix(data: dict):
    results = {}

    # 遍历年份
    for year, v in data.items():
        results[year] = []

        # 遍历会议
        for i in v:
            tasks = []
            loop = asyncio.new_event_loop()

            # 遍历文章
            for idx, paper in enumerate(i['papers']):
                url = paper['url']
                if 'usenix.org' in url:
                    task = asyncio.ensure_future(parse_usenix_org(paper), loop=loop)
                    tasks.append(task)
                elif 'doi.org' in url:
                    task = asyncio.ensure_future(parse_doi_org(paper), loop=loop)
                    tasks.append(task)
                else:
                    console.print(f'Unknow: {url}', style='bold red')
                    i['papers'][idx]['url'] = url

            if tasks:
                i['papers'] = []    # 清空原数据
                try:
                    done, pending = loop.run_until_complete(asyncio.wait(tasks))
                    print(f'done: {len(done)}\t pending: {len(pending)}\n')
                    for task in done:
                        i['papers'].append(task.result())
                    results[year].append(i)
                except Exception:
                    console.print_exception()
                finally:
                    loop.close()
            else:
                results[year].append(i)

    return results


async def parse_usenix_org(paper: dict):
    data = await fetch(paper['url'])
    if not data:
        paper['status'] = 'error'
        return paper

    try:
        soup = BeautifulSoup(data, 'html.parser')
        paper['bibtex'] = ''
        if bibtex := soup.select_one('div.bibtex-text-entry.bibtex-accordion-text-entry'):
            paper['bibtex'] = bibtex.text.strip()
            bibtex = bibtexparser.loads(bibtex.text)

        # authors = ''  # 目前是使用dblp的作者
        abstract = ''
        label = soup.select('div.field-label')
        for l in label:
            # if 'Authors' in l.text:
            #     authors = l.find_next('div', class_='field-item odd').text
            if 'Abstract' in l.text:
                abstract = l.find_next('div', class_='field-item odd').text

        # if not authors:
        #     odd = soup.select('div.field-item.odd')
        #     for o in odd:
        #         if o.text and o.text == abstract:
        #             if tmp := o.find_previous('div', class_='field-item odd'):
        #                 authors = tmp.text
        #             break
        #     if not authors:
        #         authors = bibtex.entries[0].get('author') if bibtex else ''
        # paper['authors'] = authors.strip() if authors else ''
        paper['abstract'] = abstract.strip() if abstract else ''

        files = {}
        file = soup.select('span.file')
        for f in file:
            if a := f.select_one('a'):
                files[a.text.strip()] = a.attrs.get('href')

        slides = {}
        slide = soup.select('div.usenix-schedule-slides')
        for s in slide:
            if a := s.select_one('a'):
                slides[a.text.strip()] = a.attrs.get('href')

        if not files or not slides:
            label = soup.select('div.field-label')
            for l in label:
                if 'paper' in l.text.lower():
                    files['Paper'] = l.find_next('div', class_='field-item odd').text
                if 'slides' in l.text.lower():
                    slides['Slides'] = l.find_next('div', class_='field-item odd').text
        paper['files'].update(files)
        paper['slides'] = slides

        video = ''
        if player := soup.select_one('div.player'):
            src = player.find_next().attrs.get('src')
            video_id = re.search(r"embed/(.+)\?", src)[1]
            video = f'https://www.youtube.com/watch?v={video_id}'
        elif player := soup.select_one('video'):
            video = player.find_next('source').attrs.get('src')
        paper['video'] = video

    except Exception:
        paper['status'] = 'error'
        console.print(paper['url'], style='bold red')
        console.print_exception()
    finally:
        return paper

crawler/libpaper/usenix.py
VulnTotal-Team/openccf
import os
import json
import contextlib
from semanticscholar import SemanticScholar
from scholarly import scholarly, ProxyGenerator, MaxTriesExceededException

from utils import *

# pg = ProxyGenerator()
# pg.FreeProxies()
# pg.ScraperAPI(ScraperAPI)
# scholarly.use_proxy(pg, pg)


async def get_scholar(paper: dict):
    """获取补充数据"""
    paper['abstract'] = ''
    paper.setdefault('files', {'openAccessPdf': ''})

    if ret := await get_semantic_scholar(paper['title']):
        paper['abstract'] = ret['abstract']
        paper['files']['openAccessPdf'] = ret['openAccessPdf']
    elif ret := get_semantic_scholar2(paper['title']):
        paper['abstract'] = ret['abstract']
        paper['files']['openAccessPdf'] = ret['openAccessPdf']
    elif ret := get_google_scholar(paper['title']):
        paper['abstract'] = ret['abstract']

    # 旧数据已经有title_zh，不再翻译
    if not paper.get('title_zh'):
        paper['title_zh'] = get_translate(paper['title'])
    paper['abstract_zh'] = get_translate(paper['abstract'])

    if not paper['abstract']:
        console.print(f'Abstract null: {paper["title"]}', style='bold red')

    return paper


async def get_semantic_scholar(title: str):
    """https://api.semanticscholar.org/api-docs/graph
    使用API获取Semantic Scholar数据
    无KEY：5000次/5分钟；有KEY：100次/秒
    """
    semantic_url = 'https://api.semanticscholar.org/graph/v1/paper/search'

    bad_character = ['(', ')', '/', '-', ':']
    for c in bad_character:
        title = title.replace(c, ' ')

    headers = {
        'x-api-key': os.getenv('S2API_KEY'),
    }
    params = {
        'query': title,
        'fields': 'abstract,openAccessPdf',
        'limit': 1
    }
    with contextlib.suppress(Exception):
        ret_text = await fetch(semantic_url, headers=headers, params=params)
        ret = json.loads(ret_text)['data'][0]
        return {
            'abstract': ret['abstract'] or '',
            'openAccessPdf': ret['openAccessPdf']['url'] if ret['openAccessPdf'] else ''
        }

    return {}


def get_semantic_scholar2(title: str):
    """使用第三方库获取Semantic Scholar数据"""
    sch = SemanticScholar(api_key=os.getenv('S2API_KEY'))

    bad_character = ['(', ')', '/', '-', ':']
    for c in bad_character:
        title = title.replace(c, ' ')

    with contextlib.suppress(Exception):
        ret = sch.search_paper(title, fields=['abstract', 'openAccessPdf'], limit=1)[0]
        return {
            'abstract': ret['abstract'] or '',
            'openAccessPdf': ret['openAccessPdf']['url'] if ret['openAccessPdf'] else ''
        }

    return {}


def get_google_scholar(title: str):
    """获取Google Scholar数据，摘要可能不完整"""
    result = {}
    return result
    try:
        if result := next(scholarly.search_pubs(title))['bib']['abstract']:
            result['abstract'] = result
    except MaxTriesExceededException:
        console.print(f'Google Max failed: {title}', style='bold red')
    except TimeoutError:
        console.print(f'Google Timeout failed: {title}', style='bold red')
    except Exception:
        console.print(f'Google failed: {title}', style='bold red')
        console.print_exception()
    finally:
        return result

crawler/scholar.py
VulnTotal-Team/openccf
#!/usr/bin/python3

import os
import json
import argparse
import pyfiglet
from datetime import datetime
from concurrent.futures import ProcessPoolExecutor

from utils import *
from bots import *
from crawler import *
from crawler.libpaper import *


def make_bitable(data) -> dict:
    """将数据转换为多维表格格式"""
    def format_fields(data):
        """构建一条数据"""
        if title_zh := data['title_zh']:
            title = f'{data["title"]}\n\n{title_zh}'
        else:
            title = data['title']

        fields = {
            '标题': title,
            '年份': data['year'],
            '刊物': data['dblp_url'].split('/')[-2],
            '标签': data['keywords'],
            '摘要': data['abstract_zh'] or data['abstract'],
            '网址': {'text': data['url'], 'link': data['url']},
        }
        if pdf_url := data['files']['openAccessPdf']:
            fields['文件'] = {'text': pdf_url, 'link': pdf_url}

        return fields

    result = {'records': []}
    for item in data:
        result['records'].append({
            'fields': format_fields(item)
        })
    return result


def make_database(data) -> dict:
    """将数据转换为数据表格格式"""
    def format_fields(data):
        """构建一条数据"""
        if title_zh := data['title_zh']:
            title = f'{data["title"]}\n\n{title_zh}'
        else:
            title = data['title']

        fields = {
            '标题': title,
            '年份': data['year'],
            '刊物': data['dblp_url'].split('/')[-2],
            '标签': data['keywords'],
            '摘要': data['abstract_zh'] or data['abstract'],
            '网址': data['url'],
        }
        if pdf_url := data['files']['openAccessPdf']:
            fields['文件'] = pdf_url

        return fields

    result = {'rows': []}
    for item in data:
        result['rows'].append(format_fields(item))
    return result


def translate_all_empty():
    """将空的中文摘要和标题全部翻译"""
    console.print('Translating...', style='bold yellow')

    for file in dblp_path.glob('*.json'):
        empty_num = 0
        data = json.loads(file.read_text())
        for year, year_data in data.items():
            for i, item in enumerate(year_data):
                for j, paper in enumerate(item['papers']):
                    if paper['title'] and paper['title_zh'].isascii():
                        data[year][i]['papers'][j]['title_zh'] = get_translate(paper['title'])
                        empty_num += 1
                    if paper['abstract'] and paper['abstract_zh'].isascii():
                        data[year][i]['papers'][j]['abstract_zh'] = get_translate(paper['abstract'])
                        empty_num += 1

        if empty_num:
            print(file.name, empty_num)

            with open(file, 'w', encoding='utf-8') as f:
                json.dump(data, f, ensure_ascii=False, indent=4)


def func_broker(url: str):
    """选择解析函数"""
    dblp_key = '/'.join(url.split('/')[-3:-1])
    func_map = {
        'conf/sp': get_sp_data,
        'conf/ccs': get_ccs_data,
        'conf/ndss': get_ndss_data,
        'conf/uss': get_usenix_data,
    }
    return func_map.get(dblp_key)


def parse_rule(ccf_data: dict, rule: str='all:all:all:all'):
    """规则解析"""
    try:
        ccf_field, ccf_type, ccf_rank, ccf_name = rule.split(':')

        ccf_field = FIELDS if ccf_field == 'all' else ccf_field.split(',')
        ccf_type = ['journals', 'conf'] if ccf_type == 'all' else ccf_type.split(',')
        ccf_rank = ['A', 'B', 'C'] if ccf_rank == 'all' else ccf_rank.split(',')
        ccf_name = ['all'] if ccf_name == 'all' else ccf_name.split(',')

        ccf_url = []
        for a in ccf_field:
            for b in ccf_type:
                for c in ccf_rank:
                    if ccf_name[0] == 'all':
                        ccf_url.extend(e['address'] for e in ccf_data[a][b][c])
                    else:
                        for d in ccf_name:
                            ccf_url.extend(e['address'] for e in ccf_data[a][b][c] if d == e['name'])
        print(f'dblp url: {len(ccf_url)}\n{ccf_url}\n')
    except Exception:
        console.print(f'Wrong rule: {rule}', style='bold red')
        exit(1)

    return ccf_url


def parse_year(year: str):
    """解析年份"""
    parts = year.split(':')
    return parts[0], parts[1]


def init_bot(bot_name: str, conf: dict):
    """初始化机器人"""
    bot = None
    oper = None
    table = {}

    if bot_name == 'feishu':
        bot_conf = conf['feishu']

        # 群机器人
        if bot_key := os.getenv(bot_conf['bot']['name']) or bot_conf['bot']['key']:
            bot = feishuBot(bot_key)

        # 飞书开放平台->开发者后台
        app_id = os.getenv(bot_conf['app_id']['name']) or bot_conf['app_id']['key']
        app_secret = os.getenv(bot_conf['app_secret']['name']) or bot_conf['app_secret']['key']
        # 多维表格
        for key, value in bot_conf['bitable'].items():
            table[key] = os.getenv(value['name']) or value['key']

        if app_id and app_secret and table:
            oper = feishuOper(app_id, app_secret)

    elif bot_name == 'wolai':
        bot_conf = conf['wolai']

        # 应用设置
        app_id = os.getenv(bot_conf['app_id']['name']) or bot_conf['app_id']['key']
        app_secret = os.getenv(bot_conf['app_secret']['name']) or bot_conf['app_secret']['key']
        # 数据表格
        for key, value in bot_conf['database'].items():
            table[key] = os.getenv(value['name']) or value['key']

        if app_id and app_secret:
            oper = wolaiOper(app_id, app_secret)

    else:
        console.print('Wrong bot', style='bold red')
        exit(1)

    return bot, oper, table


def crawl_papers():
    """爬取论文"""
    # 获取CCF数据
    console.print('Getting ccf...', style='bold yellow')
    ccf_data = get_ccf_data(update=True)
    urls = parse_rule(ccf_data, rule)

    # 获取全量论文
    total_num = 0
    total_new_num = 0
    total_update_num = 0
    total_data = []
    total_new_data = []
    total_update_data = []

    # 获取基础数据及网址
    console.print('Getting papers...', style='bold yellow')
    executor = ProcessPoolExecutor(os.cpu_count()-1)
    tasks = [executor.submit(get_dblp_data, url, start_year, end_year) for url in urls]
    executor.shutdown(wait=True)

    for task in tasks:
        url, all_data, new_data, update_data = task.result()

        all_num = sum(len(j['papers']) for i in all_data.values() for j in i)
        new_num = sum(len(j['papers']) for i in new_data.values() for j in i)
        update_num = sum(len(j['papers']) for i in update_data.values() for j in i)
        console.print(f'{url}\nAll Papers: {all_num}\tNew Papers: {new_num}\tUpdate Papers: {update_num}\n', style='bold green')

        total_num += all_num
        total_new_num += new_num
        total_update_num += update_num
        # total_data.append(all_data)
        # total_new_data.append(new_data)
        # total_update_data.append(update_data)

        # 从论文网址获取补充数据
        # if get_paper_data := func_broker(url):
        #     paper_data = get_paper_data(all_data)

    console.print(f'All Papers: {total_num}\tNew Papers: {total_new_num}\tUpdate Papers: {total_update_num}', style='bold yellow')

    # 翻译摘要和标题
    translate_all_empty()


def filter_papers(category: str, keywords: list):
    """过滤论文"""
    console.print(f'\[{category}] Filtering papers...', style='bold yellow')

    total_data = []
    category_file = data_path.joinpath(f'{category}.json')
    for file in dblp_path.glob('*.json'):
        data = json.loads(file.read_text())
        filter_data = filter_keywords(data, keywords)
        total_data += filter_data

    # 按url排序
    sorted_data = sorted(total_data, key=lambda x: x['url'])

    with open(category_file, 'w') as f:
        json.dump(sorted_data, f, indent=4, ensure_ascii=False)

    console.print(f'\[{category}] Papers: {len(sorted_data)}', style='bold yellow')
    return sorted_data


def send_papers(category: str, total_data: list):
    """发送论文"""
    bot_data = []
    table_data = []
    table_data_temp = []

    if args.bot == 'feishu':
        bitable_history_file = history_path.joinpath(f'feishu_bitable_{category}.txt')
        bitable_history_data = bitable_history_file.read_text().splitlines() if bitable_history_file.exists() else []
        bot_history_file = history_path.joinpath(f'feishu_bot_{category}.txt')
        bot_history_data = bot_history_file.read_text().splitlines() if bot_history_file.exists() else []

        console.print(f'\[{category}] Sending bot...', style='bold yellow')
        for data in total_data:
            title = data['title']

            # 发送机器人消息
            if title not in bot_history_data and bot.send(data):
                bot_data.append(title)

            # 多维表格临时数据
            if title not in bitable_history_data:
                table_data_temp.append(data)

        # 更新多维表格
        console.print(f'\[{category}] Updating bitable...', style='bold yellow')
        if oper.check_access_valid(feishuOper.TOKEN_TENANT):
            # 每组100条数据
            split_list = [table_data_temp[i:i+100] for i in range(0, len(table_data_temp), 100)]
            for split_data in split_list:
                data = make_bitable(split_data)
                if oper.bitable_batch_create(table[category], data):
                    table_data += [i['title'] for i in split_data]
        else:
            console.print(f'\[{category}] 权限验证失败', style='bold red')

        with open(bot_history_file, 'w') as f:
            f.write('\n'.join(bot_history_data + bot_data))

        with open(bitable_history_file, 'w') as f:
            f.write('\n'.join(bitable_history_data + table_data))

    elif args.bot == 'wolai':
        database_history_file = history_path.joinpath(f'wolai_database_{category}.txt')
        database_history_data = database_history_file.read_text().splitlines() if database_history_file.exists() else []

        # 更新数据表格
        console.print(f'\[{category}] Updating database...', style='bold yellow')
        table_data_temp = [
            data
            for data in total_data
            if data['title'] not in database_history_data
        ]
        # 每组20条数据
        split_list = [table_data_temp[i:i+20] for i in range(0, len(table_data_temp), 20)][:1]
        for split_data in split_list:
            data = make_database(split_data)
            if oper.database_post(table[category], data):
                table_data += [i['title'] for i in split_data]

        with open(database_history_file, 'w') as f:
            f.write('\n'.join(database_history_data + table_data))

    console.print(f'\[{category}] Papers: {len(total_data)}\tSend Papers: {len(bot_data)}\tUpdate Papers: {len(table_data)}', style='bold yellow')


def parse_args():
    """参数解析"""
    start_year = datetime.now().year
    end_year = start_year - 5
    parser = argparse.ArgumentParser()
    parser.add_argument('--year', type=str, metavar='start:end', default=f'{start_year}:{end_year}', help='e.g. 2020:2015')
    parser.add_argument('--rule', type=str, metavar='field:type:rank:name', default='all:all:all:all', help='e.g. NIS:conf:A,B:all')
    parser.add_argument('--category', type=str, metavar='category', default='', help='e.g. vehicle,android,linux')
    parser.add_argument('--keywords', type=str, metavar='keywords', default='', help='e.g. keyword1,keyword2')
    parser.add_argument('--bot', type=str, metavar='bot', default='feishu', help='e.g. feishu')
    return parser.parse_args()


if __name__ == '__main__':
    if Path('.env').exists():
        from dotenv import load_dotenv
        load_dotenv('.env')
    print(pyfiglet.figlet_format('OpenCCF'))

    # 参数解析
    args = parse_args()
    conf = json.loads(Path('config.json').read_text())
    proxy_url = conf['proxy']

    secrets = conf['openai']['name']
    openai_key = os.getenv(secrets) or conf['openai']['key']
    os.environ[secrets] = openai_key

    secrets = conf['s2api']['name']
    s2api_key = os.getenv(secrets) or conf['s2api']['key']
    os.environ[secrets] = s2api_key

    if args.category:
        category_list = args.category.split(',')
    else:
        category_list = conf['keywords'].keys()

    keywords_dict = {}
    if args.keywords:
        keywords_dict = {i: args.keywords.split(',') for i in category_list}
    else:
        for category in category_list:
            keywords_dict[category] = [
                j
                for i in conf['keywords'][category].values()
                for j in i
                if j.isascii()
            ]

    year = args.year or conf['year']
    rule = args.rule or conf['rule']
    start_year, end_year = parse_year(year)

    # 初始化机器人
    bot, oper, table = init_bot(args.bot, conf)

    # 爬取论文
    #crawl_papers()

    for category, keywords in keywords_dict.items():
        # 过滤论文
        total_data = filter_papers(category, keywords)

        # 发送论文
        send_papers(category, total_data)

openccf.py
VulnTotal-Team/openccf
import re
import aiohttp
import asyncio
import contextlib
import translators
from pathlib import Path
from fp.fp import FreeProxy
from fake_useragent import UserAgent
from tenacity import retry, stop_after_attempt, wait_fixed

from rich import print
from rich.console import Console
from rich.progress import Progress, TextColumn, BarColumn, MofNCompleteColumn, TimeRemainingColumn


MAX_RETRY = 5
MAX_ROUTINE = 100
sem = asyncio.Semaphore(MAX_ROUTINE)

console = Console()

ScraperAPI = 'cf7e327977f1d6f56b77b48513422be9'

root_path = Path(__file__).parent
data_path = root_path.joinpath('data')
dblp_path = data_path.joinpath('dblp')
dblp_path.mkdir(exist_ok=True)
paper_path = data_path.joinpath('paper')
paper_path.mkdir(exist_ok=True)
history_path = data_path.joinpath('history')
history_path.mkdir(exist_ok=True)


def progress():
    """创建进度条"""
    return Progress(
        TextColumn("{task.description}"),
        BarColumn(),
        MofNCompleteColumn(),
        TimeRemainingColumn()
    )


def get_translate(text: str):
    """sogou速度较快，优先使用；google作为补充"""
    if text:
        result = ''
        with contextlib.suppress(Exception):
            result = translators.translate_text(text, translator='sogou')

        if result.isascii():
            with contextlib.suppress(Exception):
                result = translators.translate_text(text, translator='google', to_language='zh')

        if not result.isascii():
            return result

    return ''


async def fetch(url: str, headers: dict = None, params: dict=None, proxy: bool=False):
    if headers is None:
        headers = {}
    headers['User-Agent'] = UserAgent().random

    for _ in range(MAX_RETRY):
        try:
            async with sem, aiohttp.ClientSession() as session:
                if proxy and 'doi.org' in url:
                    # async with session.get(url, headers=headers, params=params, proxy=FreeProxy().get()) as response:
                    payload = {'api_key': ScraperAPI, 'url': url}
                    async with session.get(url, headers=headers, payload=payload) as response:
                        if response.status == 200:
                            return await response.text()
                else:
                    async with session.get(url, headers=headers, params=params) as response:
                        if response.status == 200:
                            return await response.text()

            # 超出频率限制
            if response.status == 429:
                await asyncio.sleep(1)
                continue

            console.print(f'Request failed: {response.status} {url}', style='bold red')
            return None
        except Exception:
            await asyncio.sleep(1)

    if 'api.semanticscholar.org' not in url:
        console.print(f'Request failed: max {url}', style='bold red')
    return None


def get_keywords(paper: dict, keywords: list):
    """过滤关键词"""
    key_set = set()
    for key in keywords:
        pattern = re.compile(rf'\b{re.escape(key)}(?!\w)', re.IGNORECASE)

        if paper['title'] and re.search(pattern, paper['title']):
            key_set.add(key)
        if paper['abstract'] and re.search(pattern, paper['abstract']):
            key_set.add(key)

    return sorted(key_set)


def filter_keywords(data: dict, keywords: list):
    """通过关键词筛选相关论文"""

    results = []
    for year, year_data in data.items():
        for item in year_data:
            dblp_url = item['dblp_url']
            conf_title = item.get('journals_title') or item.get('conf_title') or ''
            conf_url = item.get('journals_url') or item.get('conf_url') or ''
            for paper in item['papers']:
                if key := get_keywords(paper, keywords):
                    paper.update({
                        'year': year,
                        'dblp_url': dblp_url,
                        'conf_title': conf_title,
                        'conf_url': conf_url,
                        'keywords': key
                    })
                    results.append(paper)

    return results

utils.py
VulnTotal-Team/openccf