Rotate the avatar of Telegram based on the time in a day using Pyrogram(MTProto).

README.md

• ./avatar.png should be the targeted avatar file.
• Setup Pyrogram (MTProto) at first: https://docs.pyrogram.ml/start/ProjectSetup.

requirements.txt

Pillow==5.1.0
pyaes==1.6.1
Pyrogram==0.7.4
PySocks==1.6.8
pytz==2018.4
TgCrypto==1.0.4

telegram_live_avatar.py

#!/usr/bin/env python3
import sys
from datetime import datetime
from time import sleep
from pytz import timezone
import json
from os.path import exists as file_exists
from os import remove as remove_file
from contextlib import contextmanager
from PIL import Image
from pyrogram import Client
from pyrogram.api.functions.photos import UploadProfilePhoto, DeletePhotos
from pyrogram.api.types import InputPhoto
from pyrogram.api.types.photos.photo import Photo

AVATAR_FILE = "avatar.png"
TEMP_FILE = ".temp.png"
CACHE_FILE = ".live_avatar.cache"

TZ_CST = timezone("Asia/Taipei")


def print_running_info(app):
    me = app.get_me()
    name = me.first_name + " " + me.last_name
    username = me.username
    print("Live Avatar is up and running...",
          f"for {name} (@{username})", sep="\n")


@contextmanager
def upload_avatar(app, file_path):
    # deprecated because an unnecessary update is not avoidable
    photo = app.send(UploadProfilePhoto(app.save_file(file_path)))
    print(photo)
    yield photo
    id = photo['photo']['id']
    access_hash = photo['photo']['access_hash']
    r = app.send(DeletePhotos([InputPhoto(id, access_hash)]))
    print(r)


def angle_now():
    seconds_today = (datetime.now(tz=TZ_CST) - datetime.now(tz=TZ_CST).replace(hour=0,
                                                                               minute=0, second=0, microsecond=0)).total_seconds()
    percent = (seconds_today % (12 * 60 * 60)) / (12 * 60 * 60)  # [0, 12)
    return percent * 360


def delete_photo(app, id, access_hash):
    return app.send(DeletePhotos([InputPhoto(id, access_hash)]))


def clean_cache(app):
    if file_exists(CACHE_FILE):
        with open(CACHE_FILE) as f:
            cache = json.loads(f.read())
        id = cache['id']
        access_hash = cache['access_hash']
        print(f"Trying to delete the old avatar (id: {id})...")
        print("Result:", delete_photo(app, id, access_hash))
        remove_file(CACHE_FILE)


def run(app):
    while True:
        uploaded = False
        try:
            try:
                avt = Image.open(AVATAR_FILE)
                angle = angle_now()
                ravt = avt.rotate(-angle)
                ravt.save(TEMP_FILE)
                del avt
                del ravt

                print(f"Uploading new avatar rotated by {angle} degree...")
                photo = app.send(UploadProfilePhoto(app.save_file(TEMP_FILE)))
                if isinstance(photo, Photo):
                    id = photo['photo']['id']
                    access_hash = photo['photo']['access_hash']
                    print(
                        f"Successfully uploaded (id: {id}, access_hash:{access_hash}).")
                    clean_cache(app)
                    uploaded = True
                    with open(CACHE_FILE, "w") as f:
                        f.write(json.dumps({
                            'id': photo['photo']['id'],
                            'access_hash': photo['photo']['access_hash']
                        }))
                else:
                    print(
                        f"Failed to upload new avatar. Got: {photo}", file=sys.stderr)
            except Exception as e:
                print(f"ERROR:", e, file=sys.stderr)
            sleep(((12 * 60 * 60) / 360) * 10)
        except KeyboardInterrupt:
            if uploaded:
                print(f"Trying to delete uploaded avatar (id: {id})...")
                print("Result:", delete_photo(app, id, access_hash))
                try:
                    remove_file(CACHE_FILE)
                except FileNotFoundError:
                    pass
            raise


def main():
    app = Client("live_avatar", workers=1)
    app.start()
    print_running_info(app)
    run(app)


if __name__ == "__main__":
    main()