70 lines
1.8 KiB
Python
70 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pandas as pd
|
|
import requests
|
|
from retry import retry
|
|
|
|
from image_download.classes import Instagram
|
|
|
|
|
|
def get_sources() -> pd.DataFrame:
|
|
"""Get sources dateframe."""
|
|
input_path = Path(__file__).parent / 'sources.csv'
|
|
assert input_path.exists()
|
|
# load in csv
|
|
df = pd.read_csv(input_path)
|
|
return df
|
|
|
|
|
|
@retry(
|
|
exceptions=AssertionError,
|
|
tries=3,
|
|
delay=5,
|
|
backoff=1.2,
|
|
jitter=(1, 10),
|
|
)
|
|
def download_images(ig_inst: Instagram, filename: str, url: str) -> None:
|
|
"""
|
|
Download images from instagram post found on url
|
|
|
|
to files with base name from filename.
|
|
"""
|
|
# get image urls
|
|
url_list = ig_inst.get_image_url_list(url)
|
|
assert len(url_list) > 0
|
|
# generate filenames
|
|
chars = 'abcdefghij'
|
|
filename_list = [name + chars[i] for i in range(len(url_list))]
|
|
# download images
|
|
download_dir = Path(__file__).parent / 'imgs'
|
|
download_dir.mkdir(exist_ok=True)
|
|
for filename, url in zip(filename_list, url_list):
|
|
output_path = download_dir / f"{filename}.jpg"
|
|
response = requests.get(url, stream=True)
|
|
with open(output_path, 'wb') as fh:
|
|
shutil.copyfileobj(response.raw, fh)
|
|
logging.info('downloaded %s', output_path)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# setup logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
# get sources df
|
|
df = get_sources()
|
|
# setup Instagram interface
|
|
igi = Instagram()
|
|
igi.setup()
|
|
# begin downloading
|
|
for idx in range(len(df)):
|
|
# get info from source dataframe
|
|
name = df.loc[idx, 'name']
|
|
url = df.loc[idx, 'url']
|
|
# download post images
|
|
download_images(ig_inst=igi, filename=name, url=url)
|
|
# clean up
|
|
igi.close()
|