File size: 5,925 Bytes
5ce75da | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 | """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() |