API.get_all_items() - Code Metrics - Inspection of "/fetchlibrary: Progress Dialog" - SIMKL/script.simkl - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — feature/fetchlibrary (0a4b7d)

by David

created 2017-02-12 00:16 UTC

API.get_all_items() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
c	0
b	0
f	0
dl	0
loc	9
rs	9.6666

#!/usr/bin/python
# -*- coding: UTF-8 -*-

import sys, os, time
import json

import xbmc
import interface
import httplib

__addon__ = interface.__addon__
def getstr(strid): return interface.getstr(strid)

REDIRECT_URI = "https://simkl.com/apps/kodi/connected"
USERFILE     = os.path.join(xbmc.translatePath(__addon__.getAddonInfo("profile")).decode("utf-8"), "simkl_key")
xbmc.translatePath("special://profile/simkl_key")

if not os.path.exists(USERFILE):
    os.mkdir(os.path.dirname(USERFILE))
    with open(USERFILE, "w+") as f:
        f.write("")
else:
    with open(USERFILE, "r") as f:
        xbmc.log("Simkl: Userfile " + str(f.read()))

APIFILE = os.path.join(os.path.dirname(os.path.realpath(__file__)).strip("lib"), "data", "apikey")
xbmc.log("Simkl: APIFILE: {0}".format(APIFILE))
with open(APIFILE, "r") as f:
    rd = f.read()
    d = json.loads(rd)
    APIKEY = d["apikey"]
    SECRET = d["secret"]
    xbmc.log("Simkl: {0}".format(rd))
    xbmc.log("Simkl: APIKEY: {0}".format(APIKEY))
ATOKEN = 0 #Get atoken from file
headers = {"Content-Type": "application-json",
    "simkl-api-key": APIKEY}

class API:
    def __init__(self):
        self.scrobbled_dict = {} #So it doesn't scrobble 5 times the same chapter
        #{"Label":expiration_time}
        with open(USERFILE, "r") as f:
            self.token = f.readline().strip("\n")
            headers["authorization"] = "Bearer " + self.token
        try:
            self.con = httplib.HTTPSConnection("api.simkl.com")
            self.con.request("GET", "/users/settings", headers=headers)
            self.USERSETTINGS = json.loads(self.con.getresponse().read().decode("utf-8"))
            xbmc.log("Simkl: Usersettings: " + str(self.USERSETTINGS))
            self.internet = True
            if not os.path.exists(USERFILE):
                api.login()
        except Exception:
            xbmc.log("Simkl: {0}".format("No INTERNET"))
            interface.notify(getstr(32027))
            self.internet = False

    def login(self):
        """ Logins the API class to Simkl """
        url = "/oauth/pin?client_id="
        url += APIKEY + "&redirect=" + REDIRECT_URI

        log = httplib.HTTPSConnection("api.simkl.com")
        log.request("GET", url, headers=headers)
        r = log.getresponse().read().decode("utf-8")
        xbmc.log(r)
        rdic = json.loads(r)
        #interface.loginDialog(rdic["verification_url"],
        #  rdic["user_code"], self.check_login, log, rdic["expires_in"],
        #  rdic["interval"], self)

        pin = rdic["user_code"]
        url = rdic["verification_url"]
        exp = int(rdic["expires_in"])
        ntv = int(rdic["interval"])

        self.logindialog = interface.loginDialog("simkl-LoginDialog.xml",
            __addon__.getAddonInfo("path"), pin=pin, url=url,
            check_login=self.check_login, log=log, exp=exp, inter=ntv, api=self)
        self.logindialog.doModal()
        del self.logindialog

    def set_atoken(self, token):
        global ATOKEN
        with open(USERFILE, "w") as f:
            f.write(token)
        ATOKEN = token
        headers["authorization"] = "Bearer "+token
        self.token = token

    def check_login(self, ucode, log): #Log is the connection
        url = "/oauth/pin/" + ucode + "?client_id=" + APIKEY
        log.request("GET", url, headers=headers)
        r = json.loads(log.getresponse().read().decode("utf-8"))
        xbmc.log("Simkl:" + str(r))
        if r["result"] == "OK":
            self.set_atoken(r["access_token"])
            log.request("GET", "/users/settings", headers=headers)
            r = json.loads(log.getresponse().read().decode("utf-8"))
            self.USERSETTINGS = r
            return True
        elif r["result"] == "KO":
            return False

    def is_user_logged(self):
        """ Checks if user is logged in """
        failed = False
        if "error" in self.USERSETTINGS.keys(): failed = self.USERSETTINGS["error"]
        if self.token == "" or failed == "user_token_failed":
            xbmc.log("Simkl: User not logged in")
            interface.login(0)
            return False
        else:
            #interface.login(self.USERSETTINGS["user"]["name"])
            interface.login(1)
            return True

    ### SCROBBLING OR CHECKIN
    def lock(self, fname, duration):
        xbmc.log("Duration: %s" %duration)
        exp = self.scrobbled_dict
        exp[fname] = int(time.time() + (105 - float(__addon__.getSetting("scr-pct"))) / 100 * duration)
        xbmc.log("Simkl: Locking {0}".format(exp))
        self.scrobbled_dict = {fname:exp[fname]} #So there is always only one entry on the dict

    def is_locked(self, fname):
        exp = self.scrobbled_dict
        if not (fname in exp.keys()): return 0
        xbmc.log("Time: {0}, exp: {1}, Dif: {2}".format(int(time.time()),
            exp[fname], int(exp[fname]-time.time())))
        #When Dif reaches 0, scrobble.
        if time.time() < exp[fname]:
            xbmc.log("Simkl: Can't scrobble, file locked (alredy scrobbled)")
            xbmc.log(str(exp))
            return 1
        else:
            del exp[fname]
            return 0

    def watched(self, filename, mediatype, duration, date=time.strftime('%Y-%m-%d %H:%M:%S'), cnt=0):
        """ Filename can also be an imdb number """
        filename = filename.replace("\\", "/")
        if self.is_user_logged() and not self.is_locked(filename):
            try:
                con = httplib.HTTPSConnection("api.simkl.com")
                mediadict = {"movie": "movies", "episode":"episodes", "show":"show"}

                if filename[:2] == "tt":
                    toappend = {"ids":{"imdb":filename}, "watched_at":date}
                    media = mediadict[mediatype]
                else:
                    xbmc.log("Simkl: Filename - {0}".format(filename))
                    values = {"file":filename}
                    values = json.dumps(values)
                    xbmc.log("Simkl: Query: {0}".format(values))
                    con.request("GET", "/search/file/", body=values, headers=headers)
                    r1 = con.getresponse().read()#.decode("utf-8")
                    xbmc.log("Simkl: Response: {0}".format(r1))
                    r = json.loads(str(r1))
                    self.lastwatched = r
                    if r == []:
                        xbmc.log("Simkl: Couldn't scrobble: Null Response")
                        return 0
                    media = mediadict[r["type"]]
                    toappend = {"ids": r[r["type"]]["ids"], "watched_at":date}

                tosend = {}
                tosend[media] = []
                tosend[media].append(toappend)
                tosend = json.dumps(tosend)

                xbmc.log("Simkl: values {0}".format(tosend))
                con.request("GET", "/sync/history/", body=tosend, headers=headers)
                r = con.getresponse().read().decode("utf-8")
                xbmc.log("Simkl: {0}".format(r))

                success = max(json.loads(r)["added"].values())
                if success:
                    self.scrobbled_dict
                    self.lock(filename, duration)
                return success

            except httplib.BadStatusLine:
                xbmc.log("Simkl: {0}".format("ERROR: httplib.BadStatusLine"))
            except SSLError: #Fix #8
                xbmc.log("Simkl: ERROR: SSLError, retrying?")
                if cnt == 0: interface.notify(getstr(32029).format(cnt+1))
                if cnt <= 3:
                    self.watched(filename, mediatype, duration, date=date, cnt=cnt+1)
                else: interface.notify("SSLError")

        else:
            xbmc.log("Simkl: Can't scrobble. User not logged in or file locked")
            return 0

    @staticmethod
    def get_all_items(mediatype):
        """
            mediatype can be 'shows', 'movies' or 'anime'
            http://docs.simkl.apiary.io/#reference/sync/last-activities/get-all-items-in-the-user's-watchlist 
        """
        con = httplib.HTTPSConnection("api.simkl.com")
        con.request("GET", "/sync/all-items/{0}?extended=full".format(mediatype), headers=headers)
        return con.getresponse().read()

    @staticmethod
    def check_if_watched(item, movie=True):
        """ Checks if an item has been watched """
        con = httplib.HTTPSConnection("api.simkl.com")
        if movie:
            values = json.dumps([{
                "type": item["media"],
                "imdb": item["imdbnumber"]
                }])
            con.request("GET", "/sync/watched", body=values, headers=headers)
            r1 = con.getresponse().read()
            xbmc.log("Simkl: {}".format(r1))
            return json.loads(r1)[0]["result"]
        else:
            values = json.dumps(item)
            con.request("GET", "/sync/watched", body=values, headers=headers)
            return json.loads(con.getresponse().read())

api = API()
if __name__ == "__main__":
    if sys.argv[1] == "login":
        xbmc.log("Logging in", level=xbmc.LOGDEBUG)
        api.login()


1			#!/usr/bin/python
2			# -- coding: UTF-8 --
3
4			import sys, os, time
5			import json
6
7			import xbmc
8			import interface
9			import httplib
10
11			__addon__ = interface.__addon__
12			def getstr(strid): return interface.getstr(strid)
13
14			REDIRECT_URI = "https://simkl.com/apps/kodi/connected"
15			USERFILE = os.path.join(xbmc.translatePath(__addon__.getAddonInfo("profile")).decode("utf-8"), "simkl_key")
16			xbmc.translatePath("special://profile/simkl_key")
17
18			if not os.path.exists(USERFILE):
19			os.mkdir(os.path.dirname(USERFILE))
20			with open(USERFILE, "w+") as f:
21			f.write("")
22			else:
23			with open(USERFILE, "r") as f:
24			xbmc.log("Simkl: Userfile " + str(f.read()))
25
26			APIFILE = os.path.join(os.path.dirname(os.path.realpath(__file__)).strip("lib"), "data", "apikey")
27			xbmc.log("Simkl: APIFILE: {0}".format(APIFILE))
28			with open(APIFILE, "r") as f:
29			rd = f.read()
30			d = json.loads(rd)
31			APIKEY = d["apikey"]
32			SECRET = d["secret"]
33			xbmc.log("Simkl: {0}".format(rd))
34			xbmc.log("Simkl: APIKEY: {0}".format(APIKEY))
35			ATOKEN = 0 #Get atoken from file
36			headers = {"Content-Type": "application-json",
37			"simkl-api-key": APIKEY}
38
39			class API:
40			def __init__(self):
41			self.scrobbled_dict = {} #So it doesn't scrobble 5 times the same chapter
42			#{"Label":expiration_time}
43			with open(USERFILE, "r") as f:
44			self.token = f.readline().strip("\n")
45			headers["authorization"] = "Bearer " + self.token
46			try:
47			self.con = httplib.HTTPSConnection("api.simkl.com")
48			self.con.request("GET", "/users/settings", headers=headers)
49			self.USERSETTINGS = json.loads(self.con.getresponse().read().decode("utf-8"))
50			xbmc.log("Simkl: Usersettings: " + str(self.USERSETTINGS))
51			self.internet = True
52			if not os.path.exists(USERFILE):
53			api.login()
54			except Exception:
55			xbmc.log("Simkl: {0}".format("No INTERNET"))
56			interface.notify(getstr(32027))
57			self.internet = False
58
59			def login(self):
60			""" Logins the API class to Simkl """
61			url = "/oauth/pin?client_id="
62			url += APIKEY + "&redirect=" + REDIRECT_URI
63
64			log = httplib.HTTPSConnection("api.simkl.com")
65			log.request("GET", url, headers=headers)
66			r = log.getresponse().read().decode("utf-8")
67			xbmc.log(r)
68			rdic = json.loads(r)
69			#interface.loginDialog(rdic["verification_url"],
70			# rdic["user_code"], self.check_login, log, rdic["expires_in"],
71			# rdic["interval"], self)
72
73			pin = rdic["user_code"]
74			url = rdic["verification_url"]
75			exp = int(rdic["expires_in"])
76			ntv = int(rdic["interval"])
77
78			self.logindialog = interface.loginDialog("simkl-LoginDialog.xml",
79			__addon__.getAddonInfo("path"), pin=pin, url=url,
80			check_login=self.check_login, log=log, exp=exp, inter=ntv, api=self)
81			self.logindialog.doModal()
82			del self.logindialog
83
84			def set_atoken(self, token):
85			global ATOKEN
86			with open(USERFILE, "w") as f:
87			f.write(token)
88			ATOKEN = token
89			headers["authorization"] = "Bearer "+token
90			self.token = token
91
92			def check_login(self, ucode, log): #Log is the connection
93			url = "/oauth/pin/" + ucode + "?client_id=" + APIKEY
94			log.request("GET", url, headers=headers)
95			r = json.loads(log.getresponse().read().decode("utf-8"))
96			xbmc.log("Simkl:" + str(r))
97			if r["result"] == "OK":
98			self.set_atoken(r["access_token"])
99			log.request("GET", "/users/settings", headers=headers)
100			r = json.loads(log.getresponse().read().decode("utf-8"))
101			self.USERSETTINGS = r
102			return True
103			elif r["result"] == "KO":
104			return False
105
106			def is_user_logged(self):
107			""" Checks if user is logged in """
108			failed = False
109			if "error" in self.USERSETTINGS.keys(): failed = self.USERSETTINGS["error"]
110			if self.token == "" or failed == "user_token_failed":
111			xbmc.log("Simkl: User not logged in")
112			interface.login(0)
113			return False
114			else:
115			#interface.login(self.USERSETTINGS["user"]["name"])
116			interface.login(1)
117			return True
118
119			### SCROBBLING OR CHECKIN
120			def lock(self, fname, duration):
121			xbmc.log("Duration: %s" %duration)
122			exp = self.scrobbled_dict
123			exp[fname] = int(time.time() + (105 - float(__addon__.getSetting("scr-pct"))) / 100 * duration)
124			xbmc.log("Simkl: Locking {0}".format(exp))
125			self.scrobbled_dict = {fname:exp[fname]} #So there is always only one entry on the dict
126
127			def is_locked(self, fname):
128			exp = self.scrobbled_dict
129			if not (fname in exp.keys()): return 0
130			xbmc.log("Time: {0}, exp: {1}, Dif: {2}".format(int(time.time()),
131			exp[fname], int(exp[fname]-time.time())))
132			#When Dif reaches 0, scrobble.
133			if time.time() < exp[fname]:
134			xbmc.log("Simkl: Can't scrobble, file locked (alredy scrobbled)")
135			xbmc.log(str(exp))
136			return 1
137			else:
138			del exp[fname]
139			return 0
140
141			def watched(self, filename, mediatype, duration, date=time.strftime('%Y-%m-%d %H:%M:%S'), cnt=0):
142			""" Filename can also be an imdb number """
143			filename = filename.replace("\\", "/")
144			if self.is_user_logged() and not self.is_locked(filename):
145			try:
146			con = httplib.HTTPSConnection("api.simkl.com")
147			mediadict = {"movie": "movies", "episode":"episodes", "show":"show"}
148
149			if filename[:2] == "tt":
150			toappend = {"ids":{"imdb":filename}, "watched_at":date}
151			media = mediadict[mediatype]
152			else:
153			xbmc.log("Simkl: Filename - {0}".format(filename))
154			values = {"file":filename}
155			values = json.dumps(values)
156			xbmc.log("Simkl: Query: {0}".format(values))
157			con.request("GET", "/search/file/", body=values, headers=headers)
158			r1 = con.getresponse().read()#.decode("utf-8")
159			xbmc.log("Simkl: Response: {0}".format(r1))
160			r = json.loads(str(r1))
161			self.lastwatched = r
162			if r == []:
163			xbmc.log("Simkl: Couldn't scrobble: Null Response")
164			return 0
165			media = mediadict[r["type"]]
166			toappend = {"ids": r[r["type"]]["ids"], "watched_at":date}
167
168			tosend = {}
169			tosend[media] = []
170			tosend[media].append(toappend)
171			tosend = json.dumps(tosend)
172
173			xbmc.log("Simkl: values {0}".format(tosend))
174			con.request("GET", "/sync/history/", body=tosend, headers=headers)
175			r = con.getresponse().read().decode("utf-8")
176			xbmc.log("Simkl: {0}".format(r))
177
178			success = max(json.loads(r)["added"].values())
179			if success:
180			self.scrobbled_dict
181			self.lock(filename, duration)
182			return success
183
184			except httplib.BadStatusLine:
185			xbmc.log("Simkl: {0}".format("ERROR: httplib.BadStatusLine"))
186			except SSLError: #Fix #8
187			xbmc.log("Simkl: ERROR: SSLError, retrying?")
188			if cnt == 0: interface.notify(getstr(32029).format(cnt+1))
189			if cnt <= 3:
190			self.watched(filename, mediatype, duration, date=date, cnt=cnt+1)
191			else: interface.notify("SSLError")
192
193			else:
194			xbmc.log("Simkl: Can't scrobble. User not logged in or file locked")
195			return 0
196
197			@staticmethod
198			def get_all_items(mediatype):
199			"""
200			mediatype can be 'shows', 'movies' or 'anime'
201			http://docs.simkl.apiary.io/#reference/sync/last-activities/get-all-items-in-the-user's-watchlist
202			"""
203			con = httplib.HTTPSConnection("api.simkl.com")
204			con.request("GET", "/sync/all-items/{0}?extended=full".format(mediatype), headers=headers)
205			return con.getresponse().read()
206
207			@staticmethod
208			def check_if_watched(item, movie=True):
209			""" Checks if an item has been watched """
210			con = httplib.HTTPSConnection("api.simkl.com")
211			if movie:
212			values = json.dumps([{
213			"type": item["media"],
214			"imdb": item["imdbnumber"]
215			}])
216			con.request("GET", "/sync/watched", body=values, headers=headers)
217			r1 = con.getresponse().read()
218			xbmc.log("Simkl: {}".format(r1))
219			return json.loads(r1)[0]["result"]
220			else:
221			values = json.dumps(item)
222			con.request("GET", "/sync/watched", body=values, headers=headers)
223			return json.loads(con.getresponse().read())
224
225			api = API()
226			if __name__ == "__main__":
227			if sys.argv[1] == "login":
228			xbmc.log("Logging in", level=xbmc.LOGDEBUG)
229			api.login()
230

SIMKL / script.simkl

GitHub Access Token became invalid

Branch — feature/fetchlibrary (0a4b7d)

API.get_all_items() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like