diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..475a30e --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,32 @@ +name: CI + +on: + push: + branches: [main, modernize-packaging] + pull_request: + branches: [main] + +jobs: + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - run: pip install ruff + - run: ruff check src/ tests/ + - run: ruff format --check src/ tests/ + + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.10", "3.11", "3.12", "3.13"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - run: pip install -e ".[dev]" + - run: pytest diff --git a/.gitignore b/.gitignore index 6fbcfa7..6da5489 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,77 @@ -.Python +# Byte-compiled / optimized / DLL files __pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +*.egg + +# Virtual environments +.venv/ +venv/ +ENV/ +env/ bin/ include/ -lib/ + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt +pip-selfcheck.json + +# Unit test / coverage +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ + +# mypy / pyright +.mypy_cache/ +.pyright/ + +# ruff +.ruff_cache/ + +# Environments +*.env +.env + +# IDE +.idea/ +.vscode/ +*.swp +*.swo +*~ + +# OS +.DS_Store +Thumbs.db + +# Project-specific +hashtags/ +*.session diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..f77a9d5 --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,7 @@ +repos: + - repo: https://github.com/astral-sh/ruff-pre-commit + rev: v0.8.6 + hooks: + - id: ruff + args: [--fix] + - id: ruff-format diff --git a/README.md b/README.md index f5f979f..37c9180 100644 --- a/README.md +++ b/README.md @@ -1,35 +1,143 @@ # Instagram Hashtag Crawler -[![HitCount](http://hits.dwyl.io/simonseo/instagram-hashtag-crawler.svg)](http://hits.dwyl.io/simonseo/instagram-hashtag-crawler) -This crawler was made because most of the crawlers out there seems to either require a browser or a developer account. This Instagram crawler utilizes a private API of Instagram and thus no developer account is required. +Crawl Instagram hashtags and collect post metadata (likes, comments, captions, user profiles) without a developer account. -Refer to a similar script I wrote. It might be more helpful in terms of documentation: [simonseo/instacrawler-privateapi](https://github.com/simonseo/instagram-hashtag-crawler) +Uses [instaloader](https://instaloader.github.io/) under the hood. ## Installation -First install [Instagram Private API](https://github.com/ping/instagram_private_api). Kudos for a great project! + +```bash +pip install . +``` + +With browser cookie support (auto-extract session from Chrome, Firefox, etc.): + +```bash +pip install ".[browser]" ``` -$ pip install git+https://github.com/ping/instagram_private_api.git + +For development: + +```bash +pip install -e ".[dev,browser]" ``` -Now run `__init__.py`. It'll provide you with the command options. If this shows up, everything probably works +## Usage + +### Crawl hashtags + +```bash +# Using browser cookies (recommended — auto-extracts session from your browser) +instagram-hashtag-crawler --browser chrome -t foodporn + +# If logged in on a non-default Chrome profile, specify the cookie file +instagram-hashtag-crawler --browser chrome \ + --cookie-file ~/Library/Application\ Support/Google/Chrome/Profile\ 1/Cookies \ + -t foodporn + +# Using username/password +instagram-hashtag-crawler -u YOUR_USERNAME -p YOUR_PASSWORD -t foodporn + +# Multiple hashtags from a file +instagram-hashtag-crawler --browser chrome -f targets.txt + +# With options +instagram-hashtag-crawler --browser chrome -t foodporn \ + --max-posts 500 \ + --output-dir ./data \ + -v ``` -$ python __init__.py -usage: __init__.py [-h] -u USERNAME -p PASSWORD [-f TARGETFILE] [-t TARGET] + +### Multi-hashtag AND search + +Pass `-t` multiple times to find posts that contain **all** specified hashtags: + +```bash +# Posts tagged with BOTH #foodporn AND #pizza +instagram-hashtag-crawler --browser chrome -t foodporn -t pizza + +# Three-way AND +instagram-hashtag-crawler --browser chrome -t food -t pizza -t italy ``` -## Get Crawlin' -To get crawlin', you need to provide your Instagram username and password, and either an Instagram Hashtag without the hash (target) or a text file of the hashtags in each row (targetfile). -Wait a bit and a folder will be made with all the hashtags crawled. +Output is saved as `food_AND_pizza.json` (tags sorted alphabetically, joined by `_AND_`). + +You can also run it as a module: -## Options -Inside `__init__.py`, there is a config dictionary. Each config option is explained in the comments. -Note that `min_collect_media` and `max_collect_media` is trumped if `min_timestamp` is provided as a number. +```bash +python -m instagram_hashtag_crawler --browser chrome -t foodporn ``` -config = { - 'profile_path' : './hashtags', # Path where output data gets saved - 'min_collect_media' : 1, # how many media items to be collected per hashtag. If time is specified, this is ignored - 'max_collect_media' : 2000, # how many media items to be collected per hashtag. If time is specified, this is ignored - # 'min_timestamp' : int(time() - 60*60*24*30*2) # up to how recent you want the posts to be in seconds. If you do not want to use this, put None as value - 'min_timestamp' : None -} + +### Export to CSV + +```bash +instagram-hashtag-export --json-dir ./hashtags --csv-dir ./output ``` + +### Options + +| Flag | Description | Default | +|------|-------------|---------| +| `--browser` | Auto-extract session from browser (chrome, firefox, safari, edge, brave, etc.) | — | +| `--cookie-file` | Path to browser cookie file (for non-default profiles) | — | +| `-u`, `--username` | Instagram username (not needed with `--browser`) | — | +| `-p`, `--password` | Instagram password (not needed with `--browser`) | — | +| `-t`, `--target` | Hashtag to crawl (without `#`). Repeat for AND search. | — | +| `-f`, `--targetfile` | File with hashtags, one per line | — | +| `--output-dir` | Directory for JSON output | `./hashtags` | +| `--max-posts` | Max posts per hashtag | `100` | +| `--min-posts` | Min posts required | `1` | +| `--since` | Unix timestamp — only collect newer posts | — | +| `--session-file` | Path to save/load session (with `-u`/`-p`) | — | +| `-v`, `--verbose` | Debug logging | off | + +### Target file format + +One hashtag per line, no `#` prefix: + +``` +delicious +dish +foodpornography +``` + +See [`examples/targets.txt`](examples/targets.txt) for a sample. + +## Output + +Each hashtag produces a JSON file in the output directory: + +``` +hashtags/ + delicious.json + dish.json + food_AND_pizza.json # multi-hashtag AND result +``` + +Each JSON file contains an array of post objects with fields like `shortcode`, `user_id`, `username`, `like_count`, `comment_count`, `caption`, `tags`, `pic_url`, `date`, and profile metadata. + +## Development + +```bash +# Install dev dependencies +pip install -e ".[dev,browser]" + +# Lint +ruff check src/ tests/ +ruff format --check src/ tests/ + +# Test +pytest + +# Pre-commit hooks +pre-commit install +``` + +## Requirements + +- Python 3.10+ +- An Instagram account (no developer/API access needed) + +## License + +MIT diff --git a/__init__.py b/__init__.py deleted file mode 100644 index a0dd25b..0000000 --- a/__init__.py +++ /dev/null @@ -1,75 +0,0 @@ -#!/usr/bin/python -# -*- coding: utf-8 -*- -# @File Name: __init__.py -# @Created: 2017-05-21 12:30:29 seo (simon.seo@nyu.edu) -# @Updated: 2017-05-21 13:41:42 Simon Seo (simon.seo@nyu.edu) - -import json -import os.path -import argparse -import multiprocessing as mp -import csv -from time import time -from collections import deque -from util import file_to_list -from crawler import crawl -import pathlib -try: - from instagram_private_api import ( - Client, __version__ as client_version) -except ImportError: - import sys - sys.path.append(os.path.join(os.path.dirname(__file__), '..')) - from instagram_private_api import ( - Client, __version__ as client_version) - - -if __name__ == '__main__': - # Example command: - # python examples/savesettings_logincallback.py -u "yyy" -p "zzz" -target "names.txt" - parser = argparse.ArgumentParser(description='Crawling') - parser.add_argument('-u', '--username', dest='username', type=str, required=True) - parser.add_argument('-p', '--password', dest='password', type=str, required=True) - parser.add_argument('-f', '--targetfile', dest='targetfile', type=str, required=False) - parser.add_argument('-t', '--target', dest='target', type=str, required=False) - - args = parser.parse_args() - config = { - 'search_algorithm' : 'BFS', # Possible values: BFS, DFS - 'profile_path' : './hashtags', # Path where output data gets saved - 'min_collect_media' : 1, # how many media items to be collected per person/hashtag. If time is specified, this is ignored - 'max_collect_media' : 100, # how many media items to be collected per person/hashtag. If time is specified, this is ignored - # 'min_timestamp' : int(time() - 60*60*24*30*2) # up to how recent you want the posts to be in seconds. If you do not want to use this, put None as value - 'min_timestamp' : None - } - - try: - if args.target: - origin_names = [args.target] - elif args.targetfile: - origin_names = file_to_list(args.targetfile) - else: - raise Exception('No crawl target given. Provide a hashtag with -t option or file of hashtags with -f') - print('Client version: %s' % client_version) - print(origin_names) - api = Client(args.username, args.password) - except Exception as e: - raise Exception("Unable to initiate API:", e) - else: - print("Initiating API") - - try: - # jobs = [] - pathlib.Path(config['profile_path']).mkdir(parents=True, exist_ok=True) - for origin in origin_names: - crawl(api, origin, config) - # p = mp.Process(target=crawl, args=(api, origin, config)) - # jobs.append(p) - # p.start() - except KeyboardInterrupt: - print('Jobs terminated') - except Exception as e: - raise e - # for p in jobs: - # p.join() - diff --git a/crawler.py b/crawler.py deleted file mode 100644 index c5b8cae..0000000 --- a/crawler.py +++ /dev/null @@ -1,129 +0,0 @@ -import json -import os -from collections import deque -from re import findall -from time import time, sleep -from util import randselect, byteify, file_to_list -import csv - -def crawl(api, hashtag, config): - # print('Crawling started at origin hashtag', origin['user']['username'], 'with ID', origin['user']['pk']) - if visit_profile(api, hashtag, config): - pass - -def visit_profile(api, hashtag, config): - while True: - try: - - processed_tagfeed = { - 'posts' : [] - } - feed = get_posts(api, hashtag, config) - with open(config['profile_path'] + os.sep + str(hashtag) + '_rawfeed.json', 'w') as outfile: - json.dump(feed, outfile, indent=2) - profile_dic = {} - posts = [beautify_post(api, post, profile_dic) for post in feed] - posts = list(filter(lambda x: not x is None, posts)) - if len(posts) < config['min_collect_media']: - return False - else: - processed_tagfeed['posts'] = posts[:config['max_collect_media']] - - try: - if not os.path.exists(config['profile_path'] + os.sep): os.makedirs(config['profile_path']) - except Exception as e: - print('exception in profile path') - raise e - - try: - with open(config['profile_path'] + os.sep + str(hashtag) + '.json', 'w') as outfile: - json.dump(processed_tagfeed, outfile, indent=2) - except Exception as e: - print('exception while dumping') - raise e - except Exception as e: - print('exception while visiting profile', e) - if str(e) == '-': - raise e - return False - else: - return True - -def beautify_post(api, post, profile_dic): - try: - if post['media_type'] != 1: # If post is not a single image media - return None - keys = post.keys() - # print(post) - user_id = post['user']['pk'] - profile = profile_dic.get(user_id, False) - while True: - try: - sleep(0.05) - if not profile: - profile = api.user_info(user_id) - profile_dic[user_id] = profile - except Exception as e: - # print(post) - print('exception in getting user_info from {} {}'.format(user_id, post['user']['username']), e) - sleep(5) - # raise e - else: - break - # profile = api.username_info('simon_oncepiglet') - # print(profile) - # print('Visiting:', profile['user']['username']) - processed_media = { - 'user_id' : user_id, - 'username' : profile['user']['username'], - 'full_name' : profile['user']['full_name'], - 'profile_pic_url' : profile['user']['profile_pic_url'], - 'media_count' : profile['user']['media_count'], - 'follower_count' : profile['user']['follower_count'], - 'following_count' : profile['user']['following_count'], - 'date' : post['taken_at'], - 'pic_url' : post['image_versions2']['candidates'][0]['url'], - 'like_count' : post['like_count'] if 'like_count' in keys else 0, - 'comment_count' : post['comment_count'] if 'comment_count' in keys else 0, - 'caption' : post['caption']['text'] if 'caption' in keys and post['caption'] is not None else '' - } - processed_media['tags'] = findall(r'#[^#\s]*', processed_media['caption']) - # print(processed_media['tags']) - return processed_media - except Exception as e: - print('exception in beautify post') - raise e - -def get_posts(api, hashtag, config): - try: - feed = [] - try: - uuid = api.generate_uuid(return_hex=False, seed='0') - results = api.feed_tag(hashtag, rank_token=uuid, min_timestamp=config['min_timestamp']) - except Exception as e: - print('exception while getting feed1') - raise e - feed.extend(results.get('items', [])) - - if config['min_timestamp'] is not None: return feed - - next_max_id = results.get('next_max_id') - while next_max_id and len(feed) < config['max_collect_media']: - print("next_max_id", next_max_id, "len(feed) < max_collect_media", len(feed) < config['max_collect_media'] , len(feed)) - try: - results = api.feed_tag(hashtag, rank_token=uuid, max_id=next_max_id) - except Exception as e: - print('exception while getting feed2') - if str(e) == 'Bad Request: Please wait a few minutes before you try again.': - sleep(60) - else: - raise e - feed.extend(results.get('items', [])) - next_max_id = results.get('next_max_id') - - return feed - - except Exception as e: - print('exception while getting posts') - raise e - diff --git a/targets.txt b/examples/targets.txt similarity index 100% rename from targets.txt rename to examples/targets.txt diff --git a/pip-selfcheck.json b/pip-selfcheck.json deleted file mode 100644 index a68e0eb..0000000 --- a/pip-selfcheck.json +++ /dev/null @@ -1 +0,0 @@ -{"last_check":"2018-05-21T10:12:49Z","pypi_version":"10.0.1"} \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..6ea8420 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,45 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "instagram-hashtag-crawler" +version = "2.0.0" +description = "Instagram hashtag crawler using instaloader" +readme = "README.md" +requires-python = ">=3.10" +license = "MIT" +authors = [ + { name = "Simon Seo", email = "simonseo.doubles@gmail.com" }, +] +dependencies = [ + "instaloader>=4.10", +] + +[project.optional-dependencies] +dev = [ + "pytest>=7.0", + "ruff>=0.4", + "pre-commit>=3.0", +] +browser = [ + "browser_cookie3>=0.19", +] + +[project.scripts] +instagram-hashtag-crawler = "instagram_hashtag_crawler.cli:main" +instagram-hashtag-export = "instagram_hashtag_crawler.export:main" + +[tool.ruff] +target-version = "py310" +line-length = 100 +src = ["src"] + +[tool.ruff.lint] +select = ["E", "W", "F", "I", "UP", "B", "SIM", "N"] + +[tool.ruff.lint.isort] +known-first-party = ["instagram_hashtag_crawler"] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/read_json.py b/read_json.py deleted file mode 100644 index c13bd81..0000000 --- a/read_json.py +++ /dev/null @@ -1,68 +0,0 @@ -#!/usr/bin/python -# -*- coding: utf-8 -*- -# @File Name: read_json.py -# @Created: 2017-05-21 13:02:49 seo (simon.seo@nyu.edu) -# @Updated: 2017-05-22 13:52:54 Simon Seo (simon.seo@nyu.edu) - -# -*- coding: utf-8 -*- -""" -Created on Tue Apr 25 12:31:38 2017 - -@author: song-isong-i -""" - -import json -import os -import unicodecsv as csv - -#read all the json files in the folder and save the data sorted by posts into csv -def read_profiles(json_dir, csv_dir, output_file_name='posts.csv'): - print('reading profiles...') - with open(os.path.join(csv_dir, output_file_name), "wb") as o_posts: - writer = csv.writer(o_posts, lineterminator='\n') - if not os.path.exists(json_dir): - raise Exception('Please provide directory of profile JSONs') - for f in os.listdir(json_dir): - if f != '.DS_Store': - with open(json_dir+f) as json_data: - d = json.load(json_data) - sort_by_posts(d, writer) - -def sort_by_posts(dic, writer): - posts = dic['posts'] - threshold = 60*60*24 - max_date = 0 #most recent - - #don't save if no post - if len(posts) == 0: return - - for p in posts: - date = p['date'] #9 - if date > max_date: - max_date = date - max_threshold_date = max_date - threshold - print("max_date {}, threshold {}, max_threshold_date {}".format(max_date, threshold, max_threshold_date)) - for p in posts: - date = p['date'] #9 - if date > max_threshold_date: - continue - - post = [] - username = p['username'] #2 - user_id = p['user_id'] #3 - full_name = p['full_name'] #4 - profile_pic_url = p['profile_pic_url'] #5 - media_count = p['media_count'] #6 - follower_count = p['follower_count'] #7 - pic_url = str(p['pic_url']) #0 - like_count = p['like_count'] #1 - comment_count = p['comment_count'] #8 - date = p['date'] #9 - caption = str(p['caption']) #10 - tags = p['tags'] #11 - post = [pic_url, like_count, username, user_id, full_name, profile_pic_url, media_count, follower_count, comment_count, date, caption, tags] - writer.writerow(post) - - - - diff --git a/requirements.txt b/requirements.txt deleted file mode 100644 index 81cb401..0000000 --- a/requirements.txt +++ /dev/null @@ -1 +0,0 @@ -instagram-private-api==1.5.5 diff --git a/src/instagram_hashtag_crawler/__init__.py b/src/instagram_hashtag_crawler/__init__.py new file mode 100644 index 0000000..c55b6e0 --- /dev/null +++ b/src/instagram_hashtag_crawler/__init__.py @@ -0,0 +1,3 @@ +"""Instagram hashtag crawler using instaloader.""" + +__version__ = "2.0.0" diff --git a/src/instagram_hashtag_crawler/__main__.py b/src/instagram_hashtag_crawler/__main__.py new file mode 100644 index 0000000..130cda5 --- /dev/null +++ b/src/instagram_hashtag_crawler/__main__.py @@ -0,0 +1,4 @@ +from instagram_hashtag_crawler.cli import main + +if __name__ == "__main__": + main() diff --git a/src/instagram_hashtag_crawler/browser_session.py b/src/instagram_hashtag_crawler/browser_session.py new file mode 100644 index 0000000..c8877fd --- /dev/null +++ b/src/instagram_hashtag_crawler/browser_session.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from http.cookiejar import CookieJar + + import instaloader + +logger = logging.getLogger(__name__) + +SUPPORTED_BROWSERS = ( + "chrome", + "firefox", + "safari", + "edge", + "opera", + "brave", + "chromium", + "vivaldi", +) + +REQUIRED_COOKIES = ("sessionid", "csrftoken", "ds_user_id") + + +def _get_browser_cookies( + browser: str, + cookie_file: str | None = None, +) -> CookieJar: + """Extract Instagram cookies from the specified browser. + + Parameters + ---------- + browser: + Browser name (e.g. ``"chrome"``). + cookie_file: + Optional path to the browser's cookie database file. Useful + when the browser has multiple profiles (e.g. Chrome's + ``~/Library/Application Support/Google/Chrome/Profile 1/Cookies``). + + Returns an ``http.cookiejar.CookieJar`` containing cookies for + ``.instagram.com``. Raises ``RuntimeError`` if *browser_cookie3* + is not installed or the browser is unsupported. + """ + try: + import browser_cookie3 # noqa: PLC0415 + except ImportError as exc: + msg = ( + "browser_cookie3 is required for --browser. " + "Install it with: pip install instagram-hashtag-crawler[browser]" + ) + raise RuntimeError(msg) from exc + + browser_lower = browser.lower() + if browser_lower not in SUPPORTED_BROWSERS: + msg = f"Unsupported browser {browser!r}. Choose from: {', '.join(SUPPORTED_BROWSERS)}" + raise ValueError(msg) + + fn = getattr(browser_cookie3, browser_lower) + kwargs: dict[str, str] = {"domain_name": ".instagram.com"} + if cookie_file is not None: + kwargs["cookie_file"] = cookie_file + + try: + return fn(**kwargs) + except PermissionError as exc: + msg = ( + f"Cannot access {browser} cookies — permission denied. " + "On macOS, you may need to grant Full Disk Access to your terminal." + ) + raise RuntimeError(msg) from exc + + +def _cookiejar_to_dict(cj: CookieJar) -> dict[str, str]: + """Convert a CookieJar to a plain dict of {name: value}.""" + return {cookie.name: cookie.value for cookie in cj if cookie.value is not None} + + +def load_browser_session( + loader: instaloader.Instaloader, + browser: str, + cookie_file: str | None = None, +) -> str: + """Load an Instagram session from browser cookies into *loader*. + + Extracts cookies from the specified browser, validates that the + required Instagram cookies are present, and loads them into the + instaloader session. + + Parameters + ---------- + loader: + An initialised ``Instaloader`` instance. + browser: + Browser name (e.g. ``"chrome"``). + cookie_file: + Optional path to the browser's cookie database file. Pass this + when the logged-in session lives in a non-default profile. + + Returns the Instagram username (``ds_user_id`` is present but + ``username`` is not stored in cookies — we return the user ID + as a fallback and let the caller verify with ``test_login``). + + Raises + ------ + RuntimeError + If browser_cookie3 is not installed, the browser is inaccessible, + or the required Instagram cookies are missing. + """ + cj = _get_browser_cookies(browser, cookie_file=cookie_file) + cookie_dict = _cookiejar_to_dict(cj) + + logger.debug("Extracted cookies from %s: %s", browser, sorted(cookie_dict.keys())) + + missing = [name for name in REQUIRED_COOKIES if name not in cookie_dict] + if missing: + msg = ( + f"Missing required Instagram cookies from {browser}: {', '.join(missing)}. " + "Make sure you are logged into instagram.com in your browser." + ) + raise RuntimeError(msg) + + # load_session expects a plain dict and sets up the full requests.Session + # internally — including headers and CSRF token. + user_id = cookie_dict["ds_user_id"] + loader.context.load_session(user_id, cookie_dict) + + logger.info("Loaded Instagram session from %s (user_id=%s)", browser, user_id) + return user_id diff --git a/src/instagram_hashtag_crawler/cli.py b/src/instagram_hashtag_crawler/cli.py new file mode 100644 index 0000000..19e4fd2 --- /dev/null +++ b/src/instagram_hashtag_crawler/cli.py @@ -0,0 +1,272 @@ +from __future__ import annotations + +import argparse +import logging +import sys +from pathlib import Path + +import instaloader + +from instagram_hashtag_crawler.crawler import CrawlConfig, crawl, crawl_multi_and +from instagram_hashtag_crawler.utils import file_to_list + +logger = logging.getLogger(__name__) + + +def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + prog="instagram-hashtag-crawler", + description="Crawl Instagram hashtags and collect post metadata.", + ) + + # Authentication — either --browser or -u/-p + auth = parser.add_argument_group("authentication") + auth.add_argument( + "-u", + "--username", + default=None, + help="Instagram username (not needed with --browser)", + ) + auth.add_argument( + "-p", + "--password", + default=None, + help="Instagram password (not needed with --browser)", + ) + auth.add_argument( + "--browser", + default=None, + metavar="NAME", + help=( + "Auto-extract Instagram session from a browser. " + "Supported: chrome, firefox, safari, edge, opera, brave, chromium, vivaldi. " + "Requires pip install instagram-hashtag-crawler[browser]" + ), + ) + auth.add_argument( + "--cookie-file", + default=None, + help=( + "Path to browser cookie database file. Use with --browser when the " + "logged-in session is in a non-default profile (e.g. Chrome Profile 1)." + ), + ) + + # Targets + parser.add_argument( + "-t", + "--target", + action="append", + default=None, + dest="targets", + help=( + "Hashtag to crawl (without #). Can be specified multiple times " + "for AND search: -t foodporn -t pizza finds posts with BOTH tags." + ), + ) + parser.add_argument( + "-f", + "--targetfile", + help="Path to file with hashtags (one per line) — crawls each independently", + ) + parser.add_argument( + "--output-dir", + default="./hashtags", + help="Directory for output data (default: ./hashtags)", + ) + parser.add_argument( + "--max-posts", + type=int, + default=100, + help="Maximum posts to collect per hashtag (default: 100)", + ) + parser.add_argument( + "--min-posts", + type=int, + default=1, + help="Minimum posts required per hashtag (default: 1)", + ) + parser.add_argument( + "--since", + type=int, + default=None, + help="Unix timestamp — only collect posts newer than this", + ) + parser.add_argument( + "--session-file", + default=None, + help="Path to save/load login session file", + ) + parser.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging") + + args = parser.parse_args(argv) + + # Validate: need either --browser or both -u and -p + if args.browser is None and (args.username is None or args.password is None): + parser.error("Provide --browser, or both -u/--username and -p/--password") + + return args + + +def _setup_logging(verbose: bool) -> None: + level = logging.DEBUG if verbose else logging.INFO + logging.basicConfig( + level=level, + format="%(asctime)s %(levelname)s [%(name)s] %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + +def _login( + loader: instaloader.Instaloader, + username: str, + password: str, + session_file: str | None, +) -> None: + """Attempt session restore, fall back to username/password login.""" + if session_file: + try: + loader.load_session_from_file(username, filename=session_file) + if loader.test_login() == username: + logger.info("Restored session from %s", session_file) + return + except FileNotFoundError: + logger.debug("No session file at %s, will login fresh", session_file) + + # Try default session location + try: + loader.load_session_from_file(username) + if loader.test_login() == username: + logger.info("Restored session for %s", username) + return + except FileNotFoundError: + logger.debug("No default session file found") + + logger.info("Logging in as %s", username) + loader.login(username, password) + + # Save session for next time + if session_file: + loader.save_session_to_file(filename=session_file) + else: + loader.save_session_to_file() + logger.info("Session saved") + + +def _login_browser( + loader: instaloader.Instaloader, + browser: str, + cookie_file: str | None = None, +) -> None: + from instagram_hashtag_crawler.browser_session import load_browser_session + + user_id = load_browser_session(loader, browser, cookie_file=cookie_file) + logged_in = loader.test_login() + if logged_in: + logger.info("Browser session verified — logged in as %s", logged_in) + else: + logger.warning( + "Browser session loaded (user_id=%s) but test_login did not confirm. " + "The session may still work for hashtag queries.", + user_id, + ) + + +def main(argv: list[str] | None = None) -> None: + args = _parse_args(argv) + _setup_logging(args.verbose) + + # Resolve targets + multi_and = False + if args.targets: + hashtags = args.targets + if len(hashtags) > 1: + multi_and = True + elif args.targetfile: + hashtags = file_to_list(args.targetfile) + else: + logger.error("Provide a hashtag with -t or a file of hashtags with -f") + sys.exit(1) + + if multi_and: + logger.info("AND search for: %s", " + ".join(f"#{h}" for h in hashtags)) + else: + logger.info("Targets: %s", hashtags) + + # Initialize instaloader and login + loader = instaloader.Instaloader( + download_pictures=False, + download_videos=False, + download_video_thumbnails=False, + download_geotags=False, + download_comments=False, + save_metadata=False, + compress_json=False, + ) + + try: + if args.browser: + _login_browser(loader, args.browser, cookie_file=args.cookie_file) + else: + _login(loader, args.username, args.password, args.session_file) + except instaloader.InvalidArgumentException as exc: + logger.error("Login failed: %s", exc) + sys.exit(1) + except instaloader.TwoFactorAuthRequiredException: + logger.error( + "Two-factor auth required. Use --browser or --session-file " + "with a pre-authenticated session." + ) + sys.exit(1) + except RuntimeError as exc: + logger.error("%s", exc) + sys.exit(1) + + # Build config + from datetime import datetime, timezone + + min_ts = None + if args.since is not None: + min_ts = datetime.fromtimestamp(args.since, tz=timezone.utc) + + config = CrawlConfig( + output_dir=Path(args.output_dir), + min_posts=args.min_posts, + max_posts=args.max_posts, + min_timestamp=min_ts, + ) + + config.output_dir.mkdir(parents=True, exist_ok=True) + + # Multi-tag AND search + if multi_and: + try: + success = crawl_multi_and(loader, hashtags, config) + if success: + logger.info( + "Finished AND search for %s", + " + ".join(f"#{h}" for h in hashtags), + ) + else: + logger.warning("Insufficient posts matching all tags") + except KeyboardInterrupt: + logger.info("Interrupted by user") + sys.exit(130) + except instaloader.QueryReturnedNotFoundException as exc: + logger.warning("Hashtag not found: %s", exc) + return + + # Single-tag or file-based independent crawls + for hashtag in hashtags: + logger.info("Crawling #%s", hashtag) + try: + success = crawl(loader, hashtag, config) + if success: + logger.info("Finished #%s", hashtag) + else: + logger.warning("Insufficient posts for #%s", hashtag) + except KeyboardInterrupt: + logger.info("Interrupted by user") + sys.exit(130) + except instaloader.QueryReturnedNotFoundException: + logger.warning("Hashtag #%s not found, skipping", hashtag) diff --git a/src/instagram_hashtag_crawler/crawler.py b/src/instagram_hashtag_crawler/crawler.py new file mode 100644 index 0000000..ed72da4 --- /dev/null +++ b/src/instagram_hashtag_crawler/crawler.py @@ -0,0 +1,250 @@ +from __future__ import annotations + +import dataclasses +import json +import logging +from datetime import datetime +from pathlib import Path +from time import sleep +from typing import Any + +import instaloader +from instaloader import Hashtag, Post, Profile + +logger = logging.getLogger(__name__) + + +@dataclasses.dataclass +class CrawlConfig: + """Configuration for a hashtag crawl.""" + + output_dir: Path + min_posts: int = 1 + max_posts: int = 100 + min_timestamp: datetime | None = None + + +def _collect_posts( + loader: instaloader.Instaloader, + hashtag: str, + config: CrawlConfig, + profile_cache: dict[int, Profile] | None = None, + *, + required_tags: frozenset[str] | None = None, +) -> list[dict[str, Any]]: + """Collect posts from a single hashtag, returning them as a list. + + If *required_tags* is given, only posts whose caption contains **all** + of the specified hashtags (case-insensitive, without ``#``) are kept. + This enables efficient AND filtering: query one hashtag via the API + and check the caption for the remaining tags. + + Posts are deduplicated by shortcode within a single call. + """ + if profile_cache is None: + profile_cache = {} + + hashtag_obj = Hashtag.from_name(loader.context, hashtag) + logger.info("Hashtag #%s has %d total posts", hashtag, hashtag_obj.mediacount) + + posts: list[dict[str, Any]] = [] + seen_shortcodes: set[str] = set() + skipped = 0 + + for post in hashtag_obj.get_posts_resumable(): + if len(posts) >= config.max_posts: + break + + # Skip if older than min_timestamp + if config.min_timestamp and post.date_utc < config.min_timestamp: + if config.min_timestamp is not None: + # When filtering by time, stop iterating once we hit old posts + # (posts are returned newest-first) + break + continue + + # Only collect single-image posts + if post.typename != "GraphImage": + skipped += 1 + continue + + # Deduplicate by shortcode + if post.shortcode in seen_shortcodes: + continue + seen_shortcodes.add(post.shortcode) + + # AND filter: check caption contains all required tags + if required_tags is not None: + caption_tags = frozenset(post.caption_hashtags) # lowercase, no # + if not required_tags <= caption_tags: + skipped += 1 + continue + + processed = _process_post(loader, post, profile_cache) + if processed is not None: + posts.append(processed) + if len(posts) % 10 == 0: + logger.info("Collected %d posts so far...", len(posts)) + + logger.info( + "Collected %d posts for #%s (skipped %d)", + len(posts), + hashtag, + skipped, + ) + return posts + + +def crawl( + loader: instaloader.Instaloader, + hashtag: str, + config: CrawlConfig, +) -> bool: + """Crawl a single hashtag and save results as JSON. + + Returns True if enough posts were collected, False otherwise. + """ + posts = _collect_posts(loader, hashtag, config) + + if len(posts) < config.min_posts: + return False + + _save_posts(posts, config.output_dir / f"{hashtag}.json") + return True + + +def crawl_multi_and( + loader: instaloader.Instaloader, + hashtags: list[str], + config: CrawlConfig, +) -> bool: + """Crawl posts that contain ALL given hashtags (AND logic). + + Strategy: query each hashtag via the API and keep only posts whose + caption contains **every** tag in *hashtags*. We start with all + hashtags because Instagram's API returns different post sets for + each hashtag — a post tagged with both ``#food`` and ``#pizza`` + may appear in the ``#pizza`` feed but be buried far down in the + ``#food`` feed. By querying each hashtag and filtering the caption, + we maximise the chance of finding matching posts. + + Duplicate posts (same shortcode) across queries are merged so the + final output contains unique posts only. + + Returns True if at least *config.min_posts* were found. + """ + if len(hashtags) < 2: + msg = "crawl_multi_and requires at least 2 hashtags" + raise ValueError(msg) + + required_tags = frozenset(tag.lower() for tag in hashtags) + profile_cache: dict[int, Profile] = {} + merged: dict[str, dict[str, Any]] = {} + + for hashtag in hashtags: + logger.info("AND search: querying #%s (require all of %s)", hashtag, sorted(required_tags)) + posts = _collect_posts( + loader, + hashtag, + config, + profile_cache, + required_tags=required_tags, + ) + for post in posts: + merged.setdefault(post["shortcode"], post) + + if len(merged) >= config.max_posts: + break + + all_posts = list(merged.values())[: config.max_posts] + + logger.info( + "AND search for %s: found %d unique posts", + " + ".join(f"#{h}" for h in hashtags), + len(all_posts), + ) + + if len(all_posts) < config.min_posts: + return False + + filename = "_AND_".join(sorted(hashtags)) + ".json" + _save_posts(all_posts, config.output_dir / filename) + return True + + +def _save_posts(posts: list[dict[str, Any]], output_file: Path) -> None: + """Write posts to a JSON file.""" + output = {"posts": posts} + output_file.write_text(json.dumps(output, indent=2, default=str)) + logger.info("Saved %d posts to %s", len(posts), output_file) + + +def _process_post( + loader: instaloader.Instaloader, + post: Post, + profile_cache: dict[int, Profile], +) -> dict[str, Any] | None: + """Extract metadata from a single post. + + Returns a dict of post data, or None on failure. + """ + try: + profile = _get_profile(loader, post, profile_cache) + + return { + "shortcode": post.shortcode, + "user_id": post.owner_id, + "username": profile.username, + "full_name": profile.full_name, + "profile_pic_url": profile.profile_pic_url, + "media_count": profile.mediacount, + "follower_count": profile.followers, + "following_count": profile.followees, + "date": int(post.date_utc.timestamp()), + "pic_url": post.url, + "like_count": post.likes, + "comment_count": post.comments, + "caption": post.caption or "", + "tags": [f"#{tag}" for tag in post.caption_hashtags], + } + except instaloader.QueryReturnedNotFoundException: + logger.warning("Post %s or its owner no longer exists", post.shortcode) + return None + except instaloader.ConnectionException as exc: + logger.warning("Connection error processing post %s: %s", post.shortcode, exc) + return None + + +def _get_profile( + loader: instaloader.Instaloader, + post: Post, + cache: dict[int, Profile], +) -> Profile: + """Fetch owner profile with caching and retry.""" + owner_id = post.owner_id + + if owner_id in cache: + return cache[owner_id] + + max_retries = 3 + for attempt in range(max_retries): + try: + sleep(0.05) # Small delay to avoid rate limiting + profile = post.owner_profile + cache[owner_id] = profile + return profile + except instaloader.ConnectionException as exc: + if attempt < max_retries - 1: + wait = 5 * (attempt + 1) + logger.warning( + "Rate limited fetching profile for %s, waiting %ds: %s", + post.owner_username, + wait, + exc, + ) + sleep(wait) + else: + raise + # Unreachable, but satisfies type checker + msg = "Failed to fetch profile after retries" + raise RuntimeError(msg) diff --git a/src/instagram_hashtag_crawler/export.py b/src/instagram_hashtag_crawler/export.py new file mode 100644 index 0000000..c4894b9 --- /dev/null +++ b/src/instagram_hashtag_crawler/export.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +import argparse +import csv +import json +import logging +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +# Posts within this window (in seconds) from the most recent post are skipped +# to avoid collecting posts that are still accumulating engagement. +RECENCY_THRESHOLD = 60 * 60 * 24 # 24 hours + + +def read_profiles( + json_dir: Path, + csv_dir: Path, + output_file_name: str = "posts.csv", +) -> None: + """Read all JSON files in a directory and write post data to CSV.""" + json_dir = Path(json_dir) + csv_dir = Path(csv_dir) + + if not json_dir.exists(): + msg = f"JSON directory does not exist: {json_dir}" + raise FileNotFoundError(msg) + + csv_dir.mkdir(parents=True, exist_ok=True) + output_path = csv_dir / output_file_name + + logger.info("Reading profiles from %s", json_dir) + + with output_path.open("w", newline="") as f: + writer = csv.writer(f, lineterminator="\n") + + for json_file in sorted(json_dir.iterdir()): + if json_file.suffix != ".json" or json_file.name.endswith("_rawfeed.json"): + continue + + logger.debug("Processing %s", json_file.name) + data = json.loads(json_file.read_text()) + _write_posts(data, writer) + + logger.info("Wrote CSV to %s", output_path) + + +def _write_posts(data: dict[str, Any], writer: csv.writer) -> None: + """Write posts from a single JSON file to the CSV writer. + + Posts within the recency threshold of the most recent post are skipped. + """ + posts: list[dict[str, Any]] = data.get("posts", []) + + if not posts: + return + + max_date = max(p["date"] for p in posts) + threshold_date = max_date - RECENCY_THRESHOLD + + for post in posts: + if post["date"] > threshold_date: + continue + + row = [ + post.get("shortcode", ""), + str(post.get("pic_url", "")), + post.get("like_count", 0), + post.get("username", ""), + post.get("user_id", ""), + post.get("full_name", ""), + post.get("profile_pic_url", ""), + post.get("media_count", 0), + post.get("follower_count", 0), + post.get("comment_count", 0), + post.get("date", 0), + str(post.get("caption", "")), + post.get("tags", []), + ] + writer.writerow(row) + + +def main(argv: list[str] | None = None) -> None: + """CLI entrypoint for exporting JSON data to CSV.""" + parser = argparse.ArgumentParser( + prog="instagram-hashtag-export", + description="Export crawled hashtag data from JSON to CSV.", + ) + parser.add_argument( + "--json-dir", + required=True, + help="Directory containing crawled JSON files", + ) + parser.add_argument( + "--csv-dir", + required=True, + help="Directory to write CSV output", + ) + parser.add_argument( + "--output-file", + default="posts.csv", + help="Output CSV filename (default: posts.csv)", + ) + parser.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging") + + args = parser.parse_args(argv) + + level = logging.DEBUG if args.verbose else logging.INFO + logging.basicConfig( + level=level, + format="%(asctime)s %(levelname)s [%(name)s] %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + read_profiles( + json_dir=Path(args.json_dir), + csv_dir=Path(args.csv_dir), + output_file_name=args.output_file, + ) diff --git a/src/instagram_hashtag_crawler/utils.py b/src/instagram_hashtag_crawler/utils.py new file mode 100644 index 0000000..4be1028 --- /dev/null +++ b/src/instagram_hashtag_crawler/utils.py @@ -0,0 +1,10 @@ +from __future__ import annotations + +from pathlib import Path + + +def file_to_list(filepath: str | Path) -> list[str]: + """Read a file and return non-empty lines as a list of strings.""" + path = Path(filepath) + lines = path.read_text().splitlines() + return [line.strip() for line in lines if line.strip()] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_browser_session.py b/tests/test_browser_session.py new file mode 100644 index 0000000..c6baadc --- /dev/null +++ b/tests/test_browser_session.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +from http.cookiejar import CookieJar +from unittest.mock import MagicMock, patch + +import pytest + +from instagram_hashtag_crawler.browser_session import ( + SUPPORTED_BROWSERS, + _cookiejar_to_dict, + _get_browser_cookies, + load_browser_session, +) + +# --------------------------------------------------------------------------- +# _cookiejar_to_dict +# --------------------------------------------------------------------------- + + +def test_cookiejar_to_dict_empty() -> None: + cj = CookieJar() + assert _cookiejar_to_dict(cj) == {} + + +def test_cookiejar_to_dict_converts() -> None: + cj = CookieJar() + # CookieJar doesn't have a simple .set method; mock cookies instead + cookie1 = MagicMock() + cookie1.name = "sessionid" + cookie1.value = "abc123" + cookie2 = MagicMock() + cookie2.name = "csrftoken" + cookie2.value = "xyz789" + cj._cookies = {} # noqa: SLF001 + # Patch iteration + with patch.object(CookieJar, "__iter__", return_value=iter([cookie1, cookie2])): + result = _cookiejar_to_dict(cj) + assert result == {"sessionid": "abc123", "csrftoken": "xyz789"} + + +# --------------------------------------------------------------------------- +# _get_browser_cookies +# --------------------------------------------------------------------------- + + +def test_get_browser_cookies_unsupported_browser() -> None: + mock_bc3 = MagicMock() + with ( + patch.dict("sys.modules", {"browser_cookie3": mock_bc3}), + pytest.raises(ValueError, match="Unsupported browser"), + ): + _get_browser_cookies("netscape") + + +def test_get_browser_cookies_all_supported_names() -> None: + """Verify the SUPPORTED_BROWSERS tuple contains expected browsers.""" + assert "chrome" in SUPPORTED_BROWSERS + assert "firefox" in SUPPORTED_BROWSERS + assert "safari" in SUPPORTED_BROWSERS + assert "edge" in SUPPORTED_BROWSERS + assert "brave" in SUPPORTED_BROWSERS + + +@patch("instagram_hashtag_crawler.browser_session.browser_cookie3", create=True) +def test_get_browser_cookies_permission_error(mock_bc3: MagicMock) -> None: + mock_bc3.chrome.side_effect = PermissionError("denied") + with ( + patch.dict("sys.modules", {"browser_cookie3": mock_bc3}), + pytest.raises(RuntimeError, match="permission denied"), + ): + _get_browser_cookies("chrome") + + +# --------------------------------------------------------------------------- +# load_browser_session +# --------------------------------------------------------------------------- + + +@patch("instagram_hashtag_crawler.browser_session._get_browser_cookies") +def test_load_browser_session_missing_sessionid(mock_get: MagicMock) -> None: + """Raises RuntimeError when sessionid is missing from cookies.""" + cj = CookieJar() + # Only csrftoken, no sessionid or ds_user_id + cookie = MagicMock() + cookie.name = "csrftoken" + cookie.value = "tok" + with patch.object(CookieJar, "__iter__", return_value=iter([cookie])): + mock_get.return_value = cj + loader = MagicMock() + with pytest.raises(RuntimeError, match="Missing required Instagram cookies"): + load_browser_session(loader, "chrome") + + +@patch("instagram_hashtag_crawler.browser_session._get_browser_cookies") +def test_load_browser_session_success(mock_get: MagicMock) -> None: + """Successfully loads session when all required cookies are present.""" + cj = CookieJar() + cookies = [] + for name, value in [ + ("sessionid", "sess123"), + ("csrftoken", "csrf456"), + ("ds_user_id", "12345"), + ("mid", "mid789"), + ]: + c = MagicMock() + c.name = name + c.value = value + cookies.append(c) + + with patch.object(CookieJar, "__iter__", return_value=iter(cookies)): + mock_get.return_value = cj + loader = MagicMock() + user_id = load_browser_session(loader, "chrome") + + assert user_id == "12345" + loader.context.load_session.assert_called_once() + call_args = loader.context.load_session.call_args + assert call_args[0][0] == "12345" # username arg = user_id + cookie_dict = call_args[0][1] + assert cookie_dict["sessionid"] == "sess123" + assert cookie_dict["csrftoken"] == "csrf456" diff --git a/tests/test_crawler.py b/tests/test_crawler.py new file mode 100644 index 0000000..189d802 --- /dev/null +++ b/tests/test_crawler.py @@ -0,0 +1,368 @@ +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +from instagram_hashtag_crawler.crawler import ( + CrawlConfig, + _collect_posts, + _save_posts, + crawl, + crawl_multi_and, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _fake_post( + shortcode: str, + caption_hashtags: list[str], + *, + typename: str = "GraphImage", + date_utc: datetime | None = None, +) -> MagicMock: + """Build a minimal mock instaloader Post.""" + post = MagicMock() + post.shortcode = shortcode + post.typename = typename + post.caption_hashtags = caption_hashtags # lowercase, no # + post.caption = " ".join(f"#{t}" for t in caption_hashtags) + post.date_utc = date_utc or datetime(2025, 1, 1, tzinfo=timezone.utc) + post.owner_id = hash(shortcode) % 10_000 + post.owner_username = f"user_{shortcode}" + post.url = f"https://example.com/{shortcode}.jpg" + post.likes = 42 + post.comments = 7 + return post + + +def _fake_profile() -> MagicMock: + profile = MagicMock() + profile.username = "testuser" + profile.full_name = "Test User" + profile.profile_pic_url = "https://example.com/pic.jpg" + profile.mediacount = 100 + profile.followers = 500 + profile.followees = 200 + return profile + + +def _fake_hashtag_obj(posts: list[MagicMock], mediacount: int = 999) -> MagicMock: + ht = MagicMock() + ht.mediacount = mediacount + ht.get_posts_resumable.return_value = iter(posts) + return ht + + +def _make_config(tmp_path: Path, **kwargs: Any) -> CrawlConfig: + output_dir = tmp_path / "output" + output_dir.mkdir(exist_ok=True) + return CrawlConfig(output_dir=output_dir, **kwargs) + + +# --------------------------------------------------------------------------- +# _collect_posts +# --------------------------------------------------------------------------- + + +@patch("instagram_hashtag_crawler.crawler._get_profile") +@patch("instagram_hashtag_crawler.crawler.Hashtag") +def test_collect_posts_basic( + mock_hashtag_cls: MagicMock, + mock_get_profile: MagicMock, + tmp_path: Path, +) -> None: + """Collects posts and includes shortcode in output.""" + mock_get_profile.return_value = _fake_profile() + posts = [_fake_post("ABC", ["food", "pizza"])] + mock_hashtag_cls.from_name.return_value = _fake_hashtag_obj(posts) + + config = _make_config(tmp_path) + loader = MagicMock() + result = _collect_posts(loader, "food", config) + + assert len(result) == 1 + assert result[0]["shortcode"] == "ABC" + assert "#food" in result[0]["tags"] + + +@patch("instagram_hashtag_crawler.crawler._get_profile") +@patch("instagram_hashtag_crawler.crawler.Hashtag") +def test_collect_posts_required_tags_filters( + mock_hashtag_cls: MagicMock, + mock_get_profile: MagicMock, + tmp_path: Path, +) -> None: + """Only posts whose caption contains ALL required tags are kept.""" + mock_get_profile.return_value = _fake_profile() + posts = [ + _fake_post("A", ["food", "pizza", "italy"]), # has both required + _fake_post("B", ["food"]), # missing pizza + _fake_post("C", ["pizza", "food", "cheese"]), # has both required + ] + mock_hashtag_cls.from_name.return_value = _fake_hashtag_obj(posts) + + config = _make_config(tmp_path) + loader = MagicMock() + result = _collect_posts(loader, "food", config, required_tags=frozenset({"food", "pizza"})) + + shortcodes = [p["shortcode"] for p in result] + assert shortcodes == ["A", "C"] + + +@patch("instagram_hashtag_crawler.crawler._get_profile") +@patch("instagram_hashtag_crawler.crawler.Hashtag") +def test_collect_posts_deduplicates_by_shortcode( + mock_hashtag_cls: MagicMock, + mock_get_profile: MagicMock, + tmp_path: Path, +) -> None: + """Duplicate shortcodes within one query are skipped.""" + mock_get_profile.return_value = _fake_profile() + posts = [ + _fake_post("DUP", ["food"]), + _fake_post("DUP", ["food"]), # same shortcode + _fake_post("UNIQUE", ["food"]), + ] + mock_hashtag_cls.from_name.return_value = _fake_hashtag_obj(posts) + + config = _make_config(tmp_path) + loader = MagicMock() + result = _collect_posts(loader, "food", config) + + shortcodes = [p["shortcode"] for p in result] + assert shortcodes == ["DUP", "UNIQUE"] + + +@patch("instagram_hashtag_crawler.crawler._get_profile") +@patch("instagram_hashtag_crawler.crawler.Hashtag") +def test_collect_posts_skips_non_image( + mock_hashtag_cls: MagicMock, + mock_get_profile: MagicMock, + tmp_path: Path, +) -> None: + """Non-GraphImage posts are skipped.""" + mock_get_profile.return_value = _fake_profile() + posts = [ + _fake_post("VID", ["food"], typename="GraphVideo"), + _fake_post("IMG", ["food"]), + ] + mock_hashtag_cls.from_name.return_value = _fake_hashtag_obj(posts) + + config = _make_config(tmp_path) + loader = MagicMock() + result = _collect_posts(loader, "food", config) + + assert len(result) == 1 + assert result[0]["shortcode"] == "IMG" + + +@patch("instagram_hashtag_crawler.crawler._get_profile") +@patch("instagram_hashtag_crawler.crawler.Hashtag") +def test_collect_posts_respects_max_posts( + mock_hashtag_cls: MagicMock, + mock_get_profile: MagicMock, + tmp_path: Path, +) -> None: + """Stops collecting after max_posts.""" + mock_get_profile.return_value = _fake_profile() + posts = [_fake_post(f"P{i}", ["food"]) for i in range(10)] + mock_hashtag_cls.from_name.return_value = _fake_hashtag_obj(posts) + + config = _make_config(tmp_path, max_posts=3) + loader = MagicMock() + result = _collect_posts(loader, "food", config) + + assert len(result) == 3 + + +# --------------------------------------------------------------------------- +# crawl (single hashtag — refactored to use _collect_posts) +# --------------------------------------------------------------------------- + + +@patch("instagram_hashtag_crawler.crawler._get_profile") +@patch("instagram_hashtag_crawler.crawler.Hashtag") +def test_crawl_single_writes_json( + mock_hashtag_cls: MagicMock, + mock_get_profile: MagicMock, + tmp_path: Path, +) -> None: + """Single-hashtag crawl writes {hashtag}.json with shortcode field.""" + mock_get_profile.return_value = _fake_profile() + posts = [_fake_post("XYZ", ["travel"])] + mock_hashtag_cls.from_name.return_value = _fake_hashtag_obj(posts) + + config = _make_config(tmp_path) + loader = MagicMock() + result = crawl(loader, "travel", config) + + assert result is True + output_file = config.output_dir / "travel.json" + assert output_file.exists() + data = json.loads(output_file.read_text()) + assert len(data["posts"]) == 1 + assert data["posts"][0]["shortcode"] == "XYZ" + + +@patch("instagram_hashtag_crawler.crawler._get_profile") +@patch("instagram_hashtag_crawler.crawler.Hashtag") +def test_crawl_single_returns_false_below_min( + mock_hashtag_cls: MagicMock, + mock_get_profile: MagicMock, + tmp_path: Path, +) -> None: + """Returns False when fewer than min_posts collected.""" + mock_get_profile.return_value = _fake_profile() + mock_hashtag_cls.from_name.return_value = _fake_hashtag_obj([]) + + config = _make_config(tmp_path, min_posts=5) + loader = MagicMock() + result = crawl(loader, "empty", config) + + assert result is False + + +# --------------------------------------------------------------------------- +# crawl_multi_and +# --------------------------------------------------------------------------- + + +@patch("instagram_hashtag_crawler.crawler._get_profile") +@patch("instagram_hashtag_crawler.crawler.Hashtag") +def test_crawl_multi_and_intersection( + mock_hashtag_cls: MagicMock, + mock_get_profile: MagicMock, + tmp_path: Path, +) -> None: + """AND search keeps only posts containing all requested hashtags.""" + mock_get_profile.return_value = _fake_profile() + + # Simulate two hashtag feeds + food_posts = [ + _fake_post("A", ["food", "pizza"]), # matches both + _fake_post("B", ["food"]), # missing pizza + ] + pizza_posts = [ + _fake_post("A", ["food", "pizza"]), # same post, will be deduped + _fake_post("C", ["pizza", "food"]), # matches both + _fake_post("D", ["pizza"]), # missing food + ] + + call_count = 0 + + def from_name_side_effect(_ctx: Any, name: str) -> MagicMock: + nonlocal call_count + call_count += 1 + if name == "food": + return _fake_hashtag_obj(food_posts) + return _fake_hashtag_obj(pizza_posts) + + mock_hashtag_cls.from_name.side_effect = from_name_side_effect + + config = _make_config(tmp_path) + loader = MagicMock() + result = crawl_multi_and(loader, ["food", "pizza"], config) + + assert result is True + + # Output file: sorted hashtags joined by _AND_ + output_file = config.output_dir / "food_AND_pizza.json" + assert output_file.exists() + data = json.loads(output_file.read_text()) + + shortcodes = {p["shortcode"] for p in data["posts"]} + assert "A" in shortcodes # found in both feeds + assert "C" in shortcodes # found in pizza feed, has both tags + assert "B" not in shortcodes # only has food + assert "D" not in shortcodes # only has pizza + + +def test_crawl_multi_and_requires_two_hashtags(tmp_path: Path) -> None: + """Raises ValueError if fewer than 2 hashtags.""" + config = _make_config(tmp_path) + loader = MagicMock() + with pytest.raises(ValueError, match="at least 2 hashtags"): + crawl_multi_and(loader, ["solo"], config) + + +@patch("instagram_hashtag_crawler.crawler._get_profile") +@patch("instagram_hashtag_crawler.crawler.Hashtag") +def test_crawl_multi_and_returns_false_below_min( + mock_hashtag_cls: MagicMock, + mock_get_profile: MagicMock, + tmp_path: Path, +) -> None: + """Returns False when no posts match all tags.""" + mock_get_profile.return_value = _fake_profile() + + # No posts have both tags + mock_hashtag_cls.from_name.return_value = _fake_hashtag_obj( + [ + _fake_post("X", ["food"]), + ] + ) + + config = _make_config(tmp_path, min_posts=1) + loader = MagicMock() + result = crawl_multi_and(loader, ["food", "pizza"], config) + + assert result is False + + +@patch("instagram_hashtag_crawler.crawler._get_profile") +@patch("instagram_hashtag_crawler.crawler.Hashtag") +def test_crawl_multi_and_deduplicates_across_feeds( + mock_hashtag_cls: MagicMock, + mock_get_profile: MagicMock, + tmp_path: Path, +) -> None: + """Same post appearing in multiple feeds is only included once.""" + mock_get_profile.return_value = _fake_profile() + + shared_post = _fake_post("SHARED", ["food", "pizza"]) + feed1 = [shared_post] + feed2 = [shared_post] + + call_count = 0 + + def from_name_side_effect(_ctx: Any, name: str) -> MagicMock: + nonlocal call_count + call_count += 1 + if call_count == 1: + return _fake_hashtag_obj(feed1) + return _fake_hashtag_obj(feed2) + + mock_hashtag_cls.from_name.side_effect = from_name_side_effect + + config = _make_config(tmp_path) + loader = MagicMock() + crawl_multi_and(loader, ["food", "pizza"], config) + + output_file = config.output_dir / "food_AND_pizza.json" + data = json.loads(output_file.read_text()) + + assert len(data["posts"]) == 1 + assert data["posts"][0]["shortcode"] == "SHARED" + + +# --------------------------------------------------------------------------- +# _save_posts +# --------------------------------------------------------------------------- + + +def test_save_posts_writes_json(tmp_path: Path) -> None: + """_save_posts writes valid JSON with posts key.""" + posts = [{"shortcode": "A", "user_id": 1}] + output_file = tmp_path / "test.json" + _save_posts(posts, output_file) + + data = json.loads(output_file.read_text()) + assert data == {"posts": [{"shortcode": "A", "user_id": 1}]} diff --git a/tests/test_export.py b/tests/test_export.py new file mode 100644 index 0000000..25bb19f --- /dev/null +++ b/tests/test_export.py @@ -0,0 +1,118 @@ +from __future__ import annotations + +import csv +import json +from pathlib import Path + +from instagram_hashtag_crawler.export import RECENCY_THRESHOLD, read_profiles + + +def _make_post( + date: int, + username: str = "user1", + like_count: int = 10, +) -> dict: + return { + "user_id": 123, + "username": username, + "full_name": "Test User", + "profile_pic_url": "https://example.com/pic.jpg", + "media_count": 50, + "follower_count": 100, + "following_count": 50, + "date": date, + "pic_url": "https://example.com/post.jpg", + "like_count": like_count, + "comment_count": 5, + "caption": "hello world", + "tags": ["#test"], + } + + +def _read_csv(path: Path) -> list[list[str]]: + with path.open(newline="") as f: + return list(csv.reader(f)) + + +def test_read_profiles_basic(tmp_path: Path) -> None: + """Writes posts to CSV, skipping recent ones.""" + json_dir = tmp_path / "json" + csv_dir = tmp_path / "csv" + json_dir.mkdir() + + now = 1_700_000_000 + old_post = _make_post(date=now - RECENCY_THRESHOLD - 1, username="old_user") + recent_post = _make_post(date=now, username="recent_user") + + data = {"posts": [recent_post, old_post]} + (json_dir / "food.json").write_text(json.dumps(data)) + + read_profiles(json_dir, csv_dir) + + rows = _read_csv(csv_dir / "posts.csv") + assert len(rows) == 1 + assert rows[0][3] == "old_user" # username column + + +def test_read_profiles_empty_json(tmp_path: Path) -> None: + """An empty posts list produces an empty CSV.""" + json_dir = tmp_path / "json" + csv_dir = tmp_path / "csv" + json_dir.mkdir() + + (json_dir / "empty.json").write_text(json.dumps({"posts": []})) + + read_profiles(json_dir, csv_dir) + + rows = _read_csv(csv_dir / "posts.csv") + assert rows == [] + + +def test_read_profiles_skips_rawfeed(tmp_path: Path) -> None: + """Files ending in _rawfeed.json are ignored.""" + json_dir = tmp_path / "json" + csv_dir = tmp_path / "csv" + json_dir.mkdir() + + now = 1_700_000_000 + post = _make_post(date=now - RECENCY_THRESHOLD - 100) + (json_dir / "food_rawfeed.json").write_text(json.dumps({"posts": [post]})) + + read_profiles(json_dir, csv_dir) + + rows = _read_csv(csv_dir / "posts.csv") + assert rows == [] + + +def test_read_profiles_missing_dir(tmp_path: Path) -> None: + """Raises FileNotFoundError when JSON dir doesn't exist.""" + import pytest + + with pytest.raises(FileNotFoundError): + read_profiles(tmp_path / "nonexistent", tmp_path / "csv") + + +def test_read_profiles_multiple_files(tmp_path: Path) -> None: + """Processes multiple JSON files in sorted order.""" + json_dir = tmp_path / "json" + csv_dir = tmp_path / "csv" + json_dir.mkdir() + + now = 1_700_000_000 + old_date = now - RECENCY_THRESHOLD - 100 + for name in ("beta", "alpha"): + data = { + "posts": [ + _make_post(date=now, username="recent"), + _make_post(date=old_date, username=f"{name}_user"), + ] + } + (json_dir / f"{name}.json").write_text(json.dumps(data)) + + read_profiles(json_dir, csv_dir) + + rows = _read_csv(csv_dir / "posts.csv") + assert len(rows) == 2 + # alpha.json sorts before beta.json + assert rows[0][3] == "alpha_user" + assert rows[1][3] == "beta_user" diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..3a7ae9a --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from pathlib import Path + +from instagram_hashtag_crawler.utils import file_to_list + + +def test_file_to_list_basic(tmp_path: Path) -> None: + """Read a file with normal lines.""" + f = tmp_path / "tags.txt" + f.write_text("alpha\nbeta\ngamma\n") + result = file_to_list(f) + assert result == ["alpha", "beta", "gamma"] + + +def test_file_to_list_skips_blank_lines(tmp_path: Path) -> None: + """Blank lines and whitespace-only lines are excluded.""" + f = tmp_path / "tags.txt" + f.write_text("alpha\n\n \nbeta\n") + result = file_to_list(f) + assert result == ["alpha", "beta"] + + +def test_file_to_list_strips_whitespace(tmp_path: Path) -> None: + """Leading/trailing whitespace is stripped from each line.""" + f = tmp_path / "tags.txt" + f.write_text(" alpha \n\tbeta\t\n") + result = file_to_list(f) + assert result == ["alpha", "beta"] + + +def test_file_to_list_empty_file(tmp_path: Path) -> None: + """An empty file returns an empty list.""" + f = tmp_path / "empty.txt" + f.write_text("") + result = file_to_list(f) + assert result == [] + + +def test_file_to_list_accepts_string_path(tmp_path: Path) -> None: + """Accepts a string path, not just Path objects.""" + f = tmp_path / "tags.txt" + f.write_text("one\ntwo\n") + result = file_to_list(str(f)) + assert result == ["one", "two"] diff --git a/util.py b/util.py deleted file mode 100644 index c30fc4b..0000000 --- a/util.py +++ /dev/null @@ -1,42 +0,0 @@ -from random import sample, shuffle -import csv - -def randselect(list, num): - l = len(list) - if l <= num: - return shuffle(list) - if l > 5*num: - return sample(list[:5*num], num) - -def byteify(input): - if isinstance(input, dict): - return {byteify(key): byteify(value) - for key, value in input.iteritems()} - elif isinstance(input, list): - return [byteify(element) for element in input] - elif isinstance(input, unicode): - return input.encode('utf-8') - else: - return input - -def file_to_list(file): - '''1 Dimensional''' - data = [] - f = open(file, 'r') - contents = csv.reader(f.read().splitlines()) - count = 0 - try: - for c in contents: - count += 1 - data.append(c) - except Exception as e: - print("count",count) - raise e - - if len(data) >= 2: - return [d[0] for d in data] - elif len(data) == 1: - return data[0] - else: - return data -