|
import tkinter as tk
from tkinter import ttk, messagebox
import os
import winreg
from datetime import datetime
from win32com.shell import shell, shellcon
import win32api
import win32con
import win32evtlog
import win32evtlogutil
import win32file
import struct
from ctypes import *
import sys
import ctypes
def is_admin():
"""检查是否具有管理员权限"""
try:
return ctypes.windll.shell32.IsUserAnAdmin()
except:
return False
class PCActivityMonitor:
def __init__(self, root):
self.root = root
self.root.title("用户操作记录查看器")
self.root.geometry("1200x800")
# 检查管理员权限
if not is_admin():
messagebox.showwarning("警告",
"程序未以管理员权限运行，某些功能可能受限。\n"
"建议以管理员身份重新运行程序以获取完整功能。")
# 创建主框架
self.main_frame = ttk.Frame(root)
self.main_frame.pack(expand=True, fill="both", padx=5, pady=5)
# 创建工具栏
self.create_toolbar()
# 创建Treeview
self.create_treeview()
# 初始加载数据
self.load_all_logs()
def create_toolbar(self):
"""创建工具栏"""
toolbar = ttk.Frame(self.main_frame)
toolbar.pack(fill="x", padx=5, pady=5)
# 刷新按钮
ttk.Button(toolbar, text="刷新", command=self.load_all_logs).pack(side="left", padx=5)
# 删除按钮
ttk.Button(toolbar, text="删除选中记录", command=self.delete_selected).pack(side="left", padx=5)
# 导出按钮
ttk.Button(toolbar, text="导出记录", command=self.export_logs).pack(side="left", padx=5)
# 搜索框
ttk.Label(toolbar, text="搜索:").pack(side="left", padx=5)
self.search_var = tk.StringVar()
ttk.Entry(toolbar, textvariable=self.search_var, width=30).pack(side="left", padx=5)
ttk.Button(toolbar, text="搜索", command=self.search_logs).pack(side="left", padx=5)
ttk.Button(toolbar, text="清除搜索", command=self.clear_search).pack(side="left", padx=5)
def create_treeview(self):
"""创建Treeview控件"""
frame = ttk.Frame(self.main_frame)
frame.pack(expand=True, fill="both")
# 创建Treeview和滚动条
self.tree = ttk.Treeview(frame, columns=("time", "action", "object", "path"), show="headings")
# 设置列标题
self.tree.heading("time", text="操作时间", command=lambda: self.treeview_sort_column("time", False))
self.tree.heading("action", text="操作类型", command=lambda: self.treeview_sort_column("action", False))
self.tree.heading("object", text="操作对象", command=lambda: self.treeview_sort_column("object", False))
self.tree.heading("path", text="完整路径", command=lambda: self.treeview_sort_column("path", False))
# 设置列宽
self.tree.column("time", width=150)
self.tree.column("action", width=100)
self.tree.column("object", width=250)
self.tree.column("path", width=400)
# 添加滚动条
y_scrollbar = ttk.Scrollbar(frame, orient="vertical", command=self.tree.yview)
x_scrollbar = ttk.Scrollbar(frame, orient="horizontal", command=self.tree.xview)
self.tree.configure(yscrollcommand=y_scrollbar.set, xscrollcommand=x_scrollbar.set)
# 布局
self.tree.pack(side="left", fill="both", expand=True)
y_scrollbar.pack(side="right", fill="y")
x_scrollbar.pack(side="bottom", fill="x")
# 绑定右键菜单
self.tree.bind("<Button-3>", self.show_context_menu)
# 绑定双击事件
self.tree.bind("<Double-1>", self.on_double_click)
def load_all_logs(self):
"""加载所有日志记录"""
self.tree.delete(*self.tree.get_children())
# 创建临时列表存储所有记录
all_records = []
# 获取文件访问记录
all_records.extend(self.get_file_access_logs())
# 获取程序执行记录
all_records.extend(self.get_program_execution_logs())
# 获取应用程序日志
all_records.extend(self.get_application_logs())
# 按时间倒序排序
all_records.sort(key=lambda x: x[0], reverse=True)
# 插入排序后的记录
for record in all_records:
self.tree.insert("", "end", values=record)
def get_file_access_logs(self):
"""获取文件访问记录"""
records = []
try:
# 读取最近访问文件
recent_path = os.path.join(os.environ['USERPROFILE'],
'AppData\\Roaming\\Microsoft\\Windows\\Recent')
for file in os.listdir(recent_path):
if file.endswith('.lnk'):
file_path = os.path.join(recent_path, file)
access_time = datetime.fromtimestamp(os.path.getatime(file_path))
try:
shortcut = shell.SHCreateItemFromParsingName(
file_path, None, shell.IID_IShellItem
)
target_path = shortcut.GetDisplayName(shellcon.SIGDN_FILESYSPATH)
records.append((
access_time.strftime("%Y-%m-%d %H:%M:%S"),
"文件访问",
file[:-4],
target_path
))
except:
continue
# 读取Office文档记录
office_path = os.path.join(os.environ['APPDATA'],
'Microsoft\\Office\\Recent')
if os.path.exists(office_path):
for file in os.listdir(office_path):
if file.endswith('.lnk'):
file_path = os.path.join(office_path, file)
access_time = datetime.fromtimestamp(os.path.getatime(file_path))
records.append((
access_time.strftime("%Y-%m-%d %H:%M:%S"),
"Office文档",
file[:-4],
file_path
))
except Exception as e:
messagebox.showerror("错误", f"读取文件访问记录出错: {str(e)}")
return records
def get_program_execution_logs(self):
"""获取程序执行记录"""
records = []
try:
key_paths = [
(winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall"),
(winreg.HKEY_CURRENT_USER, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall")
]
for hkey, key_path in key_paths:
try:
with winreg.OpenKey(hkey, key_path, 0,
winreg.KEY_READ | winreg.KEY_WOW64_64KEY) as key:
for i in range(winreg.QueryInfoKey(key)[0]):
try:
subkey_name = winreg.EnumKey(key, i)
with winreg.OpenKey(key, subkey_name) as subkey:
try:
display_name = winreg.QueryValueEx(subkey, "DisplayName")[0]
install_date = winreg.QueryValueEx(subkey, "InstallDate")[0]
install_location = winreg.QueryValueEx(subkey, "InstallLocation")[0]
date_obj = datetime.strptime(install_date, "%Y%m%d")
records.append((
date_obj.strftime("%Y-%m-%d %H:%M:%S"),
"软件安装",
display_name,
install_location
))
except:
continue
except:
continue
except:
continue
except Exception as e:
messagebox.showerror("错误", f"读取程序执行记录出错: {str(e)}")
return records
def get_application_logs(self):
"""获取应用程序日志"""
records = []
try:
hand = win32evtlog.OpenEventLog(None, "Application")
flags = win32evtlog.EVENTLOG_BACKWARDS_READ | win32evtlog.EVENTLOG_SEQUENTIAL_READ
events = win32evtlog.ReadEventLog(hand, flags, 0)
for event in events:
if event.EventType in [win32evtlog.EVENTLOG_AUDIT_SUCCESS]:
records.append((
event.TimeGenerated.Format(),
"程序执行",
event.SourceName,
"用户操作记录"
))
except:
pass
return records
def load_file_system_journal(self):
"""读取文件系统日志"""
if not is_admin():
print("需要管理员权限才能读取文件系统日志")
return
try:
drives = self.get_system_drives()
for drive in drives:
try:
# 检查驱动器类型
drive_type = win32file.GetDriveType(drive + "\\")
if drive_type != win32file.DRIVE_FIXED:
continue # 跳过非固定磁盘
# 尝试启用USN日志
self.enable_usn_journal(drive)
# 读取USN日志
self.read_usn_journal(drive)
except Exception as e:
print(f"处理驱动器 {drive} 时出错: {str(e)}")
continue
except Exception as e:
print(f"读取文件系统日志出错: {str(e)}")
def get_system_drives(self):
"""获取系统所有驱动器"""
drives = []
bitmask = win32api.GetLogicalDrives()
for letter in 'ABCDEFGHIJKLMNOPQRSTUVWXYZ':
if bitmask & 1:
drives.append(f'{letter}:')
bitmask >>= 1
return drives
def read_usn_journal(self, drive):
"""读取USN日志"""
try:
# 打开驱动器
handle = win32file.CreateFile(
f"\\\\.\\{drive}",
win32con.GENERIC_READ | win32con.GENERIC_WRITE,
win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE,
None,
win32con.OPEN_EXISTING,
0,
None
)
# 获取USN日志信息
if handle == win32file.INVALID_HANDLE_VALUE:
return
# 定义USN日志数据结构
class USN_JOURNAL_DATA(Structure):
_fields_ = [
("UsnJournalID", c_ulonglong),
("FirstUsn", c_ulonglong),
("NextUsn", c_ulonglong),
("LowestValidUsn", c_ulonglong),
("MaxUsn", c_ulonglong),
("MaximumSize", c_ulonglong),
("AllocationDelta", c_ulonglong),
]
# 获取USN日志信息
buf = win32file.DeviceIoControl(
handle,
win32file.FSCTL_QUERY_USN_JOURNAL,
None,
sizeof(USN_JOURNAL_DATA),
None
)
journal_data = USN_JOURNAL_DATA()
memmove(byref(journal_data), buf, sizeof(journal_data))
# 读取最近的USN记录
read_data = struct.pack("QQI", journal_data.FirstUsn, journal_data.NextUsn, 0)
data = win32file.DeviceIoControl(
handle,
win32file.FSCTL_READ_USN_JOURNAL,
read_data,
8192,
None
)
# 解析USN记录
usn_records = self.parse_usn_records(data)
# 添加到界面
for record in usn_records:
if record['reason'] & win32file.USN_REASON_FILE_DELETE:
self.tree.insert("", 0, values=(
record['time'].strftime("%Y-%m-%d %H:%M:%S"),
"文件删除",
record['filename'],
f"{drive}\\{record['filepath']}"
))
win32file.CloseHandle(handle)
except Exception as e:
print(f"读取USN日志出错 {drive}: {str(e)}")
def parse_usn_records(self, data):
"""解析USN记录"""
records = []
offset = 0
while offset < len(data):
# 解析记录头
record_length = struct.unpack_from("I", data, offset)[0]
if record_length == 0:
break
# 解析文件名
filename_offset = struct.unpack_from("H", data, offset + 56)[0]
filename_length = struct.unpack_from("H", data, offset + 58)[0]
filename = data[offset + filename_offset:offset + filename_offset + filename_length].decode('utf-16')
# 获取时间戳
timestamp = struct.unpack_from("Q", data, offset + 48)[0]
time = datetime.fromtimestamp((timestamp - 116444736000000000) // 10000000)
# 获取原因
reason = struct.unpack_from("I", data, offset + 40)[0]
# 获取文件引用号
file_ref = struct.unpack_from("Q", data, offset + 8)[0]
records.append({
'filename': filename,
'filepath': self.get_file_path(file_ref),
'time': time,
'reason': reason
})
offset += record_length
return records
def get_file_path(self, file_ref):
"""获取文件路径"""
try:
return "Unknown Path" # 实际实现需要更复杂的逻辑
except:
return "Unknown Path"
def treeview_sort_column(self, col, reverse):
"""排序表格列"""
l = [(self.tree.set(k, col), k) for k in self.tree.get_children('')]
l.sort(reverse=reverse)
# 重新排列项目
for index, (val, k) in enumerate(l):
self.tree.move(k, '', index)
# 切换排序方向
self.tree.heading(col, command=lambda: self.treeview_sort_column(col, not reverse))
def on_double_click(self, event):
"""双击打开文件或目录"""
item = self.tree.selection()[0]
path = self.tree.item(item)['values'][3]
try:
os.startfile(path)
except:
pass
def export_logs(self):
"""导出日志记录"""
try:
with open('user_activities.csv', 'w', encoding='utf-8') as f:
# 写入表头
f.write("操作时间,操作类型,操作对象,完整路径\n")
# 写入数据
for item in self.tree.get_children():
values = self.tree.item(item)['values']
f.write(f"{values[0]},{values[1]},{values[2]},{values[3]}\n")
messagebox.showinfo("成功", "日志已导出到 user_activities.csv")
except Exception as e:
messagebox.showerror("错误", f"导出日志失败: {str(e)}")
def clear_search(self):
"""清除搜索结果"""
self.search_var.set("")
self.tree.selection_remove(*self.tree.selection())
def delete_selected(self):
"""删除选中的记录"""
selected = self.tree.selection()
if not selected:
messagebox.showwarning("警告", "请先选择要删除的记录")
return
if messagebox.askyesno("确认", "确定要删除选中的记录吗？"):
for item in selected:
self.tree.delete(item)
def search_logs(self):
"""搜索日志"""
search_text = self.search_var.get().lower()
if not search_text:
return
for item in self.tree.get_children():
values = self.tree.item(item)['values']
if any(search_text in str(value).lower() for value in values):
self.tree.selection_add(item)
else:
self.tree.selection_remove(item)
def show_context_menu(self, event):
"""显示右键菜单"""
menu = tk.Menu(self.root, tearoff=0)
menu.add_command(label="复制", command=self.copy_selected)
menu.add_command(label="删除", command=self.delete_selected)
menu.post(event.x_root, event.y_root)
def copy_selected(self):
"""复制选中的记录到剪贴板"""
selected = self.tree.selection()
if not selected:
return
text = ""
for item in selected:
values = self.tree.item(item)['values']
text += "\t".join(str(v) for v in values) + "\n"
self.root.clipboard_clear()
self.root.clipboard_append(text)
def enable_usn_journal(self, drive):
"""启用USN日志"""
try:
handle = win32file.CreateFile(
f"\\\\.\\{drive}",
win32con.GENERIC_READ | win32con.GENERIC_WRITE,
win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE,
None,
win32con.OPEN_EXISTING,
0,
None
)
if handle == win32file.INVALID_HANDLE_VALUE:
return False
try:
# 启用USN日志
win32file.DeviceIoControl(
handle,
win32file.FSCTL_CREATE_USN_JOURNAL,
struct.pack("QQ", 0, 0),
None,
None
)
return True
finally:
win32file.CloseHandle(handle)
except:
return False
def main():
# 如果不是管理员权限，则请求提升权限
if not is_admin():
ctypes.windll.shell32.ShellExecuteW(
None, "runas", sys.executable, " ".join(sys.argv), None, 1)
sys.exit()
root = tk.Tk()
app = PCActivityMonitor(root)
root.mainloop()
if __name__ == "__main__":
main()
|
