实现自动输入密码

.gitignore

@@ -0,0 +1,4 @@
+__pycache__/
+scr_path/
+*.png
+log.txt

ReadMe.txt

@@ -0,0 +1,5 @@
+
+
+
+2025.3.1
+  自动识别 多屏协同 窗口 抓取截图

cv2_test.py

@@ -0,0 +1,42 @@
+import cv2
+import numpy as np
+
+
+
+
+
+#定义形状检测函数
+def ShapeDetection(img:cv2.typing.MatLike):
+    contours,hierarchy = cv2.findContours(img,cv2.RETR_EXTERNAL,cv2.CHAIN_APPROX_NONE)  #寻找轮廓点
+    all_cont=len(contours)
+    # print(f"all_cont:{all_cont}")
+    boll_count=0
+    for obj in contours:
+        perimeter = cv2.arcLength(obj,True)  #计算轮廓周长
+        approx = cv2.approxPolyDP(obj,0.02*perimeter,True)  #获取轮廓角点坐标
+        # CornerNum = len(approx)   #轮廓角点的数量
+        # area = cv2.contourArea(obj)  #计算轮廓内区域的面积
+        # print(f"area:{area}, perimeter:{perimeter}, CornerNum:{CornerNum}")
+        x, y, w, h = cv2.boundingRect(approx)  #获取坐标值和宽度、高度
+        # print(f"{x},{y},{x+w},{y+h}")
+        img_obj=img[y:y+h,x:x+w]
+        # print(img_obj)
+        nonzero_num=np.count_nonzero(img_obj)
+        percentage=nonzero_num/img_obj.size
+        # print(f"nonzero_num:{nonzero_num},size={img_obj.size}, pos:{percentage}")
+        # 通过计算图像白值的比例来判断实心还是空心
+        if(percentage>0.55):
+            boll_count+=1
+    return (all_cont,boll_count)
+
+def calc_boll_count(path:str):
+    img = cv2.imread(path)
+    imgGray = cv2.cvtColor(img,cv2.COLOR_RGB2GRAY)  # 转为灰度图
+    ret, binary = cv2.threshold(imgGray,60,255,cv2.THRESH_BINARY)  # 转为二值图
+    return ShapeDetection(binary)  #形状检测
+
+if __name__ == "__main__":
+    ret=calc_boll_count("1.png")
+    print(ret)
+    ret=calc_boll_count("1644-440 172x27.png")
+    print(ret)

index.json

@@ -0,0 +1,3 @@
+{
+  "index": 1681
+}

log.py

@@ -0,0 +1,41 @@
+import time
+import sys
+
+
+# 同一个进程中所有调用这个文件的  .py 文件都使用这个变量
+_log_fp=None
+
+def _time():
+    return '['+time.strftime("%Y-%m-%d %H:%M:%S")+']'
+
+def myprint_dec(func):
+    def wrapper(*args, **kwargs):
+        # 在这里添加额外的功能
+        print(*args, **kwargs)
+        if(_log_fp is not None):
+            _log_fp.write(_time())
+            kwargs["file"]=_log_fp
+            result = func(*args, **kwargs)
+            _log_fp.flush()
+        else:
+            result=None
+        return result
+    return wrapper
+
+myprint=myprint_dec(print)
+
+def mywrite(data:str):
+    if(_log_fp is not None):
+        txt=data.replace('\r','\n')
+        txt=txt.replace('\n\n','\n')
+        txt=txt.replace('\n\n','\n')
+        _log_fp.write(txt)
+        _log_fp.flush()
+    sys.stdout.write(data)
+    sys.stdout.flush()
+
+def log_init(file_name:str):
+    global _log_fp
+    if _log_fp is None:
+        _log_fp=open(file_name,mode="a+",encoding="utf-8")
+

moter.py

@@ -0,0 +1,53 @@
+#pip install pyserial
+
+#查看可用的端口
+
+# coding:utf-8
+
+import time
+
+import serial.tools.list_ports
+
+plist = list(serial.tools.list_ports.comports())
+
+print("端口数>>>",len(plist))
+
+if len(plist) <= 0:
+
+    print("没有发现端口!")
+
+else:
+    plist_0 = list(plist[0])    #在第一个端口的情况
+
+    #plist_0 = list(plist[1])   #在第二个端口的情况
+      
+    serialName = "com8"
+
+    serialFd = serial.Serial(serialName, 93450, timeout=60)
+
+    print("端口名>>>", serialFd.name)
+
+    while True:
+        try:
+            time.sleep(2)                        #等待机器准备好，重要！！！！！
+            n = serialFd.write("X100Y130\r".encode()) #移动到50*50毫米位置
+            time.sleep(1)
+            n = serialFd.write("X50Y0\r".encode()) #移动到50*50毫米位置
+            time.sleep(1)
+            n = serialFd.write("X50Y51.5\r".encode()) #移动到50*50毫米位置
+            time.sleep(1)
+            for i in range(10):
+                n = serialFd.write("Z6\r".encode())  #按下
+                time.sleep(0.05)
+                n = serialFd.write("Z0\r".encode())  #抬起
+                time.sleep(1)
+            # time.sleep(1)    
+            n = serialFd.write("X0Y0\r".encode()) #回原点
+            time.sleep(1)
+        except KeyboardInterrupt as k:
+            time.sleep(1)
+            n = serialFd.write("Z0\r".encode())  #抬起
+            time.sleep(0.1)
+            n = serialFd.write("X0Y0\r".encode()) #回原点
+            break
+    serialFd.close() 

ocr.py

@@ -0,0 +1,47 @@
+import os
+import sys
+import time
+import easyocr
+import pyautogui
+
+from cv2_test import calc_boll_count
+
+
+reader = easyocr.Reader(['ch_sim','en'], gpu = True) # need to run only once to load model into memory
+tmp_path=os.getenv("TEMP")
+
+if not os.path.exists("scr_path"):
+  os.mkdir("scr_path")
+
+def get_text(left:int,top:int,wid:int,hit:int)->list[str]:
+  img = pyautogui.screenshot(region=[left, top, wid, hit])
+  img.save(f"{tmp_path}\\screenshot.png")
+  result = reader.readtext(f"{tmp_path}\\screenshot.png")
+  ret=[]
+  for item in result:
+    ret.append(item[1])
+  return ret
+
+def get_scr(left:int,top:int,wid:int,hit:int,text:str):
+  img = pyautogui.screenshot(region=[left, top, wid, hit])
+  img.save(f"scr_path\\{time.strftime('%Y%m%d_%H%M%S')}_{text}.png")
+
+def get_boll_count(left:int,top:int,wid:int,hit:int):
+  path=f"{tmp_path}\\boll_scr.png"
+  img = pyautogui.screenshot(region=[left, top, wid, hit])
+  img.save(path)
+  return calc_boll_count(path)
+
+
+def mouse_touch(left:int,top:int,wid:int,hit:int):
+  x,y=pyautogui.position()
+  if(x>left and x<left+wid and y>top and y<top+hit):
+    print("点击手机屏幕")
+    pyautogui.click()
+
+if __name__ == "__main__":
+  print("start")
+  # get_text(1560,198,363,331)
+  result = reader.readtext(f"test.png")
+  print(result)
+

password.py

@@ -0,0 +1,10 @@
+# -*- coding: utf-8 -*-
+#生成全部的六位数字密码\
+
+print("start")
+f = open('passdict6.txt','w+')
+for id in range(1000000):
+    password = str(id).zfill(6)+'\n'
+    f.write(password)
+f.close()
+print("end")

touch.py

@@ -0,0 +1,120 @@
+import os
+import sys
+import time
+import json
+import serial
+
+from ocr import get_text
+from log import log_init, myprint
+
+
+class ser(object):
+  gpio_table={
+     "enter":"pa0",
+     "1":"pa4",
+     "2":"pa5",
+     "3":"pa6",
+     "4":"pb0",
+     "5":"pb1",
+     "6":"pb10",
+     "7":"pa12",
+     "8":"pa11",
+     "9":"pb7",
+     "0":"pb6"
+  }
+  def __init__(self):
+      self.ser=None
+  def open(self,com:str):
+      try:
+          self.ser = serial.Serial(port=com, baudrate=115200, timeout=0.3)
+      except Exception:
+          print(f"serial com:{com} open failed.")
+          sys.exit(-1)
+  def close(self):
+    if not (self.ser is None):
+      self.ser.close()
+      self.ser=None
+  def write(self,t:str):
+    # print("send:",t.encode('utf-8').hex(' '))
+    if not (self.ser is None):
+      self.ser.write(t.encode(encoding="utf-8"))
+  def read(self):
+    data=bytearray()
+    if not (self.ser is None):
+      data=self.ser.read(4096)
+    try:
+      t=data.decode(encoding='utf-8')
+    except Exception as e:
+      print(e)
+      print(data.hex(' '))
+      t=""
+    return t
+  def touch(self,key:str):
+    if not (key in self.gpio_table):
+      return
+    gpio=self.gpio_table[key]
+    print(f"touch {key}:{gpio}")
+    self.write(f"output on {gpio}\r\n")
+    ret=self.read()
+    print(ret)
+    time.sleep(0.05)
+    self.write(f"output off {gpio}\r\n")
+    ret=self.read()
+    print(ret)
+    time.sleep(0.05)
+  def enter(self):
+    s.touch("enter")
+  def passsword(self,p:str):
+    for i in p:
+      self.touch(i)
+
+
+def find_text(t:str):
+  text=get_text(1560,198,363,331)
+  print(text)
+  for item in text:
+    if(item[1]==t):
+      return True
+  return False
+
+
+def data_save(data:dict):
+  data_file="index.json"
+  with open(data_file,mode='w+') as f:
+    f.write(json.dumps(data,sort_keys=True, indent=2, separators=(',', ': ')))
+
+
+def data_get()->dict:
+  data_file="index.json"
+  if os.path.exists(data_file):
+    with open(data_file,mode='r') as f:
+      return json.loads(f.read())
+  return {"index":0}
+
+if __name__ == "__main__":
+  s=ser()
+  s.open("com6")
+  password=data_get().get("index",0)
+  log_init("log.txt")
+  myprint(f"start_index:{password}")
+  while(True):
+    try:
+      if(find_text("手机存储")):
+        myprint("enter")
+        s.enter()
+      if(find_text("请输入解锁密码")):
+        myprint(f"try:{password:06d}")
+        s.passsword(f"{password:06d}")
+    except Exception as e:
+      myprint(e)
+      data_save({"index":password})
+      break
+    except KeyboardInterrupt as k:
+      myprint(k)
+      data_save({"index":password})
+      break
+    if password>999999:
+      myprint("try end")
+      break
+  myprint(f"end_index:{password}")
+  s.close()

touch2.py

@@ -0,0 +1,239 @@
+import os
+import sys
+import time
+import json
+import serial
+
+from ocr import get_text, mouse_touch, get_scr, get_boll_count
+from log import log_init, myprint
+from win32 import get_wind_pos
+
+# 识别文字的截图区域
+_ocr_pos=[1545,157,367,384]
+
+# 识别小圆点的截图区域
+_ocr_boll=[1644,440+3,172,27]
+
+# 自动点击屏幕的区域
+_mouse_touch=[1550,760,140,70]
+
+# 获取最新的窗口位置
+def flush_wind_pos():
+  global _ocr_pos,_ocr_boll,_mouse_touch
+  not_find=False
+  while(True):
+    pos=get_wind_pos()
+    if(pos[0]<=0 and pos[1]<=0 and pos[2]<=0 and pos[3]<=0):
+      if not_find==False:
+        myprint("未找到相机窗口")
+        not_find=True
+      time.sleep(1)
+      continue
+    _ocr_pos=[pos[0]+0,pos[1]+180,375,296]
+    _ocr_boll=[pos[0]+99,pos[1]+440,172,27]
+    _mouse_touch=[pos[0]+0,pos[1]+603,145,70]
+    return
+
+# 识别屏幕文字
+def identify_text():
+  flush_wind_pos()
+  mouse_touch(_mouse_touch[0],_mouse_touch[1],_mouse_touch[2],_mouse_touch[3])
+  text=get_text(_ocr_pos[0],_ocr_pos[1],_ocr_pos[2],_ocr_pos[3])
+  print(text)
+  return text
+def find_text(text:str,key:str):
+  for item in text:
+    if(item.find(key)>=0):
+      return True
+  return False
+
+# 识别已输入数量
+def find_input_count():
+  flush_wind_pos()
+  ret=get_boll_count(_ocr_boll[0],_ocr_boll[1],_ocr_boll[2],_ocr_boll[3])
+  if(ret[0]==6):
+    return ret[1]
+  myprint("未识别到已输入个数")
+  return 0
+
+def data_save(data:dict):
+  data_file="index.json"
+  with open(data_file,mode='w+') as f:
+    f.write(json.dumps(data,sort_keys=True, indent=2, separators=(',', ': ')))
+
+
+
+class ser(object):
+  # 1 x_step 18 y_step 15
+  step_x=16
+  step_y=14
+  touch_table={
+    "1":(53,70),
+    "2":(53+step_x,70),
+    "3":(53+step_x*2,70),
+    "4":(53,70+step_y),
+    "5":(53+step_x,70+step_y),
+    "6":(53+step_x*2,70+step_y),
+    "7":(53,70+step_y*2),
+    "8":(53+step_x,70+step_y*2),
+    "9":(53+step_x*2,70+step_y*2),
+    "0":(53+step_x,70+step_y*3),
+  }
+  def __init__(self):
+    self.ser=None
+    self.x=0
+    self.y=0
+  def open(self,com:str):
+    try:
+      self.ser = serial.Serial(port=com, baudrate=93450, timeout=60)
+    except Exception:
+      print(f"serial com:{com} open failed.")
+      sys.exit(-1)
+    time.sleep(2)
+  def close(self):
+    if not (self.ser is None):
+      self.ser.close()
+      self.ser=None
+  def write(self,t:str):
+    # print("send:",t)
+    if not (self.ser is None):
+      self.ser.write(t.encode(encoding="utf-8"))
+  def togo(self,x:int,y:int):
+    if(x<0 or x>100 or y<0 or y>150):
+      print(f"位置超出范围 x={x}, y={y}")
+      return
+    self.write(f"X{x}Y{y}\r")
+    # 电机运行速度160mm/s
+    disx=abs(self.x-x)
+    disy=abs(self.y-y)
+    time.sleep((disx+disy)/160+0.1)
+    self.x=x
+    self.y=y
+  def touch(self):
+    self.write(f"Z6\r")
+    time.sleep(0.1)
+    self.write(f"Z0\r")
+    time.sleep(0.05)
+  def reset(self):
+    self.write(f"Z0\r")
+    time.sleep(0.05)
+    self.togo(0,0)
+  def enter(self):
+    self.togo(70+20,27)
+    self.touch()
+  def back(self):
+    self.togo(70,132)
+    self.touch()
+  # 避让摄像头 也是在取消按钮的位置
+  def avoid(self):
+    self.togo(70,132)
+  # 按键位置测试
+  def password_test(self,p:str):
+    for i in p:
+      pos=self.touch_table[i]
+      myprint(f"点击 {i}:{pos}")
+      self.togo(pos[0],pos[1])
+      time.sleep(2)
+  def password(self,p:str):
+    index=0
+    p_len=len(p)
+    while index<p_len:
+      pos=self.touch_table[p[index]]
+      myprint(f"点击 index={index},{p[index]}")
+      self.togo(pos[0],pos[1])
+      self.touch()
+      if p[index] not in "890":
+        # 890 不需要避让
+        self.avoid()
+      if(find_text(identify_text(),"手机已锁定")):
+        if(index==5):
+          myprint(f"手机锁定中，已全部输入")
+          return True
+        myprint(f"密码输入状态异常 index={index},password={p}")
+        return False
+      num=find_input_count()
+      myprint(f"已输入个数为 {num}")
+      if(num==index+1):
+        index+=1
+      elif(num==0 and index==5):
+        return True
+      else:
+        myprint(f"点击未识别，重试 index={index}")
+
+
+
+def data_get()->dict:
+  data_file="index.json"
+  if os.path.exists(data_file):
+    with open(data_file,mode='r') as f:
+      return json.loads(f.read())
+  return {"index":0}
+
+
+if __name__ == "__main1__":
+  s=ser()
+  s.open("com8")
+  password=data_get().get("index",0)
+  log_init("log.txt")
+  myprint(f"起始密码:{password}")
+  s.avoid()
+  unknown_scr=False
+  while(True):
+    try:
+      text=identify_text()
+      if(find_text(text,"安装系统软件")):
+        unknown_scr=False
+        myprint("进入密码界面")
+        s.enter()
+        # s.avoid()
+      elif(find_text(text,"请输入解锁密码")):
+        unknown_scr=False
+        pw=f"{password:06d}"
+        myprint(f"开始尝试:{pw}")
+        if(s.password(pw)):
+          myprint(f"密码已输入")
+          password+=1
+        s.avoid()
+        flush_wind_pos()
+        get_scr(_ocr_pos[0],_ocr_pos[1],_ocr_pos[2],_ocr_pos[3],f"{pw}")
+      elif(find_text(text,"手机已锁定")):
+        time.sleep(1)
+      else:
+        if(unknown_scr==False):
+          myprint("进入未知界面")
+          myprint(f"界面信息为；{text}")
+          unknown_scr=True
+        time.sleep(1)
+    except Exception as e:
+      myprint(e)
+      data_save({"index":password})
+      print("终止运行")
+      break
+    except KeyboardInterrupt as k:
+      myprint(k)
+      data_save({"index":password})
+      print("终止运行")
+      break
+  myprint(f"终止密码:{password}")
+  s.reset()
+  s.close()
+
+
+if __name__ == "__main__":
+  # identify_text()
+  # get_scr(_ocr_pos[0],_ocr_pos[1],_ocr_pos[2],_ocr_pos[3],f"test")
+  # find_input_count()
+  s=ser()
+  s.open("com8")
+  s.avoid()
+  while True:
+    try:
+      s.password_test("1234567890")
+    except Exception as e:
+      myprint(e)
+      break
+    except KeyboardInterrupt as k:
+      myprint(k)
+      break
+  s.reset()
+  s.close()

win32.py

@@ -0,0 +1,22 @@
+import os
+import sys
+import time
+import win32gui
+
+
+
+def get_wind_pos():
+  hwnd = win32gui.FindWindow(None, "多屏协同")
+  if hwnd == 0:
+    return (0,0,0,0)
+  else:
+    # print("找到相机窗口")
+    # print("hwnd:",hwnd)
+    # print("title:",win32gui.GetWindowText(hwnd))
+    # print("class:",win32gui.GetClassName(hwnd))
+    # print("pos:",win32gui.GetWindowRect(hwnd))
+    return win32gui.GetWindowRect(hwnd)
+  
+
+if __name__=="__main__":
+  print(get_wind_pos())