不知不觉已经入职3个月了,同事很好,工作充实,学到了很多东西,大大小小的需求也实现了接近20个。负责2个主要component,数据抓取和利用GenAI做数据提取。
1 背景
提取新闻中事件关键信息,比如人名,时间,事件等,并与可信数据进行比较,根据比较结果将新闻事件进行相应处理,减少人工审查成本。
2 主要组件
- 数据抓取:一键抓取内部可信数据,输入到系统前端界面,免去用户人工填写的成本和差错。
- 前端:将可信数据和新闻事件提交到后台,等待分析结果。
- 后端:权限控制,调用GenAI模块进行提取,将提取结果返回前端,并做数据持久化。
- GenAI:提取新闻中的关键信息,并返回给后端。
- Dashboard:监控整个系统健康状态,如响应时间,消息队列长度,用户反馈,安全检测等。
3 数据抓取
3.1 win32应用程序数据抓取
3.1.1 Windows程序窗口定位
有两种方式定位到Windows程序窗口,1.通过窗口标题,2.通过进程名字找pid,进而定位窗口。第一种适合窗口标题固定,第二种适配多语言系统。
- 通过窗口标题定位窗口
import pygetwindow,win32gui
all_windows = pygetwindow.getAllWindows() # 获取所有窗口
app_title_list = [app.title for app in all_windows if app!=''] # 获取所有非空窗口标题
for item in app_title_list:
if "窗口标题前缀" in item:
find_window = pygetwindow.getWindowsWithTitle(item) # 如果窗口标题是固定的,直接调用这句,这里使用app_title_list再遍历是因为要获取的窗口标题仅有前缀固定
find_window.activate() # 激活窗口
find_window.maximize() # 窗口最大化
win32gui.SetForegroundWindow(find_window._hWnd) # 窗口最前
- 根据进程名字找到pid,进而定位窗口
import psutil,win32process,win32gui
def get_pid_by_name(process_name): # 通过任务管理器中的进程名字定位,一般是固定的,并且是英语不会随系统语言变化
for proc in psutil.process_iter(['pid','name']) # 拉出系统所有进程
if proc.info['name']==process_name: # 名字匹配
return proc.info['pid'] # 返回名字匹配进程的pid
return None
def get_window_title_by_pid(pid): # 根据pid获取窗口名
def callback(hwnd, hwnd_list):
_, process_id = win32process.GetWindowThreadProcessId(hwnd) # 根据hwnd获取窗口的进程PID
if process_id = pid:
window_title = win32gui.GetWindowText(hwnd) # 通过窗口的标题名获取窗口句柄
hwnd_list.append(window_title)
return True # 继续枚举下一个窗口
windows_titles = []
win32gui.EnumWindows(callback, window_titles)# 遍历系统中所有的顶层窗口,并针对每个窗口调用一次 callback 函数。遍历过程中,符合条件(进程 ID 与目标 PID 相等)的窗口标题会被添加到 windows_titles 列表中。
return window_titles
pid = get_pid_by_name("xxx.exe")
window_titles = get_window_title_by_pid(pid)
for item in window_titles:
if "窗口标题前缀" in item:
进行处理咯
参考python win32gui 获取句柄
win32api 鼠标定位及点击操作:
import win32con,win32api,pygetwindow
find_window = pygetwindow.getWindowsWithTitle(item) #
x,y = int(find_window.left),int(find_window.right)
win32api.SetCursorPos([x,y]) # 鼠标移动到到程序窗口左下角
win32api.mouse_event(win32con.MOUSEEVENTF_RIGHTUP | win32con.MOUSEEVENTF_RIGHTDOWN,0,0,0,0) # 右键点击
win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP | win32con.MOUSEEVENTF_LEFTDOWN,0,0,0,0) # 左键点击
pyautogui键盘操作, pyperclip剪切板
from pyautogui import press, hotkey
import pyyperclip
pyperclip.copy("") # 剪贴板置为空字符串
prees("tab")
hotkey("ctrl","a")
hotkey("ctrl","c")
text = pyperclip.waitForPaste() # 一直等待,获取剪贴板内容
3.2 利用selenium将数据传入前端
两种方法:
- 前端设置1个隐藏元素demo,用js脚本把json传过去
driver.execute_script("return document.getElementById('demo').value='"+data+"';")
- 通过Id定位元素,然后sendkeys发送值
element = driver.find_element_by_id('demo1') # 定位到id为demo1的元素 element.sendkeys(data) # 把data传给demo1元素
3.3 pyinstall打包成exe
如果涉及到一些配置文件打包,比如excel文件,可以用–add-data指定
pyinstall --noconfirm --onefile --nowindow --noconsole --add-data "demo.xlsx:." --name="app." main.py
这是打包成一个exe,运行时可以读取demo.xlsx内容
4 GenAI
4.1 prompt编写
langchain的PromptTemplate使用
4.2 asycio异步并发
4.3 日志记录
默认的logging
5 Dashboard
5.1 定时任务apscheduler
防止并发情况下多次执行导致结果错误,可以用max_instance参数控制
5.2 redis统计最大队列长度
双key,每次put队列,用k1记录最大长度定时清零,清零前赋值给k2,后端来拿k2