From 820cb6a231673aef6c485120fb53849d5086b1bb Mon Sep 17 00:00:00 2001 From: Jieyab89 Date: Fri, 31 Jul 2026 21:21:15 +0700 Subject: [PATCH] update scripts for socmint twitter --- .gitignore | 12 + Script/SOCMINT-Twitter/Readme.md | 27 + Script/SOCMINT-Twitter/app.py | 94 ++++ Script/SOCMINT-Twitter/config.ini.example | 14 + Script/SOCMINT-Twitter/cookie_client.py | 236 +++++++++ Script/SOCMINT-Twitter/requirements.txt | 3 + Script/SOCMINT-Twitter/templates/index.html | 550 ++++++++++++++++++++ Script/SOCMINT-Twitter/xquik_client.py | 109 ++++ 8 files changed, 1045 insertions(+) create mode 100644 .gitignore create mode 100644 Script/SOCMINT-Twitter/Readme.md create mode 100644 Script/SOCMINT-Twitter/app.py create mode 100644 Script/SOCMINT-Twitter/config.ini.example create mode 100644 Script/SOCMINT-Twitter/cookie_client.py create mode 100644 Script/SOCMINT-Twitter/requirements.txt create mode 100644 Script/SOCMINT-Twitter/templates/index.html create mode 100644 Script/SOCMINT-Twitter/xquik_client.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..4a853fa --- /dev/null +++ b/.gitignore @@ -0,0 +1,12 @@ +.env +.env.* +!.env.example +*.ini +__pycache__/ +*.py[cod] +*.egg-info/ +.venv/ +venv/ +.vscode/ +.idea/ +*.swp diff --git a/Script/SOCMINT-Twitter/Readme.md b/Script/SOCMINT-Twitter/Readme.md new file mode 100644 index 0000000..bbee22a --- /dev/null +++ b/Script/SOCMINT-Twitter/Readme.md @@ -0,0 +1,27 @@ +# Jieyab ft Xquik + +## Setup + +```bash +pip install -r requirements.txt +``` + +Edit `config.ini.example` to config.ini + +## Run Local Web Server + +```bash +python app.py +``` + +Open `http://127.0.0.1:5000` + +# Results + +Xquik Dashboard + +image + +Jieyab SOCMINT Twitter Dashboard + + diff --git a/Script/SOCMINT-Twitter/app.py b/Script/SOCMINT-Twitter/app.py new file mode 100644 index 0000000..79ee549 --- /dev/null +++ b/Script/SOCMINT-Twitter/app.py @@ -0,0 +1,94 @@ +import threading +from flask import Flask, jsonify, render_template, request + +from xquik_client import XquikClient, XquikError, load_config +from cookie_client import ( + cookie_tweet_search, + cookie_follower_explorer, + cookie_post_extractor, + cookie_article_extractor, + cookie_community_post_extractor, + CookieClientError, +) + +app = Flask(__name__) +config = load_config() + +MAX_CONCURRENT_REQUESTS = 3 # parallel execution slots +ACQUIRE_TIMEOUT = 15 # seconds to wait before returning 429 + +_sem = threading.Semaphore(MAX_CONCURRENT_REQUESTS) + + +@app.route("/") +def index(): + return render_template("index.html") + + +@app.route("/api/run", methods=["POST"]) +def run_tool(): + body = request.get_json(silent=True) or {} + tool_type = body.get("toolType") + mode = body.get("mode", "api") # "api" | "cookie" + count = max(1, min(int(body.get("count", 20)), 200)) + + if not _sem.acquire(blocking=True, timeout=ACQUIRE_TIMEOUT): + return jsonify({ + "ok": False, + "error": "Server is busy — max concurrent requests reached. Please try again shortly.", + }), 429 + + try: + if tool_type == "tweet_search_extractor": + query = body.get("searchQuery", "") + if mode == "cookie": + data = cookie_tweet_search(query, count=count, config=config) + else: + data = XquikClient(config).tweet_search(query) + + elif tool_type == "follower_explorer": + username = body.get("targetUsername", "") + if mode == "cookie": + data = cookie_follower_explorer(username, count=count, config=config) + else: + data = XquikClient(config).follower_explorer(username) + + elif tool_type == "article_extractor": + tweet_id = body.get("targetTweetId", "") + if mode == "cookie": + data = cookie_article_extractor(tweet_id, config=config) + else: + data = XquikClient(config).article_extractor(tweet_id) + + elif tool_type == "community_post_extractor": + community_id = body.get("targetCommunityId", "") + if mode == "cookie": + data = cookie_community_post_extractor(community_id, count=count, config=config) + else: + data = XquikClient(config).community_post_extractor(community_id) + + elif tool_type == "post_extractor": + username = body.get("targetUsername", "") + if mode == "cookie": + data = cookie_post_extractor(username, count=count, config=config) + else: + data = XquikClient(config).post_extractor(username) + + else: + return jsonify({"ok": False, "error": f"Unknown toolType: {tool_type}"}), 400 + + return jsonify({"ok": True, "data": data}) + + except (XquikError, CookieClientError) 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 + finally: + _sem.release() + + +if __name__ == "__main__": + host = config.get("server", "host", fallback="127.0.0.1") + port = config.getint("server", "port", fallback=5000) + debug = config.getboolean("server", "debug", fallback=True) + app.run(host=host, port=port, debug=debug, threaded=True) diff --git a/Script/SOCMINT-Twitter/config.ini.example b/Script/SOCMINT-Twitter/config.ini.example new file mode 100644 index 0000000..42245d8 --- /dev/null +++ b/Script/SOCMINT-Twitter/config.ini.example @@ -0,0 +1,14 @@ +[xquik] + +api_key = xxxxxxxxxxxxxxxxxxxxxxxxxxx +base_url = https://xquik.com/api/v1/extractions + +[twitter_cookies] + +auth_token = xxxxxxxxxxxxxxxxxx +ct0 = xxxxxxxxxxxx + +[server] +host = 127.0.0.1 +port = 5000 +debug = true \ No newline at end of file diff --git a/Script/SOCMINT-Twitter/cookie_client.py b/Script/SOCMINT-Twitter/cookie_client.py new file mode 100644 index 0000000..f154c99 --- /dev/null +++ b/Script/SOCMINT-Twitter/cookie_client.py @@ -0,0 +1,236 @@ +""" +cookie_client.py +Cookie-based wrapper for all five xquik tools, hitting X.com directly via +twikit/twifork — no xquik API quota consumed. + +Install: pip install twifork + (NOT the upstream `twikit` package — it is broken since X changed ondemand.s.js. + twifork is a drop-in replacement; imports remain `from twikit import Client`.) + +How to obtain cookies: + 1. Log in to x.com in your browser. + 2. Open DevTools > Application > Cookies > https://x.com + 3. Copy the values of `auth_token` and `ct0`. + 4. Paste them into config.ini under [twitter_cookies]. +""" + +import asyncio +import configparser +import os + +CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.ini") + + +class CookieClientError(Exception): + pass + + +def load_config(path: str = CONFIG_PATH) -> configparser.ConfigParser: + cfg = configparser.ConfigParser() + cfg.read(path) + return cfg + + +def _get_creds(config: configparser.ConfigParser) -> tuple[str, str]: + auth_token = config.get("twitter_cookies", "auth_token", fallback="").strip() + ct0 = config.get("twitter_cookies", "ct0", fallback="").strip() + if not auth_token or not ct0: + raise CookieClientError( + "auth_token / ct0 are not set in config.ini [twitter_cookies]" + ) + return auth_token, ct0 + + +async def _make_client(auth_token: str, ct0: str): + try: + from twikit import Client + except ImportError as e: + raise CookieClientError( + "Required package not installed. Run: pip install twifork " + "(not `twikit` — the upstream package is broken due to X's ondemand.s.js change)" + ) from e + client = Client(language="en-US") + client.set_cookies({"auth_token": auth_token, "ct0": ct0}) + return client + + +# ── Serialisers ─────────────────────────────────────────────────────────────── + +def _extract_media(t: object) -> list: + """Return [{type, thumb, url}] for each media item attached to a tweet.""" + result = [] + for m in getattr(t, "media", None) or []: + mtype = getattr(m, "type", "photo") + thumb = ( + getattr(m, "media_url_https", None) + or getattr(m, "media_url", None) + or getattr(m, "url", None) + ) + if not thumb: + continue + url = thumb + if mtype in ("video", "animated_gif"): + vid = getattr(m, "video_info", None) + if isinstance(vid, dict): + mp4s = [ + v for v in vid.get("variants", []) + if isinstance(v, dict) and v.get("content_type") == "video/mp4" + ] + if mp4s: + url = max(mp4s, key=lambda v: v.get("bitrate", 0)).get("url", thumb) + result.append({"type": mtype, "thumb": thumb, "url": url}) + return result + + +def _tweet_to_dict(t: object) -> dict: + d = { + "id": getattr(t, "id", None), + "created_at": getattr(t, "created_at", None), + "text": getattr(t, "text", None), + "user": getattr(t.user, "screen_name", None) if getattr(t, "user", None) else None, + "reply_count": getattr(t, "reply_count", None), + "retweet_count": getattr(t, "retweet_count", None), + "favorite_count": getattr(t, "favorite_count", None), + "view_count": getattr(t, "view_count", None), + } + media = _extract_media(t) + if media: + d["media"] = media + return d + + +def _user_to_dict(u: object) -> dict: + return { + "id": getattr(u, "id", None), + "name": getattr(u, "name", None), + "screen_name": getattr(u, "screen_name", None), + "description": getattr(u, "description", None), + "followers_count": getattr(u, "followers_count", None), + "following_count": getattr(u, "following_count", None), + "tweet_count": getattr(u, "statuses_count", None), + "created_at": getattr(u, "created_at", None), + "verified": getattr(u, "verified", None), + "is_blue_verified": getattr(u, "is_blue_verified", None), + } + + +# ── Async implementations ───────────────────────────────────────────────────── + +async def _tweet_search_async(query: str, auth_token: str, ct0: str, count: int) -> list: + client = await _make_client(auth_token, ct0) + results = await client.search_tweet(query, "Latest", count=count) + return [_tweet_to_dict(t) for t in results] + + +async def _follower_explorer_async(username: str, auth_token: str, ct0: str, count: int) -> list: + client = await _make_client(auth_token, ct0) + user = await client.get_user_by_screen_name(username) + followers = await user.get_followers(count=count) + return [_user_to_dict(u) for u in followers] + + +async def _post_extractor_async(username: str, auth_token: str, ct0: str, count: int) -> list: + client = await _make_client(auth_token, ct0) + user = await client.get_user_by_screen_name(username) + tweets = await user.get_tweets("Tweets", count=count) + return [_tweet_to_dict(t) for t in tweets] + + +async def _article_extractor_async(tweet_id: str, auth_token: str, ct0: str) -> dict: + client = await _make_client(auth_token, ct0) + tweet = await client.get_tweet_by_id(tweet_id) + result = _tweet_to_dict(tweet) + note = getattr(tweet, "note_tweet", None) + if note: + result["article_text"] = note + card = getattr(tweet, "card", None) + if card: + result["card"] = str(card) + return result + + +async def _community_posts_async(community_id: str, auth_token: str, ct0: str, count: int) -> list: + client = await _make_client(auth_token, ct0) + posts = await client.get_community_tweets(community_id, "Latest", count=count) + return [_tweet_to_dict(t) for t in posts] + + +# ── Public sync wrappers ────────────────────────────────────────────────────── + +def cookie_tweet_search(query: str, count: int = 20, config: configparser.ConfigParser = None) -> list: + cfg = config or load_config() + auth, ct0 = _get_creds(cfg) + return asyncio.run(_tweet_search_async(query, auth, ct0, count)) + + +def cookie_follower_explorer(username: str, count: int = 20, config: configparser.ConfigParser = None) -> list: + cfg = config or load_config() + auth, ct0 = _get_creds(cfg) + return asyncio.run(_follower_explorer_async(username, auth, ct0, count)) + + +def cookie_post_extractor(username: str, count: int = 20, config: configparser.ConfigParser = None) -> list: + cfg = config or load_config() + auth, ct0 = _get_creds(cfg) + return asyncio.run(_post_extractor_async(username, auth, ct0, count)) + + +def cookie_article_extractor(tweet_id: str, config: configparser.ConfigParser = None) -> dict: + cfg = config or load_config() + auth, ct0 = _get_creds(cfg) + return asyncio.run(_article_extractor_async(tweet_id, auth, ct0)) + + +def cookie_community_post_extractor( + community_id: str, count: int = 20, config: configparser.ConfigParser = None +) -> list: + cfg = config or load_config() + auth, ct0 = _get_creds(cfg) + return asyncio.run(_community_posts_async(community_id, auth, ct0, count)) + + +# Legacy alias — kept for any external scripts that import this name directly +fetch_user_timeline = cookie_post_extractor + + +# ── CLI entry-point ─────────────────────────────────────────────────────────── + +if __name__ == "__main__": + import argparse + import json + + parser = argparse.ArgumentParser(description="Cookie-based X data extraction CLI") + sub = parser.add_subparsers(dest="cmd", required=True) + + sub.add_parser("tweet_search").add_argument("query") + + p = sub.add_parser("follower_explorer") + p.add_argument("username") + p.add_argument("--count", type=int, default=20) + + p = sub.add_parser("post_extractor") + p.add_argument("username") + p.add_argument("--count", type=int, default=20) + + sub.add_parser("article_extractor").add_argument("tweet_id") + + p = sub.add_parser("community_post_extractor") + p.add_argument("community_id") + p.add_argument("--count", type=int, default=20) + + args = parser.parse_args() + try: + if args.cmd == "tweet_search": + out = cookie_tweet_search(args.query) + elif args.cmd == "follower_explorer": + out = cookie_follower_explorer(args.username, count=args.count) + elif args.cmd == "post_extractor": + out = cookie_post_extractor(args.username, count=args.count) + elif args.cmd == "article_extractor": + out = cookie_article_extractor(args.tweet_id) + elif args.cmd == "community_post_extractor": + out = cookie_community_post_extractor(args.community_id, count=args.count) + print(json.dumps(out, indent=2, ensure_ascii=False)) + except CookieClientError as e: + print(f"[ERROR] {e}") + raise SystemExit(1) diff --git a/Script/SOCMINT-Twitter/requirements.txt b/Script/SOCMINT-Twitter/requirements.txt new file mode 100644 index 0000000..0045a81 --- /dev/null +++ b/Script/SOCMINT-Twitter/requirements.txt @@ -0,0 +1,3 @@ +flask +requests +twifork \ No newline at end of file diff --git a/Script/SOCMINT-Twitter/templates/index.html b/Script/SOCMINT-Twitter/templates/index.html new file mode 100644 index 0000000..5fd603d --- /dev/null +++ b/Script/SOCMINT-Twitter/templates/index.html @@ -0,0 +1,550 @@ + + + + + +Jieyab89 SOCMINT Twitter / X + + + + +
+

Jieyab89 SOCMINT X

+ | + Retrieve data using Xquik API or Cookie +
+ +
+
+ +
+ + +
+
+ +
+ +
+ + +
+ +
+
+
idle
+ +
+
+
+
Nothing found :(
+
+
+
+ + + + + diff --git a/Script/SOCMINT-Twitter/xquik_client.py b/Script/SOCMINT-Twitter/xquik_client.py new file mode 100644 index 0000000..de90df9 --- /dev/null +++ b/Script/SOCMINT-Twitter/xquik_client.py @@ -0,0 +1,109 @@ +import configparser +import os +import requests + +CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.ini") + + +class XquikError(Exception): + pass + + +def load_config(path: str = CONFIG_PATH) -> configparser.ConfigParser: + if not os.path.exists(path): + raise XquikError(f"Config tidak ditemukan: {path}") + cfg = configparser.ConfigParser() + cfg.read(path) + return cfg + + +class XquikClient: + def __init__(self, config: configparser.ConfigParser = None): + self.config = config or load_config() + self.api_key = self.config.get("xquik", "api_key", fallback="").strip() + self.base_url = self.config.get( + "xquik", "base_url", fallback="https://xquik.com/api/v1/extractions" + ).strip() + + if not self.api_key or self.api_key == "xq_YOUR_KEY": + raise XquikError("api_key belum diisi di config.ini [xquik]") + + def _post(self, payload: dict) -> dict: + headers = { + "x-api-key": self.api_key, + "Content-Type": "application/json", + } + try: + resp = requests.post(self.base_url, headers=headers, json=payload, timeout=30) + except requests.RequestException as e: + raise XquikError(f"Request gagal: {e}") from e + + if resp.status_code >= 400: + raise XquikError(f"xquik API error {resp.status_code}: {resp.text}") + + try: + return resp.json() + except ValueError as e: + raise XquikError(f"Response bukan JSON valid: {resp.text[:300]}") from e + + # --- 5 tool type sesuai docs --- + + def tweet_search(self, search_query: str) -> dict: + return self._post({"toolType": "tweet_search_extractor", "searchQuery": search_query}) + + def follower_explorer(self, target_username: str) -> dict: + return self._post({"toolType": "follower_explorer", "targetUsername": target_username}) + + def article_extractor(self, target_tweet_id: str) -> dict: + return self._post({"toolType": "article_extractor", "targetTweetId": target_tweet_id}) + + def community_post_extractor(self, target_community_id: str) -> dict: + return self._post( + {"toolType": "community_post_extractor", "targetCommunityId": target_community_id} + ) + + def post_extractor(self, target_username: str) -> dict: + """User timeline lewat xquik (pakai kuota API).""" + return self._post({"toolType": "post_extractor", "targetUsername": target_username}) + + +if __name__ == "__main__": + import argparse + import json + + parser = argparse.ArgumentParser(description="xquik.com extraction CLI") + sub = parser.add_subparsers(dest="command", required=True) + + p1 = sub.add_parser("tweet_search") + p1.add_argument("query") + + p2 = sub.add_parser("follower_explorer") + p2.add_argument("username") + + p3 = sub.add_parser("article_extractor") + p3.add_argument("tweet_id") + + p4 = sub.add_parser("community_post_extractor") + p4.add_argument("community_id") + + p5 = sub.add_parser("post_extractor") + p5.add_argument("username") + + args = parser.parse_args() + client = XquikClient() + + try: + if args.command == "tweet_search": + out = client.tweet_search(args.query) + elif args.command == "follower_explorer": + out = client.follower_explorer(args.username) + elif args.command == "article_extractor": + out = client.article_extractor(args.tweet_id) + elif args.command == "community_post_extractor": + out = client.community_post_extractor(args.community_id) + elif args.command == "post_extractor": + out = client.post_extractor(args.username) + print(json.dumps(out, indent=2, ensure_ascii=False)) + except XquikError as e: + print(f"[ERROR] {e}") + raise SystemExit(1) \ No newline at end of file