Skip to content

Update kinozal.py #428

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zalessky wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

Update kinozal.py #428

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
249 changes: 182 additions & 67 deletions monitorrent/plugins/trackers/kinozal.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ class KinozalCredentials(Base):
password = Column(String, primary_key=True)
c_uid = Column(String, nullable=True)
c_pass = Column(String, nullable=True)
domain = Column(String, nullable=True, server_default='kinozal.tv')


class KinozalTopic(Topic):
Expand All @@ -38,14 +39,23 @@ class KinozalTopic(Topic):
}


# noinspection PyUnusedLocal
def get_current_version(engine):
m = MetaData(engine)
topics = Table(KinozalTopic.__tablename__, m, autoload=True)
creds = Table(KinozalCredentials.__tablename__, m, autoload=True)
if 'last_torrent_update' not in topics.columns:
return 0
if 'domain' not in creds.columns:
return 1
return 2


def upgrade(engine, operations_factory):
if not engine.dialect.has_table(engine.connect(), KinozalTopic.__tablename__):
return
version = get_current_version(engine)
if version == 0:
with operations_factory() as operations:
# remove capital later
operations.rename_table('Kinozal_topics', 'kinozal_topics1')
operations.rename_table('kinozal_topics1', KinozalTopic.__tablename__)

Expand All @@ -55,14 +65,11 @@ def upgrade(engine, operations_factory):
last_torrent_update = Column('last_torrent_update', UTCDateTime, nullable=True)
operations.add_column(KinozalTopic.__tablename__, last_torrent_update)
version = 1


def get_current_version(engine):
m = MetaData(engine)
topics = Table(KinozalTopic.__tablename__, m, autoload=True)
if 'last_torrent_update' not in topics.columns:
return 0
return 1
if version == 1:
with operations_factory() as operations:
domain = Column('domain', String, nullable=True, server_default='kinozal.tv')
operations.add_column(KinozalCredentials.__tablename__, domain)
version = 2


class KinozalDateParser(object):
Expand Down Expand Up @@ -124,18 +131,37 @@ def __init__(self, code, message):

class KinozalTracker(object):
tracker_settings = None
login_url = "https://kinozal.tv/takelogin.php"
profile_page = "https://kinozal.tv/inbox.php"
url_regex = re.compile(six.text_type(r'^https?://kinozal\.tv/details\.php\?id=(\d+)$'))
date_parser = KinozalDateParser()

url_regex = re.compile(six.text_type(r'^https?://([^/]*kinozal[^/]*)/details\.php\?id=(\d+)$'))

def __init__(self, c_uid=None, c_pass=None):
def __init__(self, c_uid=None, c_pass=None, domain='kinozal.tv'):
self.c_uid = c_uid
self.c_pass = c_pass
self.domain = domain or 'kinozal.tv'

def setup(self, c_uid, c_pass):
def setup(self, c_uid, c_pass, domain='kinozal.tv'):
self.c_uid = c_uid
self.c_pass = c_pass
self.domain = domain or 'kinozal.tv'

def _get_request_kwargs(self):
# Минимально необходимый набор для пробития простых защит
kwargs = self.tracker_settings.get_requests_kwargs() if self.tracker_settings else {}
headers = kwargs.get('headers', {})
if not headers:
headers = {}
headers['User-Agent'] = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36'
kwargs['headers'] = headers
return kwargs

@property
def login_url(self):
return "https://{}/takelogin.php".format(self.domain)

@property
def profile_page(self):
return "https://{}/inbox.php".format(self.domain)

def can_parse_url(self, url):
return self.url_regex.match(url) is not None
Expand All @@ -144,45 +170,70 @@ def parse_url(self, url):
match = self.url_regex.match(url)
if match is None:
return None

r = requests.get(url, allow_redirects=False, **self.tracker_settings.get_requests_kwargs())

soup = get_soup(r.text)
if soup.h1 is None:
# Kinozal doesn't return 404 for not existing topic
# it return regular page with text 'Тема не найдена'
# and we can check it by not existing heading of the requested topic

url_domain = match.group(1)
torrent_id = match.group(2)
real_url = "https://{}/details.php?id={}".format(url_domain, torrent_id)

try:
cookies = self.get_cookies() or {}
r = requests.get(real_url, allow_redirects=True, cookies=cookies, **self._get_request_kwargs())
r.encoding = 'windows-1251'
r.raise_for_status()

soup = get_soup(r.text)
if soup.h1 is None:
return None
title = soup.h1.text.strip()
return {'original_name': title}
except Exception:
return None
title = soup.h1.text.strip()

return {'original_name': title}

def login(self, username, password):
s = Session()
data = {"username": username, "password": password, 'returnto': ''}
login_result = s.post(self.login_url, data, **self.tracker_settings.get_requests_kwargs())
if login_result.url.startswith(self.login_url):
# TODO get error info (although it shouldn't contain anything useful
# it can contain request to enter capture, so we should handle it

try:
# Оригинальный метод логина через POST с разрешением на редирект
login_result = s.post(self.login_url, data, allow_redirects=True, **self._get_request_kwargs())
login_result.encoding = 'windows-1251'
except Exception:
raise KinozalLoginFailedException(3, "Connection failed")

# Если остался на странице логина или на странице ошибка - пароль неверный
if login_result.url.startswith(self.login_url) or u'ОШИБКА' in login_result.text:
raise KinozalLoginFailedException(1, "Invalid login or password")
else:
c_pass = s.cookies.get('pass')
c_uid = s.cookies.get('uid')
if not c_pass or not c_uid:
raise KinozalLoginFailedException(2, "Failed to retrieve cookie")

self.c_pass = c_pass
self.c_uid = c_uid
c_pass = s.cookies.get('pass')
c_uid = s.cookies.get('uid')

# Исправление ложной авторизации: отсеиваем куки "deleted"
if not c_pass or not c_uid or c_pass == 'deleted' or c_uid == 'deleted':
raise KinozalLoginFailedException(2, "Failed to retrieve cookie")

self.c_pass = c_pass
self.c_uid = c_uid

def verify(self):
if not self.c_uid:
if not self.c_uid or not self.c_pass or self.c_uid == 'deleted' or self.c_pass == 'deleted':
return False

cookies = self.get_cookies()
if not cookies:
return False
profile_page_result = requests.get(self.profile_page, cookies=cookies,
**self.tracker_settings.get_requests_kwargs())
return profile_page_result.url == self.profile_page

try:
profile_page_result = requests.get(self.profile_page, cookies=cookies, allow_redirects=True,
**self._get_request_kwargs())
profile_page_result.encoding = 'windows-1251'

# Проверка контента вместо старой проверки URL
uid_str = six.text_type(self.c_uid)
if (u'userdetails.php?id=' + uid_str) in profile_page_result.text or u'Выход' in profile_page_result.text:
return True
return False
except Exception:
return False

def get_cookies(self):
if not self.c_pass or not self.c_uid:
Expand All @@ -193,45 +244,67 @@ def get_id(self, url):
match = self.url_regex.match(url)
if match is None:
return None

return match.group(1)
return match.group(2)

def get_last_torrent_update(self, url):
response = requests.get(url, **self.tracker_settings.get_requests_kwargs())
response.raise_for_status()

soup = get_soup(response.text)
content = soup.find("div", {"class": "mn1_menu"})
text_element = content.find(lambda tag: (tag.name == 'li') and (u'Обновлен' in tag.contents))
date_text = None
if text_element is not None:
text_element = text_element.find("span")
if text_element is not None:
date_text = text_element.string
if date_text is None:
text_element = content.find(lambda tag: (tag.name == 'li') and (u'Залит' in tag.contents))
match = self.url_regex.match(url)
if match is None:
return None

url_domain = match.group(1)
torrent_id = match.group(2)
real_url = "https://{}/details.php?id={}".format(url_domain, torrent_id)

try:
cookies = self.get_cookies() or {}
response = requests.get(real_url, allow_redirects=True, cookies=cookies, **self._get_request_kwargs())
response.encoding = 'windows-1251'
response.raise_for_status()

soup = get_soup(response.text)
content = soup.find("div", {"class": "mn1_menu"})
if content is None:
return None

text_element = content.find(lambda tag: (tag.name == 'li') and (u'Обновлен' in tag.contents))
date_text = None
if text_element is not None:
text_element = text_element.find("span")
if text_element is not None:
date_text = text_element.string
if date_text is None:
if date_text is None:
text_element = content.find(lambda tag: (tag.name == 'li') and (u'Залит' in tag.contents))
if text_element is not None:
text_element = text_element.find("span")
if text_element is not None:
date_text = text_element.string
if date_text is None:
return None

parsed_datetime = self.date_parser.parse(date_text)
return parsed_datetime.astimezone(pytz.utc)
except Exception:
return None

parsed_datetime = self.date_parser.parse(date_text)
return parsed_datetime.astimezone(pytz.utc)

def get_download_url(self, url):
torrent_id = self.get_id(url)
if torrent_id is None:
match = self.url_regex.match(url)
if match is None:
return None

return "https://dl.kinozal.tv/download.php?id=" + torrent_id

url_domain = match.group(1)
torrent_id = match.group(2)

return "https://{}/download.php?id={}".format(url_domain, torrent_id)


class KinozalPlugin(WithCredentialsMixin, ExecuteWithHashChangeMixin, TrackerPluginBase):
tracker = KinozalTracker()
topic_class = KinozalTopic
credentials_class = KinozalCredentials

credentials_public_fields = ['username', 'domain']
credentials_private_fields = ['username', 'password', 'domain']

topic_form = [{
'type': 'row',
'content': [{
Expand All @@ -242,28 +315,60 @@ class KinozalPlugin(WithCredentialsMixin, ExecuteWithHashChangeMixin, TrackerPlu
}]
}]

credentials_form = [{
'type': 'row',
'content': [{
'type': 'text',
'model': 'username',
'label': 'Логин',
'flex': 50
}, {
'type': 'password',
'model': 'password',
'label': 'Пароль',
'flex': 50
}]
}, {
'type': 'row',
'content': [{
'type': 'text',
'model': 'domain',
'label': 'Домен (варианты: kinozal.tv, kinozal.guru, kinozal.me)',
'flex': 100
}]
}]

def _get_domain(self):
with DBSession() as db:
cred = db.query(self.credentials_class).first()
if cred and cred.domain:
return cred.domain
return 'kinozal.tv'

def login(self):
with DBSession() as db:
cred = db.query(self.credentials_class).first()
if not cred:
return LoginResult.CredentialsNotSpecified
username = cred.username
password = cred.password
domain = cred.domain or 'kinozal.tv'
if not username or not password:
return LoginResult.CredentialsNotSpecified
try:
self.tracker.setup(None, None, domain)
self.tracker.login(username, password)
with DBSession() as db:
cred = db.query(self.credentials_class).first()
cred.c_uid = self.tracker.c_uid
cred.c_pass = self.tracker.c_pass
cred.domain = domain
return LoginResult.Ok
except KinozalLoginFailedException as e:
if e.code == 1:
return LoginResult.IncorrentLoginPassword
return LoginResult.Unknown
except Exception as e:
# TODO: Log unexpected excepton
return LoginResult.Unknown

def verify(self):
Expand All @@ -273,18 +378,26 @@ def verify(self):
return False
username = cred.username
password = cred.password
domain = cred.domain or 'kinozal.tv'
if not username or not password or not cred.c_uid or not cred.c_pass:
return False
self.tracker.setup(cred.c_uid, cred.c_pass)
self.tracker.setup(cred.c_uid, cred.c_pass, domain)
return self.tracker.verify()

def can_parse_url(self, url):
return self.tracker.can_parse_url(url)

def parse_url(self, url):
self.tracker.domain = self._get_domain()
with DBSession() as db:
cred = db.query(self.credentials_class).first()
if cred and cred.c_uid and cred.c_pass:
self.tracker.setup(cred.c_uid, cred.c_pass, cred.domain)

return self.tracker.parse_url(url)

def check_changes(self, topic):
self.tracker.domain = self._get_domain()
last_torrent_update = self.tracker.get_last_torrent_update(topic.url)
topic_last_torrent_update = topic.last_torrent_update
min_date = pytz.utc.localize(datetime.datetime.min)
Expand All @@ -296,7 +409,9 @@ def check_changes(self, topic):
return False

def _prepare_request(self, topic):
headers = {'referer': topic.url}
self.tracker.domain = self._get_domain()
headers = self.tracker._get_request_kwargs().get('headers', {})
headers['referer'] = topic.url
cookies = self.tracker.get_cookies()
request = requests.Request('GET', self.tracker.get_download_url(topic.url), headers=headers, cookies=cookies)
return request.prepare()
Expand Down