-
Notifications
You must be signed in to change notification settings - Fork 117
auto_sign_in
IvonWei edited this page May 16, 2022
·
38 revisions
自动签到 包含了 签到，读取收件箱未读消息，统计信息 功能
验证码签到功能需安装baidu-aip
统计信息生成图片需安装pandas matplotlib
dmhy签到需要 baidu-aip fuzzywuzzy python-Levenshtein
aipocr参数获取: https://cloud.baidu.com/doc/OCR/s/dk3iqnq51
auto_sign_in:
user-agent: <user-agent>
max_workers: 4 #线程数
get_messages: yes #设为no跳过获取未读信息。默认yes
get_details: yes #设为no跳过获取统计。默认yes
aipocr:
app_id: <app_id>
api_key: <api_key>
secret_key: <secret_key>
sites:
<site_name>: <cookie>如 ptsites.sites 目录下存在需要的站点 只需配置相应的 主域名: cookie，例：
sites:
1ptba: xxxxxxxx但是也有因cookie过期太快，而改用模拟登陆的站点，如以下站点:
sites:
filelist:
login:
username: xxxxxxxx
password: xxxxxxxx
ourbits:
# ourbits 在同时配置了cookie 和 login 时，会忽略login 使用cookie签到
cookie: xxxxxxxx
login:
username: xxxxxxxx
password: xxxxxxxx
#二次验证, 扫描二维码得到文本 otpauth://totp/*****:****?secret=[secret_key]&issuer=***** (只取[secret_key]部分)
secret_key: <secret_key>
skyey2:
login:
username: xxxxxxxx
password: xxxxxxxx
#二次验证, 扫描二维码得到文本 otpauth://totp/*****:****?secret=[secret_key]&issuer=***** (只取[secret_key]部分)
secret_key: <secret_key>
m-team:
login:
username: 'xxxxxxxx'
password: 'xxxxxxxx'
#二次验证, 扫描二维码得到文本 otpauth://totp/*****:****?secret=[secret_key]&issuer=***** (只取[secret_key]部分)
secret_key: <secret_key>
dmhy:
sites:
dmhy:
username: <username>
cookie: 'xxxxxxx'
# 五个字符以上签到留言
comment: <comment>
ocr_config:
# 重试次数
retry: 10
# 最低识别字符数
char_count: 3
# 最低匹配分数
score: 50gazellegames:
sites:
gazellegames:
cookie: 'xxxxxxx'
# api key 需要pm 管理申请 (如未申请 api key 请删除该项)
key: 'xxxxxxx'
# 用户名 (如未申请 api key 请删除该项)
name: 'xxxxxxx'qbittorrent 信息获取
sites:
qbittorrent:
# 为客户端命名
- name: <qbittorrent_name>
host: qbittorrent.example.com
port: 443
use_ssl: yes
username: xxxxxxx
password: xxxxxxx
- name: <qbittorrent_name2>
host: qbittorrent2.example.com
port: 443
use_ssl: yes
username: xxxxxxx
password: xxxxxxx签到结果保存在 entry['result'] 属性中
未读信息保存在 entry['messages'] 属性中
生成的 统计信息 保存在 plugins 目录下 details_report.png
import re
from urllib.parse import urljoin
from flexget.utils.soup import get_soup
from ..schema.site_base import SignState, Work, NetworkState
from ..schema.unit3d import Unit3D
from ..utils.net_utils import NetUtils
"""
类名必须为 MainClass, 站点名用文件名区别，继承于 ptsites.schema.site_base
NexusPHP子类: 签到页面: AttendanceHR, 答题: BakatestHR, 访问: VisitHR, 无HR版本：Attendance, Bakatest, Visit
"""
class MainClass(Unit3D):
URL = 'https://pt.abc.com'
"""
用户等级
name: uploaded, downloaded, share_ratio, points, days
需保证所有数组长度一致, 即使数值相同
value: [(保号), 毕业]
uploaded, downloaded 单位 byte
"""
USER_CLASSES = {
'uploaded': [10000000, 10000000], # 可选
'downloaded': [1099511627776, 16492674416640], # 可选
'share_ratio': [3.05, 4.55], # 可选
'points': [100, 200] # 可选
'days': [280, 700] # 可选
}
"""
构造flexget auto_sign_in 配置
"""
@classmethod
def build_sign_in_schema(cls):
return {
cls.get_module_name(): {
'type': 'object',
'properties': {
'cookie': {'type': 'string'},
'login': {
'type': 'object',
'properties': {
'username': {'type': 'string'},
'password': {'type': 'string'}
},
'additionalProperties': False
}
},
'additionalProperties': False
}
}
"""
构造flexget iyuu 配置
"""
@classmethod
def build_reseed_schema(cls):
return {
cls.get_module_name(): {
'type': 'object',
'properties': {
'rsskey': {'type': 'string'}
},
'additionalProperties': False
}
}
"""
构造辅种地址
"""
@classmethod
def build_reseed(cls, entry, config, site, passkey, torrent_id):
download_page = site['download_page'].format(torrent_id=torrent_id, rsskey=passkey['rsskey'])
entry['url'] = urljoin(MainClass.URL, download_page)
"""
不存在cookie时的登录流程，附加到正常签到之前
"""
def build_login_workflow(self, entry, config):
return [
Work(
url='/login',
method='get',
check_state=('network', NetworkState.SUCCEED),
),
Work(
url='/login',
# 调用的方法 sign_in_by_{method}
method='password',
check_state=('network', NetworkState.SUCCEED),
response_urls=['', '/pages/1'],
token_regex=r'(?<=name="_token" value=").+?(?=")',
captcha_regex=r'(?<=name="_captcha" value=").+?(?=")',
)
]
"""
签到流程
"""
def build_workflow(self, entry, config):
return [
Work(
# 请求的url
url='/',
# 调用的方法 sign_in_by_{method}
method='get',
# 成功判断正则
succeed_regex=('<a class="top-nav__username" href="https://pt.hdpost.top/users/(.*?)">', 1),
'''
判断请求执行的状况(网络错误会导致entry直接失败)
network: 网络
sign_in: 使用 succeed_regex 对响应做正则判断
final: 调用sign_in进行最终确认
sign_in 和 final 的区别在于 sign_in 在正则判断失败时 返回 SignState.NO_SIGN_IN，而 final 返回 SignState.SIGN_IN_FAILED
'''
check_state=('final', SignState.SUCCEED),
is_base_content=True,
# 跳转地址在在集合中时不会失败
response_urls=['', '/']
)
]
'''
自定义方法
每个 sign_in_by_{method} 方法都会接收4个参数
entry: entry
config: auto_sign_in 的总配置, 当前站点配置存在 entry['site_config'] 下
work: 当前调用的work
last_content: 上次请求返回的结果
返回请求的 response，供 check_state 检查
'''
def sign_in_by_password(self, entry, config, work, last_content):
login = entry['site_config'].get('login')
if not login:
entry.fail_with_prefix('Login data not found!')
return
login_page = get_soup(last_content)
hidden_input = login_page.select_one('#formContent > form > input[type=hidden]:nth-child(7)')
name = hidden_input.attrs['name']
value = hidden_input.attrs['value']
data = {
'_token': re.search(work.token_regex, last_content).group(),
'username': login['username'],
'password': login['password'],
'remember': 'on',
'_captcha': re.search(work.captcha_regex, last_content).group(),
'_username': '',
name: value,
}
login_response = self._request(entry, 'post', work.url, data=data)
login_network_state = self.check_network_state(entry, work, login_response)
if login_network_state != NetworkState.SUCCEED:
return
return login_response
'''
构造统计的选择器
'''
def build_selector(self):
selector = super(MainClass, self).build_selector()
NetUtils.dict_merge(selector, {
# 选择用户id
'user_id': '/users/(.*?)/',
# 数据源
'detail_sources': {
'default': {
# 不提取text，false时只会保留element的text
'do_not_strip': True,
'elements': {
# css选择器
'bar': 'ul.top-nav__ratio-bar',
'header': '.header',
'data_table': '.user-info'
}
}
},
'details': {
'uploaded': {
# 可使用元组指定提取的分组序号，使用str时，默认提取第一个
'regex': '上传.+?([\\d.]+ ?[ZEPTGMK]?iB)'
},
'downloaded': {
'regex': '下载.+?([\\d.]+ ?[ZEPTGMK]?iB)'
},
'share_ratio': {
'regex': '分享率.+?([\\d.]+)',
# 指定字符串的处理方法，以返回正确的数字或者日期格式
'handle': self.handle_share_ratio
},
'points': {
'regex': '魔力.+?(\\d[\\d,. ]*)',
'handle': self.handle_points
},
'join_date': {
'regex': '注册日期 (.*?\\d{4})',
'handle': self.handle_join_date
},
'seeding': {
'regex': '做种.+?(\\d+)'
},
'leeching': {
'regex': '吸血.+?(\\d+)'
},
'hr': {
'regex': '有效.+?(\\d+)'
}
}
})
return selector
def handle_share_ratio(self, value):
if value in ['.']:
return '0'
else:
return value