94 lines
3.2 KiB
Python
Executable File
94 lines
3.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import logging
|
|
import yaml
|
|
import os
|
|
import re
|
|
|
|
from asyncio import sleep
|
|
from argparse import ArgumentParser
|
|
from telethon import TelegramClient
|
|
from telethon.extensions.html import (
|
|
unparse as entities_to_html,
|
|
parse as html_to_entities
|
|
)
|
|
from telethon.tl.patched import Message
|
|
from telethon.tl.types import InputPeerChannel, InputMediaWebPage
|
|
from telethon.tl.functions.messages import EditMessageRequest, SendScheduledMessagesRequest, GetScheduledHistoryRequest
|
|
|
|
# Set up logging
|
|
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
|
|
|
|
CONFIG_PATH = os.path.expanduser('~/.config/tgschedpub/config.yaml')
|
|
DOT_IMAGE_PATTERN = r'\.\<\/a\>$'
|
|
SESSION_FILE = os.path.join(os.path.dirname(CONFIG_PATH), 'tgschedpub.session')
|
|
client: TelegramClient
|
|
|
|
|
|
def load_config():
|
|
with open(CONFIG_PATH, 'r') as file:
|
|
return yaml.safe_load(file)
|
|
|
|
|
|
async def main(mode, channel_id=None):
|
|
channel = await client.get_entity(channel_id)
|
|
messages = reversed((await client(GetScheduledHistoryRequest(peer=channel.id, hash=0))).messages)
|
|
input_peer = InputPeerChannel(channel.id, channel.access_hash)
|
|
send_date = None
|
|
|
|
for m in messages:
|
|
if not isinstance(m, Message):
|
|
continue
|
|
|
|
if mode == 'send':
|
|
if send_date is None or m.date.date() == send_date:
|
|
send_date = m.date.date()
|
|
await client(SendScheduledMessagesRequest(
|
|
peer=input_peer,
|
|
id=[m.id]
|
|
))
|
|
continue
|
|
else:
|
|
break
|
|
|
|
elif mode == 'images':
|
|
html = entities_to_html(m.message, m.entities)
|
|
if re.search(DOT_IMAGE_PATTERN, html):
|
|
html = re.sub(DOT_IMAGE_PATTERN, '‌</a>', html)
|
|
text, entities = html_to_entities(html)
|
|
await client(EditMessageRequest(
|
|
peer=input_peer,
|
|
id=m.id,
|
|
media=InputMediaWebPage(
|
|
url=m.media.webpage.url,
|
|
force_large_media=False,
|
|
force_small_media=False
|
|
),
|
|
reply_markup=m.reply_markup,
|
|
schedule_date=m.date,
|
|
invert_media=m.invert_media,
|
|
message=text,
|
|
entities=entities
|
|
))
|
|
logging.info(f'Message {m.id} seems to be edited')
|
|
|
|
await sleep(2)
|
|
else:
|
|
logging.warning(f'Message {m.id}: could not find link with a dot!')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = ArgumentParser()
|
|
parser.add_argument('--channel-id', type=str)
|
|
parser.add_argument('mode', choices=['auth', 'images', 'send'])
|
|
args = parser.parse_args()
|
|
|
|
config = load_config()
|
|
client = TelegramClient(SESSION_FILE, config['api_id'], config['api_hash'])
|
|
|
|
if args.mode == 'auth':
|
|
client.start(phone=config['phone'], password=config.get('password'))
|
|
logging.info('Authenticated successfully.')
|
|
else:
|
|
client.start()
|
|
client.loop.run_until_complete(main(args.mode, args.channel_id))
|