repo.py
3.73 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
# -*- coding: utf-8 -*-
import os
import shutil
import pysvn
class Repo:
def __init__(self, path, login, passwd):
self.path = path
def get_login(_realm, _username, _may_save):
return True, login, passwd, False
self.svn = pysvn.Client()
self.svn.callback_get_login = get_login
self.svn.update(self.path)
def add(self, src_path):
part_name = os.path.basename(src_path)
dest_path = os.path.join(self.new_path(), part_name)
if (not self.svn.info(dest_path) is None) or (os.path.exists(dest_path)):
raise Exception(src_path+" already present in repository.")
shutil.copy(src_path, dest_path)
self.svn.add(dest_path)
def remove(self, filename, anno_per_file):
paths = set()
paths.add(os.path.join(self.new_path(), filename))
for i in range(anno_per_file):
paths.add(self.upload_path(filename, i))
paths.add(os.path.join(self.finito_path(), filename))
removed = []
for path in paths:
if os.path.exists(path):
self.svn.remove(path)
removed.append(path)
return removed
def db_path(self):
return os.path.join(self.path, "db.xml")
def new_path(self):
return os.path.join(self.path, "new")
def ann_path(self, idx):
ids = chr(idx+ord('A'))
tail = os.path.join("annotation", ids)
return os.path.join(self.path, tail)
def finito_path(self):
return os.path.join(self.path, "adjudication")
def upload_path(self, file_name, idx):
return os.path.join(self.ann_path(idx), file_name)
def upload_prim_path(self, file_name):
return os.path.join(self.finito_path(), file_name)
def read_data(self, path):
with open(path) as src:
return src.read()
def read_contents(self, path, exts):
contents = {}
for ext in exts:
contents[ext] = self.read_data(path + ext)
return contents
def write_data(self, path, data):
with open(path, "w") as dest:
dest.write(data)
def write_contents(self, path, contents):
for ext, data in contents.iteritems():
dest_path = path + ext
self.write_data(dest_path, data)
if self.svn.info(dest_path) is None:
self.svn.add(dest_path)
def upload(self, file_id, idx, contents):
self.write_contents(self.upload_path(file_id, idx), contents)
def upload_prim(self, file_id, contents):
self.write_contents(self.upload_prim_path(file_id), contents)
def download(self, file_id, exts):
return self.read_contents(os.path.join(self.new_path(), file_id), exts)
def download_prim(self, file_id, exts, anno_per_file):
result = []
for i in range(anno_per_file):
conts = self.read_contents(os.path.join(self.ann_path(i), file_id), exts)
result.append(conts)
return result
def checkout(self, file_id, idx, exts):
src_paths = [ self.upload_path(file_id, idx)
, os.path.join(self.new_path(), file_id) ]
for path in src_paths:
if all(os.path.exists(path + ext) for ext in exts):
return self.read_contents(path, exts)
def checkout_prim(self, file_id, exts):
path = os.path.join(self.finito_path(), file_id)
if all(os.path.exists(path + ext) for ext in exts):
return self.read_contents(path, exts)
def commit(self, message):
self.svn.checkin(self.path, message, recurse=True)
def revert(self):
self.svn.revert(self.path, recurse=True)