start work on remote syncing; full up/down implemented

This commit is contained in:
Damien Elmes 2011-09-29 20:58:42 +09:00
parent 9b8e949a72
commit aabc884341
3 changed files with 173 additions and 184 deletions

View file

@ -178,6 +178,16 @@ crt=?, mod=?, scm=?, dty=?, usn=?, ls=?, conf=?""",
def usn(self): def usn(self):
return self._usn if self.server else -1 return self._usn if self.server else -1
def beforeUpload(self):
"Called before a full upload."
tbls = "facts", "cards", "revlog", "graves"
for t in tbls:
self.db.execute("update %s set usn=0 where usn=-1" % t)
self._usn = 0
self.modSchema()
self.ls = self.scm
self.close()
# Object creation helpers # Object creation helpers
########################################################################## ##########################################################################

View file

@ -3,9 +3,10 @@
# License: GNU AGPL, version 3 or later; http://www.gnu.org/licenses/agpl.html # License: GNU AGPL, version 3 or later; http://www.gnu.org/licenses/agpl.html
import zlib, re, urllib, urllib2, socket, simplejson, time, shutil import zlib, re, urllib, urllib2, socket, simplejson, time, shutil
import os, base64, sys, httplib, types import os, base64, sys, httplib2, types, zipfile
from cStringIO import StringIO
from datetime import date from datetime import date
import anki, anki.deck, anki.cards from anki.db import DB
from anki.errors import * from anki.errors import *
from anki.utils import ids2str, checksum, intTime from anki.utils import ids2str, checksum, intTime
from anki.consts import * from anki.consts import *
@ -21,12 +22,14 @@ SYNC_HOST = os.environ.get("SYNC_HOST") or "dev.ankiweb.net"
SYNC_PORT = int(os.environ.get("SYNC_PORT") or 80) SYNC_PORT = int(os.environ.get("SYNC_PORT") or 80)
SYNC_URL = "http://%s:%d/sync/" % (SYNC_HOST, SYNC_PORT) SYNC_URL = "http://%s:%d/sync/" % (SYNC_HOST, SYNC_PORT)
# fixme: status() should be using the hooks instead
# todo: # todo:
# - ensure all urllib references are converted to urllib2 for proxies # - ensure all urllib references are converted to urllib2 for proxies
# - ability to cancel # - ability to cancel
# - need to make sure syncing doesn't bump the deck modified time if nothing was # - need to make sure syncing doesn't bump the deck modified time if nothing was
# changed, since by default closing the deck bumps the mod time # changed, since by default closing the deck bumps the mod time
# - syncing with #/&/etc in password # - ensure the user doesn't add foreign chars to passsword
# - timeout on all requests (issue 2625) # - timeout on all requests (issue 2625)
# - ditch user/pass in favour of session key? # - ditch user/pass in favour of session key?
@ -58,8 +61,8 @@ class Syncer(object):
"Returns 'noChanges', 'fullSync', or 'success'." "Returns 'noChanges', 'fullSync', or 'success'."
# step 1: login & metadata # step 1: login & metadata
self.status("login") self.status("login")
self.rmod, rscm, self.maxUsn = self.server.times() self.rmod, rscm, self.maxUsn = self.server.meta()
self.lmod, lscm, self.minUsn = self.times() self.lmod, lscm, self.minUsn = self.meta()
if self.lmod == self.rmod: if self.lmod == self.rmod:
return "noChanges" return "noChanges"
elif lscm != rscm: elif lscm != rscm:
@ -98,7 +101,7 @@ class Syncer(object):
self.finish(mod) self.finish(mod)
return "success" return "success"
def times(self): def meta(self):
return (self.deck.mod, self.deck.scm, self.deck._usn) return (self.deck.mod, self.deck.scm, self.deck._usn)
def changes(self): def changes(self):
@ -385,9 +388,53 @@ class LocalServer(Syncer):
l = simplejson.loads; d = simplejson.dumps l = simplejson.loads; d = simplejson.dumps
return l(d(Syncer.applyChanges(self, minUsn, lnewer, l(d(changes))))) return l(d(Syncer.applyChanges(self, minUsn, lnewer, l(d(changes)))))
# not yet ported
class RemoteServer(Syncer): class RemoteServer(Syncer):
pass def __init__(self, user, hkey):
self.user = user
self.hkey = hkey
def meta(self):
h = httplib2.Http(timeout=60)
resp, cont = h.request(
SYNC_URL+"meta?" + urllib.urlencode(dict(u=self.user)))
if resp['status'] != '200':
raise Exception("Invalid response code: %s" % resp['status'])
return simplejson.loads(cont)
def hostKey(self, pw):
h = httplib2.Http(timeout=60)
resp, cont = h.request(
SYNC_URL+"hostKey?" + urllib.urlencode(dict(u=self.user,p=pw)))
if resp['status'] != '200':
raise Exception("Invalid response code: %s" % resp['status'])
self.hkey = cont
return cont
def _run(self, action, **args):
data = {"p": self.password,
"u": self.username,
"v": 2}
if self.deckName:
data['d'] = self.deckName.encode("utf-8")
else:
data['d'] = None
data.update(args)
data = urllib.urlencode(data)
try:
f = urllib2.urlopen(SYNC_URL + action, data)
except (urllib2.URLError, socket.error, socket.timeout,
httplib.BadStatusLine), e:
raise SyncError(type="connectionError",
exc=`e`)
ret = f.read()
if not ret:
raise SyncError(type="noResponse")
try:
return self.unstuff(ret)
except Exception, e:
raise SyncError(type="connectionError",
exc=`e`)
# def unstuff(self, data): # def unstuff(self, data):
# return simplejson.loads(unicode(zlib.decompress(data), "utf8")) # return simplejson.loads(unicode(zlib.decompress(data), "utf8"))
# def stuff(self, data): # def stuff(self, data):
@ -462,190 +509,73 @@ class HttpSyncServerProxy(object):
def finish(self): def finish(self):
assert self.runCmd("finish") == "OK" assert self.runCmd("finish") == "OK"
def runCmd(self, action, **args):
data = {"p": self.password,
"u": self.username,
"v": 2}
if self.deckName:
data['d'] = self.deckName.encode("utf-8")
else:
data['d'] = None
data.update(args)
data = urllib.urlencode(data)
try:
f = urllib2.urlopen(SYNC_URL + action, data)
except (urllib2.URLError, socket.error, socket.timeout,
httplib.BadStatusLine), e:
raise SyncError(type="connectionError",
exc=`e`)
ret = f.read()
if not ret:
raise SyncError(type="noResponse")
try:
return self.unstuff(ret)
except Exception, e:
raise SyncError(type="connectionError",
exc=`e`)
# Full syncing # Full syncing
########################################################################## ##########################################################################
# not yet ported # not yet ported
# make sure it resets any usn == -1 before uploading! # make sure it resets any usn == -1 before uploading!
# make sure webserver is sending gzipped
class FullSyncer(object): class FullSyncer(object):
def __init__(self, deck): def __init__(self, deck, hkey):
self.deck = deck self.deck = deck
self.hkey = hkey
def prepareFullSync(self): def download(self):
t = time.time()
# ensure modified is not greater than server time
self.deck.modified = min(self.deck.modified, self.server.timestamp)
self.deck.db.commit()
self.deck.close() self.deck.close()
fields = { h = httplib2.Http(timeout=60)
"p": self.server.password, resp, cont = h.request(
"u": self.server.username, SYNC_URL+"download?" + urllib.urlencode(dict(k=self.hkey)))
"d": self.server.deckName.encode("utf-8"), if resp['status'] != '200':
} raise Exception("Invalid response code: %s" % resp['status'])
if self.localTime > self.remoteTime: tpath = self.deck.path + ".tmp"
return ("fromLocal", fields, self.deck.path) open(tpath, "wb").write(cont)
else: os.unlink(self.deck.path)
return ("fromServer", fields, self.deck.path) os.rename(tpath, self.deck.path)
d = DB(self.deck.path)
assert d.scalar("pragma integrity_check") == "ok"
self.deck = None
def fullSync(self): def upload(self):
ret = self.prepareFullSync() self.deck.beforeUpload()
if ret[0] == "fromLocal": # compressed post body support is flaky, so bundle into a zip
self.fullSyncFromLocal(ret[1], ret[2]) f = StringIO()
else: z = zipfile.ZipFile(f, mode="w", compression=zipfile.ZIP_DEFLATED)
self.fullSyncFromServer(ret[1], ret[2]) z.write(self.deck.path, "col.anki")
z.close()
def fullSyncFromLocal(self, fields, path): # build an upload body
global sendProgressHook f2 = StringIO()
try: fields = dict(k=self.hkey)
# write into a temporary file, since POST needs content-length
src = open(path, "rb")
name = namedtmp("fullsync.anki")
tmp = open(name, "wb")
# post vars # post vars
for (key, value) in fields.items(): for (key, value) in fields.items():
tmp.write('--' + MIME_BOUNDARY + "\r\n") f2.write('--' + MIME_BOUNDARY + "\r\n")
tmp.write('Content-Disposition: form-data; name="%s"\r\n' % key) f2.write('Content-Disposition: form-data; name="%s"\r\n' % key)
tmp.write('\r\n') f2.write('\r\n')
tmp.write(value) f2.write(value)
tmp.write('\r\n') f2.write('\r\n')
# file header # file header
tmp.write('--' + MIME_BOUNDARY + "\r\n") f2.write('--' + MIME_BOUNDARY + "\r\n")
tmp.write( f2.write(
'Content-Disposition: form-data; name="deck"; filename="deck"\r\n') 'Content-Disposition: form-data; name="deck"; filename="deck"\r\n')
tmp.write('Content-Type: application/octet-stream\r\n') f2.write('Content-Type: application/octet-stream\r\n')
tmp.write('\r\n') f2.write('\r\n')
# data f2.write(f.getvalue())
comp = zlib.compressobj() f.close()
while 1: f2.write('\r\n--' + MIME_BOUNDARY + '--\r\n\r\n')
data = src.read(CHUNK_SIZE) size = f2.tell()
if not data: # connection headers
tmp.write(comp.flush())
break
tmp.write(comp.compress(data))
src.close()
tmp.write('\r\n--' + MIME_BOUNDARY + '--\r\n\r\n')
size = tmp.tell()
tmp.seek(0)
# open http connection
runHook("fullSyncStarted", size)
headers = { headers = {
'Content-type': 'multipart/form-data; boundary=%s' % 'Content-type': 'multipart/form-data; boundary=%s' %
MIME_BOUNDARY, MIME_BOUNDARY,
'Content-length': str(size), 'Content-length': str(size),
'Host': SYNC_HOST, 'Host': SYNC_HOST,
} }
req = urllib2.Request(SYNC_URL + "fullup?v=2", tmp, headers) body = f2.getvalue()
try: f2.close()
sendProgressHook = fullSyncProgressHook h = httplib2.Http(timeout=60)
res = urllib2.urlopen(req).read() resp, cont = h.request(
assert res.startswith("OK") SYNC_URL+"upload", "POST", headers=headers, body=body)
# update lastSync if resp['status'] != '200':
c = sqlite.connect(path) raise Exception("Invalid response code: %s" % resp['status'])
c.execute("update decks set lastSync = ?", assert cont == "OK"
(res[3:],))
c.commit()
c.close()
finally:
sendProgressHook = None
tmp.close()
finally:
runHook("fullSyncFinished")
def fullSyncFromServer(self, fields, path):
try:
runHook("fullSyncStarted", 0)
fields = urllib.urlencode(fields)
src = urllib.urlopen(SYNC_URL + "fulldown", fields)
tmpname = namedtmp("fullsync.anki")
tmp = open(tmpname, "wb")
decomp = zlib.decompressobj()
cnt = 0
while 1:
data = src.read(CHUNK_SIZE)
if not data:
tmp.write(decomp.flush())
break
tmp.write(decomp.decompress(data))
cnt += CHUNK_SIZE
runHook("fullSyncProgress", "fromServer", cnt)
src.close()
tmp.close()
os.close(fd)
# if we were successful, overwrite old deck
os.unlink(path)
os.rename(tmpname, path)
# reset the deck name
c = sqlite.connect(path)
c.execute("update decks set syncName = ?",
[checksum(path.encode("utf-8"))])
c.commit()
c.close()
finally:
runHook("fullSyncFinished")
##########################################################################
# Monkey-patch httplib to incrementally send instead of chewing up large
# amounts of memory, and track progress.
sendProgressHook = None
def incrementalSend(self, strOrFile):
if self.sock is None:
if self.auto_open:
self.connect()
else:
raise NotConnected()
if self.debuglevel > 0:
print "send:", repr(str)
try:
if (isinstance(strOrFile, str) or
isinstance(strOrFile, unicode)):
self.sock.sendall(strOrFile)
else:
cnt = 0
t = time.time()
while 1:
if sendProgressHook and time.time() - t > 1:
sendProgressHook(cnt)
t = time.time()
data = strOrFile.read(CHUNK_SIZE)
cnt += len(data)
if not data:
break
self.sock.sendall(data)
except socket.error, v:
if v[0] == 32: # Broken pipe
self.close()
raise
httplib.HTTPConnection.send = incrementalSend
def fullSyncProgressHook(cnt):
runHook("fullSyncProgress", "fromLocal", cnt)

View file

@ -6,7 +6,7 @@ from tests.shared import assertException
from anki.errors import * from anki.errors import *
from anki import Deck from anki import Deck
from anki.utils import intTime from anki.utils import intTime
from anki.sync import Syncer, LocalServer from anki.sync import Syncer, FullSyncer, LocalServer, RemoteServer
from anki.facts import Fact from anki.facts import Fact
from anki.cards import Card from anki.cards import Card
from tests.shared import getEmptyDeck from tests.shared import getEmptyDeck
@ -248,5 +248,54 @@ def _test_speed():
assert client.sync() == "success" assert client.sync() == "success"
print "sync %d" % ((time.time() - t)*1000); t = time.time() print "sync %d" % ((time.time() - t)*1000); t = time.time()
# Remote tests
##########################################################################
import anki.sync
anki.sync.SYNC_URL = "http://localhost:8001/sync/"
TEST_USER = "synctest@ichi2.net"
TEST_PASS = "synctest"
TEST_HKEY = "k14LvSaEtXFITCJz"
def setup_remote():
global server
setup_basic()
# mark deck1 as changed
deck1.save()
server = RemoteServer(TEST_USER, TEST_HKEY)
client.server = server
@nose.with_setup(setup_remote)
def test_meta():
(mod, scm, usn) = server.meta()
assert mod
assert scm
assert mod != client.deck.mod
@nose.with_setup(setup_remote)
def test_hkey():
assertException(Exception, lambda: server.hostKey("wrongpass"))
server.hkey = "abc"
k = server.hostKey(TEST_PASS)
assert k == server.hkey == TEST_HKEY
@nose.with_setup(setup_remote)
def test_download():
f = FullSyncer(client.deck, "abc")
assertException(Exception, f.download)
f.hkey = TEST_HKEY
f.download()
@nose.with_setup(setup_remote)
def test_remoteSync():
# not yet associated, so will require a full sync
assert client.sync() == "fullSync"
# upload
f = FullSyncer(client.deck, TEST_HKEY)
f.upload()
client.deck.reopen()
# should report no changes
assert client.sync() == "noChanges"
# bump local deck
client.deck.save()
print client.sync()