diff --git a/.gitignore b/.gitignore
index d9c12f9..85c0735 100644
--- a/.gitignore
+++ b/.gitignore
@@ -11,4 +11,6 @@ venv/
.idea/
*.swp
*.log
-flask_session/
+*flask_session/
+*.sessions
+*.cookie
diff --git a/Script/SOCMINT-Twitter/Readme.md b/Script/SOCMINT-Twitter/Readme.md
index ff8f39f..2729079 100644
--- a/Script/SOCMINT-Twitter/Readme.md
+++ b/Script/SOCMINT-Twitter/Readme.md
@@ -2,21 +2,45 @@
+# Sett up
+
+1. Config your Google api console here, enable and manage API Custom Search by Google
+
+
+
+2. Sett the api key in web console Google
+
+
+
+Check the result in the table
+
+
+
+3. Settings CSE Google to put the cx key
+
+
+
+4. Add site want to crawll e.g twitter.com and x.com
+
+
+
## Data Source
-1. Xquik API
-2. Cookie (your account cookie session)
+1. Xquik API (subs there is a price)
+2. Cookie (your account cookie session)
3. Wayback Machine (Cdx API)
+4. Goole CSE API (free quota 100 per day u can increase u limit with buy the service)
## Update Note
-1. Update infinity scroll and load new data
+1. Update infinity scroll and load new data for twitter reply and retweets
2. Update data corelation
3. Fix business logic flow
4. Monitoring (Soon)
5. MCP (Soon)
6. Add more parameter for enrichment
7. Add no rate limit (throttle)
+8. Add Google CSE data source
## Setup
diff --git a/Script/SOCMINT-Twitter/app.py b/Script/SOCMINT-Twitter/app.py
index 98a401d..3e03083 100644
--- a/Script/SOCMINT-Twitter/app.py
+++ b/Script/SOCMINT-Twitter/app.py
@@ -12,6 +12,7 @@ from flask import Flask, g, jsonify, render_template, request, Response, stream_
import archive as _archive
from xquik_client import XquikClient, XquikError, load_config
from wayback_client import wayback_search, WaybackError
+from google_cse_client import google_cse_search, GoogleCSEError
from id_forensics import enrich_account_age
from cookie_client import (
cookie_tweet_search,
@@ -142,6 +143,7 @@ SOURCE_LABELS = {
"cookie": "Twitter Cookie",
"xquik": "Xquik API",
"wayback": "Wayback Machine",
+ "cse": "Google CSE",
}
@@ -151,6 +153,39 @@ def _tag_source(items, label):
return [{**it, "source": label} if isinstance(it, dict) else it for it in items]
+def _stamp_fetched_at(data):
+ """Mutates every dict in `data` (list or single dict) in place, adding
+ fetched_at — when *this app* pulled the record, as opposed to created_at
+ (a tweet's own post time) or iso_date (a Wayback snapshot's capture time).
+ Applied uniformly across every source so results are comparable no
+ matter which tool/source produced them."""
+ now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+ items = data if isinstance(data, list) else [data]
+ for item in items:
+ if isinstance(item, dict):
+ item["fetched_at"] = now
+ return data
+
+
+def _stamp_tweet_url(data):
+ """Mutates every dict in `data` (list or single dict) in place, adding
+ tweet_url wherever id+user identify an actual tweet — the citable-link
+ field Wayback (archive_url/original) and Google CSE (result_url) already
+ carry directly in their own data. Cookie/xquik never set this on the raw
+ tweet dict themselves — previously it only got built at archive time, so a
+ live card, a JSON dump, and an archive of the same search could each show
+ a different answer for "what's the URL." Building it once here means all
+ three read the identical value. archive.py's build_tweet_url() reuses this
+ same logic rather than recomputing it separately."""
+ items = data if isinstance(data, list) else [data]
+ for item in items:
+ if isinstance(item, dict) and not item.get("tweet_url"):
+ url = _archive.build_tweet_url(item)
+ if url:
+ item["tweet_url"] = url
+ return data
+
+
# ── Optional date-range narrowing ───────────────────────────────────────────
# Dates come in from the client as free-text fields, so every value is run
# through this strict YYYYMMDD check before it touches a search query string
@@ -240,11 +275,15 @@ def _multi_source_search(query: str, count: int, from_date: str = "", to_date: s
"cookie": lambda: cookie_tweet_search(twitter_query, count=count, config=config)[0],
"xquik": lambda: XquikClient(config).tweet_search(twitter_query),
"wayback": lambda: wayback_search(query, count=count, from_date=from_date, to_date=to_date)[0],
+ # Google has no since:/until: query syntax like Twitter/Wayback do, so
+ # this lane runs unbounded by date — _filter_by_date below keeps
+ # results whose own timestamp it can't verify rather than dropping them.
+ "cse": lambda: google_cse_search(query, count=count, config=config)[0],
}
with ThreadPoolExecutor(max_workers=len(jobs)) as pool:
futures = {key: pool.submit(fn) for key, fn in jobs.items()}
results = []
- for key in ("cookie", "xquik", "wayback"): # deterministic display order
+ for key in ("cookie", "xquik", "wayback", "cse"): # deterministic display order
try:
data = futures[key].result()
except Exception:
@@ -267,7 +306,7 @@ def video_proxy():
url,
stream=True,
timeout=20,
- headers={"Referer": "https://x.com/", "User-Agent": "Mozilla/5.0"},
+ headers={"Referer": "https://x.com/", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/42.0.2311.135 Safari/537.36 Edge/12.10240"},
)
headers = {"Content-Type": upstream.headers.get("Content-Type", "video/mp4")}
if "Content-Length" in upstream.headers:
@@ -390,14 +429,16 @@ def run_tool():
return jsonify({"ok": False, "error": f"Unknown toolType: {tool_type}"}), 400
# Single choke point: every tool's output passes through here, so the
- # account-age label shows up everywhere downstream for free — cards,
- # graph nodes, JSON dump, and archives (once the fields are whitelisted
- # in archive.py's _pick_fields).
+ # account-age label, fetch timestamp, and tweet_url all show up
+ # everywhere downstream for free — cards, graph nodes, JSON dump, and
+ # archives (once the fields are whitelisted in archive.py's _pick_fields).
data = enrich_account_age(data)
+ data = _stamp_fetched_at(data)
+ data = _stamp_tweet_url(data)
return jsonify({"ok": True, "data": data, "nextCursor": next_cursor})
- except (XquikError, CookieClientError, WaybackError) as e:
+ except (XquikError, CookieClientError, WaybackError, GoogleCSEError) as e:
return jsonify({"ok": False, "error": str(e)}), 400
except Exception as e: # noqa: BLE001
return jsonify({"ok": False, "error": f"Unexpected error: {e}"}), 500
diff --git a/Script/SOCMINT-Twitter/archive.py b/Script/SOCMINT-Twitter/archive.py
index 29ad216..27d4788 100644
--- a/Script/SOCMINT-Twitter/archive.py
+++ b/Script/SOCMINT-Twitter/archive.py
@@ -23,7 +23,10 @@ def _make_id(tool_type: str) -> str:
return f"{tool_type}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
-def _tweet_url(item: dict) -> str | None:
+def build_tweet_url(item: dict) -> str | None:
+ """Not prefixed private — app.py reuses this so a live result and its
+ later archive always compute the identical tweet_url, instead of two
+ separate copies of the same logic drifting apart."""
tid = str(item.get("id", "")).strip()
user = str(item.get("user", "")).strip()
if tid and user:
@@ -31,6 +34,46 @@ def _tweet_url(item: dict) -> str | None:
return None
+def _extract_media(item: dict) -> list[dict]:
+ """Same normalization the frontend's extractMedia() applies for card
+ display. Cookie mode already returns [{type, thumb, url}] under `media`.
+ xquik/API mode instead carries raw Twitter API shape under
+ entities.media / extended_entities.media, which needs unpacking first —
+ otherwise xquik-sourced photos/videos never enter the download queue below."""
+ media = item.get("media")
+ if isinstance(media, list) and media and isinstance(media[0], dict) \
+ and ("thumb" in media[0] or "url" in media[0]):
+ return media
+
+ src = None
+ ext_ent = item.get("extended_entities")
+ if isinstance(ext_ent, dict):
+ src = ext_ent.get("media")
+ if not isinstance(src, list):
+ ent = item.get("entities")
+ if isinstance(ent, dict):
+ src = ent.get("media")
+ if not isinstance(src, list):
+ return []
+
+ result = []
+ for m in src:
+ if not isinstance(m, dict):
+ continue
+ mtype = m.get("type", "photo")
+ thumb = m.get("media_url_https") or m.get("media_url") or ""
+ if not thumb:
+ continue
+ url = thumb
+ if mtype in ("video", "animated_gif"):
+ variants = ((m.get("video_info") or {}).get("variants")) or []
+ mp4s = [v for v in variants if isinstance(v, dict) and v.get("content_type") == "video/mp4"]
+ if mp4s:
+ url = max(mp4s, key=lambda v: v.get("bitrate", 0) or 0).get("url", thumb)
+ result.append({"type": mtype, "thumb": thumb, "url": url})
+ return result
+
+
def _media_ext(url: str, mtype: str) -> str:
if mtype in ("video", "animated_gif"):
return "mp4"
@@ -46,7 +89,7 @@ def _download_file(url: str, dest: Path) -> bool:
r = requests.get(
url,
timeout=REQUEST_TIMEOUT,
- headers={"Referer": "https://x.com/", "User-Agent": "Mozilla/5.0"},
+ headers={"Referer": "https://x.com/", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/42.0.2311.135 Safari/537.36 Edge/12.10240"},
stream=True,
)
r.raise_for_status()
@@ -71,6 +114,7 @@ def _pick_fields(item: dict) -> dict:
"retweeted_by_bio", "retweeted_at", "retweeted_tweet_id",
"iso_date", "original", "statuscode", "mimetype", "length",
"archive_url", "post_title", "post_text", "preview_image",
+ "result_url", "display_link", "fetched_at",
"source", "account_created", "account_age", "account_age_flag",
"account_age_precision"]
return {k: item[k] for k in keys if k in item and item[k] is not None}
@@ -96,10 +140,10 @@ def _run(archive_id: str, tool_type: str, data, query_info: dict) -> None:
continue
record = _pick_fields(item)
- record["tweet_url"] = _tweet_url(item)
+ record["tweet_url"] = item.get("tweet_url") or build_tweet_url(item)
local_media = []
- for midx, m in enumerate(item.get("media", [])):
+ for midx, m in enumerate(_extract_media(item)):
url = m.get("url") or m.get("thumb", "")
if not url:
continue
diff --git a/Script/SOCMINT-Twitter/config.ini.example b/Script/SOCMINT-Twitter/config.ini.example
index 7e2b784..775649c 100644
--- a/Script/SOCMINT-Twitter/config.ini.example
+++ b/Script/SOCMINT-Twitter/config.ini.example
@@ -1,4 +1,6 @@
[xquik]
+; check dasboard Xquik to put the api key
+
api_key = xxxxxxxxxxxxxxxxxxxxxxxxxxx
; Soon there is a array for base_url and endpoint are listed in offc docs
@@ -9,11 +11,25 @@ base_url = https://xquik.com/api/v1/extractions
;end arr endpoint
+[google_cse]
+; put the api key in google console api https://console.cloud.google.com/ also dont forget check the security config
+
+api_key = AIzaSxxxxxxxxx
+
+; put the cx key in https://programmablesearchengine.google.com/ google cse and sett the site want to crawll check the readme.md
+
+cx = cxxxxxxxxxx
+
[twitter_cookies]
+; Inspect element then check the tab application then click the cookie in storage or u can use burpsuite
+
auth_token = xxxxxxxxxxxxxxxxxx
ct0 = xxxxxxxxxxxx
[server]
+
+; config flask
+
host = 127.0.0.1
port = 5000
debug = true
\ No newline at end of file
diff --git a/Script/SOCMINT-Twitter/google_cse_client.py b/Script/SOCMINT-Twitter/google_cse_client.py
new file mode 100644
index 0000000..bcd9922
--- /dev/null
+++ b/Script/SOCMINT-Twitter/google_cse_client.py
@@ -0,0 +1,266 @@
+"""Look up general web results via the Google Custom Search JSON API
+(https://developers.google.com/custom-search/v1/overview) — a 4th data
+source alongside Cookie/Xquik (live X data) and Wayback (archived X pages).
+
+Unlike the other three, this source isn't X-specific: it searches whatever
+scope the Custom Search Engine (cx) itself is configured for on Google's
+side, so it's the one lane that can surface a username/keyword showing up
+*off* X entirely — news mentions, forum posts, cached pages, other social
+platforms — which is what makes it worth adding to Multi Source Search.
+
+Requires two values in config.ini [google_cse]:
+ api_key — issued via Google Cloud/API Console (enable "Custom Search API")
+ cx — the Search Engine ID from https://programmablesearchengine.google.com/
+
+Free tier: 100 queries/day. Each page here costs exactly one query
+(Google caps `num` at 10 results/request), so `count` is served in
+10-result pages up to a hard ceiling well under the daily quota.
+"""
+
+import configparser
+import html
+import re
+from concurrent.futures import ThreadPoolExecutor
+
+import requests
+
+from id_forensics import decode_snowflake
+
+CSE_URL = "https://www.googleapis.com/customsearch/v1"
+REQUEST_TIMEOUT = 20
+PAGE_SIZE = 10 # Google's hard max for `num`
+MAX_RESULTS = 50 # ceiling on results served per call, regardless of `count`
+
+# Google truncates BOTH the title and the snippet it hands back in the SERP
+# JSON — title gets clipped just like the snippet does (see the module intro
+# reasoning). For text-intel use we want the page's own full title and
+# description instead, so every result with a link gets a best-effort live
+# fetch to pull its real og:/twitter: tags — same technique wayback_client.py
+# uses on archived snapshots, just against the live page instead.
+ENRICH_TIMEOUT = 10
+ENRICH_WORKERS = 8
+META_PARSE_CAP = 300_000 # bytes of HTML scanned for meta tags
+
+_SAFE_URL_RE = re.compile(r"^https?://", re.IGNORECASE)
+_META_TAG_RE = re.compile(r"]*>", re.IGNORECASE)
+_ATTR_RE = re.compile(r'''([\w:-]+)\s*=\s*"([^"]*)"|([\w:-]+)\s*=\s*'([^']*)\'''')
+_TITLE_TAG_RE = re.compile(r"]*>(.*?)", re.IGNORECASE | re.DOTALL)
+_TWEET_ID_RE = re.compile(r"/status/(\d+)")
+
+
+def _tweet_created_at(url: str) -> str | None:
+ """When `url` is a tweet permalink (…/status/), decode the actual
+ post-creation time straight out of the id's Snowflake bits. Same field
+ name/format cookie and xquik already populate (`created_at`, Twitter's
+ own classic timestamp string), so every source is consistent — and
+ absent entirely for non-X results, same as it's absent for anything
+ without a usable id."""
+ m = _TWEET_ID_RE.search(url or "")
+ if not m:
+ return None
+ dt = decode_snowflake(m.group(1))
+ if not dt:
+ return None
+ return dt.strftime("%a %b %d %H:%M:%S +0000 %Y")
+
+
+class GoogleCSEError(Exception):
+ pass
+
+
+def _get_client_config(config: configparser.ConfigParser) -> tuple[str, str]:
+ api_key = config.get("google_cse", "api_key", fallback="").strip()
+ cx = config.get("google_cse", "cx", fallback="").strip()
+ if not api_key or api_key == "YOUR_GOOGLE_API_KEY":
+ raise GoogleCSEError("api_key belum diisi di config.ini [google_cse]")
+ if not cx or cx == "YOUR_SEARCH_ENGINE_ID":
+ raise GoogleCSEError("cx (Search Engine ID) belum diisi di config.ini [google_cse]")
+ return api_key, cx
+
+
+def _extract_thumbnail(item: dict) -> str | None:
+ pagemap = item.get("pagemap") or {}
+ for key in ("cse_image", "cse_thumbnail"):
+ candidates = pagemap.get(key) or []
+ if candidates and isinstance(candidates, list):
+ src = (candidates[0] or {}).get("src", "").strip()
+ # Third-party page metadata — only trust it if it's a plain http(s)
+ # link, since the frontend renders this straight into an .
+ if src and _SAFE_URL_RE.match(src):
+ return src
+ return None
+
+
+def _row_to_record(item: dict) -> dict:
+ record = {}
+ title = item.get("title")
+ if title:
+ record["post_title"] = html.unescape(title).strip()
+ snippet = item.get("snippet")
+ if snippet:
+ record["post_text"] = html.unescape(snippet).strip()
+ link = item.get("link")
+ if link and _SAFE_URL_RE.match(link):
+ record["result_url"] = link
+ created_at = _tweet_created_at(link)
+ if created_at:
+ record["created_at"] = created_at # when the post itself was actually made
+ display_link = item.get("displayLink")
+ if display_link:
+ record["display_link"] = display_link
+ thumb = _extract_thumbnail(item)
+ if thumb:
+ record["preview_image"] = thumb
+ return record
+
+
+def _parse_meta_tags(text: str) -> dict:
+ tags = {}
+ for tag in _META_TAG_RE.findall(text):
+ attrs = {}
+ for m in _ATTR_RE.finditer(tag):
+ if m.group(1):
+ attrs[m.group(1).lower()] = m.group(2)
+ else:
+ attrs[m.group(3).lower()] = m.group(4)
+ key = attrs.get("property") or attrs.get("name")
+ val = attrs.get("content")
+ if key and val is not None:
+ tags[key.lower()] = val
+ return tags
+
+
+def _extract_title_tag(text: str) -> str | None:
+ m = _TITLE_TAG_RE.search(text)
+ if not m:
+ return None
+ title = re.sub(r"\s+", " ", m.group(1)).strip()
+ return title or None
+
+
+def _fetch_live_meta(url: str) -> dict:
+ """Best-effort live fetch of the result's own page — pulls its real
+ title and og:/twitter:/meta description to replace Google's clipped SERP
+ title+snippet. Any failure (timeout, non-200, no usable tags) is
+ swallowed — the caller just keeps Google's own (possibly truncated)
+ values as a fallback."""
+ try:
+ r = requests.get(url, timeout=ENRICH_TIMEOUT, headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/42.0.2311.135 Safari/537.36 Edge/12.10240"})
+ if r.status_code != 200 or not r.text:
+ return {}
+ except requests.RequestException:
+ return {}
+
+ body = r.text[:META_PARSE_CAP]
+ tags = _parse_meta_tags(body)
+ out = {}
+
+ title = tags.get("og:title") or tags.get("twitter:title") or _extract_title_tag(body)
+ if title:
+ out["post_title"] = html.unescape(title).strip()
+
+ desc = tags.get("og:description") or tags.get("twitter:description") or tags.get("description")
+ if desc:
+ out["post_text"] = html.unescape(desc).strip()
+
+ image = tags.get("og:image") or tags.get("twitter:image")
+ if image:
+ image = html.unescape(image).strip()
+ if _SAFE_URL_RE.match(image):
+ out["preview_image"] = image
+
+ return out
+
+
+def _enrich_records(records: list[dict]) -> None:
+ """Mutates each record in place. Runs in parallel — one slow/dead site
+ shouldn't hold up the rest of the result set."""
+ candidates = [r for r in records if r.get("result_url")]
+ if not candidates:
+ return
+
+ def _job(rec):
+ extra = _fetch_live_meta(rec["result_url"])
+ if extra.get("post_title"):
+ rec["post_title"] = extra["post_title"]
+ if extra.get("post_text"):
+ rec["post_text"] = extra["post_text"]
+ if extra.get("preview_image") and not rec.get("preview_image"):
+ rec["preview_image"] = extra["preview_image"]
+
+ with ThreadPoolExecutor(max_workers=ENRICH_WORKERS) as pool:
+ list(pool.map(_job, candidates))
+
+
+def _fetch_page(query: str, api_key: str, cx: str, start: int) -> list[dict]:
+ params = {
+ "key": api_key,
+ "cx": cx,
+ "q": query,
+ "num": PAGE_SIZE,
+ "start": start,
+ }
+ try:
+ r = requests.get(CSE_URL, params=params, timeout=REQUEST_TIMEOUT)
+ except requests.RequestException as e:
+ raise GoogleCSEError(f"Google CSE request failed: {e}") from e
+
+ if r.status_code == 429:
+ raise GoogleCSEError("Google CSE daily quota exceeded (100 free queries/day)")
+ if r.status_code == 403:
+ raise GoogleCSEError("Google CSE request forbidden — check api_key/cx and that the "
+ "Custom Search API is enabled for that key's project")
+ if r.status_code >= 400:
+ raise GoogleCSEError(f"Google CSE API error {r.status_code}: {r.text[:300]}")
+
+ try:
+ body = r.json()
+ except ValueError as e:
+ raise GoogleCSEError(f"Response bukan JSON valid: {r.text[:300]}") from e
+
+ return body.get("items") or []
+
+
+# ── Public API ───────────────────────────────────────────────────────────────
+
+def google_cse_search(raw_query: str, count: int = 20, config: configparser.ConfigParser = None,
+ cursor: str | None = None) -> tuple[list[dict], str | None]:
+ """Returns (records, next_cursor). `cursor` is the opaque 1-based `start`
+ index from a previous call's next_cursor — pass it back to fetch the next
+ page. None once there's nothing more to load (or the MAX_RESULTS ceiling
+ is hit, to keep one Multi Source Search from burning the whole daily quota)."""
+ query = (raw_query or "").strip()
+ if not query:
+ raise GoogleCSEError("Search query is required")
+
+ api_key, cx = _get_client_config(config)
+
+ try:
+ start = int(cursor) if cursor else 1
+ except ValueError:
+ start = 1
+
+ target = min(max(1, count), MAX_RESULTS)
+ records: list[dict] = []
+ next_cursor = None
+
+ while len(records) < target:
+ items = _fetch_page(query, api_key, cx, start)
+ if not items:
+ break
+ for item in items:
+ records.append(_row_to_record(item))
+ if len(records) >= target:
+ break
+ start += PAGE_SIZE
+ if len(items) < PAGE_SIZE:
+ break # Google itself signaled this was the last page
+ if start > MAX_RESULTS:
+ break
+
+ _enrich_records(records)
+
+ if records and start <= MAX_RESULTS:
+ next_cursor = str(start)
+
+ return records, next_cursor
diff --git a/Script/SOCMINT-Twitter/static/js/card_constants.js b/Script/SOCMINT-Twitter/static/js/card_constants.js
index b30dd59..ad45db0 100644
--- a/Script/SOCMINT-Twitter/static/js/card_constants.js
+++ b/Script/SOCMINT-Twitter/static/js/card_constants.js
@@ -4,6 +4,9 @@
// same data. (Previously included via Jinja {% include %} directly inside a
//