android_scripts/devhost.py at kitkat · AndroidOpenSourceXperia/android_scripts · GitHub

History

371 lines (339 loc) · 15.4 KB

Raw

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

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

#!/usr/bin/env python2

# Requirements: python2

# python2-requests

# Dev-Host API documentation

# http://d-h.st/api

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

# GNU General Public License for more details.

# You should have received a copy of the GNU General Public License

# along with this program. If not, see <http://www.gnu.org/licenses/>.

from __future__ import division

import xml.etree.ElementTree as ET

from getpass import getpass

import os

import binascii

import argparse

import time

import json

import threading

import signal

import sys

try:

from requests import get, post

import requests.exceptions

except ImportError:

print "The requests module is required to use this script."

exit(1)

def arg_parser():

"""Parse command line arguments, and return a dict containing them."""

# Create the top level parser

parser = argparse.ArgumentParser(description=("d-h.st (Dev-Host) command"

"line tool"))

parser.add_argument('-u', "--username",

help=("Username. If none is provided, uploads are"

" done anonymously, and only public files are"

" accessible"))

parser.add_argument('-p', "--password",

help=("Password. If only a username is provided, the"

" user will be prompted for one without it"

" appearing on the screen"))

subparsers = parser.add_subparsers(metavar="ACTION", dest="action",

help="Use %(prog)s ACTION -h for help")

# Parent parsers

# We use a parent parser to get the user's info again so that args can be

# after the actions too. For example, this would raise an unknown arg error

# otherwise:

# devhost.py upload file.txt -u myusername -p mypassword

parser_u = argparse.ArgumentParser(add_help=False)

parser_u.add_argument('-u', "--username",

help=("Username. If none is provided, uploads are"

" done anonymously, and only public files are"

" accessible"))

parser_u.add_argument('-p', "--password",

help=("Password. If only a username is provided, the"

" user will be prompted for one without it"

" appearing on the screen"))

# Other parent parsers

parser_c = argparse.ArgumentParser(add_help=False)

parser_c.add_argument("file_code", metavar="file-code",

help=("File code of an existing file. Multiple file"

" code may be specified for certain commands"

" (separated by commas, without spaces)"))

parser_fo = argparse.ArgumentParser(add_help=False)

parser_fo.add_argument("folder_id", metavar="folder-id",

help="ID of an existing folder")

# Create the parser for the "upload" command

parser_upload = subparsers.add_parser("upload", parents=[parser_u],

help="Upload file")

parser_upload.add_argument("my_file", type=argparse.FileType('rb'),

metavar="file", help="File to upload")

parser_upload.add_argument('-d', "--file-desc", help="Description of file")

parser_upload.add_argument('-c', "--file-code", help=("File code of an"

"existing file to update/replace"))

parser_upload.add_argument('-pb', "--public", choices=['0', '1'],

default='0', help=("File is public or private,"

" 0 - private, 1 - public"))

parser_upload.add_argument('-f', "--upload-folder", dest="uploadfolder",

default='0',

help=("Folder id to upload file to. The root"

" folder is chosen by default"))

# Create the parser for the "get-file-info" command

parser_getf = subparsers.add_parser("file-get-info",

parents=[parser_c, parser_u],

help="Return file info")

# Create the parser for the "set-file-info" command

parser_setf = subparsers.add_parser("file-set-info",

parents=[parser_c, parser_u],

help="Set file info")

parser_setf.add_argument('-n', "--file-name", dest="name",

help=h_empty("name"))

parser_setf.add_argument('-d', "--file-desc", dest="description",

help=h_empty("description"))

parser_setf.add_argument('-pb', "--public", choices=['0', '1'],

default='0', help=h_empty("public status, 0 -"

" private, 1 - public"))

parser_setf.add_argument('-f', "--folder-id",

help="Use to change the file's folder")

# Create the parser for the "file-delete" command

parser_delf = subparsers.add_parser("file-delete",

parents=[parser_c, parser_u],

help="Delete file")

# Create the parser for the "file-move" command

parser_mvf = subparsers.add_parser("file-move",

parents=[parser_c, parser_u],

help="Move file")

parser_mvf.add_argument('-f', "--folder-id",

help=("Use if you want to change the folder."

" Specify folder_id or 0 for root directory"))

# Create the parser for the "get-folder-info" command

parser_getfo = subparsers.add_parser("folder-get-info",

parents=[parser_fo, parser_u],

help="Return folder info")

# Create the parser for the "set-folder-info" command

parser_setfo = subparsers.add_parser("folder-set-info",

parents=[parser_fo, parser_u],

help="Set folder info")

parser_setfo.add_argument('-n', "--folder-name", dest="name",

help=h_empty("name"))

parser_setfo.add_argument('-d', "--folder-desc", dest="description",

help=h_empty("description"))

parser_setfo.add_argument('-f', "--parent-folder-id",

help=("Use to change the parent folder. Specify"

"the folder ID or 0 for root directory"))

# Create the parser for the "folder-delete" command

parser_delfo = subparsers.add_parser("folder-delete",

parents=[parser_fo, parser_u],

help="Delete folder")

# Create the parser for the "folder-move" command

parser_mvfo = subparsers.add_parser("folder-move",

parents=[parser_fo, parser_u],

help="Move folder")

parser_mvfo.add_argument('-f', "--parent-folder-id",

help=("Use if you want to change the folder."

" Specify the folder ID or 0 for root directory"))

# Create the parser for the "folder-create" command

parser_cfo = subparsers.add_parser("folder-create",

parents=[parser_u],

help="Create folder")

parser_cfo.add_argument("name", metavar="folder-name",

help="Folder name")

parser_cfo.add_argument('-d', "--folder-desc", dest="description",

help="Folder description")

parser_cfo.add_argument('-f', "--parent-folder-id",

help="Create the folder inside this one")

# Create the parser for the "folder-content" command

parser_confo = subparsers.add_parser("folder-content",

parents=[parser_fo, parser_u],

help="Get folder content")

parser_confo.add_argument("--user",

help=("Username of the person you want to"

" retrieve the folder's content for"))

parser_confo.add_argument("--user-id",

help=("User ID of the person you want to"

" retrieve the folder content for"))

# TODO: merge some of the duplicate items into a parent parser

# Parse the args and return them as a dict

args = parser.parse_args()

if args.action is None:

parser.print_help()

exit(0)

return vars(args)

def h_empty(s):

"""Substitute keyword and return repetitive help message for arg parser"""

s = ("Use to change the %s. Choosing an empty value \"\" will"

" clear the data.") % s

return s

def pretty_print(result):

"""Print XML object line by line, capitalizing the tag"""

for field in parse_info(result):

print "%s: %s" % (field.tag.capitalize(), field.text)

def login(username, password):

"""Login and return the token, which is used for identification.

The token lasts for one hour.

"""

args = {'action': "user/auth", 'user': username, 'pass': password}

resp = api_do(args)

resp = ET.XML(resp)

try:

token = resp.findall(".//token")[0].text

except IndexError:

pretty_print(ET.tostring(resp))

exit(1)

return token

def parse_info(xml):

"""Parse XML and return its elements as a list"""

xml = ET.XML(xml)

return xml.findall(".//*")

def upload(args):

"""Handle file upload

Generate XID, build request and call the upload_file method.

Also run get_progress as a thread to print the progress from the server.

"""

xid = binascii.hexlify(os.urandom(8))

files_data = {'file': args.pop('my_file')}

if 'file_desc' in args:

args['file_description[]'] = args.pop('file_desc')

if 'file_code' in args:

args['file_code[]'] = args.pop('file_code')

upload_data = args

# Get and print the progress using a daemon thread

t = threading.Thread(target=get_progress, args=(xid,))

t.daemon = True

t.start()

result = upload_file(files_data, upload_data, xid)

return result

def upload_file(files_data, upload_data, xid=None):

"""Upload file and return the parsed response"""

# xid is optional, and can be used to track progress

url = 'http://api.d-h.st/upload'

if xid is not None:

url = '%s?X-Progress-ID=%s' % (url, xid)

r = post(url, data=upload_data, files=files_data)

return r.content

def get_progress(xid):

"""Prints the upload's progress, using the xid

This should be run in a separate thread.

"""

url = 'http://api.d-h.st/progress?X-Progress-ID=%s' % xid

# Wait a bit more before getting the progress for the first time. This is

# to (hopefully) avoid the "Max retries exceeded" error, which seems to

# happen when we request the progress too many times while the the upload

# is still starting.

time.sleep(5)

while True:

# We're getting the progress from the website, so there's a slight

# traffic overhead, which is why we're waiting a few seconds between

# refreshes.

time.sleep(5)

try:

request = get(url)

# It doesn't matter if we fail to get the progress, as long as

# the upload is still going on. Should that fail, this thread will

# terminate anyway.

except requests.exceptions:

continue

except Exception, e:

print("An error has occured: %s" % repr(e))

print("Continuing...")

continue

resp = request.content.strip()[1:-2]

progress = json.loads(resp.decode())

if progress.get('state') == "uploading":

percentage = progress.get('received') / progress.get('size') * 100

percentage = '{n:.{d}f}'.format(n=percentage, d=2)

print "Progress: %s%%" % percentage,

sys.stdout.write('\r')

sys.stdout.flush()

elif progress.get('state') == "starting":

pass

else:

print progress.get('state')

def api_do(args):

"""Generates a URL using the passed args, gets the data from it,

and returns the content of the response

Refer to the gen_url docstring for more info.

"""

url = gen_url(args)

try:

r = get(url)

except requests.exceptions, err:

print err

exit(1)

return r.content

def gen_url(args):

"""Generate a URL using the keys/values of args, and return it

args is a dict that has the following items:

'action': one of "file/getinfo", "file/setinfo", "file/delete",

"file/move", "folder/getinfo", "folder/setinfo", "folder/delete",

"folder/move", "folder/create", "folder/content"

'parameter': parameter value

For example, to delete a file:

args = {'action': "file/delete", 'token': token, 'file_code': "ygH"}

Refer to the Dev-Host API for more information.

"""

url = ["http://d-h.st/api/%s" % args['action']]

del args['action']

first = True

for key, value in args.items():

if first == True:

url.append("?")

first = False

else:

url.append("&")

url.append("%s=%s" % (key, value))

url = ''.join(url)

return url

def signal_handler(signal, frame):

"""Handle SIGINT"""

print "\nAborted by user."

exit(0)

def clean_dict(args):

"""Remove None items from the dict and return it"""

result = {}

result.update((k, v) for k, v in args.items() if v is not None)

return result

def main():

args = arg_parser()

args = clean_dict(args)

signal.signal(signal.SIGINT, signal_handler)

methods = {'upload': "uploadapi", 'file-get-info': "file/getinfo",

'file-set-info': "file/setinfo", 'file-delete': "file/delete",

'file-move': "file/move", 'folder-get-info': "folder/getinfo",

'folder-set-info': "folder/setinfo", 'folder-delete':

"folder/delete", 'folder-move': "folder/move", 'folder-create':

"folder/create", 'folder-content': "folder/content"}

token = None

if 'username' in args:

if 'password' not in args:

args['password'] = getpass("Password? ")

print "Logging in..."

args['token'] = login(args['username'], args['password'])

del args['password']

del args['username']

elif 'password' in args:

del args['password']

result = None

if args['action'] in methods:

print "Starting...\n"

args['action'] = methods[args['action']]

if args['action'] == "uploadapi":

result = upload(args)

else:

result = api_do(args)

else:

print "Action not recognized."

if result is not None:

pretty_print(result)

if __name__ == "__main__":

main()

android_scripts/devhost.py at kitkat · AndroidOpenSourceXperia/android_scripts · GitHub

Uh oh!

Expand file tree

Breadcrumbs

devhost.py

Latest commit

History

Breadcrumbs

devhost.py

File metadata and controls