1 # irgramd: IRC-Telegram gateway
  2 # utils.py: Helper functions
  3 #
  4 # Copyright (c) 2019 Peter Bui <pbui@bx612.space>
  5 # Copyright (c) 2020-2026 E. Bosch <presidev@AT@gmail.com>
  6 #
  7 # Use of this source code is governed by a MIT style license that
  8 # can be found in the LICENSE file included in this project.
  9 
 10 import itertools
 11 import textwrap
 12 import re
 13 import datetime
 14 import zoneinfo
 15 import difflib
 16 import logging
 17 import random
 18 
 19 # Constants
 20 
 21 FILENAME_INVALID_CHARS = re.compile('[\0-\x1F/{}<>"\'\\|*&#%?\x7F]')
 22 SIMPLE_URL = re.compile('http(|s)://[^ ]+')
 23 
 24 from include import MAX_LINE
 25 
 26 # Utilities
 27 
 28 class command:
 29     async def parse_command(self, line, nick):
 30         command = line.partition(' ')[0].lower()
 31         self.tmp_ircnick = nick
 32         if command in self.commands.keys():
 33             handler, min_args, max_args, maxsplit = self.commands[command]
 34             words = line.split(maxsplit=maxsplit)[1:]
 35             num_words = len(words)
 36             if num_words < min_args or num_words > max_args:
 37                 reply = ('Wrong number of arguments',)
 38             else:
 39                 reply = await handler(*words)
 40         else:
 41             reply = ('Unknown command',)
 42 
 43         return reply
 44 
 45 class HELP:
 46     desc = 1
 47     brief = 2
 48 
 49 class LOGL:
 50     debug = False
 51 
 52 def chunks(iterable, n, fillvalue=None):
 53     ''' Return iterable consisting of a sequence of n-length chunks '''
 54     args = [iter(iterable)] * n
 55     return itertools.zip_longest(*args, fillvalue=fillvalue)
 56 
 57 def set_replace(set, item, new_item):
 58     if item in set:
 59         set.remove(item)
 60         set.add(new_item)
 61 
 62 def get_continued(items, mark, length):
 63     # Add "continued" mark to lines, except last one
 64     return (x + mark if n != length else x for n, x in enumerate(items, start=1))
 65 
 66 def split_lines(message):
 67     messages_limited = []
 68     wr = textwrap.TextWrapper(width=MAX_LINE)
 69 
 70     # Split when Telegram original message has breaks
 71     messages = message.splitlines()
 72     lm = len(messages)
 73     if lm > 1:
 74         # Add "continued line" mark (\) for lines that belong to the same message
 75         # (split previously)
 76         messages = get_continued(messages, ' \\', lm)
 77     for m in messages:
 78         wrapped = wr.wrap(text=m)
 79         lw = len(wrapped)
 80         if lw > 1:
 81             # Add double "continued line" mark (\\) for lines that belong to the same message
 82             # and have been wrapped to not exceed IRC limits
 83             messages_limited += get_continued(wrapped, ' \\\\', lw)
 84         else:
 85             messages_limited += wrapped
 86     del wr
 87     return messages_limited
 88 
 89 def sanitize_filename(fn):
 90     def hexize(m):
 91         return '-{:x}-'.format(ord(m.group(0)))
 92 
 93     new_fn = FILENAME_INVALID_CHARS.sub(hexize, fn)
 94     return new_fn.lstrip('-').replace(' ','_')
 95 
 96 def add_filename(filename, add):
 97     if add:
 98         aux = filename.rsplit('.', 1)
 99         name = aux[0]
100         last_dot = '.'
101         try:
102             ext = aux[1]
103         except:
104             ext = ''
105             last_dot = ''
106         return '{}-{}{}{}'.format(name, add, last_dot, ext)
107     else:
108         return filename
109 
110 def remove_slash(url):
111     return url[:-1] if url[-1:] == '/' else url
112 
113 def remove_http_s(url):
114     if url[:8] == 'https://':
115         surl = url[8:]
116     elif url[:7] == 'http://':
117         surl = url[7:]
118     else:
119         surl = url
120     return remove_slash(surl)
121 
122 def is_url_equiv(url1, url2):
123     if url1 and url2:
124         return url1 == url2 or remove_slash(remove_http_s(url1)) == remove_slash(remove_http_s(url2))
125     else:
126         return False
127 
128 def extract_url(text):
129     url = SIMPLE_URL.search(text)
130     return url.group() if url else None
131 
132 def get_human_size(size):
133     human_units = ('', 'K', 'M', 'G', 'T', 'P', 'E', 'Z', 'Y')
134 
135     def get_human_size_values(size, unit_pos=0):
136         aux = size / 1024.0
137         if aux > 1: return get_human_size_values(aux, unit_pos + 1)
138         else: return size, human_units[unit_pos]
139 
140     if size <= 1237940039285380274899124224:    # 1024Y
141         num, unit = get_human_size_values(size)
142     else:
143         num = size / 1208925819614629174706176  # 1Y
144         unit = 'Y'
145 
146     fs = '{:.1f}{}' if num < 10 else '{:.0f}{}'
147 
148     return fs.format(num, unit)
149 
150 def get_human_duration(duration):
151     res = ''
152     x, s = divmod(duration, 60)
153     h, m = divmod(x, 60)
154 
155     if h > 0: res = str(h) + 'h'
156     if m > 0: res += str(m) + 'm'
157     if s > 0 or duration < 60: res += str(s) + 's'
158     return res
159 
160 def compact_date(date, tz):
161     delta = current_date() - date
162     date_local = date.astimezone(zoneinfo.ZoneInfo(tz))
163 
164     if delta.days < 1:
165         compact_date = date_local.strftime('%H:%M')
166     elif delta.days < 365:
167         compact_date = date_local.strftime('%d-%b')
168     else:
169         compact_date = date_local.strftime('%Y')
170 
171     return compact_date
172 
173 def current_date():
174     return datetime.datetime.now(datetime.timezone.utc)
175 
176 def get_highlighted(a, b):
177     awl = len(a.split())
178     bwl = len(b.split())
179     delta_size = abs(awl - bwl)
180     highlighted = True
181 
182     if not a:
183         res = '> {}'.format(b)
184     elif delta_size > 5:
185         res = b
186         highlighted = False
187     else:
188         al = a.split(' ')
189         bl = b.split(' ')
190         diff = difflib.ndiff(al, bl)
191         ld = list(diff)
192         res = ''
193         d = ''
194         eq = 0
195 
196         for i in ld:
197             if i == '- ' or i[0] == '?':
198                 continue
199             elif i == '  ' or i == '+ ':
200                 res += ' '
201                 continue
202             # deletion of words
203             elif i[0] == '-':
204                 res += '-{}- '.format(i[2:])
205             # addition of words
206             elif i[0] == '+':
207                 res += '+{}+ '.format(i[2:])
208             else:
209                 res += '{} '.format(i[2:])
210                 eq += 1
211 
212         delta_eq = bwl - eq
213         if delta_eq > 3:
214             res = b
215             highlighted = False
216 
217     return res, highlighted
218 
219 def fix_braces(text):
220     # Remove braces not closed, if the text was truncated
221     if text.endswith(' {...'):
222         subtext = text[:-5]
223         if not '{}' in subtext:
224             return '{}...'.format(subtext)
225     return text
226 
227 def format_timestamp(format, tz, date):
228     date_local = date.astimezone(zoneinfo.ZoneInfo(tz))
229     return date_local.strftime(format)
230 
231 def parse_loglevel(level):
232     levelu = level.upper()
233     if levelu == 'DEBUG':
234         LOGL.debug = True
235     if levelu == 'NONE':
236         l = None
237     elif levelu in ('DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'):
238         l = getattr(logging, levelu)
239     else:
240         l = False
241     return l
242 
243 def pretty(object):
244     return object.stringify() if LOGL.debug and object else object
245 
246 class token:
247     def __init__(self, alpha):
248         self.alpha = alpha
249         self.long_alpha = len(alpha)
250 
251     def gen_token(self, long_token):
252         if long_token == 1:
253             return self.alpha[random.randrange(self.long_alpha)]
254         else:
255             aux = self.gen_token(long_token - 1)
256             return aux + self.alpha[random.randrange(self.long_alpha)]