python實戰（開發新浪微博應用）
#coding=utf-8
"""
本應用主要功能
1.用戶選擇喜歡的標籤加關注
2.獲取用戶粉絲中本身尚未關注的,->加關注,提升粉絲穩定性
3.獲取用戶關注列表中沒有回粉的,並能夠一鍵取消關注
2,3兩個功能基本實現,有一缺點,數據量一大,很慢很慢
1功能不太好,主要是經過一個線程去搜索數據,把感興趣的用戶放入數據庫,當用戶選擇加關注標籤時,從數據庫中取數據,
之前用sqlite3比較少,一用發現仍是有好多值得研究的地方,主要是線程安全....,慢慢研究....
"""
import os
import sys
import key
import web
import threading
import sqlite3
import time
from weibopy.auth import OAuthHandler
from weibopy.api import API
from jinja2 import Environment,FileSystemLoader
"""
url配置
"""
urls = (
'/', 'Login',
'/index','Index',
'/callback','CallBack',
'/logout','LogOut',
'/(.*)/', 'Redirect',
'/tag', 'Tag',
'/noattention','Noattention',
'/fensiattention','Fensiattention',
'/minusattention','Minusattention',
'/delattention','Delattention'
)
app=web.application(urls,globals())
if web.config.get('_session') is None:
session = web.session.Session(app,web.session.DiskStore("sina"))
web.config._session = session
else:
session = web.config._session
"""
用jinja2模板渲染文件
"""
def render_template(template_name,**context):
extensions=context.pop('extensions',[])
globals=context.pop("globals",{})
jinja_env=Environment(
loader=FileSystemLoader(os.path.join(os.path.dirname(__file__),'templates')),
extensions=extensions)
jinja_env.globals.update(globals)
return jinja_env.get_template(template_name).render(context)
"""
定義404請求頁面
"""
def notfound():
info="親,您所請求的頁面不存在,系統在3秒後自動返回..."
return web.notfound(render_template('error.html',info=info.decode('utf-8')))
db=web.database(dbn='sqlite',db='tag.s3db')
"""
要求tag
"""
tagdict={'水瓶座':1,'雙魚座':2,'白羊座':3,'金牛座':4,'雙子座':5,'巨蟹座':6,'獅子座':7,'處女座':8,'天秤座':9,'天蠍座':10,'射手座':11,
'摩羯座':12,'吃':13,'聽歌':14,'淘寶':15,'網購':16,'數碼':17,'攝影':18,'睡覺':19,'旅遊':20,'體育':21,'動漫':22,'遊戲':23,
'股票':24,'交友':25,'宅女':26,'宅男':27,'時尚':28,'浪漫':29,'美女':30,'創業':31,'IT':32,'媒體':33,'營銷':34}
systag=['水瓶座','雙魚座','白羊座','金牛座','雙子座','巨蟹座','獅子座','處女座','天秤座','天蠍座','射手座','摩羯座','吃','聽歌','淘寶',
'網購','數碼','攝影','睡覺','旅遊','體育','動漫','遊戲','股票','交友','宅女','宅男','時尚','浪漫','美女','創業','IT','媒體','營銷']
conn=sqlite3.connect('tag.s3db',check_same_thread=False) #容許其它線程使用這個鏈接
conn.text_factory = str
cursor=conn.cursor()
class Setag(threading.Thread):
"""
這個線程主要是用來搜索用戶tag,若是知足tag要求就寫入數據庫中,
"""
def authorization(self):
"""
開發者認證
"""
auth = OAuthHandler(key.CONSUME_KEY, key.CONSUME_SECRET)
auth.setToken(key.TOKEN,key.TOKEN_SECRET)
self.api = API(auth)
self.cursor=cursor
def adduser(self,uid):
"""
遍歷uid用戶的tag,知足條件加入數據庫
"""
try:
fan=self.api.followers_ids(uid)
fanuid=fan.ids
for id in fanuid:
tags=self.api.tags(id)
tt=[]
for t in tags:
tagid=t.__getattribute__('id')
value=t.__getattribute__(tagid)
tt.append(value.encode('utf-8'))
"""
獲取用戶tag與要求標籤的交集
"""
common=set(tt).intersection(set(systag))
if len(common)==0:
continue
else:
for t in common:
"""
獲取tag對應的tagid
"""
tagindex=tagdict[t]
try:
self.cursor.execute("insert into taginfo(uid,tagid) values(%d,%d)" %(int(id),int(tagindex)))
conn.commit()
except:
continue
except:
time.sleep(120)
pass
finally:
time.sleep(60)
"""
將uid用戶的第一個粉絲uid傳給adduser
"""
return self.adduser(fanuid[0])
def run(self):
self.authorization()
me=self.api.verify_credentials()
"""
將我本身的uid給adduser
"""
self.adduser(me.id)
"""
定義404請求頁面
"""
app.notfound= notfound
#首頁
#首先從session中獲取access_token，沒有就轉向新浪微博頁面認證
#認證成功後將access_token保存在session中
"""
首頁是登錄頁面,經過新浪微博受權
"""
class Login:
def GET(self):
return render_template('login.html')
"""
新浪微博受權原理:
首先首頁判斷有無用戶session信息,若是有則跳轉到相應地址,
沒有則引導用戶跳轉到受權uri,受權後自動跳轉到永遠自定義的回調地址,
回調地址保存用戶session信息,跳轉到首頁,這時已有用戶session信息,幹壞事吧....
"""
class Index:
def GET(self):
access_token=session.get('access_token',None)
if not access_token:
"""
key.py中放置了開發者的信息
"""
auth = OAuthHandler(key.CONSUME_KEY, key.CONSUME_SECRET,web.ctx.get('homedomain')+'/callback')
#獲取受權url
auth_url = auth.get_authorization_url()
session.request_token=auth.request_token
web.seeother(auth_url)
else:
auth = OAuthHandler(key.CONSUME_KEY, key.CONSUME_SECRET)
auth.access_token=access_token
api=API(auth)
user=api.verify_credentials()
return render_template('index.html',user=user)
"""
頁面回調，新浪微博驗證成功後會返回本頁面
"""
class CallBack:
def GET(self):
try:
ins=web.input()
oauth_verifier=ins.get('oauth_verifier',None)
request_token=session.get('request_token',None)
auth=OAuthHandler(key.CONSUME_KEY, key.CONSUME_SECRET)
auth.request_token=request_token
access_token=auth.get_access_token(oauth_verifier)
session.access_token=access_token
web.seeother("/index")
except Exception:
info="親,系統繁忙,請稍後再試......,系統在3秒後自動返回..."
return render_template('error.html',info=info.decode('utf-8'))
"""
重定向用戶輸入uri後的/
"""
class Redirect:
def GET(self,path):
web.seeother('/'+path)
class Tag:
"""
獲取用戶選擇加關注的tag
"""
def GET(self):
data=web.input()
try:
select=data.star
except:
try:
select=data.hobby
except:
try:
select=data.personality
except:
select=data.job
try:
auth = OAuthHandler(key.CONSUME_KEY, key.CONSUME_SECRET)
auth.access_token=session['access_token']
api=API(auth)
seuid=[]
nu=0
"""
這裏寫的很很差.....
"""
while True:
re=cursor.execute('select uid from taginfo where tagid=%d limit 20' %select).fetchall()
for r in re:
seuid.append(r[0])
for s in seuid:
try:
api.create_friendship(user_id=s)
nu+=1
except:
continue
if nu>=50:
break
info="恭喜您已成功關注%d位用戶....." %nu
return render_template('success.html',info=info.decode('utf-8'))
except:
info="親,系統繁忙,請稍後再試......,系統在3秒後自動返回..."
return render_template('error.html',info=info.decode('utf-8'))
class Noattention:
"""
獲取個人粉絲中我沒有加關注的,把數據傳給noattention.html頁面顯示...
"""
def GET(self):
try:
auth=OAuthHandler(key.CONSUME_KEY,key.CONSUME_SECRET)
auth.access_token=session['access_token']
api=API(auth)
user=api.verify_credentials()
fan=[]
next_cursor=-1
while next_cursor!=0:
timeline=api.followers(user.id,'','','',next_cursor)
if isinstance(timeline,tuple):
next_cursor=timeline[1]
for line in timeline[0]:
fid=line.__getattribute__("id")
fname=line.__getattribute__("screen_name")
fan.append((fid,fname))
else:
next_cursor=0
for line in timeline:
fid=line.__getattribute__("id")
fname=line.__getattribute__("screen_name")
fan.append((fid,fname))
friend=[]
next_cursor=-1
while next_cursor!=0:
timeline=api.friends(user.id,'','','',next_cursor)
if isinstance(timeline,tuple):
next_cursor=timeline[1]
for line in timeline[0]:
frid=line.__getattribute__("id")
frname=line.__getattribute__("screen_name")
friend.append((frid,frname))
else:
next_cursor=0
for line in timeline:
frid=line.__getattribute__("id")
frname=line.__getattribute__("screen_name")
friend.append((frid,frname))
#獲取個人粉絲中還不是個人關注對象
fanNotAttention=list(set(fan).difference(set(friend)))
nu=len(fanNotAttention)
if nu==0:
return render_template('noattentionok.html',nu=nu)
else:
return render_template('noattention.html',nu=nu,fanNotAttention=fanNotAttention)
except:
info="親,系統繁忙,請稍後再試......,系統在3秒後自動返回..."
return render_template('error.html',info=info.decode('utf-8'))
class Fensiattention:
"""
對未加關注的粉絲加關注
"""
def GET(self):
#獲取noattentionok.html傳過來的數據
data=web.input()
on=[]
try:
auth=OAuthHandler(key.CONSUME_KEY,key.CONSUME_SECRET)
auth.access_token=session['access_token']
api=API(auth)
"""
獲取noattention.html頁面傳過來的uid,經過checkbox,因爲有一個全選按鈕,若是點擊,則去掉
"""
for x in data:
on.append(x)
try:
on.remove('checkbox2')
except:
pass
nu=len(on)
if nu==0:
pass
if nu>60:
on=on[:60]
nu=60
"""
一次最多加60次關注
"""
map(api.create_friendship,on)
info="恭喜您已成功關注%d位用戶....." %nu
return render_template('success.html',info=info.decode('utf-8'))
except:
info="親,系統繁忙,請稍後再試......,系統在3秒後自動返回..."
return render_template('error.html',info=info.decode('utf-8'))
class Minusattention:
"""
獲取個人關注中不是我粉絲的,即不回粉的傢伙,把數據傳給attentionnotfan.html顯示...
"""
def GET(self):
try:
auth=OAuthHandler(key.CONSUME_KEY,key.CONSUME_SECRET)
auth.access_token=session['access_token']
api=API(auth)
user=api.verify_credentials()
fan=[]
next_cursor=-1
while next_cursor!=0:
timeline=api.followers(user.id,'','','',next_cursor)
if isinstance(timeline,tuple):
next_cursor=timeline[1]
for line in timeline[0]:
fid=line.__getattribute__("id")
fname=line.__getattribute__("screen_name")
fan.append((fid,fname))
else:
next_cursor=0
for line in timeline:
fid=line.__getattribute__("id")
fname=line.__getattribute__("screen_name")
fan.append((fid,fname))
friend=[]
next_cursor=-1
while next_cursor!=0:
timeline=api.friends(user.id,'','','',next_cursor)
if isinstance(timeline,tuple):
next_cursor=timeline[1]
for line in timeline[0]:
frid=line.__getattribute__("id")
frname=line.__getattribute__("screen_name")
friend.append((frid,frname))
else:
next_cursor=0
for line in timeline:
frid=line.__getattribute__("id")
frname=line.__getattribute__("screen_name")
friend.append((frid,frname))
attentionNotFan=list(set(friend).difference(set(fan)))
nu=len(attentionNotFan)
if nu==0:
return render_template('attentionnotfanok.html',nu=nu)
else:
return render_template('attentionnotfan.html',nu=nu,attentionNotFan=attentionNotFan)
except:
info="親,系統繁忙,請稍後再試......,系統在3秒後自動返回..."
return render_template('error.html',info=info.decode('utf-8'))
class Delattention:
"""
獲取attentionnotfan.html頁面選擇的用戶,並一鍵取消關注
"""
def GET(self):
#獲取attentionnotfan.html傳過來的數據
data=web.input()
on=[]
try:
auth=OAuthHandler(key.CONSUME_KEY,key.CONSUME_SECRET)
auth.access_token=session['access_token']
api=API(auth)
for x in data:
on.append(x)
try:
#同理,因爲有全選按鈕.....
on.remove('checkbox2')
except:
pass
nu=len(on)
if nu==0:
pass
#取消關注
map(api.destroy_friendship,on)
info="恭喜您已成功取消關注%d位用戶....." %nu
return render_template('success.html',info=info.decode('utf-8'))
except:
info="親,系統繁忙,請稍後再試......,系統在3秒後自動返回..."
return render_template('error.html',info=info.decode('utf-8'))
if __name__=='__main__':
"""
啓動app,啓動s線程去搜索數據
"""
s=Setag()
s.start()
app.run()
相關文章
- 1. python調用新浪微博API實踐
- 2. Python3網絡爬蟲開發實戰，Scrapy 爬取新浪微博
- 3. python 爬取新浪微博
- 4. python項目實戰:獲取實時新浪微博熱搜
- 5. 【CF 應用開發大賽】新浪微博rss訂閱生成
- 6. 如何開發優秀的新浪微博應用
- 7. 新浪微博開發（三）Oauth認證
- 8. IOS開發之新浪微博OAuth2
- 9. Oauth認證-新浪微博開發(上)
- 10. 新浪微博-開發過程
- 更多相關文章...
- • Maven Web 應用 - Maven教程
- • XML 應用程序 - XML 教程
- • Spring Cloud 微服務實戰(三) - 服務註冊與發現
- • PHP開發工具
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章