feat(scripts): 增加 Instagram 同步至 mastodon 脚本

增加 Instagram 同步至 mastodon 脚本

Signed-off-by: Ching <loooching@gmail.com>

develop_requirements.txt

@@ -32,3 +32,4 @@ wcwidth==0.2.5
 zipp==3.5.0
 redis==4.1.0
 black
+instagram_private_api

dsite/settings_default.py

@@ -142,3 +142,8 @@ CORS_ALLOW_CREDENTIALS = True  # 允许携带cookie
 
 LARK_WEBHOOK_URL = ''
 LARK_WEBHOOK_SECRET = ''
+
+MASTODON_SYNCED_IMAGES_LOG = ''
+IG_PRIVATE_API_SETTINGS = ''
+IG_LOGIN_USERNAME = ''
+IG_LOGIN_PASSWORD = ''

scripts/ins2mastodon.py

@@ -13,13 +13,16 @@ get_wsgi_application()
 
 import pickle
 from urllib.parse import urlparse
+import time
 
 from mastodon import Mastodon
 from instagram_private_api import Client
 import requests
+import logging
 
 from dsite import settings
 
+logger = logging.getLogger('dsite.' + __name__)
 mastodon_cli = Mastodon(access_token='Ug_bUMWCk3RLamOnqYIytmeB0nO6aNfjdmf06mAj2bE', api_base_url='https://nofan.xyz')
 
 
@@ -34,6 +37,7 @@ def send_image_to_mastodon(image_url, text):
     if toot_resp.get('id'):
         media_ids = [toot_resp['id']]
         mastodon_cli.status_post(text, media_ids=media_ids)
+        print(text)
 
 
 # write binary file with api.settings
@@ -51,23 +55,28 @@ def readSettings(settings_file):
     return cache
 
 
-if not os.path.exists(settings.IG_PRIVATE_API_SETTINGS):
-    writeSettings(settings.IG_LOGIN_USERNAME, settings.IG_LOGIN_PASSWORD, settings_file)
+if __name__ == '__main__':
 
-cache_settings = readSettings(settings.IG_PRIVATE_API_SETTINGS)
-api = Client(settings.IG_LOGIN_USERNAME, settings.IG_LOGIN_PASSWORD, settings=cache_settings)
+    if not os.path.exists(settings.IG_PRIVATE_API_SETTINGS):
+        writeSettings(settings.IG_LOGIN_USERNAME, settings.IG_LOGIN_PASSWORD, settings.IG_PRIVATE_API_SETTINGS)
 
-results = api.self_feed()
-for item in results['items']:
-    text = item['caption']['text']
-    image_url = item['image_versions2']['candidates'][0]['url']
-    try:
-        with open(settings.MASTODON_SYNCED_IMAGES_LOG, 'r') as f:
-            send_images = f.readlines()
-            send_images = [x.strip() for x in send_images]
-        if image_url not in send_images:
-            send_image_to_mastodon(image_url, text)
-            with open(settings.MASTODON_SYNCED_IMAGES_LOG, 'a') as f:
-                f.write(image_url + '\n')
-    except:
-        print('error')
+    cache_settings = readSettings(settings.IG_PRIVATE_API_SETTINGS)
+    api = Client(settings.IG_LOGIN_USERNAME, settings.IG_LOGIN_PASSWORD, settings=cache_settings)
+
+    while True:
+        results = api.self_feed()
+        logger.info('getting %s posts', len(results['items']))
+        for item in results['items']:
+            text = item['caption']['text']
+            image_url = item['image_versions2']['candidates'][0]['url']
+            try:
+                with open(settings.MASTODON_SYNCED_IMAGES_LOG, 'r') as f:
+                    send_images = f.readlines()
+                    send_images = [x.strip() for x in send_images]
+                if image_url not in send_images:
+                    send_image_to_mastodon(image_url, text)
+                    with open(settings.MASTODON_SYNCED_IMAGES_LOG, 'a') as f:
+                        f.write(image_url + '\n')
+            except Exception as e:
+                logger.error(e)
+        time.sleep(60)