Python實(shí)現(xiàn)號碼歸屬地查詢功能
使用場景
對手機(jī)號碼進(jìn)行地域分析,需要查詢歸屬地;
問題描述
針對數(shù)據(jù)集比較大的情況,通過腳本來處理,使用多線程的方法來加快查詢速度
pool = multiprocessing.Pool(processes=pool_count) for i in data_cut(data,pool_count): data_log_list.append(pool.apply_async(main, (i,))) pool.close() pool.join()
解決方案
創(chuàng)建一個 pool 進(jìn)程池,然后通過data_cut將數(shù)據(jù)讀取并且等分成數(shù)據(jù)組,設(shè)置好pool_count進(jìn)程數(shù)量就可以開始,每個數(shù)據(jù)組獨(dú)立查詢?nèi)缓髮⒔Y(jié)果匯總給push_log進(jìn)行最終處理,保存為csv文件。
# 電話號碼歸屬地查詢 import os import sys import time import json import warnings import pandas as pd import multiprocessing from phone import Phone warnings.filterwarnings("ignore") path = os.path.abspath(".") def data_cut(data_list,data_cut=4): #將任務(wù)拆分,建議拆分?jǐn)?shù)為CPU核心數(shù),默認(rèn)為4 #分組數(shù)據(jù),分組間隔 data_all=[] if data_cut > len(data_list): data_cut = len(data_list) data_cut_num = int((len(data_list)+1)/data_cut) for i in range(1,data_cut+1): if i < data_cut: data_1=data_list[data_cut_num*(i-1):data_cut_num*i] else: data_1=data_list[data_cut_num*(i-1):] data_all.append(data_1) return data_all def push_log(data_log_list,file_name): data_all = [] data_list = [i.get() for i in data_log_list] for i in data_list: for j in i: data_all.append(j) data_all = pd.DataFrame(data_all) data_all.to_csv(path + "/phone_{}.csv".format(file_name),index=False,encoding='gbk') print('成功查詢:',data_all.shape[0]) def main(data): resp = [] for i in data: try: if type(Phone().find(i)) == dict: resp.append(Phone().find(i)) except: pass return resp if __name__ == '__main__': start_time= time.time() file_name = name = sys.argv[1] data = pd.read_table(path + "/{}".format(file_name),header=None) data=list(data[0]) pool_count = 12 data_log_list = [] pool = multiprocessing.Pool(processes=pool_count) for i in data_cut(data,pool_count): data_log_list.append(pool.apply_async(main, (i,))) pool.close() pool.join() push_log(data_log_list,file_name) print(time.time()-start_time)
方法補(bǔ)充
除了上文的方法,小編還為大家整理了一些其他Python號碼歸屬地查詢的方法,需要的可以參考一下
方法一:
import requests def get_callerloc(phone): url = f"https://www.qvdv.com/tools/qvdv-api-mobile.html?f=json&mobile={phone}" headers = { 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/100.0.4896.75 Safari/537.36', } response = requests.request("GET", url, headers=headers).json() return response["message"] if __name__ == '__main__': phone = input("請輸入手機(jī)號(查詢歸屬地):") res = get_callerloc(phone) print("手機(jī)號碼:"+res["mobile"]) print("歸屬地:"+res["province"]) print("運(yùn)營商:"+res["supplier"])
截圖
方法二
GUI
from tkinter import * from tkinter import ttk from tkinter import messagebox from Get_Attr import Get_Infos import re import threading class App: def __init__(self): self.root=Tk() self.root.title('手機(jī)號碼歸屬地查詢-v1.0') self.root.resizable(0,0) width=410 height=390 left=(self.root.winfo_screenwidth()-width)/2 top=(self.root.winfo_screenheight()-height)/2 self.root.geometry('%dx%d+%d+%d'%(width,height,left,top)) self.create_widet() self.set_widget() self.place_widget() self.root.mainloop() def create_widet(self): self.l1=ttk.Label(self.root) self.e1=ttk.Entry(self.root) self.b1=ttk.Button(self.root) self.lf=ttk.LabelFrame(self.root) self.l2=ttk.Label(self.lf) self.e2=ttk.Entry(self.lf) self.l3=ttk.Label(self.lf) self.e3=ttk.Entry(self.lf) self.l4=ttk.Label(self.lf) self.e4=ttk.Entry(self.lf) self.l5=ttk.Label(self.lf) self.e5=ttk.Entry(self.lf) self.l6=ttk.Label(self.lf) self.e6=ttk.Entry(self.lf) self.l7=ttk.Label(self.lf) self.e7=ttk.Entry(self.lf) self.b1.config(command=lambda:self.thread_it(self.search_infos)) def set_widget(self): self.e2_var=StringVar() self.e3_var=StringVar() self.e4_var=StringVar() self.e5_var=StringVar() self.e6_var=StringVar() self.e7_var=StringVar() self.l1.config(text='請輸入手機(jī)號:') self.b1.config(text='查詢') self.lf.config(text='查詢結(jié)果') self.l2.config(text='手機(jī)號碼:') self.l3.config(text='所屬省份:') self.l4.config(text='所屬城市:') self.l5.config(text='區(qū) 號:') self.l6.config(text='郵 編:') self.l7.config(text='類 型:') #將字符串變量綁定Entry組件 self.e2.config(textvariable=self.e2_var) self.e3.config(textvariable=self.e3_var) self.e4.config(textvariable=self.e4_var) self.e5.config(textvariable=self.e5_var) self.e6.config(textvariable=self.e6_var) self.e7.config(textvariable=self.e7_var) self.root.bind('<Escape>',self.escape) self.root.bind('<Return>',self.do_search) def place_widget(self): self.l1.place(x=30,y=20) self.e1.place(x=130,y=20) self.b1.place(x=290,y=20) self.lf.place(x=30,y=60,width=350,height=300) self.l2.place(x=60,y=10) self.e2.place(x=150,y=10) self.l3.place(x=60,y=50) self.e3.place(x=150,y=50) self.l4.place(x=60,y=90) self.e4.place(x=150,y=90) self.l5.place(x=60,y=130) self.e5.place(x=150,y=130) self.l6.place(x=60,y=170) self.e6.place(x=150,y=170) self.l7.place(x=60,y=210) self.e7.place(x=150,y=210) def search_infos(self): pn=self.e1.get() #判斷輸入類型,必須為11位數(shù)字 if re.match('\d{11}',pn): result=Get_Infos().get_infos(pn) self.e2_var.set(pn) self.e3_var.set(result['province']) self.e4_var.set(result['city']) self.e5_var.set(result['areacode']) self.e6_var.set(result['zip']) self.e7_var.set(result['company']) else: messagebox.showwarning('警告','輸入有誤,請檢查!') #使用線程防止UI界面卡死 def thread_it(self,func,*args): t=threading.Thread(target=func,args=args) t.setDaemon(True) t.start() def escape(self,event): self.root.destroy() def do_search(self,event): self.thread_it(self.search_infos()) if __name__ == '__main__': a=App()
Get_Attr
import json import requests from urllib.parse import urlencode class Get_Infos(): def __init__(self): self.url='http://apis.juhe.cn/mobile/get?' self.headers={ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/81.0.4044.138 Safari/537.36' } def get_infos(self,phone_num): params={ 'phone':phone_num, 'key':'7a2b367a62fa24108b1f27ed4c84c97a', 'dtype':'' } r=requests.get(self.url+urlencode(params),headers=self.headers) _json=json.loads(r.text) if _json.get('resultcode')=='200': result=_json.get('result') item={} item['province']=result.get('province') item['city']=result.get('city') item['areacode']=result.get('areacode') item['zip']=result.get('zip') item['company']=result.get('company') return item else: return False
到此這篇關(guān)于Python實(shí)現(xiàn)號碼歸屬地查詢功能的文章就介紹到這了,更多相關(guān)Python號碼歸屬地查詢內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Python練習(xí)之操作SQLite數(shù)據(jù)庫
這篇文章主要介紹了Python練習(xí)之操作SQLite數(shù)據(jù)庫,主要通過三個問題如何創(chuàng)建SQLite數(shù)據(jù)庫?如何向SQLite表中插入數(shù)據(jù)?如何查詢SQLite表中的數(shù)據(jù)?展開文章主題詳情,需要的朋友可以參考一下2022-06-06Python自動化辦公實(shí)現(xiàn)數(shù)據(jù)自動填充需求
這篇文章主要為大家介紹了Python自動化辦公實(shí)現(xiàn)數(shù)據(jù)自動填充需求,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-06-06