Browse Source

v0.0.1

master v0.0.1
cofob 3 months ago
parent
commit
7366f0e239
Signed by: cofob GPG Key ID: 702FAC7C5F43C2E6
  1. 5
      .gitignore
  2. 64
      main.py
  3. 3
      register_masto.py
  4. 2
      requirements.txt

5
.gitignore

@ -1,5 +1,10 @@
.gitignore
.idea
*.tgs
*.png
*.jpg
*.jpeg
*.mp4
venv
*.secret
*.session

64
main.py

@ -13,38 +13,78 @@ def save_config(j):
json.dump(j, file)
def is_bullshit(text: str):
text = text.lower()
bullshit = ['otus.pw', 't.me', 'selectel', 'slc.tl', '#промо', '#реклама', 'sber', 'beclick.cc', '#текстприслан',
'bit.ly', 'clck.ru', 'clc.to']
if 1 in [(1 if text.find(i) >= 0 else 0) for i in bullshit]:
return True
return False
def unpack_links(text: str):
text = text.replace('](', ' (').replace('[', '')
return text
async def post(message: Message, account: str, channel: str):
mastodon = Mastodon(
access_token=f"{account}-account.secret", api_base_url=settings.HOMESERVER
)
print(message.stringify())
media = await client.download_media(message)
posted_media = mastodon.media_post(media)
mastodon.status_post(
status=message.text
+ f"\n\nCrostooted from https://t.me/{channel}/{str(message.id)}",
media_ids=[posted_media["id"]],
)
os.remove(media)
media_ids = []
if message.media is not None:
try:
print('Downloading media from telegram...')
media = await client.download_media(message)
if media.find('.tgs') >= 0:
print('Media is sticker, which unsupported in mastodon. Skipping.')
else:
print('Uploading media to mastodon...')
posted_media = mastodon.media_post(media)
media_ids.append(posted_media['id'])
except Exception as e:
print(e)
text = unpack_links(message.text)
if text == '' and media_ids == []:
print('Toot is empty. Skipping.')
else:
mastodon.status_post(
status=text
+ f"\n\nCrostooted from https://t.me/{channel}/{str(message.id)}",
media_ids=media_ids,
)
print('Toot!')
if message.media is not None:
try:
os.remove(media)
except Exception as e:
print(e)
async def main(client: TelegramClient):
with open("subscriptions.secret", "r") as file:
j = json.load(file)
print('Getting dialogs...')
for i in await client.get_dialogs():
i: Dialog
if isinstance(i.entity, Channel):
obj = j.get(i.entity.username.lower())
if obj is not None:
print(f'Processing channel {i.entity.username}...')
messages: [Message] = await client.get_messages(i, 10)
messages = messages[::-1]
for msg in messages:
print(f'Processing message {msg.id}...')
if msg.id not in obj["viewed"]:
await post(msg, obj["name"], i.entity.username)
if is_bullshit(msg.text) is False:
await post(msg, obj["name"], i.entity.username)
else:
print('Is bullshit. Skipping.')
obj["viewed"].append(msg.id)
j[i.entity.username.lower()] = obj
save_config(j)
print(i.message)
else:
print('Already viewed. Skipping.')
client = TelegramClient(
"tg2masto", api_id=settings.API_ID, api_hash=settings.API_HASH

3
register_masto.py

@ -3,5 +3,6 @@ import settings
Mastodon.create_app(
"telegram2masto", api_base_url=settings.HOMESERVER, to_file="app.secret"
"telegram2masto", api_base_url=settings.HOMESERVER, to_file="app.secret",
website='https://git.cofob.ru/cofob/Telegram2Masto'
)

2
requirements.txt

@ -1,2 +1,2 @@
telethon
telethon~=1.23.0
Mastodon.py
Loading…
Cancel
Save