#!/usr/bin/python # -*- coding: utf-8 -*- # --------------------------------------------------------------------------- # # # # Plugin for iSida Jabber Bot # # Copyright (C) diSabler # # Copyright (C) Vit@liy # # # # This program is free software: you can redistribute it and/or modify # # it under the terms of the GNU General Public License as published by # # the Free Software Foundation, either version 3 of the License, or # # (at your option) any later version. # # # # This program is distributed in the hope that it will be useful, # # but WITHOUT ANY WARRANTY; without even the implied warranty of # # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # # GNU General Public License for more details. # # # # You should have received a copy of the GNU General Public License # # along with this program. If not, see . # # # # --------------------------------------------------------------------------- # import urllib.parse import urllib.request import re import time import socket filename_chars_limit = 48 last_url_watch = {} url_watch_ignore = ['pdf','sig','spl','class','ps','torrent','dvi','gz','pac','swf','tar','tgz','tar','zip','mp3','m3u','wma', 'wax','ogg','wav','gif','jar','jpg','jpeg','png','xbm','xpm','xwd','css','asc','c','cpp','log','conf','text', 'txt','dtd','xml','mpeg','mpg','mov','qt','avi','asf','asx','wmv','bz2','tbz','tar','so','dll','exe','bin', 'img','usbimg','rar','deb','rpm','iso','ico','apk','patch','svg','7z','tcl'] # --------------------------------------------------------------------------- # # NETWORK HELPERS (Python 3) # # --------------------------------------------------------------------------- # def _fetch_url(url, limit=None): """Загружает URL и возвращает (content, headers, status).""" try: req = urllib.request.Request(url, headers={'User-Agent': GT('user_agent')}) with urllib.request.urlopen(req, timeout=GT('rss_get_timeout')) as resp: if limit: data = resp.read(limit) else: data = resp.read(GT('size_overflow')) headers = resp.headers status = resp.getcode() try: content = data.decode('utf-8', errors='replace') except: content = data.decode('latin-1', errors='replace') return content, headers, status except Exception as e: return None, None, None def _ensure_url(url): if not re.findall('^http(s?)://', url[:10]): return 'http://%s' % url return url # --------------------------------------------------------------------------- # # ISDOWN # # --------------------------------------------------------------------------- # def www_isdown(type, jid, nick, text): if not text: send_msg(type, jid, nick, L('What?', '%s/%s' % (jid, nick))) return url = _ensure_url(text.lower()) url = enidna(url) content, headers, status = _fetch_url(url) if content is not None and status and status < 400: msg = L('It\'s just you. %s is up.', '%s/%s' % (jid, nick)) % url else: msg = L('It\'s not just you! %s looks down from here.', '%s/%s' % (jid, nick)) % url send_msg(type, jid, nick, msg) # --------------------------------------------------------------------------- # # HEADER # # --------------------------------------------------------------------------- # def netheader(type, jid, nick, text): if not text: send_msg(type, jid, nick, L('What?', '%s/%s' % (jid, nick))) return try: regex = text.split('\n')[0].replace('*', '*?') url = text.split('\n')[1] except: regex = None url = text url = _ensure_url(url) url = enidna(url) content, headers, status = _fetch_url(url) if content is None: send_msg(type, jid, nick, L('Error fetching %s', '%s/%s' % (jid, nick)) % url) return output = '%s\n%s' % (url, '\n'.join(['%s: %s' % (k, v) for k, v in headers.items()])) if regex: try: matches = re.findall(regex, output, re.S | re.I | re.U) if matches: output = ''.join(matches[0]) else: output = L('RegExp not found!', '%s/%s' % (jid, nick)) except: output = L('Error in RegExp!', '%s/%s' % (jid, nick)) send_msg(type, jid, nick, deidna(output)) # --------------------------------------------------------------------------- # # WWW # # --------------------------------------------------------------------------- # def netwww(type, jid, nick, text): if not text: send_msg(type, jid, nick, L('What?', '%s/%s' % (jid, nick))) return regex = None n = 1 try: parts = text.split('\n') first_line = parts[0].strip() rest = '\n'.join(parts[1:]) tmp = first_line.split(' ', 1) if tmp[0].strip().isdigit(): n = int(tmp[0]) regex = tmp[1] if len(tmp) > 1 else None else: if rest: regex = first_line url = rest else: url = first_line except: regex = None n = 1 url = text if 'url' not in locals(): url = text url = _ensure_url(url) url = enidna(url) content, headers, status = _fetch_url(url, GT('size_overflow')) if content is None: send_msg(type, jid, nick, L('Error fetching %s', '%s/%s' % (jid, nick)) % url) return page = content if regex: try: matches = re.findall(regex, page, re.S | re.I | re.U) if matches: if n: output = unhtml_hard('\n'.join([''.join(m) for m in matches[:n]])) else: output = unhtml_hard('\n'.join([''.join(m) for m in matches])) else: output = L('RegExp not found!', '%s/%s' % (jid, nick)) except: output = L('Error in RegExp!', '%s/%s' % (jid, nick)) else: output = urllib.parse.unquote(unhtml_hard(page)) title = rss_replace(get_tag(page, 'title')) if title: output = '%s\n%s' % (title, output) send_msg(type, jid, nick, output[:msg_limit]) # --------------------------------------------------------------------------- # # URL TITLE / CONTENT LENGTH (оригинал) # # --------------------------------------------------------------------------- # def get_content_title(link): try: ll = link.lower() for t in url_watch_ignore: if ll.endswith('.%s' % t): raise Exception('Ignored extension') link = enidna(link) content, headers, status = _fetch_url(link, 16384) if content is None: return '' if '= len(text) / 3: text = remove_sub_space(html_encode(get_tag(content, tag)).replace('\n', ' ').replace('\r', ' ').replace('\t', ' ')) return text.strip() except: return '' def parse_url_in_message(room, jid, nick, type, text): global last_url_watch if type != 'groupchat' or text == 'None' or nick == '' or getRoom(jid) == getRoom(selfjid): return if get_level(room, nick)[0] < 4: return content_title = None if get_config(getRoom(room), 'store_users_url'): rjid = getRoom(jid) for t in text.split(): link = re.findall(r'(http[s]?://.*)', t) if link: link = link[0].split(' ')[0].split('"')[0].split('\'')[0] if not cur_execute_fetchone('select * from url where room=%s and jid=%s and url=%s', (room, rjid, link)): ttext = get_content_title(link) if ttext: ttext = to_censore(rss_del_html(rss_replace(ttext)), room) content_title = [link, ttext] else: is_file = False ll = link.lower() for ext in url_watch_ignore: if ll.endswith('.%s' % ext): is_file = True break if is_file: content, headers, status = _fetch_url(link) if content is not None: mt = float(headers.get('Content-Length', 0)) if mt: ttext = L('Content length %s', '%s/%s' % (jid, nick)) % get_size_human(mt) pprint('Store url: %s in %s/%s' % (link, room, nick), 'white') cur_execute('insert into url values (%s,%s,%s,%s,%s,%s);', (room, rjid, nick, int(time.time()), link, ttext)) was_shown = False if get_config(getRoom(room), 'url_title'): try: link = re.findall(r'(http[s]?://.*)', text)[0].split(' ')[0].split('"')[0].split('\'')[0] if link and last_url_watch.get(getRoom(room), '') != link and pasteurl not in link: if content_title and content_title[0] == link: ttext = content_title[1] else: ttext = get_content_title(link) if ttext: pprint('Show url-title: %s in %s' % (link, room), 'white') was_shown = True send_msg(type, room, '', L('Title: %s', '%s/%s' % (jid, nick)) % to_censore(rss_del_html(rss_replace(ttext)), room)) last_url_watch[getRoom(room)] = link except: pass if not was_shown and get_config(getRoom(room), 'content_length'): try: link = re.findall(u'(http[s]?://[-0-9a-zа-я.]+\/[-a-zа-я0-9._?#=@%/]+\.[a-z0-9]{2,7})', text, re.I + re.U + re.S)[0] if link and last_url_watch.get(getRoom(room), '') != link and pasteurl not in link: is_file = False ll = link.lower() for ext in url_watch_ignore: if ll.endswith('.%s' % ext): is_file = True break if is_file: last_url_watch[getRoom(room)] = enidna(link) content, headers, status = _fetch_url(last_url_watch[getRoom(room)]) pprint('Show content length: %s in %s' % (link, room), 'white') if content is not None: mt = float(headers.get('Content-Length', 0)) if mt: link_end = urllib.parse.unquote(last_url_watch[getRoom(room)].rsplit('/', 1)[-1]) link_end = u'…%s%s' % (['/', ''][len(link_end) > filename_chars_limit], link_end[-filename_chars_limit:]) send_msg(type, room, '', L('Length of %s is %s', '%s/%s' % (jid, nick)) % (to_censore(link_end, room), get_size_human(mt))) except: pass global execute message_act_control = [parse_url_in_message] execute = [ (3, 'www', netwww, 2, 'Show web page.\nwww [count (0 for all)] regexp\n[http://]url - page after regexp\nwww [http://]url - without html tags'), (3, 'header', netheader, 2, 'Show net header'), (3, 'isdown', www_isdown, 2, 'Check works site'), ]