import threading import pymysql from tools import * class UploadThread(threading.Thread): def __init__(self, uploadqqnumber): super().__init__() self.uploadqqnumber = uploadqqnumber self.conn = pymysql.connect( host='192.168.9.1', # 你的 MySQL 主机 user='root', # 你的 MySQL 用户 password='123456', # 你的 MySQL 密码 database='qqinfo' # 你要连接的数据库 ) def conn_close(self): self.conn.close() def run(self): print(f'我是上传线程{self.uploadqqnumber}') if download_image(self.uploadqqnumber): Hash(self.conn).tomysql(self.uploadqqnumber) print('上传成功') remove_image(self.uploadqqnumber) class FindThread(threading.Thread): def __init__(self, TargetImageHash): super().__init__() self.TargetImageHash = TargetImageHash self.conn = pymysql.connect( host='192.168.9.1', # 你的 MySQL 主机 user='root', # 你的 MySQL 用户 password='123456', # 你的 MySQL 密码 database='qqinfo' # 你要连接的数据库 ) def conn_close(self): self.conn.close() def run(self): print(f'我是查询线程{self.TargetImageHash}') res = Hash(self.conn).getqq(self.TargetImageHash) if res != 'error': print(f'查询成功,QQ号是: {res}') exit() class ByNetFindThread(threading.Thread): def __init__(self, TargetImageHash, findqqnumber): super().__init__() self.TargetImageHash = TargetImageHash self.findqqnumber = findqqnumber def run(self): if download_image(self.findqqnumber): FindImage = Image.open(str(f'./img/{self.findqqnumber}.jpg')) FindImageHash = bytes.fromhex(str(imagehash.average_hash(FindImage))) res = compare(self.TargetImageHash, FindImageHash) if res-99.9 >= 0 : print(f'找到QQ号了:{self.findqqnumber}') os.rename(f'./img/{self.findqqnumber}.jpg',f'./img/congratulations_{self.findqqnumber}.jpg') exit() print(f'[{self.findqqnumber}]的相似度是:{res}') remove_image(self.findqqnumber)