| """Export helpers: post identifier parsing/resolution, CSV/JSON output.""" |
|
|
| import csv |
| import json |
| import re |
| import urllib.parse |
|
|
| from .crawler import CommentCrawler, CommentCrawlerError, GRAPH as _GRAPH |
|
|
|
|
| class ExportError(RuntimeError): |
| pass |
|
|
|
|
| def parse_post_identifier(url_or_id: str) -> str: |
| """Return the Graph API object id for a Facebook post URL. |
| |
| Accepts: |
| - ``https://www.facebook.com/<page>/posts/<story_fbid>`` |
| - ``https://www.facebook.com/<page>/posts/pfbid0...`` |
| - legacy ``https://www.facebook.com/permalink.php?story_fbid=...&id=...`` |
| - ``https://www.facebook.com/photo.php?fbid=...&set=a...`` (photo comments) |
| - plain ``<page_id>_<post_id>`` or ``<post_id>`` |
| Returns e.g. ``123456789_101579...``. |
| """ |
| s = (url_or_id or "").strip() |
| if not s: |
| raise ExportError("Empty post URL/id.") |
| if re.fullmatch(r"\d+_\d+", s): |
| return s |
| if re.fullmatch(r"\d+", s): |
| return s |
| if re.fullmatch(r"pfbid\w+", s) or re.fullmatch(r"pfbidp\w+", s): |
| return s |
| parsed = urllib.parse.urlparse(s if s.startswith("http") else "https://" + s) |
| host = (parsed.netloc or "").lower() |
| if "facebook.com" not in host: |
| raise ExportError(f"Not a Facebook URL: {s}") |
| path = parsed.path.strip("/") |
| qs = urllib.parse.parse_qs(parsed.query) |
| if path.startswith("permalink.php"): |
| story = qs.get("story_fbid", [None])[0] |
| page_id = qs.get("id", [None])[0] |
| if story and page_id: |
| return f"{page_id}_{story}" |
| raise ExportError("permalink.php URL missing story_fbid/id parameters.") |
| if path.startswith("photo.php"): |
| fbid = qs.get("fbid", [None])[0] |
| page_id = qs.get("id", [None])[0] |
| if fbid and page_id: |
| return f"{page_id}_{fbid}" |
| raise ExportError("photo.php URL missing fbid/id parameters.") |
| parts = path.split("/") |
| if len(parts) >= 3 and parts[1] == "posts": |
| return f"{parts[0]}_{parts[2]}" |
| if len(parts) == 3 and parts[1] == "videos": |
| return f"{parts[0]}_{parts[2]}" |
| raise ExportError(f"Unrecognized Facebook post URL pattern: {s} (path={path})") |
|
|
|
|
| def resolve_post_id(post_identifier: str, crawler: CommentCrawler) -> str: |
| """Best-effort resolution of a handle to a stable object id. |
| |
| If the identifier is already ``<page_id>_<post_id>`` it is returned as-is. |
| If it is a page handle + pfbid (e.g. ``MaiAnhTaiSinger_pfbid0...``) we call |
| the post node and read ``id``. Falls back gracefully when the API can't. |
| """ |
| if re.fullmatch(r"\d+_\d+", post_identifier): |
| return post_identifier |
| page, _, story = post_identifier.partition("_") |
| if not story: |
| return post_identifier |
| try: |
| data = crawler._page(f"{_GRAPH}/{crawler.api_version}/{post_identifier}?" |
| f"fields=id&access_token={crawler.token}") |
| if data.get("id"): |
| return data["id"] |
| except CommentCrawlerError: |
| pass |
| return post_identifier |
|
|
|
|
| def export(comments, out_csv: str | None = None, out_json: str | None = None): |
| """Write comments (iterable of dicts) to CSV and/or JSON. |
| |
| Returns the list of dicts written (useful to count). |
| """ |
| rows = [] |
| for c in comments: |
| author = c.get("from") or {} |
| rows.append( |
| { |
| "comment_id": c.get("id"), |
| "parent_id": (c.get("parent") or {}).get("id"), |
| "name": author.get("name"), |
| "uid": author.get("id"), |
| "message": c.get("message") or "", |
| "created_time": c.get("created_time"), |
| "like_count": c.get("like_count"), |
| } |
| ) |
| if out_csv: |
| with open(out_csv, "w", newline="", encoding="utf-8-sig") as f: |
| writer = csv.DictWriter( |
| f, |
| fieldnames=["comment_id", "parent_id", "name", "uid", "message", "created_time", "like_count"], |
| ) |
| writer.writeheader() |
| writer.writerows(rows) |
| if out_json: |
| with open(out_json, "w", encoding="utf-8") as f: |
| json.dump(rows, f, ensure_ascii=False, indent=2) |
| if not out_csv and not out_json: |
| import sys |
|
|
| for r in rows: |
| sys.stdout.write( |
| f"[{r['comment_id']}] {r['name']} ({r['uid']}): {r['message']}\n" |
| ) |
| return rows |
