1from telegram import Update, InlineKeyboardMarkup, InlineKeyboardButton
  2from telegram.constants import ParseMode
  3from Config import INLINE_SEP, FEEDBACK_TIMEOUT
  4import logging, re, json
  5from asyncio import sleep
  6logger = logging.getLogger(__name__)
  7
  8re_flags = re.I | re.M
  9
 10def get_human_readable(input_str: str, indicator: str, offset: int = 0) -> (int, int):
 11    try:
 12        result = int(input_str[offset:].split(indicator, maxsplit=1)[0])
 13        return result, len(str(result)) + len(indicator)
 14    except ValueError:
 15        return 0, 0
 16
 17youtube_timestamp_regex = re.compile(r"(?:&|\?)t=(\d*h?\d*m?\d*s?)", re_flags)
 18def youtube_timestamp(input_str: str) -> str:
 19    result = youtube_timestamp_regex.findall(input_str)
 20    try:
 21        input_string = result[0]
 22        seconds = int(input_string)
 23        hours, remainder = divmod(seconds, 3600)
 24        minutes, seconds = divmod(remainder, 60)
 25    except IndexError: # nothing to parse
 26        return ""
 27    except ValueError: # human-readable number
 28        hours, offset = get_human_readable(input_string, "h")
 29        minutes, offset = get_human_readable(input_string, "m", offset)
 30        seconds, offset = get_human_readable(input_string, "s", offset)
 31
 32    if hours == 0:
 33        return '{}:{:02}'.format(minutes, seconds)
 34    return '{}:{:02}:{:02}'.format(hours, minutes, seconds)
 35
 36replacers = [
 37    {
 38        "regex": re.compile(r"(?:(?:https?:)?\/\/)?(?:(?:www|m)\.)?(?:(?:youtube(?:-nocookie)?\.com|youtu.be))(?:\/(?:[\w\-]+\?v=|embed\/|live\/|v\/)?)([\w\-]+)(\S+)?", re_flags),
 39        "becomes": "https://y.outube.duckdns.org/{}",
 40        "timestamp": youtube_timestamp
 41    },
 42    {
 43        "regex": re.compile(r"(?:https?:\/\/)?(?:www\.)?twitter\.com\/(?:#!\/)?(.*)\/status(?:es)?\/([^\/\?\s]+)", re_flags),
 44        "becomes": "https://fxtwitter.com/{}/status/{}",
 45    },
 46    {
 47        "regex": re.compile(r"(?:https?:\/\/)?(?:www\.)?x\.com\/(?:#!\/)?(.*)\/status(?:es)?\/([^\/\?\s]+)", re_flags),
 48        "becomes": "https://fixupx.com/{}/status/{}",
 49    },
 50    {
 51        "regex": re.compile(r"(?:https?:\/\/)?(?:www\.)?instagram\.com\/((?:reel)|p)\/([A-Za-z0-9_]{11})[\/\?\w=&]*", re_flags),
 52        "becomes": "https://ddinstagram.com/{}/{}",
 53    },
 54]
 55
 56link_message = "Da {}[\.]({}) {}"
 57
 58def get_callback_data(feedback: bool) -> str:
 59    payload = { "feedback": feedback }
 60    return "feedback" + INLINE_SEP + json.dumps(payload)
 61
 62def get_message_markup() -> InlineKeyboardMarkup:
 63    buttons = [
 64        [
 65            InlineKeyboardButton(text="✅", callback_data=get_callback_data(True)),
 66            InlineKeyboardButton(text="❌", callback_data=get_callback_data(False)),
 67        ]
 68    ]
 69    return InlineKeyboardMarkup(buttons)
 70
 71def format_template(template: str, regex_result) -> str:
 72    result_type = type(regex_result)
 73    if result_type is str:
 74        return template.format(regex_result)
 75    elif result_type is tuple or result_type is list:
 76        return template.format(*regex_result)
 77    elif result_type is dict:
 78        return template.format(**regex_result)
 79    else:
 80        return ""
 81
 82def parse_text(message: str) -> list:
 83    output = []
 84    for site in replacers:
 85        regex = site["regex"]
 86        res = regex.findall(message)
 87        for r in res:
 88            link = format_template(site["becomes"], r)
 89            
 90            try:
 91                timestamp = site["timestamp"](r[-1])
 92            except KeyError:
 93                timestamp = None
 94
 95            output.append([link, timestamp])
 96    return output
 97
 98async def replace(update: Update, _) -> None:
 99    try:
100        links = parse_text(update.message.text)
101    except TypeError:
102        links = parse_text(update.message.caption)
103
104    for link in links:
105        logger.info(link)
106
107        user = update.effective_user.mention_markdown_v2(update.effective_user.name)
108        text = link_message.format(user, link[0], link[1])
109        message = await update.effective_chat.send_message(text, parse_mode=ParseMode.MARKDOWN_V2)
110        await sleep(FEEDBACK_TIMEOUT)
111        await message.edit_reply_markup(reply_markup=get_message_markup())
112
113async def feedback(update: Update, _, data_json: str) -> None:
114    data = json.loads(data_json)
115
116    if data["feedback"]:
117        await update.callback_query.answer("Bene!")
118        await update.effective_message.edit_reply_markup()
119        return
120
121    await update.callback_query.answer("Ci ho provato...")
122    await update.effective_message.delete()
123    return