File size: 4,446 Bytes
df0daf5 | 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 | """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"), # PSID for page comments
"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 |