| """Command-line entry point for the Facebook comment crawler (used by the Space's |
| fb_comments package import chain; the Space UI itself uses crawler/export only). |
| """ |
|
|
| import argparse |
| import json |
| import os |
| import sys |
|
|
| from .crawler import CommentCrawler, CommentCrawlerError, RateLimitError |
| from .export import ExportError, export, parse_post_identifier, resolve_post_id |
| from .oauth import ( |
| DEFAULT_SCOPE, |
| OAuthError, |
| exchange_code, |
| exchange_long_lived, |
| list_pages, |
| ) |
|
|
|
|
| def _cmd_login(args): |
| """Interactive login: print a URL, capture the code, exchange to a page token.""" |
| app_id = args.app_id or os.environ.get("FB_APP_ID") |
| app_secret = args.app_secret or os.environ.get("FB_APP_SECRET") |
| if not app_id or not app_secret: |
| print("--app-id / --app-secret (or FB_APP_ID / FB_APP_SECRET) required for login.") |
| sys.exit(2) |
|
|
| from .oauth import login_url |
| from .oauth import start_callback_server |
|
|
| url, state = login_url(app_id, app_secret, DEFAULT_SCOPE.split(","), port=args.port) |
| print("Open this URL in a browser and log in as the Page's admin:") |
| print(" " + url) |
| print("\nWaiting for the redirect to localhost ...") |
|
|
| holder = start_callback_server(port=args.port, timeout=args.timeout) |
| if holder.get("error"): |
| sys.exit(f"Facebook login failed: {holder['error']}") |
| if not holder.get("code"): |
| sys.exit("No login code received (timed out).") |
| if holder.get("state") != state: |
| sys.exit("state mismatch - possible CSRF, aborting.") |
|
|
| short = None |
| try: |
| data = exchange_code(app_id, app_secret, holder["code"], |
| redirect_uri=f"http://localhost:{args.port}/callback") |
| short = data["access_token"] |
| except OAuthError as e: |
| sys.exit(f"Code exchange failed: {e}") |
|
|
| try: |
| long_data = exchange_long_lived(app_id, app_secret, short) |
| long_token = long_data["access_token"] |
| except OAuthError as e: |
| sys.exit(f"Long-lived exchange failed: {e}") |
|
|
| pages = list_pages(long_token) |
| print(f"\nUser has {len(pages)} managed Page(s):") |
| for p in pages: |
| print(f" - {p.get('name')} (id={p.get('id')})") |
| if not pages: |
| sys.exit("User manages no Pages; comments can only be read for pages you manage.") |
|
|
| if len(pages) > 1 and not args.page: |
| print("Multiple Pages — pass --page <name-or-id> to pick one.") |
| sys.exit(2) |
| page = pages[0] if len(pages) == 1 else next(p for p in pages if p.get("id") == args.page or p.get("name") == args.page) |
|
|
| token = page["access_token"] |
| out = { |
| "app_id": app_id, |
| "page_name": page.get("name"), |
| "page_id": page.get("id"), |
| "user_long_lived_token": long_token, |
| "page_access_token": token, |
| "token_type": "page", |
| } |
| if args.token_cache: |
| with open(args.token_cache, "w", encoding="utf-8") as f: |
| json.dump(out, f, indent=2, ensure_ascii=False) |
| print(f"\n[OK] Saved token config to {args.token_cache}") |
| print("Use: fbcomments crawl --token <page_access_token> --url <post-url> --csv out.csv") |
|
|
|
|
| def _cmd_crawl(args): |
| token = args.token or os.environ.get("FB_TOKEN") |
| if not token: |
| sys.exit("--token required (Page access token) or FB_TOKEN env var. Run 'login' first.") |
| post_id = parse_post_identifier(args.url) |
| crawler = CommentCrawler(token) |
| print(f"Resolving post identifier: {post_id}") |
| try: |
| post_id = resolve_post_id(post_id, crawler) |
| print(f"Using object id: {post_id}") |
| except Exception: |
| pass |
| print("Crawling comments ...\n") |
| try: |
| comments = list(crawler.comments(post_id, include_replies=not args.no_replies)) |
| except CommentCrawlerError as e: |
| sys.exit(f"Crawl failed: {e}") |
| print(f"Collected {len(comments)} comments") |
| export(comments, out_csv=args.csv, out_json=args.json) |
|
|
|
|
| def _cmd_list_tokens(args): |
| token = args.token or os.environ.get("FB_TOKEN") |
| if not token: |
| sys.exit("--token required.") |
| try: |
| for p in list_pages(token): |
| print(f"{p.get('name')}\t{p.get('id')}") |
| except OAuthError as e: |
| sys.exit(f"Error listing pages: {e}") |
|
|
|
|
| def _build_parser(): |
| parser = argparse.ArgumentParser( |
| prog="fbcomments", |
| description="Fetch 100% of comments (with replies) from Facebook posts you can read, with commenter UID export.", |
| ) |
| parser.add_argument("--version", action="store_true") |
| sub = parser.add_subparsers(dest="cmd") |
|
|
| p_login = sub.add_parser("login", help="OAuth login -> page access token") |
| p_login.add_argument("--app-id", default=None) |
| p_login.add_argument("--app-secret", default=None) |
| p_login.add_argument("--port", type=int, default=8080) |
| p_login.add_argument("--timeout", type=float, default=300) |
| p_login.add_argument("--page", default=None) |
| p_login.add_argument("--token-cache", default=None) |
|
|
| p_crawl = sub.add_parser("crawl", help="Fetch comments for a post URL/id") |
| p_crawl.add_argument("--token", default=None) |
| p_crawl.add_argument("--url", required=True, help="Facebook post URL or id") |
| p_crawl.add_argument("--csv", default=None) |
| p_crawl.add_argument("--json", default=None) |
| p_crawl.add_argument("--no-replies", action="store_true") |
|
|
| p_tokens = sub.add_parser("tokens", help="List pages of the current token") |
| p_tokens.add_argument("--token", default=None) |
|
|
| return parser |
|
|
|
|
| def main(argv=None): |
| parser = _build_parser() |
| args = parser.parse_args(argv) |
| if args.version: |
| from . import __version__ |
|
|
| print(__version__) |
| return |
| if args.cmd == "login": |
| _cmd_login(args) |
| elif args.cmd == "crawl": |
| _cmd_crawl(args) |
| elif args.cmd == "tokens": |
| _cmd_list_tokens(args) |
| else: |
| parser.print_help() |
|
|
|
|
| if __name__ == "__main__": |
| main() |
