mirror of
https://github.com/sossuOrkesteri/sossunVerkkosivut.git
synced 2026-10-12 19:23:37 +03:00
62 lines
2.5 KiB
Python
62 lines
2.5 KiB
Python
import json, time, requests, re, os
|
|
|
|
secrets = {
|
|
"user_id": os.getenv("USER_ID"),
|
|
"access_token": os.getenv("ACCESS_TOKEN")
|
|
}
|
|
|
|
def read_data_from_file(filename):
|
|
with open(filename, "r") as file:
|
|
return json.loads(file.read())
|
|
|
|
def write_data_to_file(filename, data):
|
|
with open(filename, "w") as file:
|
|
file.write(json.dumps(data, indent=4))
|
|
|
|
def call(url, payload):
|
|
""" Calls the instagram Basic Display API with the given url and payload.
|
|
Returns the response body as a dict.
|
|
Raises HTTPError if the HTTP response code is 4XX or 5XX. """
|
|
# append access token to payload
|
|
payload["access_token"] = secrets["access_token"]
|
|
response = requests.get("https://graph.instagram.com" + url, params=payload, timeout=1)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def refresh_instagram_access_token():
|
|
""" Refreshes the long-lived access token of the instagram account. The access token
|
|
will expire in 60 days if not refreshed. A refresh is possible when at least 24 hours
|
|
have passed since the previous one. """
|
|
response_body = call("/refresh_access_token", {"grant_type": "ig_refresh_token"})
|
|
expiration_date = time.time() + response_body["expires_in"]
|
|
expiration_date = time.strftime("%Y.%m.%d %H:%M", time.gmtime(expiration_date))
|
|
print(f"Instagram access token refreshed. Expires on {expiration_date}")
|
|
|
|
def format_media(media):
|
|
""" Writes the media returned by the IG API in the desired format. """
|
|
for entry in media:
|
|
# date format
|
|
timestamp = time.strptime(entry["timestamp"], "%Y-%m-%dT%H:%M:%S%z")
|
|
entry["timestamp"] = time.strftime("%d.%m.%Y", timestamp)
|
|
|
|
# remove hashtags from the ends of captions
|
|
entry["caption"] = re.sub("(\s?#\S+)*$", "", entry["caption"])
|
|
# replace @ signs in email addesses to avoid spam
|
|
entry["caption"] = re.sub("@gmail.com", "\[at\]gmail.com", entry["caption"])
|
|
return media
|
|
|
|
def fetch_instagram_media():
|
|
""" Fetches the most recent media of the instagram user. """
|
|
print("Fetching instagram media.")
|
|
fields = ",".join([
|
|
"caption", "media_type", "media_url", "permalink",
|
|
"thumbnail_url", "timestamp", "children"
|
|
])
|
|
response_body = call("/{:s}/media".format(secrets["user_id"]), {"fields": fields})
|
|
return format_media(response_body["data"])
|
|
|
|
if __name__ == "__main__":
|
|
refresh_instagram_access_token()
|
|
media = fetch_instagram_media()
|
|
write_data_to_file("instagram_media.json", media)
|