使用Python实现获取本机当前用户登陆过的微信ID

import os
import platform
import re
from utils.utils_cmd import cmdutils


class deviceutils:
    """
    设备工具类
    """

    @staticmethod
    def get_wechat_wxid(folder=none):
        """
        获取设备上登陆的 微信 wxid
        :param folder:仅针对windows平台自定义数据存储路径生效
        :return:
        """
        wxids = []
        system = deviceutils().get_system()
        if system == "windows":
            if not folder:
                file = os.path.join(
                    os.environ['userprofile'],
                    'documents',
                    'wechat files',
                    'all users',
                    'config',
                    'config.data'
                )
            else:
                file = os.path.join(
                    folder,
                    'all users',
                    'config',
                    'config.data'
                )
            with open(file, "rb") as f:
                data = f.read()
            data_matches = re.findall(r'(wxid_.*?)\\\\config', str(data))
            if len(data_matches) > 0:
                wxids.extend(data_matches)
        elif system == "linux":
            user = cmdutils.run_await_results(cmd=['whoami'])
            for folder in ('文档', 'documents'):
                file = f'/home/{user}/{folder}/xwechat_files/all_users/login'
                if os.path.exists(file):
                    file_list = os.listdir(file)
                    if len(file_list) > 0:
                        for filename in file_list:
                            if filename.startswith('wxid_'):
                                wxids.append(filename)
        elif system == "darwin":
            user = cmdutils.run_await_results(cmd=['whoami'])
            file = f"/users/{user}//library/containers/com.tencent.xinwechat/data/documents/xwechat_files/all_users/login"
            if os.path.exists(file):
                file_list = os.listdir(file)
                if len(file_list) > 0:
                    for filename in file_list:
                        if filename.startswith('wxid_'):
                            wxids.append(filename)

        msg = f"{system} 识别到微信{len(wxids)}个" if len(wxids) > 0 else f"{system} 未识别到微信"
        print(msg)
        return wxids

    @staticmethod
    def get_system():
        system = platform.system()
        if system in ('windows', 'linux', 'darwin'):
            return system
        return "unsupported os"


if __name__ == '__main__':
    folder = input("input folder:")
    print(deviceutils.get_wechat_wxid(folder))

# !/usr/bin/python3
# -*- coding:utf-8 -*-
"""
@author: jhc000abc@gmail.com
@file: utils_cmd.py
@time: 2025/7/8 21:53 
@desc: 

"""
import subprocess
from typing import union, list, callable, optional
import threading


class cmdutils:
    """

    """

    @staticmethod
    def run_await_results(cmd: union[str, list[str]],
                          timeout: optional[int] = none) -> str:
        """
        阻塞执行命令并返回结果（自动剥离首尾空格）

        :param cmd: 命令（字符串或列表形式）
        :param timeout: 超时时间（秒）
        :return: 命令输出（stdout 或 stderr）
        :raises: subprocess.calledprocesserror 当命令返回非零状态码时
        """
        result = subprocess.run(
            cmd,
            capture_output=true,
            text=true,
            check=true,
            shell=isinstance(cmd, str),
            timeout=timeout
        )
        return result.stdout.strip()

    @staticmethod
    def run_background(cmd: union[str, list[str]],
                       **kwargs) -> subprocess.popen:
        """
        后台非阻塞执行命令

        :param cmd: 命令（字符串或列表形式）
        :param kwargs: 传递给 subprocess.popen 的额外参数
        :return: popen 对象（可通过 returncode 属性检查状态）
        """
        return subprocess.popen(
            cmd,
            stdout=subprocess.devnull,
            stderr=subprocess.devnull,
            shell=isinstance(cmd, str),
            **kwargs
        )

    @staticmethod
    def run_realtime_output(cmd: union[str, list[str]],
                            output_handler: callable[[str], none] = print,
                            **kwargs) -> subprocess.popen:
        """
        实时输出命令执行结果（非阻塞）

        :param cmd: 命令（字符串或列表形式）
        :param output_handler: 自定义输出处理函数（默认打印到控制台）
        :param kwargs: 传递给 subprocess.popen 的额外参数
        :return: popen 对象
        """

        def _enqueue_output(pipe, handler):
            for line in iter(pipe.readline, ''):
                handler(line.strip())
            pipe.close()

        proc = subprocess.popen(
            cmd,
            stdout=subprocess.pipe,
            stderr=subprocess.stdout,
            text=true,
            shell=isinstance(cmd, str),
            **kwargs
        )

        # 启动独立线程处理输出
        threading.thread(
            target=_enqueue_output,
            args=(proc.stdout, output_handler),
            daemon=true
        ).start()

        return proc

import tkinter as tk
from tkinter import ttk
from tkinter import messagebox
import threading
import requests
import json
import time
import base64
import io
try:
    from pil import image, imagetk
    pil_available = true
except importerror:
    pil_available = false
 
# 全局变量定义
token_id = none        # 接口token
app_id = none          # 设备id（appid）
logged_in = false      # 是否已登录
account_nick = none    # 登录账户昵称（用于提示）
user_wxid = none       # 登录账户的微信id
 
# 初始化tk窗口
root = tk.tk()
root.title("gewechat 工具")
root.geometry("800x600")
 
# 界面布局：登录区、群列表区、批量加好友区、日志区
login_frame = ttk.frame(root, padding=5)
group_frame = ttk.frame(root, padding=5)
friend_frame = ttk.frame(root, padding=5)
log_frame = ttk.frame(root, padding=5)
 
login_frame.grid(row=0, column=0, sticky="nsw")
group_frame.grid(row=0, column=1, sticky="ns")
friend_frame.grid(row=0, column=2, sticky="ns")
log_frame.grid(row=1, column=0, columnspan=3, sticky="nsew")
 
root.grid_columnconfigure(0, weight=1)
root.grid_columnconfigure(1, weight=1)
root.grid_columnconfigure(2, weight=1)
root.grid_rowconfigure(1, weight=1)
 
# 登录区域
lbl_login_status = ttk.label(login_frame, text="登录状态：未登录")
lbl_login_status.pack(pady=5)
qr_label = ttk.label(login_frame, text="（请扫描二维码登录）", justify="center")
qr_label.pack()
 
# 群列表操作区域
group_buttons_frame = ttk.frame(group_frame)
group_buttons_frame.pack(fill="x", pady=2)
btn_refresh_groups = ttk.button(group_buttons_frame, text="获取群列表")
btn_export_members = ttk.button(group_buttons_frame, text="导出群成员")
btn_refresh_groups.pack(side="left", padx=2)
btn_export_members.pack(side="left", padx=2)
group_listbox = tk.listbox(group_frame, height=15)
scrollbar_groups = ttk.scrollbar(group_frame, orient="vertical", command=group_listbox.yview)
group_listbox.config(yscrollcommand=scrollbar_groups.set)
group_listbox.pack(side="left", fill="y", expand=true)
scrollbar_groups.pack(side="left", fill="y")
 
# 批量加好友区域
friend_label = ttk.label(friend_frame, text="批量加好友（每行一个id）：")
friend_label.pack(anchor="w")
friend_text = tk.text(friend_frame, width=30, height=10)
friend_text.pack(fill="both", expand=true)
btn_add_friends = ttk.button(friend_frame, text="开始添加好友")
btn_add_friends.pack(pady=5)
 
# 日志输出区域
log_label = ttk.label(log_frame, text="日志输出：")
log_label.pack(anchor="w")
log_text = tk.text(log_frame, height=8)
log_text.pack(side="left", fill="both", expand=true)
log_scroll = ttk.scrollbar(log_frame, orient="vertical", command=log_text.yview)
log_text.config(yscrollcommand=log_scroll.set, state="normal")
log_scroll.pack(side="right", fill="y")
 
def log_message(message: str):
    """在日志文本框追加消息（超过500字截断）"""
    truncated = message if len(message) <= 500 else message[:500] + "...(截断)"
    def _append():
        log_text.insert(tk.end, truncated + "\n")
        log_text.see(tk.end)
    root.after(0, _append)
 
def save_credentials():
    """将当前凭据保存到本地文件"""
    cred = {"appid": app_id, "tokenid": token_id}
    if user_wxid:
        cred["wxid"] = user_wxid
    try:
        with open("gewe_cred.json", "w", encoding="utf-8") as f:
            json.dump(cred, f, ensure_ascii=false, indent=4)
    except exception as e:
        log_message(f"保存凭据失败: {e}")
 
def load_credentials():
    """从本地文件加载上次保存的凭据"""
    global app_id, token_id
    try:
        with open("gewe_cred.json", "r", encoding="utf-8") as f:
            cred = json.load(f)
            app_id = cred.get("appid", "")
            token_id = cred.get("tokenid", "")
    except filenotfounderror:
        app_id = ""
        token_id = ""
    except exception as e:
        log_message(f"读取凭据异常: {e}")
        app_id = ""
        token_id = ""
 
def http_post(endpoint: str, data: dict):
    """发送post请求到gewechat api并返回结果json（同时日志记录请求和响应）"""
    url = base_url + endpoint
    try:
        payload_str = json.dumps(data, ensure_ascii=false)
    except exception:
        payload_str = str(data)
    log_message(f"请求: post {endpoint} {payload_str}")
    headers = {"content-type": "application/json"}
    if token_id:
        headers["x-gewe-token"] = token_id
    try:
        resp = requests.post(url, json=data, headers=headers, timeout=10)
    except exception as e:
        log_message(f"请求异常: {e}")
        return none
    resp_text = resp.text
    log_message(f"响应: {resp_text}")
    try:
        result = resp.json()
    except exception:
        log_message("响应非json格式或解析失败")
        return none
    return result
 
def enable_ui_after_login():
    """登录成功后，启用各功能按钮，并更新登录状态显示"""
    btn_refresh_groups.config(state="normal")
    btn_export_members.config(state="normal")
    btn_add_friends.config(state="normal")
    status_text = "登录成功"
    if account_nick:
        status_text = f"已登录：{account_nick}"
    lbl_login_status.config(text=status_text)
    try:
        qr_label.config(image="", text="")  # 清除二维码显示
    except exception:
        pass
 
def login_process():
    """登录线程：检查在线状态或执行扫码登录"""
    global logged_in, token_id, app_id, account_nick, user_wxid
    # 1. 尝试使用已有token检查是否在线
    if token_id:
        result = http_post("/login/checkonline", {"appid": app_id})
        if result:
            ret = result.get("ret")
            data = result.get("data")
            if (ret == 200 or ret == 0) and data is true:
                # 当前已在线
                logged_in = true
                log_message("当前已在线，无需重复登录")
                root.after(0, enable_ui_after_login)
                return
    # 2. 若不存在token或检查结果离线，获取新token
    if not token_id or (result and result.get("ret") in [-1, 401, 404, 500] or result is none):
        log_message("获取新的token...")
        token_res = http_post("/tools/gettokenid", {})
        if token_res:
            if token_res.get("ret") in (200, 0):
                new_token = token_res.get("data") or token_res.get("token") or token_res.get("tokenid")
                if not new_token:
                    data_field = token_res.get("data")
                    if data_field and isinstance(data_field, dict):
                        new_token = data_field.get("token") or data_field.get("tokenid")
                if new_token:
                    token_id = new_token
                    log_message("token获取成功")
                else:
                    log_message("token获取失败: 未返回token值")
            else:
                log_message(f"获取token失败: {token_res.get('msg')}")
        else:
            log_message("获取token失败: 无响应")
        if not token_id:
            messagebox.showerror("登录失败", "无法获取token，登录终止。")
            return
    # 3. 获取登录二维码
    log_message("请求登录二维码...")
    req_app_id = app_id if app_id else ""
    payload = {"appid": req_app_id, "regionid": "320000"}
    qr_res = http_post("/login/getloginqrcode", payload)
    if not qr_res or qr_res.get("ret") not in [200, 0]:
        log_message(f"获取登录二维码失败: {qr_res.get('msg') if qr_res else '无响应'}")
        messagebox.showerror("登录失败", "获取登录二维码失败，请检查服务器。")
        return
    qr_data = qr_res.get("data", {})
    new_app_id = qr_data.get("appid", "")
    if new_app_id:
        app_id = new_app_id
        log_message(f"使用设备id: {app_id}")
    uuid = qr_data.get("uuid")
    if not uuid:
        qr_link = qr_data.get("qrdata")
        if qr_link and "/x/" in qr_link:
            uuid = qr_link.split("/x/")[-1]
    qr_img_b64 = qr_data.get("qrimgbase64", "")
    if qr_img_b64:
        if qr_img_b64.startswith("data:image"):
            comma_index = qr_img_b64.find("base64,")
            if comma_index != -1:
                qr_img_b64 = qr_img_b64[comma_index+7:]
        try:
            img_data = base64.b64decode(qr_img_b64)
            if pil_available:
                image = image.open(io.bytesio(img_data))
                tk_img = imagetk.photoimage(image)
            else:
                tk_img = tk.photoimage(data=qr_img_b64)
            def show_qr():
                qr_label.config(image=tk_img)
                qr_label.image = tk_img
                lbl_login_status.config(text="请使用微信扫描二维码登录")
            root.after(0, show_qr)
        except exception as e:
            log_message(f"二维码显示失败: {e}")
    # 4. 轮询检查扫码状态
    log_message("等待扫码确认...")
    start_time = time.time()
    refresh_count = 0
    while not logged_in and refresh_count < 2:
        check_res = http_post("/login/checklogin", {"appid": app_id, "uuid": uuid, "captchcode": ""})
        if check_res:
            ret = check_res.get("ret")
            data = check_res.get("data", {})
            if (ret == 200 or ret == 0) and data:
                status = data.get("status")
                if status == 0:
                    # 未扫码
                    pass
                elif status == 1:
                    # 已扫码，待确认
                    nick = data.get("nickname")
                    if nick:
                        account_nick = nick
                        log_message(f"已扫码，等待确认登录... (微信昵称: {nick})")
                elif status == 2:
                    # 登录成功
                    logged_in = true
                    login_info = data.get("logininfo")
                    if login_info and isinstance(login_info, dict):
                        user_name = login_info.get("username") or login_info.get("wxid")
                        if user_name:
                            log_message(f"登录账号id: {user_name}")
                            user_wxid = user_name
                    if not account_nick:
                        account_nick = data.get("nickname") or (login_info.get("nickname") if login_info else none)
                    log_message("微信登录成功")
                    break
            else:
                err_msg = check_res.get("msg") or (check_res.get("data", {}).get("msg") if isinstance(check_res.get("data"), dict) else "")
                log_message(f"登录检查失败: {err_msg}")
        # 若超过约220秒未确认且未登录，尝试刷新二维码一次
        if time.time() - start_time > 220 and not logged_in and refresh_count < 1:
            log_message("二维码已过期，正在获取新的二维码...")
            refresh_count += 1
            start_time = time.time()
            qr_res2 = http_post("/login/getloginqrcode", {"appid": app_id, "regionid": "320000"})
            if qr_res2 and (qr_res2.get("ret") == 200 or qr_res2.get("ret") == 0):
                qr_data2 = qr_res2.get("data", {})
                uuid = qr_data2.get("uuid") or uuid
                qr_img_b64_new = qr_data2.get("qrimgbase64")
                if qr_img_b64_new:
                    if qr_img_b64_new.startswith("data:image"):
                        idx = qr_img_b64_new.find("base64,")
                        if idx != -1:
                            qr_img_b64_new = qr_img_b64_new[idx+7:]
                    try:
                        img_data = base64.b64decode(qr_img_b64_new)
                        if pil_available:
                            image = image.open(io.bytesio(img_data))
                            tk_img2 = imagetk.photoimage(image)
                        else:
                            tk_img2 = tk.photoimage(data=qr_img_b64_new)
                        def update_qr():
                            qr_label.config(image=tk_img2)
                            qr_label.image = tk_img2
                        root.after(0, update_qr)
                    except exception as e:
                        log_message(f"新二维码显示失败: {e}")
                log_message("请扫描新的二维码")
            else:
                log_message("无法刷新二维码，登录失败")
                break
        time.sleep(2)
    if not logged_in:
        log_message("登录超时或失败")
        messagebox.showwarning("登录失败", "二维码超时，请重新运行程序重试。")
        return
    # 5. 登录成功后，启用功能并保存凭据
    root.after(0, enable_ui_after_login)
    save_credentials()
 
# 获取群列表线程函数
def refresh_groups():
    global group_ids
    log_message("获取通讯录列表...")
    res = http_post("/contacts/fetchcontactslist", {"appid": app_id})
    if not res or res.get("ret") not in [0, 200]:
        log_message("获取通讯录列表失败")
        root.after(0, lambda: messagebox.showerror("错误", "获取群列表失败"))
        return
    data = res.get("data", {})
    chatrooms = data.get("chatrooms", [])
    if not chatrooms:
        log_message("未获取到任何群聊")
        root.after(0, lambda: messagebox.showinfo("结果", "通讯录中没有群聊"))
        return
    log_message(f"获取到{len(chatrooms)}个群id，正在获取群名称...")
    # 获取群/好友简要信息以获得群名称
    info_res = http_post("/contacts/getcontactinfo", {"appid": app_id, "usernames": chatrooms})
    group_name_list = []
    group_ids = []
    if info_res and (info_res.get("ret") == 0 or info_res.get("ret") == 200):
        info_data = info_res.get("data")
        if info_data:
            if isinstance(info_data, list):
                infos = info_data
            elif isinstance(info_data, dict) and info_data.get("contacts"):
                infos = info_data.get("contacts")
            elif isinstance(info_data, dict):
                infos = []
                for val in info_data.values():
                    if isinstance(val, dict):
                        infos.append(val)
            else:
                infos = []
            for info in infos:
                name = info.get("nickname") or info.get("username")
                uid = info.get("username") or info.get("wxid")
                if name:
                    group_name_list.append(name)
                    group_ids.append(uid)
    else:
        group_name_list = chatrooms
        group_ids = chatrooms
    def update_listbox():
        group_listbox.delete(0, tk.end)
        for name in group_name_list:
            group_listbox.insert(tk.end, name)
    root.after(0, update_listbox)
    log_message("群列表更新完成")
 
# 启动群成员导出线程（由按钮触发）
def start_refresh_groups():
    threading.thread(target=refresh_groups, daemon=true).start()
 
def start_export_members():
    # 主线程读取选中项
    sel = group_listbox.curselection()
    if not sel:
        messagebox.showinfo("提示", "请先选择一个群组")
        return
    index = sel[0]
    if index >= len(group_ids):
        messagebox.showerror("错误", "群组选择无效")
        return
    gid = group_ids[index]
    gname = group_listbox.get(index)
    def export_thread(group_id, group_name):
        log_message(f"正在导出群成员: {group_name} ({group_id})")
        res = http_post("/group/getchatroommemberlist", {"appid": app_id, "chatroomid": group_id})
        if not res or res.get("ret") not in [0, 200]:
            log_message("获取群成员列表失败")
            root.after(0, lambda: messagebox.showerror("错误", f"获取群成员失败: {res.get('msg') if res else '无响应'}"))
            return
        data = res.get("data", {})
        members = []
        if data:
            if "memberlist" in data:
                members = data["memberlist"]
            elif "members" in data and isinstance(data["members"], list) and data["members"] and isinstance(data["members"][0], dict):
                members = data["members"]
            elif isinstance(data, list):
                members = data
        if not members:
            log_message("群成员列表为空或解析失败")
            root.after(0, lambda: messagebox.showinfo("提示", "未获取到群成员信息"))
            return
        safe_name = "".join([c if c not in r'\/:*?"<>|' else "_" for c in group_name])
        if not safe_name:
            safe_name = group_id.replace("@chatroom", "")
        filename = f"{safe_name}_members.csv"
        try:
            with open(filename, "w", encoding="utf-8-sig") as f:
                f.write("微信id,昵称\n")
                for member in members:
                    wxid_val = member.get("wxid") or member.get("username") or ""
                    nick_val = member.get("nickname") or ""
                    wxid_val = str(wxid_val).replace(",", " ")
                    nick_val = str(nick_val).replace(",", " ")
                    f.write(f"{wxid_val},{nick_val}\n")
            log_message(f"群成员已导出到文件: {filename}")
            root.after(0, lambda: messagebox.showinfo("导出完成", f"群成员列表已保存至 {filename}"))
        except exception as e:
            log_message(f"导出文件写入失败: {e}")
            root.after(0, lambda: messagebox.showerror("错误", f"导出文件失败: {e}"))
    threading.thread(target=export_thread, args=(gid, gname), daemon=true).start()
 
def start_add_friends():
    # 主线程获取文本框内容
    lines = friend_text.get("1.0", tk.end).strip().splitlines()
    ids_to_add = [line.strip() for line in lines if line.strip()]
    if not ids_to_add:
        messagebox.showinfo("提示", "请在文本框中输入要添加的好友id，每行一个")
        return
    verify_message = "你好，我是通过群聊添加您为好友。"
    def add_thread(id_list):
        for wxid in id_list:
            log_message(f"搜索好友: {wxid}")
            search_res = http_post("/contacts/search", {"appid": app_id, "contactsinfo": wxid})
            if not search_res or search_res.get("ret") not in [0, 200]:
                log_message(f"搜索失败: 无法找到 {wxid}")
                continue
            search_data = search_res.get("data", {})
            v3 = search_data.get("v3")
            v4 = search_data.get("v4")
            nick = search_data.get("nickname")
            if not v3 or not v4:
                log_message(f"无法添加 {wxid}: 未获取到v3/v4")
                continue
            log_message(f"发送好友申请给 {wxid} (昵称: {nick})")
            add_payload = {"appid": app_id, "v3": v3, "v4": v4, "scene": 3, "content": verify_message, "option": 2}
            add_res = http_post("/contacts/addcontacts", add_payload)
            if add_res and (add_res.get("ret") == 0 or add_res.get("ret") == 200):
                log_message(f"已发送好友请求给 {wxid}")
            else:
                msg = add_res.get("msg") if add_res else "无响应"
                log_message(f"添加好友失败: {wxid} - {msg}")
        root.after(0, lambda: messagebox.showinfo("完成", "批量加好友操作完成，请查看日志获取详情。"))
    threading.thread(target=add_thread, args=(ids_to_add,), daemon=true).start()
 
# 将按钮点击事件绑定到相应线程启动函数
btn_refresh_groups.config(command=start_refresh_groups, state="disabled")
btn_export_members.config(command=start_export_members, state="disabled")
btn_add_friends.config(command=start_add_friends, state="disabled")
 
# 设置gewechat服务的api地址（本地默认端口2531）
base_url = "http://192.168.175.145:2531/v2/api"
group_ids = []  # 全局群id列表（与列表框序号对应）
 
# 尝试使用已保存的凭据自动登录，然后启动登录线程
load_credentials()
threading.thread(target=login_process, daemon=true).start()
root.mainloop()