#!/usr/bin/env python3 """ Author: Michal Szymanski v1.6 Tool implementing real-time tracking of Sony PlayStation (PSN) players activities: https://github.com/misiektoja/psn_monitor/ Python pip3 requirements: PSNAWP requests python-dateutil pytz tzlocal (optional) python-dotenv (optional) """ VERSION = "1.6" # --------------------------- # CONFIGURATION SECTION START # --------------------------- CONFIG_BLOCK = """ # Log in to your PSN account: # https://my.playstation.com/ # # In another tab, visit: # https://ca.account.sony.com/api/v1/ssocookie # # Copy the value of the npsso code # # Provide the PSN_NPSSO secret using one of the following methods: # - Pass it at runtime with -n / --npsso-key # - Set it as an environment variable (e.g. export PSN_NPSSO=...) # - Add it to ".env" file (PSN_NPSSO=...) for persistent use # Fallback: # - Hard-code it in the code or config file # # The refresh token generated from the npsso should remain valid for about 2 months PSN_NPSSO = "your_psn_npsso_code" # SMTP settings for sending email notifications # If left as-is, no notifications will be sent # # Provide the SMTP_PASSWORD secret using one of the following methods: # - Set it as an environment variable (e.g. export SMTP_PASSWORD=...) # - Add it to ".env" file (SMTP_PASSWORD=...) for persistent use # Fallback: # - Hard-code it in the code or config file SMTP_HOST = "your_smtp_server_ssl" SMTP_PORT = 587 SMTP_USER = "your_smtp_user" SMTP_PASSWORD = "your_smtp_password" SMTP_SSL = True SENDER_EMAIL = "your_sender_email" RECEIVER_EMAIL = "your_receiver_email" # Whether to send an email when user goes online/offline # Can also be enabled via the -a flag ACTIVE_INACTIVE_NOTIFICATION = False # Whether to send an email on game start/change/stop # Can also be enabled via the -g flag GAME_CHANGE_NOTIFICATION = False # Whether to send an email on errors # Can also be disabled via the -e flag ERROR_NOTIFICATION = True # How often to check for player activity when the user is offline; in seconds # Can also be set using the -c flag PSN_CHECK_INTERVAL = 180 # 3 min # How often to check for player activity when the user is online; in seconds # Can also be set using the -k flag PSN_ACTIVE_CHECK_INTERVAL = 60 # 1 min # Set your local time zone so that PSN API timestamps are converted accordingly (e.g. 'Europe/Warsaw'). # Use this command to list all time zones supported by pytz: # python3 -c "import pytz; print('\\n'.join(pytz.all_timezones))" # If set to 'Auto', the tool will try to detect your local time zone automatically (requires tzlocal) LOCAL_TIMEZONE = 'Auto' # If the user disconnects (offline) and reconnects (online) within OFFLINE_INTERRUPT seconds, # the online session start time will be restored to the previous session's start time (short offline interruption), # and previous session statistics (like total playtime and number of played games) will be preserved OFFLINE_INTERRUPT = 420 # 7 mins # How often to print a "liveness check" message to the output; in seconds # Set to 0 to disable LIVENESS_CHECK_INTERVAL = 43200 # 12 hours # URL used to verify internet connectivity at startup CHECK_INTERNET_URL = 'https://ca.account.sony.com/' # Timeout used when checking initial internet connectivity; in seconds CHECK_INTERNET_TIMEOUT = 5 # CSV file to write all status & game changes # Can also be set using the -b flag CSV_FILE = "" # Location of the optional dotenv file which can keep secrets # If not specified it will try to auto-search for .env files # To disable auto-search, set this to the literal string "none" # Can also be set using the --env-file flag DOTENV_FILE = "" # Base name for the log file. Output will be saved to psn_monitor_.log # Can include a directory path to specify the location, e.g. ~/some_dir/psn_monitor PSN_LOGFILE = "psn_monitor" # Whether to disable logging to psn_monitor_.log # Can also be disabled via the -d flag DISABLE_LOGGING = False # Width of horizontal line HORIZONTAL_LINE = 113 # Whether to clear the terminal screen after starting the tool CLEAR_SCREEN = True # Value used by signal handlers increasing/decreasing the check for player activity # when user is online (PSN_ACTIVE_CHECK_INTERVAL); in seconds PSN_ACTIVE_CHECK_SIGNAL_VALUE = 30 # 30 seconds """ # ------------------------- # CONFIGURATION SECTION END # ------------------------- # Default dummy values so linters shut up # Do not change values below - modify them in the configuration section or config file instead PSN_NPSSO = "" SMTP_HOST = "" SMTP_PORT = 0 SMTP_USER = "" SMTP_PASSWORD = "" SMTP_SSL = False SENDER_EMAIL = "" RECEIVER_EMAIL = "" ACTIVE_INACTIVE_NOTIFICATION = False GAME_CHANGE_NOTIFICATION = False ERROR_NOTIFICATION = False PSN_CHECK_INTERVAL = 0 PSN_ACTIVE_CHECK_INTERVAL = 0 LOCAL_TIMEZONE = "" OFFLINE_INTERRUPT = 0 LIVENESS_CHECK_INTERVAL = 0 CHECK_INTERNET_URL = "" CHECK_INTERNET_TIMEOUT = 0 CSV_FILE = "" DOTENV_FILE = "" PSN_LOGFILE = "" DISABLE_LOGGING = False HORIZONTAL_LINE = 0 CLEAR_SCREEN = False PSN_ACTIVE_CHECK_SIGNAL_VALUE = 0 exec(CONFIG_BLOCK, globals()) # Default name for the optional config file DEFAULT_CONFIG_FILENAME = "psn_monitor.conf" # List of secret keys to load from env/config SECRET_KEYS = ("PSN_NPSSO", "SMTP_PASSWORD") # Default value for timeouts in alarm signal handler; in seconds FUNCTION_TIMEOUT = 15 LIVENESS_CHECK_COUNTER = LIVENESS_CHECK_INTERVAL / PSN_CHECK_INTERVAL stdout_bck = None csvfieldnames = ['Date', 'Status', 'Game name'] CLI_CONFIG_PATH = None # to solve the issue: 'SyntaxError: f-string expression part cannot include a backslash' nl_ch = "\n" import sys if sys.version_info < (3, 10): print("* Error: Python version 3.10 or higher required !") sys.exit(1) import time import string import json import os from datetime import datetime, timezone from dateutil import relativedelta from dateutil.parser import isoparse import calendar import requests as req import signal import smtplib import ssl from email.header import Header from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText import argparse import csv try: import pytz except ModuleNotFoundError: raise SystemExit("Error: Couldn't find the pytz library !\n\nTo install it, run:\n pip3 install pytz\n\nOnce installed, re-run this tool") try: from tzlocal import get_localzone except ImportError: get_localzone = None import platform import re import ipaddress try: from psnawp_api import PSNAWP from psnawp_api.core.psnawp_exceptions import PSNAWPAuthenticationError except ModuleNotFoundError: raise SystemExit("Error: Couldn't find the PSNAWP library !\n\nTo install it, run:\n pip3 install PSNAWP\n\nOnce installed, re-run this tool. For more help, visit:\nhttps://github.com/isFakeAccount/psnawp") import shutil from pathlib import Path # Logger class to output messages to stdout and log file class Logger(object): def __init__(self, filename): self.terminal = sys.stdout self.logfile = open(filename, "a", buffering=1, encoding="utf-8") def write(self, message): self.terminal.write(message) self.logfile.write(message) self.terminal.flush() self.logfile.flush() def flush(self): pass # Class used to generate timeout exceptions class TimeoutException(Exception): pass # Signal handler for SIGALRM when the operation times out def timeout_handler(sig, frame): raise TimeoutException # Signal handler when user presses Ctrl+C def signal_handler(sig, frame): sys.stdout = stdout_bck print('\n* You pressed Ctrl+C, tool is terminated.') sys.exit(0) # Checks internet connectivity def check_internet(url=CHECK_INTERNET_URL, timeout=CHECK_INTERNET_TIMEOUT): try: _ = req.get(url, timeout=timeout) return True except req.RequestException as e: print(f"* No connectivity, please check your network:\n\n{e}") return False # Clears the terminal screen def clear_screen(enabled=True): if not enabled: return try: if platform.system() == 'Windows': os.system('cls') else: os.system('clear') except Exception: print("* Cannot clear the screen contents") # Converts absolute value of seconds to human readable format def display_time(seconds, granularity=2): intervals = ( ('years', 31556952), # approximation ('months', 2629746), # approximation ('weeks', 604800), # 60 * 60 * 24 * 7 ('days', 86400), # 60 * 60 * 24 ('hours', 3600), # 60 * 60 ('minutes', 60), ('seconds', 1), ) result = [] if seconds > 0: for name, count in intervals: value = seconds // count if value: seconds -= value * count if value == 1: name = name.rstrip('s') result.append(f"{value} {name}") return ', '.join(result[:granularity]) else: return '0 seconds' # Calculates time span between two timestamps, accepts timestamp integers, floats and datetime objects def calculate_timespan(timestamp1, timestamp2, show_weeks=True, show_hours=True, show_minutes=True, show_seconds=True, granularity=3): result = [] intervals = ['years', 'months', 'weeks', 'days', 'hours', 'minutes', 'seconds'] ts1 = timestamp1 ts2 = timestamp2 if isinstance(timestamp1, str): try: timestamp1 = isoparse(timestamp1) except Exception: return "" if isinstance(timestamp1, int): dt1 = datetime.fromtimestamp(int(ts1), tz=timezone.utc) elif isinstance(timestamp1, float): ts1 = int(round(ts1)) dt1 = datetime.fromtimestamp(ts1, tz=timezone.utc) elif isinstance(timestamp1, datetime): dt1 = timestamp1 if dt1.tzinfo is None: dt1 = pytz.utc.localize(dt1) else: dt1 = dt1.astimezone(pytz.utc) ts1 = int(round(dt1.timestamp())) else: return "" if isinstance(timestamp2, str): try: timestamp2 = isoparse(timestamp2) except Exception: return "" if isinstance(timestamp2, int): dt2 = datetime.fromtimestamp(int(ts2), tz=timezone.utc) elif isinstance(timestamp2, float): ts2 = int(round(ts2)) dt2 = datetime.fromtimestamp(ts2, tz=timezone.utc) elif isinstance(timestamp2, datetime): dt2 = timestamp2 if dt2.tzinfo is None: dt2 = pytz.utc.localize(dt2) else: dt2 = dt2.astimezone(pytz.utc) ts2 = int(round(dt2.timestamp())) else: return "" if ts1 >= ts2: ts_diff = ts1 - ts2 else: ts_diff = ts2 - ts1 dt1, dt2 = dt2, dt1 if ts_diff > 0: date_diff = relativedelta.relativedelta(dt1, dt2) years = date_diff.years months = date_diff.months days_total = date_diff.days if show_weeks: weeks = days_total // 7 days = days_total % 7 else: weeks = 0 days = days_total hours = date_diff.hours if show_hours or ts_diff <= 86400 else 0 minutes = date_diff.minutes if show_minutes or ts_diff <= 3600 else 0 seconds = date_diff.seconds if show_seconds or ts_diff <= 60 else 0 date_list = [years, months, weeks, days, hours, minutes, seconds] for index, interval in enumerate(date_list): if interval > 0: name = intervals[index] if interval == 1: name = name.rstrip('s') result.append(f"{interval} {name}") return ', '.join(result[:granularity]) else: return '0 seconds' # Sends email notification def send_email(subject, body, body_html, use_ssl, smtp_timeout=15): fqdn_re = re.compile(r'(?=^.{4,253}$)(^((?!-)[a-zA-Z0-9-]{1,63}(? 0: PSN_ACTIVE_CHECK_INTERVAL = PSN_ACTIVE_CHECK_INTERVAL - PSN_ACTIVE_CHECK_SIGNAL_VALUE sig_name = signal.Signals(sig).name print(f"* Signal {sig_name} received") print(f"* PSN timers: [active check interval: {display_time(PSN_ACTIVE_CHECK_INTERVAL)}]") print_cur_ts("Timestamp:\t\t\t") # Signal handler for SIGHUP allowing to reload secrets from .env def reload_secrets_signal_handler(sig, frame): sig_name = signal.Signals(sig).name print(f"* Signal {sig_name} received") # disable autoscan if DOTENV_FILE set to none if DOTENV_FILE and DOTENV_FILE.lower() == 'none': env_path = None else: # reload .env if python-dotenv is installed try: from dotenv import load_dotenv, find_dotenv if DOTENV_FILE: env_path = DOTENV_FILE else: env_path = find_dotenv() if env_path: load_dotenv(env_path, override=True) else: print("* No .env file found, skipping env-var reload") except ImportError: env_path = None print("* python-dotenv not installed, skipping env-var reload") if env_path: for secret in SECRET_KEYS: old_val = globals().get(secret) val = os.getenv(secret) if val is not None and val != old_val: globals()[secret] = val print(f"* Reloaded {secret} from {env_path}") print_cur_ts("Timestamp:\t\t\t") # Finds an optional config file def find_config_file(cli_path=None): """ Search for an optional config file in: 1) CLI-provided path (must exist if given) 2) ./{DEFAULT_CONFIG_FILENAME} 3) ~/.{DEFAULT_CONFIG_FILENAME} 4) script-directory/{DEFAULT_CONFIG_FILENAME} """ if cli_path: p = Path(os.path.expanduser(cli_path)) return str(p) if p.is_file() else None candidates = [ Path.cwd() / DEFAULT_CONFIG_FILENAME, Path.home() / f".{DEFAULT_CONFIG_FILENAME}", Path(__file__).parent / DEFAULT_CONFIG_FILENAME, ] for p in candidates: if p.is_file(): return str(p) return None # Resolves an executable path by checking if it's a valid file or searching in $PATH def resolve_executable(path): if os.path.isfile(path) and os.access(path, os.X_OK): return path found = shutil.which(path) if found: return found raise FileNotFoundError(f"Could not find executable '{path}'") # Normalizes Unicode punctuation, symbols and spacing in a string to plain ASCII def normalize_ascii(s): if not isinstance(s, str): return s # punctuation & symbols to ASCII s = (s.replace("\u2018", "'").replace("\u2019", "'") # ‘ ’ -> ' .replace("\u201C", '"').replace("\u201D", '"') # “ ” -> " .replace("\u2013", "-") # – -> - .replace("\u2026", "...") # … -> ... .replace("\u00A0", " ")) # NBSP -> space # remove trademark symbols for ch in ("\u00AE", "\u2122"): # ® ™ s = s.replace(ch, "") # collapse doubled single quotes that often appear after smart-quote normalization s = s.replace("''", "'") # collapse multiple spaces while " " in s: s = s.replace(" ", " ") return s.strip() # Prints the last N earned trophies across titles with game, type and earn date def print_last_earned_trophies(psn_user, max_items=5, title_limit=15): PT = None try: from psnawp_api.models.trophies import PlatformType as PT # 3.x except Exception: PT = None # fallback to string platforms later def _get(obj, *names, default=None): for n in names: if hasattr(obj, n): v = getattr(obj, n) if v is not None: return v return default def _platforms_to_try(title): raw = getattr(title, "platform", None) raw_val = getattr(raw, "value", raw) s = (str(raw_val).lower() if raw_val else "") if PT: if "ps5" in s: return [PT.PS5, PT.PS4] if "ps4" in s: return [PT.PS4, PT.PS5] return [PT.PS5, PT.PS4] # string fallback if "ps5" in s: return ["ps5", "ps4"] if "ps4" in s: return ["ps4", "ps5"] return ["ps5", "ps4"] def _earn_dt(tr): return _get(tr, "earned_date_time", "earnedDateTime", default=None) def _trophy_type_str(tr): raw = _get(tr, "trophy_type", "trophyType", default=None) if raw is None: return "UNKNOWN" if hasattr(raw, "name"): return raw.name return str(raw).upper() # title-name resolver (cache) _title_name_cache = {} def _resolve_title_name(npcomm, platform): key = (npcomm, str(platform)) if key in _title_name_cache: return _title_name_cache[key] def _first_name_like(obj): # Try common fields first for fld in ("trophy_title_name", "trophyTitleName", "title_name", "titleName", "name"): if hasattr(obj, fld): val = getattr(obj, fld) if isinstance(val, str) and val.strip(): return val.strip() # Fallback: scan attributes that look like "*name" for attr in dir(obj): if attr.startswith("_"): continue if "name" in attr.lower(): try: val = getattr(obj, attr) except Exception: continue if isinstance(val, str) and val.strip(): return val.strip() return None name = None # A) groups often carry the title name try: for g in psn_user.trophy_groups(np_communication_id=npcomm, platform=platform): name = _first_name_like(g) if name: break except Exception: pass # B) per-title summary if not name: try: summ = psn_user.trophy_summary(np_communication_id=npcomm, platform=platform) name = _first_name_like(summ) except Exception: pass # C) scan titles if not name: try: for tt in psn_user.trophy_titles(limit=title_limit): nc = getattr(tt, "np_communication_id", None) or getattr(tt, "npCommunicationId", None) if nc == npcomm: name = _first_name_like(tt) if name: break except Exception: pass if not name: name = npcomm # last resort _title_name_cache[key] = name return name # ------------------------------------- items = [] # 1) list titles (no special args for cross-version compat) try: titles_iter = psn_user.trophy_titles(limit=title_limit) except Exception: titles_iter = [] for tt in titles_iter: npcomm = _get(tt, "np_communication_id", "npCommunicationId", default=None) if not npcomm: continue for plat in _platforms_to_try(tt): try: it = psn_user.trophies( np_communication_id=npcomm, platform=plat, include_progress=True, trophy_group_id="all", ) except Exception: continue got_any_for_title = False for tr in it: got_any_for_title = True if not getattr(tr, "earned", False): continue dt = _earn_dt(tr) if not dt: continue game_name = normalize_ascii(_resolve_title_name(npcomm, plat)) ttype = _trophy_type_str(tr) tname = _get(tr, "trophy_name", "trophyName", default=None) if not tname: tname = "(hidden)" if getattr(tr, "hidden", False) else "(unknown)" tname = normalize_ascii(tname) items.append((dt, game_name, ttype, tname)) if got_any_for_title: break # this platform works for this title if len(items) >= max_items: break if not items: print("- (no recent trophies found or trophy visibility is restricted)") return # 2) sort & print try: items.sort(key=lambda x: x[0], reverse=True) except Exception: def _ts(dt): try: return int(dt.timestamp()) except Exception: return -1 items.sort(key=lambda x: _ts(x[0]), reverse=True) for dt, game, ttype, tname in items[:max_items]: try: ts = int(dt.timestamp()) dt_fmt = get_date_from_ts(ts) except Exception: dt_fmt = "n/a" print(f"- {dt_fmt} | {game} | {ttype} | {tname}") # Gets detailed user information and displays it (for -i/--info mode) def get_user_info(psn_user_id, include_trophies=False, show_recent_games=True): print(f"* Fetching details for PlayStation user '{psn_user_id}'... this may take a moment\n") try: psnawp = PSNAWP(PSN_NPSSO) psn_user = psnawp.user(online_id=psn_user_id) accountid = psn_user.account_id profile = psn_user.profile() aboutme = profile.get("aboutMe") isplus = profile.get("isPlus") langs = profile.get("languages") or [] is_verified = profile.get("isOfficiallyVerified") fs = psn_user.friendship() share = psn_user.get_shareable_profile_link() except Exception as e: print(f"* Error: {e}") sys.exit(1) try: psn_user_presence = psn_user.get_presence() except Exception as e: print(f"* Error: Cannot get presence for user {psn_user_id}: {e}") sys.exit(1) status = psn_user_presence["basicPresence"]["primaryPlatformInfo"].get("onlineStatus") if not status: print(f"* Error: Cannot get status for user {psn_user_id}") sys.exit(1) status = str(status).lower() psn_platform = psn_user_presence["basicPresence"]["primaryPlatformInfo"].get("platform") psn_platform = str(psn_platform).upper() if psn_platform else "" lastonline = psn_user_presence["basicPresence"]["primaryPlatformInfo"].get("lastOnlineDate") availability = psn_user_presence["basicPresence"].get("availability") lastonline_dt = convert_iso_str_to_datetime(lastonline) if lastonline_dt: lastonline_ts = int(lastonline_dt.timestamp()) else: lastonline_ts = 0 gametitleinfolist = psn_user_presence["basicPresence"].get("gameTitleInfoList") game_name = "" launchplatform = "" if gametitleinfolist: game_name_raw = gametitleinfolist[0].get("titleName") game_name = normalize_ascii(game_name_raw) if game_name_raw else "" launchplatform = gametitleinfolist[0].get("launchPlatform") launchplatform = str(launchplatform).upper() psn_last_status_file = f"psn_{psn_user_id}_last_status.json" status_ts_old = int(time.time()) if os.path.isfile(psn_last_status_file): try: with open(psn_last_status_file, 'r', encoding="utf-8") as f: last_status_read = json.load(f) if last_status_read: last_status_ts = last_status_read[0] last_status = last_status_read[1] if lastonline_ts and status == "offline": if lastonline_ts >= last_status_ts: status_ts_old = lastonline_ts else: status_ts_old = last_status_ts elif not lastonline_ts and status == "offline": status_ts_old = last_status_ts elif status and status != "offline" and status == last_status: status_ts_old = last_status_ts except Exception: if lastonline_ts and status == "offline": status_ts_old = lastonline_ts else: if lastonline_ts and status == "offline": status_ts_old = lastonline_ts print(f"PlayStation ID:\t\t\t{psn_user_id}") print(f"PSN account ID:\t\t\t{accountid}") print(f"\nStatus:\t\t\t\t{str(status).upper()}") if availability: available_str = "Yes" if availability == "availableToPlay" else "No" print(f"Available to play:\t\t{available_str}") psn_platform_displayed = False if psn_platform: print(f"\nPlatform:\t\t\t{psn_platform}") psn_platform_displayed = True if not psn_platform_displayed: print() print(f"PS+ user:\t\t\t{isplus}") # an official account belonging to a recognised developer, publisher, community manager or another official role if is_verified is not None: print(f"Verified:\t\t\t{is_verified}") newline_needed = False if aboutme: print(f"\nAbout me:\t\t\t{aboutme}") newline_needed = True if langs: prefix = "\n" if not newline_needed else "" print(f"{prefix}Languages:\t\t\t{', '.join(langs)}") try: relation = fs.get("friendRelation") print(f"\nRelation:\t\t\t{relation}") if relation == "friend": mf = fs.get("mutualFriendsCount") if isinstance(mf, int) and mf >= 0: print(f"Mutual friends:\t\t\t{mf}") elif mf is None: print("Mutual friends:\t\t\tunknown") else: print("Mutual friends:\t\t\thidden") else: # Don't print mutual friends at all pass except Exception: pass try: print(f"\nProfile URL:\t\t\t{share.get('shareUrl')}") # print(f"Profile QR image:\t\t{share.get('shareImageUrl')}") except Exception: pass if status == "offline" and status_ts_old > 0: last_status_dt_str = get_date_from_ts(status_ts_old) print(f"\n* Last time user was available:\t{last_status_dt_str}") print(f"* User is OFFLINE for:\t\t{calculate_timespan(now_local(), int(status_ts_old), show_seconds=False)}") elif status != "offline": if os.path.isfile(psn_last_status_file): try: with open(psn_last_status_file, 'r', encoding="utf-8") as f: last_status_read = json.load(f) if last_status_read and last_status_read[1] == status: print(f"* User is {str(status).upper()} for:\t\t{calculate_timespan(now_local(), int(last_status_read[0]), show_seconds=False)}") except Exception: pass # Show trophy summary and last earned trophies only if requested if include_trophies: try: print(f"\n* Getting trophy summary ...") ts = psn_user.trophy_summary() et = ts.earned_trophies prog = int(ts.progress) if ts.progress is not None else 0 print(f"\nTrophy level:\t\t\t{ts.trophy_level} ({prog}% to next, tier {ts.tier})") print( "Trophies earned:\t\t" f"{et.platinum} Platinum, {et.gold} Gold, {et.silver} Silver, {et.bronze} Bronze " f"({et.platinum + et.gold + et.silver + et.bronze} total)" ) except Exception: pass num_trophies = 5 try: print(f"\n* Getting list of last {num_trophies} earned trophies ...\n") print_last_earned_trophies(psn_user, max_items=num_trophies, title_limit=15) except Exception: pass # Show recently played games only if requested if show_recent_games: try: # Helper function to compact duration format, convert "X day(s), HH:MM:SS" to "Xd HH:MM:SS" def _compact_duration(s): if not s: return "0:00:00" s = str(s).strip() if "day" in s.lower(): try: if "," in s: parts = s.split(",", 1) days_part = parts[0].strip() # "1 day" / "2 days" time_part = parts[1].strip() # "23:47:54" d = int(days_part.split()[0]) return f"{d}d {time_part}" else: # No comma, try to extract days anyway (unlikely but handle it) words = s.split() if len(words) >= 2 and words[1].lower().startswith("day"): d = int(words[0]) if len(words) > 2: time_part = " ".join(words[2:]) return f"{d}d {time_part}" return f"{d}d" except (ValueError, IndexError): return s # fallback to original if parsing fails return s def _shorten_middle(s, max_len, ellipsis="..."): if s is None: return "" s = str(s) if len(s) <= max_len: return s keep = max_len - len(ellipsis) if keep <= 0: return ellipsis[:max_len] left = keep // 2 right = keep - left return f"{s[:left]}{ellipsis}{s[-right:]}" recent_entries = [] print(f"\n* Getting list of recently played games ...") for i, t in enumerate(psn_user.title_stats(limit=10, page_size=50), 1): if not t: continue name_raw = t.name or "(unknown)" name = normalize_ascii(name_raw) cat = getattr(getattr(t, "category", None), "name", "UNKNOWN") last_played = ( get_date_from_ts(int(t.last_played_date_time.timestamp())) if t.last_played_date_time else "n/a" ) total_raw = str(t.play_duration) if t.play_duration else "0:00:00" # Compact duration immediately to ensure it fits in the column total = _compact_duration(total_raw) recent_entries.append(f"Recent #{i}:\t\t\t{name} | {cat} | last played {last_played} | total {total}") # Decide column widths based on terminal size try: import shutil term_width = shutil.get_terminal_size(fallback=(100, 24)).columns except Exception: term_width = 100 w_num = 3 w_platform = 8 w_last = 24 w_total = 14 # fits "999d 23:59:59" (14 chars) after compacting "X day(s), HH:MM:SS" -> "Xd HH:MM:SS" fixed = 1 + w_num + 2 + w_platform + 2 + w_last + 2 + w_total w_title = max(24, term_width - fixed) # Only print the table if we have entries if recent_entries: print() hdr = f"{'#'.ljust(w_num)} {'Title'.ljust(w_title)} {'Platform'.ljust(w_platform)} {'Last played'.ljust(w_last)} {'Total'.ljust(w_total)}" sep = f"{'-' * w_num} {'-' * w_title} {'-' * w_platform} {'-' * w_last} {'-' * w_total}" print(hdr) print(sep) for i, entry in enumerate(recent_entries, 1): try: _, rest = entry.split(":", 1) parts = rest.strip().split("|") name = parts[0].strip() cat = parts[1].strip() last_played = parts[2].replace("last played", "").strip() total = _compact_duration(parts[3].replace("total", "").strip()) except Exception: # If parsing ever fails, print raw line as a fallback print(entry) continue name_fmt = _shorten_middle(name, w_title) row = ( f"{str(i).ljust(w_num)} " f"{name_fmt.ljust(w_title)} " f"{cat.ljust(w_platform)} " f"{last_played.ljust(w_last)} " f"{total.ljust(w_total)}" ) print(row) except Exception: pass if game_name: launchplatform_str = "" if launchplatform: launchplatform_str = f" ({launchplatform})" print(f"\nUser is currently in-game:\t{game_name}{launchplatform_str}") # Main function that monitors gaming activity of the specified PSN user def psn_monitor_user(psn_user_id, csv_file_name): alive_counter = 0 status_ts = 0 status_ts_old = 0 status_online_start_ts = 0 status_online_start_ts_old = 0 game_ts = 0 game_ts_old = 0 lastonline_ts = 0 status = "" game_total_ts = 0 games_number = 0 game_total_after_offline_counted = False try: if csv_file_name: init_csv_file(csv_file_name) except Exception as e: print(f"* Error: {e}") print("Sneaking into PlayStation like a ninja ... (be patient, secrets take time)\n") try: psnawp = PSNAWP(PSN_NPSSO) psn_user = psnawp.user(online_id=psn_user_id) accountid = psn_user.account_id profile = psn_user.profile() aboutme = profile.get("aboutMe") isplus = profile.get("isPlus") langs = profile.get("languages") or [] is_verified = profile.get("isOfficiallyVerified") fs = psn_user.friendship() share = psn_user.get_shareable_profile_link() except Exception as e: print("* Error:", e) sys.exit(1) try: psn_user_presence = psn_user.get_presence() except Exception as e: print(f"* Error: Cannot get presence for user {psn_user_id}: {e}") sys.exit(1) status = psn_user_presence["basicPresence"]["primaryPlatformInfo"].get("onlineStatus") if not status: print(f"* Error: Cannot get status for user {psn_user_id}") sys.exit(1) status = str(status).lower() psn_platform = psn_user_presence["basicPresence"]["primaryPlatformInfo"].get("platform") psn_platform = str(psn_platform).upper() if psn_platform else "" lastonline = psn_user_presence["basicPresence"]["primaryPlatformInfo"].get("lastOnlineDate") availability = psn_user_presence["basicPresence"].get("availability") lastonline_dt = convert_iso_str_to_datetime(lastonline) if lastonline_dt: lastonline_ts = int(lastonline_dt.timestamp()) else: lastonline_ts = 0 gametitleinfolist = psn_user_presence["basicPresence"].get("gameTitleInfoList") game_name = "" launchplatform = "" if gametitleinfolist: game_name_raw = gametitleinfolist[0].get("titleName") game_name = normalize_ascii(game_name_raw) if game_name_raw else "" launchplatform = gametitleinfolist[0].get("launchPlatform") launchplatform = str(launchplatform).upper() status_ts_old = int(time.time()) status_ts_old_bck = status_ts_old if status and status != "offline": status_online_start_ts = status_ts_old status_online_start_ts_old = status_online_start_ts psn_last_status_file = f"psn_{psn_user_id}_last_status.json" last_status_read = [] last_status_ts = 0 last_status = "" if os.path.isfile(psn_last_status_file): try: with open(psn_last_status_file, 'r', encoding="utf-8") as f: last_status_read = json.load(f) except Exception as e: print(f"* Cannot load last status from '{psn_last_status_file}' file: {e}") if last_status_read: last_status_ts = last_status_read[0] last_status = last_status_read[1] psn_last_status_file_mdate_dt = datetime.fromtimestamp(int(os.path.getmtime(psn_last_status_file)), pytz.timezone(LOCAL_TIMEZONE)) print(f"* Last status loaded from file '{psn_last_status_file}' ({get_short_date_from_ts(psn_last_status_file_mdate_dt, show_weekday=False, always_show_year=True)})") if last_status_ts > 0: last_status_dt_str = get_short_date_from_ts(last_status_ts, show_weekday=False, always_show_year=True) last_status_str = str(last_status.upper()) print(f"* Last status read from file: {last_status_str} ({last_status_dt_str})") if lastonline_ts and status == "offline": if lastonline_ts >= last_status_ts: status_ts_old = lastonline_ts else: status_ts_old = last_status_ts if not lastonline_ts and status == "offline": status_ts_old = last_status_ts if status and status != "offline" and status == last_status: status_online_start_ts = last_status_ts status_online_start_ts_old = status_online_start_ts status_ts_old = last_status_ts if last_status_ts > 0 and status != last_status: last_status_to_save = [] last_status_to_save.append(status_ts_old) last_status_to_save.append(status) try: with open(psn_last_status_file, 'w', encoding="utf-8") as f: json.dump(last_status_to_save, f, indent=2) except Exception as e: print(f"* Cannot save last status to '{psn_last_status_file}' file: {e}") try: if csv_file_name and (status != last_status): write_csv_entry(csv_file_name, now_local_naive(), status, game_name) except Exception as e: print(f"* Error: {e}") print(f"\nPlayStation ID:\t\t\t{psn_user_id}") print(f"PSN account ID:\t\t\t{accountid}") print(f"\nStatus:\t\t\t\t{str(status).upper()}") if availability: available_str = "Yes" if availability == "availableToPlay" else "No" print(f"Available to play:\t\t{available_str}") psn_platform_displayed = False if psn_platform: print(f"\nPlatform:\t\t\t{psn_platform}") psn_platform_displayed = True if not psn_platform_displayed: print() print(f"PS+ user:\t\t\t{isplus}") # an official account belonging to a recognised developer, publisher, community manager or another official role if is_verified is not None: print(f"Verified:\t\t\t{is_verified}") newline_needed = False if aboutme: print(f"\nAbout me:\t\t\t{aboutme}") newline_needed = True if langs: prefix = "\n" if not newline_needed else "" print(f"{prefix}Languages:\t\t\t{', '.join(langs)}") try: relation = fs.get("friendRelation") print(f"\nRelation:\t\t\t{relation}") if relation == "friend": mf = fs.get("mutualFriendsCount") if isinstance(mf, int) and mf >= 0: print(f"Mutual friends:\t\t\t{mf}") elif mf is None: print("Mutual friends:\t\t\tunknown") else: print("Mutual friends:\t\t\thidden") else: # Don't print mutual friends at all pass except Exception: pass try: print(f"\nProfile URL:\t\t\t{share.get('shareUrl')}") # print(f"Profile QR image:\t\t{share.get('shareImageUrl')}") except Exception: pass if status != "offline" and game_name: launchplatform_str = "" if launchplatform: launchplatform_str = f" ({launchplatform})" print(f"\nUser is currently in-game:\t{game_name}{launchplatform_str}") game_ts_old = int(time.time()) games_number += 1 if last_status_ts == 0: if lastonline_ts and status == "offline": status_ts_old = lastonline_ts last_status_to_save = [] last_status_to_save.append(status_ts_old) last_status_to_save.append(status) try: with open(psn_last_status_file, 'w', encoding="utf-8") as f: json.dump(last_status_to_save, f, indent=2) except Exception as e: print(f"* Cannot save last status to '{psn_last_status_file}' file: {e}") if status_ts_old != status_ts_old_bck: if status == "offline": last_status_dt_str = get_date_from_ts(status_ts_old) print(f"\n* Last time user was available:\t{last_status_dt_str}") print(f"\n* User is {str(status).upper()} for:\t\t{calculate_timespan(now_local(), int(status_ts_old), show_seconds=False)}") status_old = status game_name_old = game_name print_cur_ts("\nTimestamp:\t\t\t") alive_counter = 0 email_sent = False m_subject = m_body = "" error_streak = 0 def get_sleep_interval(): return PSN_ACTIVE_CHECK_INTERVAL if status and status != "offline" else PSN_CHECK_INTERVAL sleep_interval = get_sleep_interval() time.sleep(sleep_interval) # Main loop while True: # Sometimes PSN network functions halt, so we use alarm signal functionality to kill it inevitably, not available on Windows if platform.system() != 'Windows': signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(FUNCTION_TIMEOUT) try: psn_user_presence = psn_user.get_presence() status = psn_user_presence["basicPresence"]["primaryPlatformInfo"].get("onlineStatus") gametitleinfolist = psn_user_presence["basicPresence"].get("gameTitleInfoList") game_name = "" launchplatform = "" if gametitleinfolist: game_name_raw = gametitleinfolist[0].get("titleName") game_name = normalize_ascii(game_name_raw) if game_name_raw else "" launchplatform = gametitleinfolist[0].get("launchPlatform") launchplatform = str(launchplatform).upper() if platform.system() != 'Windows': signal.alarm(0) if not status: raise ValueError('PSN user status is empty') else: status = str(status).lower() except TimeoutException: if platform.system() != 'Windows': signal.alarm(0) print(f"psn_user.get_presence() timeout, retrying in {display_time(FUNCTION_TIMEOUT)}") print_cur_ts("Timestamp:\t\t\t") time.sleep(FUNCTION_TIMEOUT) continue except PSNAWPAuthenticationError as auth_err: if platform.system() != 'Windows': signal.alarm(0) sleep_interval = get_sleep_interval() print(f"* PSN NPSSO key might not be valid anymore: {auth_err}") if ERROR_NOTIFICATION and not email_sent: m_subject = f"psn_monitor: PSN NPSSO key error! (user: {psn_user_id})" m_body = f"PSN NPSSO key might not be valid anymore: {auth_err}{get_cur_ts(nl_ch + nl_ch + 'Timestamp: ')}" print(f"Sending email notification to {RECEIVER_EMAIL}") send_email(m_subject, m_body, "", SMTP_SSL) email_sent = True print_cur_ts("Timestamp:\t\t\t") time.sleep(sleep_interval) continue except Exception as e: if platform.system() != 'Windows': signal.alarm(0) msg = str(e).lower() # Connection-related errors that can often be fixed by recreating the session connection_error = ('remote end closed connection' in msg or 'connection reset by peer' in msg or 'connection aborted' in msg) if connection_error: try: psnawp = PSNAWP(PSN_NPSSO) psn_user = psnawp.user(online_id=psn_user_id) except Exception: pass error_streak += 1 # For connection errors, retry quickly since they're often transient retry_delay = FUNCTION_TIMEOUT # However, if connection errors persist, it might indicate an expired NPSSO token # Send notification after 5+ consecutive connection errors if ERROR_NOTIFICATION and not email_sent and error_streak >= 5: print(f"* Multiple consecutive connection errors detected - this may indicate an expired NPSSO token, error streak {error_streak}: {e}") m_subject = f"psn_monitor: PSN NPSSO key might be expired! (user: {psn_user_id})" m_body = f"Multiple consecutive connection errors detected - this may indicate an expired NPSSO token.\n\nError: {e}\n\nError streak: {error_streak}{get_cur_ts(nl_ch + nl_ch + 'Timestamp: ')}" print(f"Sending email notification to {RECEIVER_EMAIL}") send_email(m_subject, m_body, "", SMTP_SSL) email_sent = True print_cur_ts("Timestamp:\t\t\t") # print(f"* Connection error, recreating session and retrying in {display_time(retry_delay)}: {e}") # print_cur_ts("Timestamp:\t\t\t") time.sleep(retry_delay) continue error_streak += 1 # Check for authentication errors (excluding connection errors which we already handled) likely_auth = ('401' in msg) or ('expired' in msg) or ('invalid' in msg) if likely_auth and ERROR_NOTIFICATION and not email_sent and error_streak >= 5: m_subject = f"psn_monitor: PSN NPSSO key error! (user: {psn_user_id})" m_body = f"PSN NPSSO key might not be valid anymore: {e}{get_cur_ts(nl_ch + nl_ch + 'Timestamp: ')}" print(f"Sending email notification to {RECEIVER_EMAIL}") send_email(m_subject, m_body, "", SMTP_SSL) email_sent = True sleep_interval = get_sleep_interval() print(f"* Error, retrying in {display_time(sleep_interval)}: {e}") print_cur_ts("Timestamp:\t\t\t") time.sleep(sleep_interval) continue else: email_sent = False error_streak = 0 finally: if platform.system() != 'Windows': signal.alarm(0) change = False act_inact_flag = False status_ts = int(time.time()) game_ts = int(time.time()) # Player status changed if status != status_old: last_status_to_save = [] last_status_to_save.append(status_ts) last_status_to_save.append(status) try: with open(psn_last_status_file, 'w', encoding="utf-8") as f: json.dump(last_status_to_save, f, indent=2) except Exception as e: print(f"* Cannot save last status to '{psn_last_status_file}' file: {e}") print(f"PSN user {psn_user_id} changed status from {status_old} to {status}") print(f"User was {status_old} for {calculate_timespan(int(status_ts), int(status_ts_old))} ({get_range_of_dates_from_tss(int(status_ts_old), int(status_ts), short=True)})") m_subject_was_since = f", was {status_old}: {get_range_of_dates_from_tss(int(status_ts_old), int(status_ts), short=True)}" m_subject_after = calculate_timespan(int(status_ts), int(status_ts_old), show_seconds=False) m_body_was_since = f" ({get_range_of_dates_from_tss(int(status_ts_old), int(status_ts), short=True)})" m_body_short_offline_msg = "" # Player got online if status_old == "offline" and status and status != "offline": print(f"*** User got ACTIVE ! (was offline since {get_date_from_ts(status_ts_old)})") game_total_after_offline_counted = False if (status_ts - status_ts_old) > OFFLINE_INTERRUPT or not status_online_start_ts_old: status_online_start_ts = status_ts game_total_ts = 0 games_number = 0 elif (status_ts - status_ts_old) <= OFFLINE_INTERRUPT and status_online_start_ts_old > 0: status_online_start_ts = status_online_start_ts_old short_offline_msg = f"Short offline interruption ({display_time(status_ts - status_ts_old)}), online start timestamp set back to {get_short_date_from_ts(status_online_start_ts_old)}" m_body_short_offline_msg = f"\n\n{short_offline_msg}" print(short_offline_msg) act_inact_flag = True m_body_played_games = "" # Player got offline if status_old and status_old != "offline" and status == "offline": if status_online_start_ts > 0: m_subject_after = calculate_timespan(int(status_ts), int(status_online_start_ts), show_seconds=False) online_since_msg = f"(after {calculate_timespan(int(status_ts), int(status_online_start_ts), show_seconds=False)}: {get_range_of_dates_from_tss(int(status_online_start_ts), int(status_ts), short=True)})" m_subject_was_since = f", was available: {get_range_of_dates_from_tss(int(status_online_start_ts), int(status_ts), short=True)}" m_body_was_since = f" ({get_range_of_dates_from_tss(int(status_ts_old), int(status_ts), short=True)})\n\nUser was available for {calculate_timespan(int(status_ts), int(status_online_start_ts), show_seconds=False)} ({get_range_of_dates_from_tss(int(status_online_start_ts), int(status_ts), short=True)})" else: online_since_msg = "" if games_number > 0: if game_name_old and not game_name: game_total_ts += (int(game_ts) - int(game_ts_old)) game_total_after_offline_counted = True m_body_played_games = f"\n\nUser played {games_number} games for total time of {display_time(game_total_ts)}" print(f"User played {games_number} games for total time of {display_time(game_total_ts)}") print(f"*** User got OFFLINE ! {online_since_msg}") status_online_start_ts_old = status_online_start_ts status_online_start_ts = 0 act_inact_flag = True m_body_user_in_game = "" if status != "offline" and game_name: launchplatform_str = "" if launchplatform: launchplatform_str = f" ({launchplatform})" print(f"User is currently in-game: {game_name}{launchplatform_str}") m_body_user_in_game = f"\n\nUser is currently in-game: {game_name}{launchplatform_str}" change = True m_subject = f"PSN user {psn_user_id} is now {status} (after {m_subject_after}{m_subject_was_since})" m_body = f"PSN user {psn_user_id} changed status from {status_old} to {status}\n\nUser was {status_old} for {calculate_timespan(int(status_ts), int(status_ts_old))}{m_body_was_since}{m_body_short_offline_msg}{m_body_user_in_game}{m_body_played_games}{get_cur_ts(nl_ch + nl_ch + 'Timestamp: ')}" if ACTIVE_INACTIVE_NOTIFICATION and act_inact_flag: print(f"Sending email notification to {RECEIVER_EMAIL}") send_email(m_subject, m_body, "", SMTP_SSL) status_ts_old = status_ts print_cur_ts("Timestamp:\t\t\t") # Player started/stopped/changed the game if game_name != game_name_old: launchplatform_str = "" if launchplatform: launchplatform_str = f" ({launchplatform})" # User changed the game if game_name_old and game_name: print(f"PSN user {psn_user_id} changed game from '{game_name_old}' to '{game_name}'{launchplatform_str} after {calculate_timespan(int(game_ts), int(game_ts_old))}") print(f"User played game from {get_range_of_dates_from_tss(int(game_ts_old), int(game_ts), short=True, between_sep=' to ')}") game_total_ts += (int(game_ts) - int(game_ts_old)) games_number += 1 m_body = f"PSN user {psn_user_id} changed game from '{game_name_old}' to '{game_name}'{launchplatform_str} after {calculate_timespan(int(game_ts), int(game_ts_old))}\n\nUser played game from {get_range_of_dates_from_tss(int(game_ts_old), int(game_ts), short=True, between_sep=' to ')}{get_cur_ts(nl_ch + nl_ch + 'Timestamp: ')}" if launchplatform: launchplatform_str = f"{launchplatform}, " m_subject = f"PSN user {psn_user_id} changed game to '{game_name}' ({launchplatform_str}after {calculate_timespan(int(game_ts), int(game_ts_old), show_seconds=False)}: {get_range_of_dates_from_tss(int(game_ts_old), int(game_ts), short=True)})" # User started playing new game elif not game_name_old and game_name: print(f"PSN user {psn_user_id} started playing '{game_name}'{launchplatform_str}") games_number += 1 m_subject = f"PSN user {psn_user_id} now plays '{game_name}'{launchplatform_str}" m_body = f"PSN user {psn_user_id} now plays '{game_name}'{launchplatform_str}{get_cur_ts(nl_ch + nl_ch + 'Timestamp: ')}" # User stopped playing the game elif game_name_old and not game_name: print(f"PSN user {psn_user_id} stopped playing '{game_name_old}' after {calculate_timespan(int(game_ts), int(game_ts_old))}") print(f"User played game from {get_range_of_dates_from_tss(int(game_ts_old), int(game_ts), short=True, between_sep=' to ')}") if not game_total_after_offline_counted: game_total_ts += (int(game_ts) - int(game_ts_old)) m_subject = f"PSN user {psn_user_id} stopped playing '{game_name_old}' (after {calculate_timespan(int(game_ts), int(game_ts_old), show_seconds=False)}: {get_range_of_dates_from_tss(int(game_ts_old), int(game_ts), short=True)})" m_body = f"PSN user {psn_user_id} stopped playing '{game_name_old}' after {calculate_timespan(int(game_ts), int(game_ts_old))}\n\nUser played game from {get_range_of_dates_from_tss(int(game_ts_old), int(game_ts), short=True, between_sep=' to ')}{get_cur_ts(nl_ch + nl_ch + 'Timestamp: ')}" change = True if GAME_CHANGE_NOTIFICATION and m_subject and m_body: print(f"Sending email notification to {RECEIVER_EMAIL}") send_email(m_subject, m_body, "", SMTP_SSL) game_ts_old = game_ts print_cur_ts("Timestamp:\t\t\t") if change: alive_counter = 0 try: if csv_file_name: write_csv_entry(csv_file_name, now_local_naive(), status, game_name) except Exception as e: print(f"* Error: {e}") status_old = status game_name_old = game_name alive_counter += 1 if LIVENESS_CHECK_COUNTER and alive_counter >= LIVENESS_CHECK_COUNTER and (status == "offline" or not status): print_cur_ts("Liveness check, timestamp:\t") alive_counter = 0 sleep_interval = get_sleep_interval() time.sleep(sleep_interval) def main(): global CLI_CONFIG_PATH, DOTENV_FILE, LOCAL_TIMEZONE, LIVENESS_CHECK_COUNTER, PSN_NPSSO, CSV_FILE, DISABLE_LOGGING, PSN_LOGFILE, ACTIVE_INACTIVE_NOTIFICATION, GAME_CHANGE_NOTIFICATION, ERROR_NOTIFICATION, PSN_CHECK_INTERVAL, PSN_ACTIVE_CHECK_INTERVAL, SMTP_PASSWORD, stdout_bck if "--generate-config" in sys.argv: print(CONFIG_BLOCK.strip("\n")) sys.exit(0) if "--version" in sys.argv: print(f"{os.path.basename(sys.argv[0])} v{VERSION}") sys.exit(0) stdout_bck = sys.stdout signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) clear_screen(CLEAR_SCREEN) print(f"PSN Monitoring Tool v{VERSION}\n") parser = argparse.ArgumentParser( prog="psn_monitor", description=("Monitor a PSN user's playing status and send customizable email alerts [ https://github.com/misiektoja/psn_monitor/ ]"), formatter_class=argparse.RawTextHelpFormatter ) # Positional parser.add_argument( "psn_user_id", nargs="?", metavar="PSN_USER_ID", help="User's PSN ID", type=str ) # Version, just to list in help, it is handled earlier parser.add_argument( "--version", action="version", version=f"%(prog)s v{VERSION}" ) # Configuration & dotenv files conf = parser.add_argument_group("Configuration & dotenv files") conf.add_argument( "--config-file", dest="config_file", metavar="PATH", help="Location of the optional config file", ) conf.add_argument( "--generate-config", action="store_true", help="Print default config template and exit", ) conf.add_argument( "--env-file", dest="env_file", metavar="PATH", help="Path to optional dotenv file (auto-search if not set, disable with 'none')", ) # API credentials creds = parser.add_argument_group("API credentials") creds.add_argument( "-n", "--npsso-key", dest="npsso_key", metavar="PSN_NPSSO", type=str, help="PlayStation NPSSO key" ) # Notifications notify = parser.add_argument_group("Notifications") notify.add_argument( "-a", "--notify-active-inactive", dest="notify_active_inactive", action="store_true", default=None, help="Email when user goes online/offline" ) notify.add_argument( "-g", "--notify-game-change", dest="notify_game_change", action="store_true", default=None, help="Email on game start/change/stop" ) notify.add_argument( "-e", "--no-error-notify", dest="notify_errors", action="store_false", default=None, help="Disable email on errors (e.g. invalid NPSSO)" ) notify.add_argument( "--send-test-email", dest="send_test_email", action="store_true", help="Send test email to verify SMTP settings" ) # User information info = parser.add_argument_group("User information") info.add_argument( "-i", "--info", dest="info_mode", action="store_true", help="Get detailed user information and display it, then exit" ) info.add_argument( "--trophies", dest="include_trophies", action="store_true", help="Show trophy summary and last earned trophies (only works with -i/--info)" ) info.add_argument( "--no-recent-games", dest="no_recent_games", action="store_true", help="Don't fetch recently played games list (only works with -i/--info)" ) # Intervals & timers times = parser.add_argument_group("Intervals & timers") times.add_argument( "-c", "--check-interval", dest="check_interval", metavar="SECONDS", type=int, help="Polling interval when user is offline" ) times.add_argument( "-k", "--active-interval", dest="active_interval", metavar="SECONDS", type=int, help="Polling interval when user is online" ) # Features & Output opts = parser.add_argument_group("Features & output") opts.add_argument( "-b", "--csv-file", dest="csv_file", metavar="CSV_FILENAME", type=str, help="Write status & game changes to CSV" ) opts.add_argument( "-d", "--disable-logging", dest="disable_logging", action="store_true", default=None, help="Disable logging to psn_monitor_.log" ) args = parser.parse_args() if len(sys.argv) == 1: parser.print_help(sys.stderr) sys.exit(1) if args.config_file: CLI_CONFIG_PATH = os.path.expanduser(args.config_file) cfg_path = find_config_file(CLI_CONFIG_PATH) if not cfg_path and CLI_CONFIG_PATH: print(f"* Error: Config file '{CLI_CONFIG_PATH}' does not exist") sys.exit(1) if cfg_path: try: with open(cfg_path, "r") as cf: exec(cf.read(), globals()) except Exception as e: print(f"* Error loading config file '{cfg_path}': {e}") sys.exit(1) if args.env_file: DOTENV_FILE = os.path.expanduser(args.env_file) else: if DOTENV_FILE: DOTENV_FILE = os.path.expanduser(DOTENV_FILE) if DOTENV_FILE and DOTENV_FILE.lower() == 'none': env_path = None else: try: from dotenv import load_dotenv, find_dotenv if DOTENV_FILE: env_path = DOTENV_FILE if not os.path.isfile(env_path): print(f"* Warning: dotenv file '{env_path}' does not exist\n") else: load_dotenv(env_path, override=True) else: env_path = find_dotenv() or None if env_path: load_dotenv(env_path, override=True) except ImportError: env_path = DOTENV_FILE if DOTENV_FILE else None if env_path: print(f"* Warning: Cannot load dotenv file '{env_path}' because 'python-dotenv' is not installed\n\nTo install it, run:\n pip3 install python-dotenv\n\nOnce installed, re-run this tool\n") if env_path: for secret in SECRET_KEYS: val = os.getenv(secret) if val is not None: globals()[secret] = val local_tz = None if LOCAL_TIMEZONE == "Auto": if get_localzone is not None: try: local_tz = get_localzone() except Exception: pass if local_tz: LOCAL_TIMEZONE = str(local_tz) else: print("* Error: Cannot detect local timezone, consider setting LOCAL_TIMEZONE to your local timezone manually !") sys.exit(1) else: if not is_valid_timezone(LOCAL_TIMEZONE): print(f"* Error: Configured LOCAL_TIMEZONE '{LOCAL_TIMEZONE}' is not valid. Please use a valid pytz timezone name.") sys.exit(1) if not check_internet(): sys.exit(1) if args.send_test_email: print("* Sending test email notification ...\n") if send_email("psn_monitor: test email", "This is test email - your SMTP settings seems to be correct !", "", SMTP_SSL, smtp_timeout=5) == 0: print("* Email sent successfully !") else: sys.exit(1) sys.exit(0) if not args.psn_user_id: print("* Error: PSN_USER_ID needs to be defined !") sys.exit(1) if args.npsso_key: PSN_NPSSO = args.npsso_key if not PSN_NPSSO or PSN_NPSSO == "your_psn_npsso_code": print("* Error: PSN_NPSSO (-n / --npsso_key) value is empty or incorrect") sys.exit(1) if args.info_mode: include_trophies = args.include_trophies if hasattr(args, 'include_trophies') and args.include_trophies else False show_recent_games = not (hasattr(args, 'no_recent_games') and args.no_recent_games) get_user_info(args.psn_user_id, include_trophies=include_trophies, show_recent_games=show_recent_games) sys.exit(0) if args.check_interval: PSN_CHECK_INTERVAL = args.check_interval LIVENESS_CHECK_COUNTER = LIVENESS_CHECK_INTERVAL / PSN_CHECK_INTERVAL if args.active_interval: PSN_ACTIVE_CHECK_INTERVAL = args.active_interval if args.csv_file: CSV_FILE = os.path.expanduser(args.csv_file) else: if CSV_FILE: CSV_FILE = os.path.expanduser(CSV_FILE) if CSV_FILE: try: with open(CSV_FILE, 'a', newline='', buffering=1, encoding="utf-8") as _: pass except Exception as e: print(f"* Error, CSV file cannot be opened for writing: {e}") sys.exit(1) if args.disable_logging is True: DISABLE_LOGGING = True if not DISABLE_LOGGING: log_path = Path(os.path.expanduser(PSN_LOGFILE)) if log_path.parent != Path('.'): if log_path.suffix == "": log_path = log_path.parent / f"{log_path.name}_{args.psn_user_id}.log" else: if log_path.suffix == "": log_path = Path(f"{log_path.name}_{args.psn_user_id}.log") log_path.parent.mkdir(parents=True, exist_ok=True) FINAL_LOG_PATH = str(log_path) sys.stdout = Logger(FINAL_LOG_PATH) else: FINAL_LOG_PATH = None if args.notify_active_inactive is True: ACTIVE_INACTIVE_NOTIFICATION = True if args.notify_game_change is True: GAME_CHANGE_NOTIFICATION = True if args.notify_errors is False: ERROR_NOTIFICATION = False if SMTP_HOST.startswith("your_smtp_server_"): ACTIVE_INACTIVE_NOTIFICATION = False GAME_CHANGE_NOTIFICATION = False ERROR_NOTIFICATION = False print(f"* PSN polling intervals:\t[offline: {display_time(PSN_CHECK_INTERVAL)}] [online: {display_time(PSN_ACTIVE_CHECK_INTERVAL)}]") print(f"* Email notifications:\t\t[online/offline status changes = {ACTIVE_INACTIVE_NOTIFICATION}] [game changes = {GAME_CHANGE_NOTIFICATION}]\n*\t\t\t\t[errors = {ERROR_NOTIFICATION}]") print(f"* Liveness check:\t\t{bool(LIVENESS_CHECK_INTERVAL)}" + (f" ({display_time(LIVENESS_CHECK_INTERVAL)})" if LIVENESS_CHECK_INTERVAL else "")) print(f"* CSV logging enabled:\t\t{bool(CSV_FILE)}" + (f" ({CSV_FILE})" if CSV_FILE else "")) print(f"* Output logging enabled:\t{not DISABLE_LOGGING}" + (f" ({FINAL_LOG_PATH})" if not DISABLE_LOGGING else "")) print(f"* Configuration file:\t\t{cfg_path}") print(f"* Dotenv file:\t\t\t{env_path or 'None'}") print(f"* Local timezone:\t\t{LOCAL_TIMEZONE}") out = f"\nMonitoring user with PSN ID {args.psn_user_id}" print(out) print("-" * len(out)) # We define signal handlers only for Linux, Unix & MacOS since Windows has limited number of signals supported if platform.system() != 'Windows': signal.signal(signal.SIGUSR1, toggle_active_inactive_notifications_signal_handler) signal.signal(signal.SIGUSR2, toggle_game_change_notifications_signal_handler) signal.signal(signal.SIGTRAP, increase_active_check_signal_handler) signal.signal(signal.SIGABRT, decrease_active_check_signal_handler) signal.signal(signal.SIGHUP, reload_secrets_signal_handler) psn_monitor_user(args.psn_user_id, CSV_FILE) sys.stdout = stdout_bck sys.exit(0) if __name__ == "__main__": main()