File size: 3,833 Bytes
4117d38
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import requests
import random
from fastapi import APIRouter, BackgroundTasks




# 添加headers模拟浏览器访问(针对juzi等特殊订阅)
headers = {
    'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
    'accept-language': 'zh-CN,zh;q=0.9',
    'cache-control': 'max-age=0',
    'priority': 'u=0, i',
    'sec-ch-ua': '"Chromium";v="128", "Not;A=Brand";v="24", "Google Chrome";v="128"',
    'sec-ch-ua-mobile': '?0',
    'sec-ch-ua-platform': '"macOS"',
    'sec-fetch-dest': 'document',
    'sec-fetch-mode': 'navigate',
    'sec-fetch-site': 'none',
    'sec-fetch-user': '?1',
    'upgrade-insecure-requests': '1',
    'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36'
}


def is_subscription_link_valid(subscribeUrl: str) -> bool:
    try:
        result = requests.get(subscribeUrl,headers=headers).text
        if "error" in result:
            return False
        return True
    except Exception as e:
        #这里如果只是请求超时那么认为正常返回true,比如juzi sub判断时候连接超时
        print('判断sub link是否有效时候发生错误:',e)
        return True


def delete_invalid_url_in_txt(fileName: str) -> "void":
    valid_lines = []
    with open(fileName, "r") as file:
        lines = file.readlines()
    for line in lines:
        email = line.strip().split(",")[0]
        subscription_url = line.strip().split(",")[1]
        if is_subscription_link_valid(subscription_url):
            valid_lines.append(line)
        else:
            print(email + "订阅已经不可用")
    # 将有效行重新写回文件
    with open(fileName, "w") as file:
        file.writelines(valid_lines)


def subscription_link_list(fileName: str) -> list:
    SubscribeUrlList = []
    with open(fileName, "r") as f:
        lines = f.readlines()
    for line in lines:
        subscription_url = line.strip().split(",")[1]
        SubscribeUrlList.append(subscription_url)
    return SubscribeUrlList


def subscription_link_valid_list(SubscribeUrlList: list) -> list:
    valid_link_list = list(filter(lambda f: is_subscription_link_valid(f), SubscribeUrlList))
    ##返回可用订阅链接前对原始文件进行删除不可用链接操作:
    return valid_link_list


def read_random_line(fileName: str) -> str:
    with open(fileName, "r") as file:
        lines = file.readlines()
    return random.choice(lines)


def getRandomSubscribeUrl(validSubscribeUrlList: list) -> str:
    # 返回一个可用的订阅链接信息
    return random.choice(validSubscribeUrlList)


router = APIRouter()

#此接口目前唯一作用是方便调试查看订阅剩余状态
@router.get('/')
def returnRandomSubscribeUrl(background_tasks: BackgroundTasks) -> str:
    # 返回一个可用的订阅链接信息
    SubscribeUrlList = subscription_link_list(router.fileName)
    #直接用现成的认为可用然后直接返回,之后再删不可用的
    result = getRandomSubscribeUrl(SubscribeUrlList)
    background_tasks.add_task(delete_invalid_url_in_txt, router.fileName)
    return result

    # return hf_test()




if __name__ == "__main__":
    print()

    ##思路是随机选取一行订阅检测,如果没问题就返回
    # 1订阅有问题,那么
    ##1->对全部订阅扫描,删除不可用的,然后重新随机一个返回,此返回时不用检测
    ##2->如果扫描后没有可用的,那么不做处理让vercel报错

    # 1订阅一个都没有,那么vercel错误提醒(不做处理考虑就行)