tongoclinh/get_wishes.py

## get_wishes.py
#!/usr/bin/env python3
"""
get_genshin_wishes.py

A Python rewrite of get_genshin_wishes.sh that extracts the Genshin Impact wish‑history URL from
the recent WebKit network cache and copies it to the macOS clipboard.

Author: <your name>
"""

from __future__ import annotations

import argparse
import os
import re
import subprocess
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Iterable, List, Optional
import sqlite3


class WishExtractor:
    HOYO_PATTERN = r"https://[^ ]+gacha[^ ]+authkey[^ ]+game_biz[^ ]+"
    """
    Extracts the wish‑history URL from a recent WebKit cache directory.

    Parameters
    ----------
    game : str
        Either ``"genshin"`` or ``"honkai"`` – the bundle id is looked up from this.
    home : Path, optional
        The user's home directory.  Defaults to ``Path.home()``.
    """

    # ---------------------------------------------------------------------------
    # Configuration
    # ---------------------------------------------------------------------------

    CONFIGs = {
        "genshin": {
            "bundle": "com.miHoYo.GenshinImpact",
            "import_instructions": (
                "Import the URL on https://paimon.moe/wish/import to view your wishes history."
            ),
        },
        "hsr": {
            "bundle": "com.HoYoverse.hkrpgoversea",
            "import_instructions": (
                'Import the URL on https://starrailstation.com/en/warp#import to view your wishes history. Uncheck "Upload to global stats" if the website is stuck forever.'
            ),
        },
        "wutheringwaves": {
            "bundle": "com.kurogame.wutheringwaves.global",
            "import_instructions": (
                "Import the URL on https://wuwatracker.com to view your wishes history."
            ),
        },
        "zenless": {
            "bundle": "com.HoYoverse.Nap",
            "import_instructions": (
                "Import the URL on https://zzz.rng.moe/en/tracker/import to view your wishes history."
            ),
        },
    }

    # ---------------------------------------------------------------------------
    # Construction
    # ---------------------------------------------------------------------------

    def __init__(self, game: str = "genshin", home: Optional[Path] = None) -> None:
        if game not in self.CONFIGs:
            raise ValueError(
                f"Unsupported game {game!r}.  Must be one of {list(self.CONFIGs)}"
            )
        self.game = game
        self.bundle_id = self.CONFIGs[game]["bundle"]
        self.home = Path(home) if home else Path.home()

    # ---------------------------------------------------------------------------
    # Public API – orchestrates the whole workflow
    # ---------------------------------------------------------------------------

    def run(self) -> None:
        if self.game == "wutheringwaves":
            url, matched_file = self.ww_run()
        elif self.game == "zenless":
            url, matched_file = self.zz_run()
        else:
            url, matched_file = self.hoyo_run()

        if not url:
            self._error(
                "URL not found!\n"
                "Are you sure you opened the wishing history screen?\n"
                "If so, the script may need to be updated."
            )

        print(f"Found Wish History URL in file:\n{matched_file}\n")
        print(url)
        print("\nURL copied to clipboard!\n")

        self.copy_to_clipboard(url)

        print(
            self.CONFIGs[self.game]["import_instructions"]
        )

    # ---------------------------------------------------------------------------
    # Wuthering Waves specific extraction
    # ---------------------------------------------------------------------------

    def ww_run(self) -> None:
        cache_file = (
            self.home
            / "Library"
            / "Containers"
            / "com.kurogame.wutheringwaves.global"
            / "Data"
            / "Library"
            / "Logs"
            / "Client"
            / "Client.log"
        )
        if not cache_file.is_file():
            self._error(
                f"Couldn't find Wuthering Waves log file in\n{cache_file}",
                exit_code=1,
            )

        URL_RE = re.compile(
            r"https://aki-gm-resources(-oversea)?\.aki-game\.(net|com)[^\"]*"
        )

        def find_last_url(path: Path) -> str | None:
            try:
                with path.open("r", encoding="utf-8") as fh:
                    # Đọc toàn bộ dòng, để có thể lấy “đầu cuối”.
                    lines = fh.readlines()
            except FileNotFoundError:
                self._error(f"File not found: {path}", exit_code=1)
                return None

            urls = [
                m.group(0) for line in reversed(lines) if (m := URL_RE.search(line))
            ]
            return urls[0] if urls else None

        url = find_last_url(cache_file)
        if not url:
            self._error(
                "URL not found!\n"
                "Are you sure you opened the wishing history screen?\n"
                "If so, the script may need to be updated."
            )
        return (url, cache_file)

    # ---------------------------------------------------------------------------
    # Zenless Zone specific extraction
    # ---------------------------------------------------------------------------

    def zz_run(self) -> None:

        def connect_db(db_path: str) -> sqlite3.Connection:
            """Return a SQLite connection to the given database file."""
            if not Path(db_path).is_file():
                self._error(
                    f"Database for Zenless Zone Zero not found: {db_path}", exit_code=1
                )
                raise FileNotFoundError(f"Database file not found: {db_path}")
            return sqlite3.connect(db_path)

        def find_latest_request(conn: sqlite3.Connection, pattern: str) -> str | None:
            """
            Return the newest `request_key` that matches the regular expression.

            Parameters
            ----------
            conn : sqlite3.Connection
                Active connection to the database.
            pattern : str
                Regular expression used to filter request keys.

            Returns
            -------
            str or None
                The matched key, or None if nothing matches.
            """
            # We let SQLite sort by the timestamp column (text type, ISO‑8601 format)
            sql = """
                SELECT request_key
                FROM cfurl_cache_response
                WHERE request_key REGEXP ?
            ORDER BY time_stamp DESC
                LIMIT 1;
            """

            # Register a custom REGEXP function so that SQLite can use Python's regex.
            conn.create_function(
                "REGEXP", 2, lambda expr, val: re.search(expr, val) is not None
            )

            cur = conn.execute(sql, (pattern,))
            row = cur.fetchone()
            return row[0] if row else None

        cached_db = (
            self.home
            / "Library"
            / "Containers"
            / self.bundle_id
            / "Data"
            / "Library"
            / "Caches"
            / self.bundle_id
            / "Cache.db"
        )

        conn = connect_db(cached_db)
        url = find_latest_request(conn, self.HOYO_PATTERN)

        if url is None:
            self._error(
                "URL not found!\n"
                "Are you sure you opened the wishing history screen?\n"
                "If so, the script may need to be updated."
            )

        return url, cached_db

    # ---------------------------------------------------------------------------
    # Genshin Impact / Honkai Star Rail extraction
    # ---------------------------------------------------------------------------

    def hoyo_run(self) -> None:
        CACHE_VERSIONS = ("Version 17", "Version 16")  # order matters – try newer first
        URL_PATTERN = re.compile(
            self.HOYO_PATTERN,
            re.IGNORECASE,
        )
        NON_URL_CHARS_RE = re.compile(r"[^A-Za-z0-9._~:/?#$@!&'()*+,;=%\-]")
        # ---------------------------------------------------------------------------
        # Individual steps – can be unit‑tested separately
        # ---------------------------------------------------------------------------

        def find_cache_dir() -> Path:
            """
            Return the first existing cache directory for the selected bundle id.

            Raises an error if none of the expected directories exist.
            """
            base = (
                self.home
                / "Library"
                / "Containers"
                / self.bundle_id
                / "Data"
                / "Library"
                / "Caches"
                / "WebKit"
                / "NetworkCache"
            )

            for version in CACHE_VERSIONS:
                candidate = base / version / "Records"
                if candidate.is_dir():
                    return candidate

            self._error(
                f"Couldn't find Genshin Impact/PlayCover cache directory in\n{self.home / 'Library' / 'Containers'}",
                exit_code=1,
            )
            # pragma: no cover – the error will raise and never reach this point
            raise RuntimeError("Unreachable")

        def candidate_files(cache_dir: Path) -> Iterable[Path]:
            """
            Yield all files under ``cache_dir`` that were modified in the last hour.
            The generator yields them sorted by modification time (most recent first).
            """
            one_hour_ago = datetime.now() - timedelta(hours=1)
            candidates = (
                p
                for p in cache_dir.rglob("*")
                if p.is_file()
                and p.name.lower() != ".ds_store"
                and datetime.fromtimestamp(p.stat().st_mtime) > one_hour_ago
            )
            # Sort by mtime descending so the newest files are examined first
            sorted_candidates = sorted(
                candidates,
                key=lambda p: p.stat().st_mtime,
                reverse=True,
            )
            yield from sorted_candidates

        def extract_url(files: List[Path]) -> tuple[Optional[str], Optional[Path]]:
            """
            Search the given list of files for a wish‑history URL.

            Returns ``(url, file)`` if found, otherwise ``(None, None)``.
            """
            for f in files:
                try:
                    content_bytes = f.read_bytes()
                except OSError as e:
                    print(f"WARNING: Couldn't read {f}: {e}", file=sys.stderr)
                    continue

                # 1️⃣ Decode the bytes – errors become � so we can safely strip them later
                decoded = content_bytes.decode("utf-8", errors="replace")

                # 2️⃣ Remove any characters that are not allowed in a URL
                cleaned = NON_URL_CHARS_RE.sub(" ", decoded)

                match = URL_PATTERN.search(cleaned)
                if match:
                    return match.group(0), f

            return None, None

        """Run the extraction pipeline and copy the URL to the clipboard."""
        if self.game == "zenless":
            candidate_files = []
        cache_dir = find_cache_dir()
        candidate_files = list(candidate_files(cache_dir))
        if not candidate_files:
            self._error(
                "Couldn't find any Genshin network cache files!\n"
                "Are you sure you opened the wishing history screen recently?"
            )

        print(f"# candidate files: {len(candidate_files)}\n")

        url, matched_file = extract_url(candidate_files)
        return url, matched_file

    def copy_to_clipboard(self, url: str) -> None:
        """Copy the given URL to macOS clipboard via `pbcopy`."""
        try:
            subprocess.run(
                ["pbcopy"],
                input=url.encode("utf-8"),
                check=True,
            )
        except Exception as e:
            print(f"WARNING: Failed to copy URL to clipboard: {e}", file=sys.stderr)

    # ---------------------------------------------------------------------------
    # Helpers
    # ---------------------------------------------------------------------------

    @staticmethod
    def _error(message: str, exit_code: int = 1) -> None:
        """Print an error message and exit."""
        print(f"ERROR: {message}\n", file=sys.stderr)
        sys.exit(exit_code)


# -------------------------------------------------------------------------------
# CLI handling – keeps the script runnable from the command line
# -------------------------------------------------------------------------------


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Extract wish‑history URL from WebKit cache."
    )
    parser.add_argument(
        "--game",
        choices=["genshin", "hsr", "wutheringwaves", "zenless"],
        # default="genshin",
        help="genshin (Genshin Impact)/hsr (Honkai Star Rail)/wutheringwaves (Wuthering Waves -- macOS version)/zenless (Zenless Zone Zero)",
    )
    return parser


def main() -> None:
    args = build_parser().parse_args()
    extractor = WishExtractor(game=args.game)
    extractor.run()


if __name__ == "__main__":
    main()
	#!/usr/bin/env python3
	"""
	get_genshin_wishes.py

	A Python rewrite of get_genshin_wishes.sh that extracts the Genshin Impact wish‑history URL from
	the recent WebKit network cache and copies it to the macOS clipboard.

	Author: <your name>
	"""

	from __future__ import annotations

	import argparse
	import os
	import re
	import subprocess
	import sys
	from datetime import datetime, timedelta
	from pathlib import Path
	from typing import Iterable, List, Optional
	import sqlite3


	class WishExtractor:
	HOYO_PATTERN = r"https://[^ ]+gacha[^ ]+authkey[^ ]+game_biz[^ ]+"
	"""
	Extracts the wish‑history URL from a recent WebKit cache directory.

	Parameters
	----------
	game : str
	Either ``"genshin"`` or ``"honkai"`` – the bundle id is looked up from this.
	home : Path, optional
	The user's home directory. Defaults to ``Path.home()``.
	"""

	# ---------------------------------------------------------------------------
	# Configuration
	# ---------------------------------------------------------------------------

	CONFIGs = {
	"genshin": {
	"bundle": "com.miHoYo.GenshinImpact",
	"import_instructions": (
	"Import the URL on https://paimon.moe/wish/import to view your wishes history."
	),
	},
	"hsr": {
	"bundle": "com.HoYoverse.hkrpgoversea",
	"import_instructions": (
	'Import the URL on https://starrailstation.com/en/warp#import to view your wishes history. Uncheck "Upload to global stats" if the website is stuck forever.'
	),
	},
	"wutheringwaves": {
	"bundle": "com.kurogame.wutheringwaves.global",
	"import_instructions": (
	"Import the URL on https://wuwatracker.com to view your wishes history."
	),
	},
	"zenless": {
	"bundle": "com.HoYoverse.Nap",
	"import_instructions": (
	"Import the URL on https://zzz.rng.moe/en/tracker/import to view your wishes history."
	),
	},
	}

	# ---------------------------------------------------------------------------
	# Construction
	# ---------------------------------------------------------------------------

	def __init__(self, game: str = "genshin", home: Optional[Path] = None) -> None:
	if game not in self.CONFIGs:
	raise ValueError(
	f"Unsupported game {game!r}. Must be one of {list(self.CONFIGs)}"
	)
	self.game = game
	self.bundle_id = self.CONFIGs[game]["bundle"]
	self.home = Path(home) if home else Path.home()

	# ---------------------------------------------------------------------------
	# Public API – orchestrates the whole workflow
	# ---------------------------------------------------------------------------

	def run(self) -> None:
	if self.game == "wutheringwaves":
	url, matched_file = self.ww_run()
	elif self.game == "zenless":
	url, matched_file = self.zz_run()
	else:
	url, matched_file = self.hoyo_run()

	if not url:
	self._error(
	"URL not found!\n"
	"Are you sure you opened the wishing history screen?\n"
	"If so, the script may need to be updated."
	)

	print(f"Found Wish History URL in file:\n{matched_file}\n")
	print(url)
	print("\nURL copied to clipboard!\n")

	self.copy_to_clipboard(url)

	print(
	self.CONFIGs[self.game]["import_instructions"]
	)

	# ---------------------------------------------------------------------------
	# Wuthering Waves specific extraction
	# ---------------------------------------------------------------------------

	def ww_run(self) -> None:
	cache_file = (
	self.home
	/ "Library"
	/ "Containers"
	/ "com.kurogame.wutheringwaves.global"
	/ "Data"
	/ "Library"
	/ "Logs"
	/ "Client"
	/ "Client.log"
	)
	if not cache_file.is_file():
	self._error(
	f"Couldn't find Wuthering Waves log file in\n{cache_file}",
	exit_code=1,
	)

	URL_RE = re.compile(
	r"https://aki-gm-resources(-oversea)?\.aki-game\.(net\|com)[^\"]*"
	)

	def find_last_url(path: Path) -> str \| None:
	try:
	with path.open("r", encoding="utf-8") as fh:
	# Đọc toàn bộ dòng, để có thể lấy “đầu cuối”.
	lines = fh.readlines()
	except FileNotFoundError:
	self._error(f"File not found: {path}", exit_code=1)
	return None

	urls = [
	m.group(0) for line in reversed(lines) if (m := URL_RE.search(line))
	]
	return urls[0] if urls else None

	url = find_last_url(cache_file)
	if not url:
	self._error(
	"URL not found!\n"
	"Are you sure you opened the wishing history screen?\n"
	"If so, the script may need to be updated."
	)
	return (url, cache_file)

	# ---------------------------------------------------------------------------
	# Zenless Zone specific extraction
	# ---------------------------------------------------------------------------

	def zz_run(self) -> None:

	def connect_db(db_path: str) -> sqlite3.Connection:
	"""Return a SQLite connection to the given database file."""
	if not Path(db_path).is_file():
	self._error(
	f"Database for Zenless Zone Zero not found: {db_path}", exit_code=1
	)
	raise FileNotFoundError(f"Database file not found: {db_path}")
	return sqlite3.connect(db_path)

	def find_latest_request(conn: sqlite3.Connection, pattern: str) -> str \| None:
	"""
	Return the newest `request_key` that matches the regular expression.

	Parameters
	----------
	conn : sqlite3.Connection
	Active connection to the database.
	pattern : str
	Regular expression used to filter request keys.

	Returns
	-------
	str or None
	The matched key, or None if nothing matches.
	"""
	# We let SQLite sort by the timestamp column (text type, ISO‑8601 format)
	sql = """
	SELECT request_key
	FROM cfurl_cache_response
	WHERE request_key REGEXP ?
	ORDER BY time_stamp DESC
	LIMIT 1;
	"""

	# Register a custom REGEXP function so that SQLite can use Python's regex.
	conn.create_function(
	"REGEXP", 2, lambda expr, val: re.search(expr, val) is not None
	)

	cur = conn.execute(sql, (pattern,))
	row = cur.fetchone()
	return row[0] if row else None

	cached_db = (
	self.home
	/ "Library"
	/ "Containers"
	/ self.bundle_id
	/ "Data"
	/ "Library"
	/ "Caches"
	/ self.bundle_id
	/ "Cache.db"
	)

	conn = connect_db(cached_db)
	url = find_latest_request(conn, self.HOYO_PATTERN)

	if url is None:
	self._error(
	"URL not found!\n"
	"Are you sure you opened the wishing history screen?\n"
	"If so, the script may need to be updated."
	)

	return url, cached_db

	# ---------------------------------------------------------------------------
	# Genshin Impact / Honkai Star Rail extraction
	# ---------------------------------------------------------------------------

	def hoyo_run(self) -> None:
	CACHE_VERSIONS = ("Version 17", "Version 16") # order matters – try newer first
	URL_PATTERN = re.compile(
	self.HOYO_PATTERN,
	re.IGNORECASE,
	)
	NON_URL_CHARS_RE = re.compile(r"[^A-Za-z0-9._~:/?#$@!&'()*+,;=%\-]")
	# ---------------------------------------------------------------------------
	# Individual steps – can be unit‑tested separately
	# ---------------------------------------------------------------------------

	def find_cache_dir() -> Path:
	"""
	Return the first existing cache directory for the selected bundle id.

	Raises an error if none of the expected directories exist.
	"""
	base = (
	self.home
	/ "Library"
	/ "Containers"
	/ self.bundle_id
	/ "Data"
	/ "Library"
	/ "Caches"
	/ "WebKit"
	/ "NetworkCache"
	)

	for version in CACHE_VERSIONS:
	candidate = base / version / "Records"
	if candidate.is_dir():
	return candidate

	self._error(
	f"Couldn't find Genshin Impact/PlayCover cache directory in\n{self.home / 'Library' / 'Containers'}",
	exit_code=1,
	)
	# pragma: no cover – the error will raise and never reach this point
	raise RuntimeError("Unreachable")

	def candidate_files(cache_dir: Path) -> Iterable[Path]:
	"""
	Yield all files under ``cache_dir`` that were modified in the last hour.
	The generator yields them sorted by modification time (most recent first).
	"""
	one_hour_ago = datetime.now() - timedelta(hours=1)
	candidates = (
	p
	for p in cache_dir.rglob("*")
	if p.is_file()
	and p.name.lower() != ".ds_store"
	and datetime.fromtimestamp(p.stat().st_mtime) > one_hour_ago
	)
	# Sort by mtime descending so the newest files are examined first
	sorted_candidates = sorted(
	candidates,
	key=lambda p: p.stat().st_mtime,
	reverse=True,
	)
	yield from sorted_candidates

	def extract_url(files: List[Path]) -> tuple[Optional[str], Optional[Path]]:
	"""
	Search the given list of files for a wish‑history URL.

	Returns ``(url, file)`` if found, otherwise ``(None, None)``.
	"""
	for f in files:
	try:
	content_bytes = f.read_bytes()
	except OSError as e:
	print(f"WARNING: Couldn't read {f}: {e}", file=sys.stderr)
	continue

	# 1️⃣ Decode the bytes – errors become � so we can safely strip them later
	decoded = content_bytes.decode("utf-8", errors="replace")

	# 2️⃣ Remove any characters that are not allowed in a URL
	cleaned = NON_URL_CHARS_RE.sub(" ", decoded)

	match = URL_PATTERN.search(cleaned)
	if match:
	return match.group(0), f

	return None, None

	"""Run the extraction pipeline and copy the URL to the clipboard."""
	if self.game == "zenless":
	candidate_files = []
	cache_dir = find_cache_dir()
	candidate_files = list(candidate_files(cache_dir))
	if not candidate_files:
	self._error(
	"Couldn't find any Genshin network cache files!\n"
	"Are you sure you opened the wishing history screen recently?"
	)

	print(f"# candidate files: {len(candidate_files)}\n")

	url, matched_file = extract_url(candidate_files)
	return url, matched_file

	def copy_to_clipboard(self, url: str) -> None:
	"""Copy the given URL to macOS clipboard via `pbcopy`."""
	try:
	subprocess.run(
	["pbcopy"],
	input=url.encode("utf-8"),
	check=True,
	)
	except Exception as e:
	print(f"WARNING: Failed to copy URL to clipboard: {e}", file=sys.stderr)

	# ---------------------------------------------------------------------------
	# Helpers
	# ---------------------------------------------------------------------------

	@staticmethod
	def _error(message: str, exit_code: int = 1) -> None:
	"""Print an error message and exit."""
	print(f"ERROR: {message}\n", file=sys.stderr)
	sys.exit(exit_code)


	# -------------------------------------------------------------------------------
	# CLI handling – keeps the script runnable from the command line
	# -------------------------------------------------------------------------------


	def build_parser() -> argparse.ArgumentParser:
	parser = argparse.ArgumentParser(
	description="Extract wish‑history URL from WebKit cache."
	)
	parser.add_argument(
	"--game",
	choices=["genshin", "hsr", "wutheringwaves", "zenless"],
	# default="genshin",
	help="genshin (Genshin Impact)/hsr (Honkai Star Rail)/wutheringwaves (Wuthering Waves -- macOS version)/zenless (Zenless Zone Zero)",
	)
	return parser


	def main() -> None:
	args = build_parser().parse_args()
	extractor = WishExtractor(game=args.game)
	extractor.run()


	if __name__ == "__main__":
	main()
No results found