main.py

import time
from datetime import timezone, timedelta, datetime

import requests
import urllib3

from util import *

urllib3.disable_warnings()

user_agent = (
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
    'Chrome/77.0.3865.120 Safari/537.36'
)
sub_folders = [
    str(i * 100).zfill(4) + '-' + str(i * 100 + 99).zfill(4) for i in range(100)
]
cn_graph_url = 'https://leetcode.cn/graphql'
difficulty = dict(Easy='简单', Medium='中等', Hard='困难')
weekly_range = range(83, 500)
biweekly_range = range(1, 300)
WEEKLY_URL = 'https://leetcode.cn/contest/api/info/weekly-contest-{}/'
BIWEEKLY_URL = 'https://leetcode.cn/contest/api/info/biweekly-contest-{}/'
WEEKLY_SLUG = 'weekly-contest-{}'
BIWEEKLY_SLUG = 'biweekly-contest-{}'


class Spider:
    def __init__(self, cookie1: str, cookie2: str):
        self.cookie_cn = cookie1
        self.cookie_en = cookie2
        self.session = requests.session()

    def get_all_questions(self, retry: int = 3) -> List:
        """获取所有题目"""
        headers = {
            'accept': 'application/json, text/javascript, */*; q=0.01',
            'content-type': 'application/json',
            'user-agent': user_agent,
            'x-requested-with': 'XMLHttpRequest',
            'cookie': self.cookie_en,
        }
        try:
            resp = self.session.get(
                url='https://leetcode.com/api/problems/all/',
                headers=headers,
                allow_redirects=False,
                timeout=10,
                verify=False,
            )
            return resp.json()['stat_status_pairs']
        except Exception as e:
            print(e)
            time.sleep(2)
            return self.get_all_questions(retry - 1) if retry > 0 else []

    def get_question_detail_en(self, question_title_slug: str, retry: int = 3) -> dict:
        headers = {
            'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;'
                      'q=0.8,application/signed-exchange;v=b3;q=0.7',
            'user-agent': user_agent,
            'cookie': self.cookie_en
        }
        question_url = 'https://leetcode.com/problems' + question_title_slug
        en_graph_url = 'https://leetcode.com/graphql'
        form = {
            'operationName': 'questionData',
            'variables': {'titleSlug': question_title_slug},
            'query': 'query questionData($titleSlug: String!) {\n  question(titleSlug: $titleSlug) {\n    '
                     'questionId\n    questionFrontendId\n    categoryTitle\n    boundTopicId\n    title\n    '
                     'titleSlug\n    content\n    translatedTitle\n    translatedContent\n    isPaidOnly\n    '
                     'difficulty\n    likes\n    dislikes\n    isLiked\n    similarQuestions\n    '
                     'contributors {\n      username\n      profileUrl\n      avatarUrl\n      __typename\n    '
                     '}\n    langToValidPlayground\n    topicTags {\n      name\n      slug\n      '
                     'translatedName\n      __typename\n    }\n    companyTagStats\n    codeSnippets {\n      '
                     'lang\n      langSlug\n      code\n      __typename\n    }\n    stats\n    hints\n    '
                     'solution {\n      id\n      canSeeDetail\n      __typename\n    }\n    status\n    '
                     'sampleTestCase\n    metaData\n    judgerAvailable\n    judgeType\n    mysqlSchemas\n    '
                     'exampleTestcases\n    __typename\n  }\n}\n',
        }
        for _ in range(max(0, retry) + 1):
            try:
                self.session.get(question_url, headers=headers, timeout=10, verify=False)
                headers = {
                    'User-Agent': user_agent,
                    'Connection': 'keep-alive',
                    'Content-Type': 'application/json',
                    'Referer': 'https://leetcode.com/problems/' + slug,
                    'cookie': self.cookie_en
                }
                resp = self.session.post(en_graph_url, headers=headers, data=json.dumps(form), timeout=10, verify=False)
                res = resp.json()
                return res['data']['question'] or {}
            except Exception as e:
                print(e)
                time.sleep(2)
        return {}

    def get_question_detail(self, question_title_slug: str, retry: int = 3) -> dict:
        """获取题目详情"""
        form1 = {
            'operationName': 'globalData',
            'query': 'query globalData {\n  feature {\n    questionTranslation\n    subscription\n    signUp\n    '
                     'discuss\n    mockInterview\n    contest\n    store\n    book\n    chinaProblemDiscuss\n    '
                     'socialProviders\n    studentFooter\n    cnJobs\n    enableLsp\n    enableWs\n    '
                     'enableDebugger\n    enableDebuggerAdmin\n    enableDarkMode\n    tasks\n    '
                     'leetbook\n    __typename\n  }\n  userStatus {\n    isSignedIn\n    isAdmin\n    '
                     'isStaff\n    isSuperuser\n    isTranslator\n    isPremium\n    isVerified\n    '
                     'isPhoneVerified\n    isWechatVerified\n    checkedInToday\n    username\n    '
                     'realName\n    userSlug\n    groups\n    avatar\n    optedIn\n    '
                     'requestRegion\n    region\n    activeSessionId\n    permissions\n    notificationStatus {\n      '
                     'lastModified\n      numUnread\n      __typename\n    }\n    completedFeatureGuides\n    '
                     'useTranslation\n    accountStatus {\n      isFrozen\n      inactiveAfter\n      __typename\n    '
                     '}\n    __typename\n  }\n  siteRegion\n  chinaHost\n  websocketUrl\n  userBannedInfo {\n    '
                     'bannedData {\n      endAt\n      bannedType\n      __typename\n    }\n    __typename\n  }\n}\n',
            'variables': {},
        }
        headers = {
            'User-Agent': user_agent,
            'Connection': 'keep-alive',
            'Content-Type': 'application/json',
            'Referer': 'https://leetcode.cn/problems/' + question_title_slug,
            'cookie': self.cookie_cn,
        }

        form2 = {
            'operationName': 'questionData',
            'variables': {'titleSlug': question_title_slug},
            'query': 'query questionData($titleSlug: String!) {\n  question(titleSlug: $titleSlug) {\n    '
                     'questionId\n    questionFrontendId\n    categoryTitle\n    boundTopicId\n    title\n    '
                     'titleSlug\n    content\n    translatedTitle\n    translatedContent\n    isPaidOnly\n    '
                     'difficulty\n    likes\n    dislikes\n    isLiked\n    similarQuestions\n    '
                     'contributors {\n      username\n      profileUrl\n      avatarUrl\n      __typename\n    '
                     '}\n    langToValidPlayground\n    topicTags {\n      name\n      slug\n      '
                     'translatedName\n      __typename\n    }\n    companyTagStats\n    codeSnippets {\n      '
                     'lang\n      langSlug\n      code\n      __typename\n    }\n    stats\n    hints\n    '
                     'solution {\n      id\n      canSeeDetail\n      __typename\n    }\n    status\n    '
                     'sampleTestCase\n    metaData\n    judgerAvailable\n    judgeType\n    mysqlSchemas\n    '
                     'enableRunCode\n    envInfo\n    book {\n      id\n      bookName\n      pressName\n      '
                     'source\n      shortDescription\n      fullDescription\n      bookImgUrl\n      '
                     'pressImgUrl\n      productUrl\n      __typename\n    }\n    isSubscribed\n    '
                     'isDailyQuestion\n    dailyRecordStatus\n    editorType\n    ugcQuestionId\n    style\n    '
                     'exampleTestcases\n    __typename\n  }\n}\n',
        }
        for _ in range(max(0, retry) + 1):
            try:
                self.session.post(
                    url=cn_graph_url,
                    data=json.dumps(form1),
                    headers=headers,
                    timeout=10,
                    verify=False,
                )
                # get question detail
                resp = self.session.post(
                    url=cn_graph_url,
                    data=json.dumps(form2).encode('utf-8'),
                    headers=headers,
                    timeout=10,
                    verify=False,
                )
                res = resp.json()
                return res['data']['question'] or {}
            except Exception as e:
                print(e)
                time.sleep(2)
        return {}

    @staticmethod
    def format_question_detail(question_detail: dict) -> dict:
        question_title_slug = question_detail.get('titleSlug')
        url_cn = f'https://leetcode.cn/problems/{question_title_slug}'
        url_en = f'https://leetcode.com/problems/{question_title_slug}'
        frontend_question_id = str(question_detail['questionFrontendId']).zfill(4)
        no = int(frontend_question_id) // 100
        question_title_en = question_detail['title']
        question_title_en = re.sub(r'[\\/:*?"<>|]', '', question_title_en).strip()
        path_cn = f'/solution/{sub_folders[no]}/{frontend_question_id}.{quote(question_title_en)}/README.md'
        path_en = f'/solution/{sub_folders[no]}/{frontend_question_id}.{quote(question_title_en)}/README_EN.md'

        print(f'{frontend_question_id}. {question_title_en}')
        topic_tags = question_detail.get('topicTags')

        item = {
            'sub_folder': sub_folders[no],
            'question_id': str(question_detail['questionId']).zfill(4),
            'frontend_question_id': frontend_question_id,
            'paid_only': question_detail.get('isPaidOnly'),
            'paid_only_cn': question_detail.get('isPaidOnly'),
            # Shell Database Algorithms Concurrency
            'category': question_detail.get('categoryTitle'),
            'url_cn': url_cn,
            'url_en': url_en,
            'relative_path_cn': path_cn,
            'relative_path_en': path_en,
            'title_cn': question_detail.get('translatedTitle')
                        or question_title_en
                        or '',
            'title_en': question_title_en or '',
            'question_title_slug': question_title_slug,
            'content_en': question_detail.get('content'),
            'content_cn': question_detail.get('translatedContent')
                          or question_detail.get('content')
                          or '',
            'tags_en': [e['name'] for e in topic_tags if e['name']] or [],
            'tags_cn': [e['translatedName'] for e in topic_tags if e['translatedName']]
                       or [],
            'difficulty_en': question_detail.get('difficulty'),
            'difficulty_cn': difficulty.get(question_detail.get('difficulty')),
            'code_snippets': question_detail.get('codeSnippets') or [],
        }

        col1_cn = frontend_question_id
        col2_cn = (
            f'[{item["title_cn"]}]({path_cn})'
            if item["title_cn"]
            else f'[{item["title_en"]}]({path_en})'
        )
        col3_cn = ','.join([f'`{tag}`' for tag in item['tags_cn']])
        col3_cn = '' if (col3_cn == 'None' or not col3_cn) else col3_cn
        col4_cn = item['difficulty_cn']
        col5_cn = '🔒' if item['paid_only_cn'] else ''
        col1_en = frontend_question_id
        col2_en = f'[{item["title_en"]}]({path_en})'
        col3_en = ','.join([f'`{tag}`' for tag in item['tags_en']])
        col3_en = '' if (col3_en == 'None' or not col3_en) else col3_en
        col4_en = item['difficulty_en']
        col5_en = '🔒' if item['paid_only'] else ''

        item['md_table_row_cn'] = [col1_cn, col2_cn, col3_cn, col4_cn, col5_cn]
        item['md_table_row_en'] = [col1_en, col2_en, col3_en, col4_en, col5_en]
        return item


class Contest:
    def __init__(self, contest_seq: int, contest_type: int = 1):
        double = contest_type % 2 == 0
        url_pattern = BIWEEKLY_URL if double else WEEKLY_URL
        slug_pattern = BIWEEKLY_SLUG if double else WEEKLY_SLUG
        self.contest_type = contest_type
        self.contest_url = url_pattern.format(contest_seq)
        self.contest_title_slug = slug_pattern.format(contest_seq)
        self.contest_title = (
            f'第 {contest_seq} 场双周赛' if double else f'第 {contest_seq} 场周赛'
        )
        self.contest_title_en = (
            f'Biweekly Contest {contest_seq}'
            if double
            else f'Weekly Contest {contest_seq}'
        )

    @staticmethod
    def format_time(timestamp: int) -> str:
        tz = timezone(timedelta(hours=+8))
        return datetime.fromtimestamp(timestamp, tz).strftime('%Y-%m-%d %H:%M')

    def get_data(self, retry: int = 3):
        try:
            res = requests.get(self.contest_url, timeout=6, verify=False).json()
            if not res or 'error' in res or not res['questions']:
                return {}
            questions = res['questions']
            question_slugs = [q['title_slug'] for q in questions]
            return {
                'contest_title': self.contest_title,
                'contest_title_en': self.contest_title_en,
                'contest_title_slug': res['contest']['title_slug'],
                'contest_id': res['contest']['id'],
                'contest_start_time': res['contest']['origin_start_time'],
                'contest_duration': res['contest']['duration'],
                'user_num': res['user_num'],
                'question_slugs': question_slugs,
            }
        except Exception as e:
            print(e)
            time.sleep(2)
            return self.get_data(retry - 1) if retry > 0 else {}

    @staticmethod
    def format(data: dict) -> List:
        if not data:
            return []
        title = data['contest_title']
        title_en = data['contest_title_en']
        start_time = data['contest_start_time']
        duration = data['contest_duration']
        cost_minutes = duration // 60
        user_num = data['user_num']
        rows = [
            f'#### {title}({Contest.format_time(start_time)}, {cost_minutes} 分钟) 参赛人数 {user_num}\n'
        ]
        rows_en = [f'#### {title_en}\n']
        for question in data['question_list']:
            (
                frontend_question_id,
                title_cn,
                title_en,
                relative_path_cn,
                relative_path_en,
            ) = question
            rows.append(f'- [{frontend_question_id}. {title_cn}]({relative_path_cn})')
            rows_en.append(
                f'- [{frontend_question_id}. {title_en}]({relative_path_en})'
            )
        return [start_time, '\n'.join(rows), '\n'.join(rows_en)]


def get_contests(fetch_new=True) -> List:
    res = [] if fetch_new else load_contest_result()
    t = 0
    d = {x.get('contest_title_slug'): x for x in res}
    for r in (weekly_range, biweekly_range):
        t += 1
        cnt = 0
        for i in r:
            c = Contest(i, contest_type=t)
            if c.contest_title_slug in d:
                continue
            contest_data = c.get_data(retry=3)
            if not contest_data:
                cnt += 1
                if cnt > 2:
                    break
                continue
            print(contest_data)
            res.append(contest_data)
            d[c.contest_title_slug] = contest_data
    save_contest_result(res)
    return res


########################################################################################
# 加载 cookies
cookie_cn, cookie_en = load_cookies()
spider = Spider(cookie_cn, cookie_en)

# 是否刷新所有题目
refresh_all = load_refresh_config()

question_details = {}
if not refresh_all:
    for item in load_result():
        slug = item.get('question_title_slug')
        if slug:
            question_details[slug] = item

for q in spider.get_all_questions(retry=4):
    slug = q['stat']['question__title_slug']
    if slug in question_details:
        print(f'skip {slug}')
        continue
    detail = spider.get_question_detail(slug, retry=4) or spider.get_question_detail_en(slug, retry=4)
    if not detail:
        continue
    time.sleep(0.3)
    question_details[slug] = Spider.format_question_detail(detail)

# 周赛场次列表
contest_list = get_contests(refresh_all)
cls = []
for contest in contest_list:
    contest_title = contest['contest_title']
    contest_title_en = contest['contest_title_en']
    contest_question_list = []
    for slug in contest['question_slugs']:
        if slug in question_details:
            detail = question_details.get(slug)
            # 给题目详情添加周赛信息
            detail['md_table_row_cn'][4] = contest_title
            detail['md_table_row_en'][4] = contest_title_en

            # 给周赛信息添加题目详情
            add = [
                detail['frontend_question_id'],
                detail['title_cn'],
                detail['title_en'],
                detail['relative_path_cn'],
                detail['relative_path_en'],
            ]
            contest_question_list.append(add)

    contest['question_list'] = contest_question_list
    cls.append(Contest.format(contest))

# 保存题目到本地，生成题目列表以及题目详情文件
ls = list(question_details.values())
save_result(ls)
generate_readme(ls)
generate_question_readme(ls)
generate_summary(ls)

# 生成周赛题目列表
generate_contest_readme(cls)

# 刷新题目文件
if refresh_all:
    refresh(ls)