forked from pyload/pyload
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathCoreApi.py
More file actions
146 lines (111 loc) · 4.76 KB
/
CoreApi.py
File metadata and controls
146 lines (111 loc) · 4.76 KB
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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from pyload.Api import Api, RequirePerm, Permission, ServerStatus, Interaction
from pyload.utils.fs import join, free_space, exists
from pyload.utils import compare_time
from ApiComponent import ApiComponent
class CoreApi(ApiComponent):
""" This module provides methods for general interaction with the core, like status or progress retrieval """
@RequirePerm(Permission.All)
def getServerVersion(self):
"""pyLoad Core version """
return self.core.version
def isWSSecure(self):
# needs to use TLS when either requested or webUI is also using encryption
if not self.core.config['ssl']['activated'] or self.core.config['webUI']['https']:
return False
if not exists(self.core.config['ssl']['cert']) or not exists(self.core.config['ssl']['key']):
self.core.log.warning(_('SSL key or certificate not found'))
return False
return True
@RequirePerm(Permission.All)
def getWSAddress(self):
"""Gets and address for the websocket based on configuration"""
if self.isWSSecure():
ws = "wss"
else:
ws = "ws"
return "%s://%%s:%d" % (ws, self.core.config['webUI']['wsPort'])
@RequirePerm(Permission.All)
def getServerStatus(self):
"""Some general information about the current status of pyLoad.
:return: `ServerStatus`
"""
queue = self.core.files.getQueueStats(self.primaryUID)
total = self.core.files.getDownloadStats(self.primaryUID)
serverStatus = ServerStatus(0,
total[0], queue[0],
total[1], queue[1],
self.isInteractionWaiting(Interaction.All),
not self.core.threadManager.pause and self.isTimeDownload(),
self.core.threadManager.pause,
self.core.config['reconnect']['activated'] and self.isTimeReconnect())
for pyfile in self.core.threadManager.getActiveDownloads(self.primaryUID):
serverStatus.speed += pyfile.getSpeed() #bytes/s
return serverStatus
@RequirePerm(Permission.All)
def getProgressInfo(self):
""" Status of all currently running tasks
:rtype: list of :class:`ProgressInfo`
"""
return self.core.threadManager.getProgressList(self.primaryUID)
def pauseServer(self):
"""Pause server: It won't start any new downloads, but nothing gets aborted."""
self.core.threadManager.pause = True
def unpauseServer(self):
"""Unpause server: New Downloads will be started."""
self.core.threadManager.pause = False
def togglePause(self):
"""Toggle pause state.
:return: new pause state
"""
self.core.threadManager.pause ^= True
return self.core.threadManager.pause
def toggleReconnect(self):
"""Toggle reconnect activation.
:return: new reconnect state
"""
self.core.config["reconnect"]["activated"] ^= True
return self.core.config["reconnect"]["activated"]
def freeSpace(self):
"""Available free space at download directory in bytes"""
return free_space(self.core.config["general"]["download_folder"])
def quit(self):
"""Clean way to quit pyLoad"""
self.core.do_kill = True
def restart(self):
"""Restart pyload core"""
self.core.do_restart = True
def getLog(self, offset=0):
"""Returns most recent log entries.
:param offset: line offset
:return: List of log entries
"""
filename = join(self.core.config['log']['log_folder'], 'log.txt')
try:
fh = open(filename, "r")
lines = fh.readlines()
fh.close()
if offset >= len(lines):
return []
return lines[offset:]
except:
return ['No log available']
@RequirePerm(Permission.All)
def isTimeDownload(self):
"""Checks if pyload will start new downloads according to time in config.
:return: bool
"""
start = self.core.config['downloadTime']['start'].split(":")
end = self.core.config['downloadTime']['end'].split(":")
return compare_time(start, end)
@RequirePerm(Permission.All)
def isTimeReconnect(self):
"""Checks if pyload will try to make a reconnect
:return: bool
"""
start = self.core.config['reconnect']['startTime'].split(":")
end = self.core.config['reconnect']['endTime'].split(":")
return compare_time(start, end) and self.core.config["reconnect"]["activated"]
if Api.extend(CoreApi):
del CoreApi