Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import json
- import re
- import time
- import unicodedata
- from itertools import combinations
- from pathlib import Path
- # need pip install requests openpyxl
- import requests
- from openpyxl import Workbook
- from openpyxl.styles import Alignment, Font
- # ============================================================
- # Config
- # ============================================================
- INPUT_FILE = "badAccount.txt"
- OUTPUT_XLSX = "mverse_analysis.xlsx"
- CACHE_FILE = "mverse_cache.json"
- PROFILE_API = "https://mverse-api.nexon.com/social/v1/profile/{ppsn}"
- FRIENDS_API = "https://mverse-api.nexon.com/social/v1/{ppsn}/friends/all"
- REQUEST_DELAY = 0
- TIMEOUT = 15
- MAX_RETRIES = 3
- MIN_COLUMN_WIDTH = 10
- MAX_COLUMN_WIDTH = 50
- COLUMN_PADDING = 3
- session = requests.Session()
- session.headers.update({
- "User-Agent": "Mozilla/5.0",
- "Accept": "application/json",
- })
- # ============================================================
- # Input
- # ============================================================
- def load_ppsns(filename):
- text = Path(filename).read_text(
- encoding="utf-8",
- errors="ignore",
- )
- ppsns = re.findall(
- r"\b\d{15,20}\b",
- text,
- )
- return list(dict.fromkeys(ppsns))
- # ============================================================
- # HTTP
- # ============================================================
- def request_json(url):
- for attempt in range(1, MAX_RETRIES + 1):
- try:
- response = session.get(
- url,
- timeout=TIMEOUT,
- )
- if response.status_code == 200:
- try:
- return response.json()
- except ValueError:
- return None
- if response.status_code in {
- 400,
- 401,
- 403,
- 404,
- }:
- return None
- except requests.RequestException:
- pass
- if attempt < MAX_RETRIES:
- time.sleep(attempt)
- return None
- # ============================================================
- # Generic JSON helper
- # ============================================================
- def recursive_find(obj, key):
- if isinstance(obj, dict):
- if key in obj:
- return obj[key]
- for value in obj.values():
- result = recursive_find(
- value,
- key,
- )
- if result is not None:
- return result
- elif isinstance(obj, list):
- for item in obj:
- result = recursive_find(
- item,
- key,
- )
- if result is not None:
- return result
- return None
- # ============================================================
- # Profile
- # ============================================================
- def get_profile(ppsn):
- url = PROFILE_API.format(
- ppsn=ppsn
- )
- data = request_json(url)
- if not data:
- return {
- "ppsn": ppsn,
- "profileCode": "",
- "profileName": "",
- "status": "FAILED",
- "friendCount": 0,
- }
- profile_code = (
- recursive_find(
- data,
- "profileCode",
- )
- or ""
- )
- profile_name = (
- recursive_find(
- data,
- "profileName",
- )
- or recursive_find(
- data,
- "nickname",
- )
- or recursive_find(
- data,
- "characterName",
- )
- or recursive_find(
- data,
- "displayName",
- )
- or recursive_find(
- data,
- "name",
- )
- or ""
- )
- return {
- "ppsn": str(ppsn),
- "profileCode": str(profile_code),
- "profileName": str(profile_name),
- "status": "OK",
- "friendCount": 0,
- }
- # ============================================================
- # Friends
- # ============================================================
- def normalize_friend(friend):
- if not isinstance(friend, dict):
- return None
- ppsn = str(
- friend.get("ppsn")
- or ""
- )
- profile_code = str(
- friend.get("profileCode")
- or ""
- )
- profile_name = str(
- friend.get("profileName")
- or ""
- )
- if not ppsn:
- return None
- return {
- "ppsn": ppsn,
- "profileCode": profile_code,
- "profileName": profile_name,
- }
- def get_friends(ppsn):
- url = FRIENDS_API.format(
- ppsn=ppsn
- )
- data = request_json(url)
- if not data:
- return {}
- if data.get("code") != 0:
- return {}
- api_data = data.get("data")
- if not isinstance(
- api_data,
- dict,
- ):
- return {}
- raw_friends = api_data.get(
- "result"
- )
- if not isinstance(
- raw_friends,
- list,
- ):
- return {}
- result = {}
- for friend in raw_friends:
- normalized = normalize_friend(
- friend
- )
- if not normalized:
- continue
- friend_ppsn = normalized[
- "ppsn"
- ]
- result[
- friend_ppsn
- ] = normalized
- return result
- # ============================================================
- # Shared friends
- # ============================================================
- def build_shared_friends(
- profiles_by_ppsn,
- friends_by_ppsn,
- ):
- rows = []
- suspicious_ppsns = list(
- friends_by_ppsn.keys()
- )
- for ppsn_a, ppsn_b in combinations(
- suspicious_ppsns,
- 2,
- ):
- friends_a = friends_by_ppsn[
- ppsn_a
- ]
- friends_b = friends_by_ppsn[
- ppsn_b
- ]
- shared_ppsns = (
- set(friends_a.keys())
- &
- set(friends_b.keys())
- )
- if not shared_ppsns:
- continue
- profile_a = profiles_by_ppsn[
- ppsn_a
- ]
- profile_b = profiles_by_ppsn[
- ppsn_b
- ]
- shared_count = len(
- shared_ppsns
- )
- for shared_ppsn in sorted(
- shared_ppsns
- ):
- friend_a = friends_a[
- shared_ppsn
- ]
- friend_b = friends_b[
- shared_ppsn
- ]
- shared_profile_code = (
- friend_a["profileCode"]
- or friend_b["profileCode"]
- )
- shared_profile_name = (
- friend_a["profileName"]
- or friend_b["profileName"]
- )
- rows.append({
- "accountA_PPSN":
- ppsn_a,
- "accountA_ProfileCode":
- profile_a[
- "profileCode"
- ],
- "accountA_ProfileName":
- profile_a[
- "profileName"
- ],
- "accountB_PPSN":
- ppsn_b,
- "accountB_ProfileCode":
- profile_b[
- "profileCode"
- ],
- "accountB_ProfileName":
- profile_b[
- "profileName"
- ],
- "sharedFriend_PPSN":
- shared_ppsn,
- "sharedFriend_ProfileCode":
- shared_profile_code,
- "sharedFriend_ProfileName":
- shared_profile_name,
- "sharedFriendCount":
- shared_count,
- })
- rows.sort(
- key=lambda x: (
- -x[
- "sharedFriendCount"
- ],
- x[
- "accountA_ProfileCode"
- ],
- x[
- "accountB_ProfileCode"
- ],
- x[
- "sharedFriend_PPSN"
- ],
- )
- )
- return rows
- # ============================================================
- # Pair summary
- # ============================================================
- def build_pair_summary(
- shared_rows,
- ):
- pair_data = {}
- for row in shared_rows:
- key = (
- row[
- "accountA_PPSN"
- ],
- row[
- "accountB_PPSN"
- ],
- )
- if key not in pair_data:
- pair_data[key] = {
- "accountA_PPSN":
- row[
- "accountA_PPSN"
- ],
- "accountA_ProfileCode":
- row[
- "accountA_ProfileCode"
- ],
- "accountA_ProfileName":
- row[
- "accountA_ProfileName"
- ],
- "accountB_PPSN":
- row[
- "accountB_PPSN"
- ],
- "accountB_ProfileCode":
- row[
- "accountB_ProfileCode"
- ],
- "accountB_ProfileName":
- row[
- "accountB_ProfileName"
- ],
- "sharedFriendCount":
- row[
- "sharedFriendCount"
- ],
- "sharedFriendPPSNs":
- [],
- "sharedFriendProfileCodes":
- [],
- "sharedFriendProfileNames":
- [],
- }
- pair_data[key][
- "sharedFriendPPSNs"
- ].append(
- row[
- "sharedFriend_PPSN"
- ]
- )
- pair_data[key][
- "sharedFriendProfileCodes"
- ].append(
- row[
- "sharedFriend_ProfileCode"
- ]
- )
- pair_data[key][
- "sharedFriendProfileNames"
- ].append(
- row[
- "sharedFriend_ProfileName"
- ]
- )
- rows = []
- for data in pair_data.values():
- rows.append({
- "accountA_PPSN":
- data[
- "accountA_PPSN"
- ],
- "accountA_ProfileCode":
- data[
- "accountA_ProfileCode"
- ],
- "accountA_ProfileName":
- data[
- "accountA_ProfileName"
- ],
- "accountB_PPSN":
- data[
- "accountB_PPSN"
- ],
- "accountB_ProfileCode":
- data[
- "accountB_ProfileCode"
- ],
- "accountB_ProfileName":
- data[
- "accountB_ProfileName"
- ],
- "sharedFriendCount":
- data[
- "sharedFriendCount"
- ],
- "sharedFriendPPSNs":
- " | ".join(
- data[
- "sharedFriendPPSNs"
- ]
- ),
- "sharedFriendProfileCodes":
- " | ".join(
- data[
- "sharedFriendProfileCodes"
- ]
- ),
- "sharedFriendProfileNames":
- " | ".join(
- data[
- "sharedFriendProfileNames"
- ]
- ),
- })
- rows.sort(
- key=lambda x: (
- -x[
- "sharedFriendCount"
- ],
- x[
- "accountA_ProfileCode"
- ],
- x[
- "accountB_ProfileCode"
- ],
- )
- )
- return rows
- # ============================================================
- # Friend clusters
- # ============================================================
- def build_friend_clusters(
- profiles_by_ppsn,
- friends_by_ppsn,
- ):
- friend_to_suspects = {}
- friend_info = {}
- for suspect_ppsn, friends in (
- friends_by_ppsn.items()
- ):
- for friend_ppsn, friend in (
- friends.items()
- ):
- friend_to_suspects.setdefault(
- friend_ppsn,
- set(),
- )
- friend_to_suspects[
- friend_ppsn
- ].add(
- suspect_ppsn
- )
- if friend_ppsn not in friend_info:
- friend_info[
- friend_ppsn
- ] = {
- "ppsn":
- friend_ppsn,
- "profileCode":
- friend[
- "profileCode"
- ],
- "profileName":
- friend[
- "profileName"
- ],
- }
- else:
- old = friend_info[
- friend_ppsn
- ]
- if not old[
- "profileCode"
- ]:
- old[
- "profileCode"
- ] = friend[
- "profileCode"
- ]
- if not old[
- "profileName"
- ]:
- old[
- "profileName"
- ] = friend[
- "profileName"
- ]
- rows = []
- for friend_ppsn, suspects in (
- friend_to_suspects.items()
- ):
- if len(suspects) < 2:
- continue
- friend = friend_info[
- friend_ppsn
- ]
- suspect_info = []
- for suspect_ppsn in sorted(
- suspects
- ):
- profile = profiles_by_ppsn[
- suspect_ppsn
- ]
- suspect_info.append({
- "ppsn":
- suspect_ppsn,
- "profileCode":
- profile[
- "profileCode"
- ],
- "profileName":
- profile[
- "profileName"
- ],
- })
- rows.append({
- "friendPPSN":
- friend_ppsn,
- "friendProfileCode":
- friend[
- "profileCode"
- ],
- "friendProfileName":
- friend[
- "profileName"
- ],
- "suspectCount":
- len(suspects),
- "suspectPPSNs":
- " | ".join(
- x["ppsn"]
- for x
- in suspect_info
- ),
- "suspectProfileCodes":
- " | ".join(
- x["profileCode"]
- for x
- in suspect_info
- ),
- "suspectProfileNames":
- " | ".join(
- x["profileName"]
- for x
- in suspect_info
- ),
- })
- rows.sort(
- key=lambda x: (
- -x[
- "suspectCount"
- ],
- x[
- "friendProfileCode"
- ],
- x[
- "friendPPSN"
- ],
- )
- )
- return rows
- # ============================================================
- # Excel display width
- # ============================================================
- def get_display_width(value):
- if value is None:
- return 0
- text = str(value)
- width = 0
- for char in text:
- east_asian_width = (
- unicodedata.east_asian_width(
- char
- )
- )
- if east_asian_width in {
- "W",
- "F",
- }:
- width += 2
- else:
- width += 1
- return width
- def auto_fit_worksheet(ws):
- for column_cells in ws.columns:
- column_letter = (
- column_cells[
- 0
- ].column_letter
- )
- max_length = 0
- for cell in column_cells:
- if cell.value is None:
- continue
- value = str(
- cell.value
- )
- lines = value.splitlines()
- if not lines:
- lines = [""]
- display_length = max(
- get_display_width(
- line
- )
- for line in lines
- )
- max_length = max(
- max_length,
- display_length,
- )
- wanted_width = (
- max_length
- + COLUMN_PADDING
- )
- if wanted_width < MIN_COLUMN_WIDTH:
- wanted_width = (
- MIN_COLUMN_WIDTH
- )
- if wanted_width > MAX_COLUMN_WIDTH:
- ws.column_dimensions[
- column_letter
- ].width = (
- MAX_COLUMN_WIDTH
- )
- for cell in column_cells:
- cell.alignment = Alignment(
- wrap_text=True,
- vertical="top",
- )
- else:
- ws.column_dimensions[
- column_letter
- ].width = (
- wanted_width
- )
- # ============================================================
- # Excel helpers
- # ============================================================
- def format_sheet(ws):
- for cell in ws[1]:
- cell.font = Font(
- bold=True
- )
- cell.alignment = Alignment(
- vertical="center",
- )
- ws.freeze_panes = "A2"
- ws.auto_filter.ref = (
- ws.dimensions
- )
- auto_fit_worksheet(
- ws
- )
- def force_text(
- ws,
- row,
- columns,
- ):
- for column in columns:
- cell = ws.cell(
- row=row,
- column=column,
- )
- cell.number_format = "@"
- # ============================================================
- # Excel output
- # ============================================================
- def write_xlsx(
- profiles_by_ppsn,
- friends_by_ppsn,
- shared_rows,
- pair_rows,
- cluster_rows,
- ):
- wb = Workbook()
- # --------------------------------------------------------
- # Profiles
- # --------------------------------------------------------
- ws = wb.active
- ws.title = "Profiles"
- ws.append([
- "PPSN",
- "ProfileCode",
- "ProfileName",
- "FriendCount",
- "Status",
- ])
- profiles = list(
- profiles_by_ppsn.values()
- )
- profiles.sort(
- key=lambda x: (
- x["profileCode"] == "",
- x["profileCode"],
- x["ppsn"],
- )
- )
- for profile in profiles:
- ws.append([
- str(
- profile[
- "ppsn"
- ]
- ),
- profile[
- "profileCode"
- ],
- profile[
- "profileName"
- ],
- profile[
- "friendCount"
- ],
- profile[
- "status"
- ],
- ])
- force_text(
- ws,
- ws.max_row,
- [1],
- )
- format_sheet(ws)
- # --------------------------------------------------------
- # All Friends
- # --------------------------------------------------------
- ws = wb.create_sheet(
- "All Friends"
- )
- ws.append([
- "Owner PPSN",
- "Owner ProfileCode",
- "Owner ProfileName",
- "Friend PPSN",
- "Friend ProfileCode",
- "Friend ProfileName",
- ])
- all_friend_rows = []
- for owner_ppsn, friends in (
- friends_by_ppsn.items()
- ):
- owner = profiles_by_ppsn[
- owner_ppsn
- ]
- for friend_ppsn, friend in (
- friends.items()
- ):
- all_friend_rows.append({
- "ownerPPSN":
- owner_ppsn,
- "ownerProfileCode":
- owner[
- "profileCode"
- ],
- "ownerProfileName":
- owner[
- "profileName"
- ],
- "friendPPSN":
- friend_ppsn,
- "friendProfileCode":
- friend[
- "profileCode"
- ],
- "friendProfileName":
- friend[
- "profileName"
- ],
- })
- all_friend_rows.sort(
- key=lambda x: (
- x[
- "ownerProfileCode"
- ] == "",
- x[
- "ownerProfileCode"
- ],
- x[
- "ownerPPSN"
- ],
- x[
- "friendProfileCode"
- ],
- x[
- "friendPPSN"
- ],
- )
- )
- for item in all_friend_rows:
- ws.append([
- str(
- item[
- "ownerPPSN"
- ]
- ),
- item[
- "ownerProfileCode"
- ],
- item[
- "ownerProfileName"
- ],
- str(
- item[
- "friendPPSN"
- ]
- ),
- item[
- "friendProfileCode"
- ],
- item[
- "friendProfileName"
- ],
- ])
- force_text(
- ws,
- ws.max_row,
- [1, 4],
- )
- format_sheet(ws)
- # --------------------------------------------------------
- # Shared Friends
- # --------------------------------------------------------
- ws = wb.create_sheet(
- "Shared Friends"
- )
- ws.append([
- "Account A PPSN",
- "Account A ProfileCode",
- "Account A ProfileName",
- "Account B PPSN",
- "Account B ProfileCode",
- "Account B ProfileName",
- "Shared Friend PPSN",
- "Shared Friend ProfileCode",
- "Shared Friend ProfileName",
- "Shared Friend Count",
- ])
- for item in shared_rows:
- ws.append([
- str(
- item[
- "accountA_PPSN"
- ]
- ),
- item[
- "accountA_ProfileCode"
- ],
- item[
- "accountA_ProfileName"
- ],
- str(
- item[
- "accountB_PPSN"
- ]
- ),
- item[
- "accountB_ProfileCode"
- ],
- item[
- "accountB_ProfileName"
- ],
- str(
- item[
- "sharedFriend_PPSN"
- ]
- ),
- item[
- "sharedFriend_ProfileCode"
- ],
- item[
- "sharedFriend_ProfileName"
- ],
- item[
- "sharedFriendCount"
- ],
- ])
- force_text(
- ws,
- ws.max_row,
- [1, 4, 7],
- )
- format_sheet(ws)
- # --------------------------------------------------------
- # Pair Summary
- # --------------------------------------------------------
- ws = wb.create_sheet(
- "Pair Summary"
- )
- ws.append([
- "Account A PPSN",
- "Account A ProfileCode",
- "Account A ProfileName",
- "Account B PPSN",
- "Account B ProfileCode",
- "Account B ProfileName",
- "Shared Friend Count",
- "Shared Friend PPSNs",
- "Shared Friend ProfileCodes",
- "Shared Friend ProfileNames",
- ])
- for item in pair_rows:
- ws.append([
- str(
- item[
- "accountA_PPSN"
- ]
- ),
- item[
- "accountA_ProfileCode"
- ],
- item[
- "accountA_ProfileName"
- ],
- str(
- item[
- "accountB_PPSN"
- ]
- ),
- item[
- "accountB_ProfileCode"
- ],
- item[
- "accountB_ProfileName"
- ],
- item[
- "sharedFriendCount"
- ],
- item[
- "sharedFriendPPSNs"
- ],
- item[
- "sharedFriendProfileCodes"
- ],
- item[
- "sharedFriendProfileNames"
- ],
- ])
- force_text(
- ws,
- ws.max_row,
- [1, 4, 8],
- )
- format_sheet(ws)
- # --------------------------------------------------------
- # Friend Clusters
- # --------------------------------------------------------
- ws = wb.create_sheet(
- "Friend Clusters"
- )
- ws.append([
- "Friend PPSN",
- "Friend ProfileCode",
- "Friend ProfileName",
- "Suspect Count",
- "Suspect PPSNs",
- "Suspect ProfileCodes",
- "Suspect ProfileNames",
- ])
- for item in cluster_rows:
- ws.append([
- str(
- item[
- "friendPPSN"
- ]
- ),
- item[
- "friendProfileCode"
- ],
- item[
- "friendProfileName"
- ],
- item[
- "suspectCount"
- ],
- item[
- "suspectPPSNs"
- ],
- item[
- "suspectProfileCodes"
- ],
- item[
- "suspectProfileNames"
- ],
- ])
- force_text(
- ws,
- ws.max_row,
- [1, 5],
- )
- format_sheet(ws)
- wb.save(
- OUTPUT_XLSX
- )
- # ============================================================
- # Cache
- # ============================================================
- def save_cache(
- profiles_by_ppsn,
- friends_by_ppsn,
- ):
- cache = {
- "profiles":
- profiles_by_ppsn,
- "friends":
- friends_by_ppsn,
- }
- with open(
- CACHE_FILE,
- "w",
- encoding="utf-8",
- ) as f:
- json.dump(
- cache,
- f,
- ensure_ascii=False,
- indent=2,
- )
- # ============================================================
- # Main
- # ============================================================
- def main():
- ppsns = load_ppsns(
- INPUT_FILE
- )
- if not ppsns:
- return
- profiles_by_ppsn = {}
- friends_by_ppsn = {}
- for ppsn in ppsns:
- profile = get_profile(
- ppsn
- )
- profiles_by_ppsn[
- ppsn
- ] = profile
- time.sleep(
- REQUEST_DELAY
- )
- for ppsn in ppsns:
- profile = profiles_by_ppsn[
- ppsn
- ]
- friends = get_friends(
- ppsn
- )
- friends_by_ppsn[
- ppsn
- ] = friends
- profile[
- "friendCount"
- ] = len(
- friends
- )
- time.sleep(
- REQUEST_DELAY
- )
- shared_rows = (
- build_shared_friends(
- profiles_by_ppsn,
- friends_by_ppsn,
- )
- )
- pair_rows = (
- build_pair_summary(
- shared_rows
- )
- )
- cluster_rows = (
- build_friend_clusters(
- profiles_by_ppsn,
- friends_by_ppsn,
- )
- )
- write_xlsx(
- profiles_by_ppsn,
- friends_by_ppsn,
- shared_rows,
- pair_rows,
- cluster_rows,
- )
- save_cache(
- profiles_by_ppsn,
- friends_by_ppsn,
- )
- if __name__ == "__main__":
- main()
- print("done!")
Advertisement
Add Comment
Please, Sign In to add comment