V2RayAggregator/utils/list_update.py

#!/usr/bin/env python3

from datetime import timedelta, datetime
import json
import re
import requests
from requests.adapters import HTTPAdapter
from urllib.parse import quote

# 文件路径定义
# sub_list_json = './sub/sub_list.json'

def url_updated(url):  # 判断远程远程链接是否已经更新
    s = requests.Session()
    s.mount('http://', HTTPAdapter(max_retries=2))
    s.mount('https://', HTTPAdapter(max_retries=2))
    try:
        resp = s.get(url, timeout=4)
        status = resp.status_code
    except Exception:
        status = 404
    if status == 200:
        url_updated = True
    else:
        url_updated = False
    return url_updated


class update_url():

    def update_main(use_airport=False, airports_id: [] = [5], sub_list_json= './sub/sub_list.json'):
        with open(sub_list_json, 'r', encoding='utf-8') as f:  # 载入订阅链接
            raw_list = json.load(f)
            f.close()
            
        for sub in raw_list:
            id = sub['id']
            current_url = sub['url']
            if use_airport == False:
                if airports_id.__contains__(id) == False and sub['update_method'] != 'update_airports':
                    try:
                        if sub['update_method'] != 'auto' and sub['enabled'] == True:
                            print(f'Finding available update for ID{id}')
                            if sub['update_method'] == 'change_date':
                                new_url = update_url.change_date(
                                    id, current_url)
                                if new_url == current_url:
                                    print(f'No available update for ID{id}\n')
                                else:
                                    sub['url'] = new_url
                                    print(f'ID{id} url updated to {new_url}\n')
                            elif sub['update_method'] == 'page_release':
                                new_url = update_url.find_link(id, current_url)
                                if new_url == current_url:
                                    print(f'No available update for ID{id}\n')
                                else:
                                    sub['url'] = new_url
                                    print(f'ID{id} url updated to {new_url}\n')
                            elif sub['update_method'] == 'update_airports':
                                new_url = update_url.update_airports(
                                    id, current_url)
                                if new_url == current_url:
                                    print(f'No available update for ID{id}\n')
                                else:
                                    sub['url'] = new_url
                                    print(f'ID{id} url updated to {new_url}\n')

                    except KeyError:
                        print(
                            f'{id} Url not changed! Please define update method.')

            else:
                if airports_id.__contains__(id) == True:
                    try:
                        if sub['update_method'] != 'auto' and sub['enabled'] == True:
                            print(f'Finding available update for ID{id}')
                            if sub['update_method'] == 'change_date':
                                new_url = update_url.change_date(
                                    id, current_url)
                                if new_url == current_url:
                                    print(f'No available update for ID{id}\n')
                                else:
                                    sub['url'] = new_url
                                    print(f'ID{id} url updated to {new_url}\n')
                            elif sub['update_method'] == 'page_release':
                                new_url = update_url.find_link(id, current_url)
                                if new_url == current_url:
                                    print(f'No available update for ID{id}\n')
                                else:
                                    sub['url'] = new_url
                                    print(f'ID{id} url updated to {new_url}\n')
                            elif sub['update_method'] == 'update_airports':
                                new_url = update_url.update_airports(
                                    id, current_url)
                                if new_url == current_url:
                                    print(f'No available update for ID{id}\n')
                                else:
                                    sub['url'] = new_url
                                    print(f'ID{id} url updated to {new_url}\n')

                    except KeyError:
                        print(
                            f'{id} Url not changed! Please define update method.')

            updated_list = json.dumps(
                raw_list, sort_keys=False, indent=2, ensure_ascii=False)
            file = open(sub_list_json, 'w', encoding='utf-8')
            file.write(updated_list)
            file.close()

    def update_airports(id, current_url):
        if id == 5:
            # try:
            s = requests.Session()
            s.mount('http://', HTTPAdapter(max_retries=2))
            s.mount('https://', HTTPAdapter(max_retries=2))
            urllist = list(set(list(filter(lambda x: x != "" and str(x).startswith("http"), s.get(
                'https://raw.githubusercontent.com/RenaLio/Mux2sub/main/urllist', timeout=4).text.split("\n")))))
            sublist = list(set(list(filter(lambda x: x != "" and str(x).startswith("http"), s.get(
                'https://raw.githubusercontent.com/RenaLio/Mux2sub/main/sub_list', timeout=4).text.split("\n")))))

            air_free = list(set(list(filter(lambda x: x != "" and str(x).startswith("http"), s.get(
                'https://raw.githubusercontent.com/rxsweet/getAirport/main/config/sublist_free', timeout=4).text.split("\n")))))
            air_mining = list(set(list(filter(lambda x: x != "" and str(x).startswith("http"), s.get(
                'https://raw.githubusercontent.com/rxsweet/getAirport/main/config/sublist_mining', timeout=4).text.split("\n")))))

            urllist.extend(sublist)
            urllist.extend(air_free)
            urllist.extend(air_mining)

            # urllist = list(map(lambda x: quote(x, safe=""), urllist))
            # urllist = list(filter(lambda x: str(x).__contains__(
            #    "getafreenode.com") == False, urllist))
            new_url = "|".join(list(set(urllist)))
            # except Exception as e:
            #     print(e)
        return new_url

    def change_date(id, current_url):
        if id == 0:
            today = datetime.today().strftime('%m%d')
            url_front = 'https://raw.githubusercontent.com/pojiezhiyuanjun/freev2/master/'
            url_end = '.txt'
            new_url = url_front + today + url_end

        if id == 1:
            today = datetime.today().strftime('%Y%m%d')
            this_year = datetime.today().strftime('%Y')
            this_month = datetime.today().strftime('%m')
            url_front = 'https://nodefree.org/dy/'
            url_end = '.yaml'
            new_url = url_front + this_year + '/' + this_month + '/' + today + url_end

        if id == 3:
            # https://v2rayshare.com/wp-content/uploads/2022/10/20221025.txt
            today = datetime.today().strftime('%Y%m%d')
            this_month = datetime.today().strftime('%m')
            this_year = datetime.today().strftime('%Y')
            url_front = 'https://v2rayshare.com/wp-content/uploads/'
            url_end = '.txt'
            new_url = url_front + \
                "/".join([this_year, this_month, today]) + url_end

        if id == 4:
            # https://clashnode.com/wp-content/uploads/2022/10/20221004.yaml
            today = datetime.today().strftime('%Y%m%d')
            this_month = datetime.today().strftime('%m')
            this_year = datetime.today().strftime('%Y')
            url_front = 'https://clashnode.com/wp-content/uploads/'
            url_end = '.txt'
            new_url = url_front + \
                "/".join([this_year, this_month, today]) + url_end

        if url_updated(new_url):
            return new_url
        else:
            return current_url

    def find_link(id, current_url):
        if id == 2:
            try:
                res_json = requests.get(
                    'https://api.github.com/repos/mianfeifq/share/contents/').json()
                for file in res_json:
                    if file['name'].startswith('data'):
                        return file['download_url']
                else:
                    return current_url
            except Exception:
                return current_url


if __name__ == '__main__':
    update_url.update_main()