import pymysql, threading from Tools import config, download_image, Hash, remove_image, compare, Image, imagehash, os, logging, clean_image Myconn = pymysql.connect( host=str(config['mysql']['host']), user=str(config['mysql']['user']), password=str(config['mysql']['password']), database=str(config['mysql']['database']) ) exit_flag = False class UploadThread(threading.Thread): def __init__(self, uploadqqnumber): super().__init__() self.uploadqqnumber = uploadqqnumber self.conn = Myconn def run(self): logging.debug(f'我是上传线程{self.uploadqqnumber}') if download_image(self.uploadqqnumber): Hash(self.conn).tomysql(self.uploadqqnumber) logging.info(f'上传成功: {self.uploadqqnumber}') remove_image(self.uploadqqnumber) self.conn.close() class FindThread(threading.Thread): def __init__(self, TargetImageHash): super().__init__() self.TargetImageHash = TargetImageHash self.conn = Myconn def run(self): global exit_flag logging.debug(f'我是查询线程{self.TargetImageHash}') res = Hash(self.conn).getqq(self.TargetImageHash) if res != 'error': logging.info(f'查询成功,QQ号是: {res}') exit_flag = True self.conn.close() class ByNetFindThread(threading.Thread): def __init__(self, TargetImageHash, findqqnumber): super().__init__() self.TargetImageHash = TargetImageHash self.findqqnumber = findqqnumber def run(self): global exit_flag if download_image(self.findqqnumber): FindImage = Image.open(str(f'./img/{self.findqqnumber}.jpg')) if FindImage.mode == 'P' and 'transparency' in FindImage.info: FindImage = FindImage.convert('RGBA') FindImageHash = bytes.fromhex(str(imagehash.average_hash(FindImage))) res = compare(self.TargetImageHash, FindImageHash) if res-99.9 >= 0 : logging.info(f'找到QQ号了:{self.findqqnumber}') if os.path.exists(f'./img/congratulations_{self.findqqnumber}.jpg'): os.remove(f'./img/congratulations_{self.findqqnumber}.jpg') os.rename(f'./img/{self.findqqnumber}.jpg',f'./img/congratulations_{self.findqqnumber}.jpg') exit_flag = True clean_image() os._exit(0) logging.info(f'[{self.findqqnumber}]的相似度是:{res}%') remove_image(self.findqqnumber)