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
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
|
# Copyright (C) 2009 Canonical
#
# Authors:
# Michael Vogt
#
# 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; version 3.
#
# 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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
from datetime import datetime
import apt_pkg
apt_pkg.init_config()
from gi.repository import GLib
from gi.repository import Gio
import glob
import gzip
import os.path
import logging
import string
import re
try:
import cPickle as pickle
pickle # pyflakes
except ImportError:
import pickle
LOG = logging.getLogger(__name__)
from softwarecenter.paths import SOFTWARE_CENTER_CACHE_DIR
from softwarecenter.utils import ExecutionTime
from softwarecenter.db.history import Transaction, PackageHistory
def ascii_lower(key):
ascii_trans_table = string.maketrans(string.ascii_uppercase,
string.ascii_lowercase)
return key.translate(ascii_trans_table)
class AptTransaction(Transaction):
PKGACTIONS = ["Install", "Upgrade", "Downgrade", "Remove", "Purge"]
def __init__(self, sec):
self.start_date = datetime.strptime(sec["Start-Date"],
"%Y-%m-%d %H:%M:%S")
# set the object attributes "install", "upgrade", "downgrade",
# "remove", "purge", error
for k in self.PKGACTIONS + ["Error"]:
# we use ascii_lower for issues described in LP: #581207
attr = ascii_lower(k)
if k in sec:
value = map(self._fixup_history_item, sec[k].split("),"))
else:
value = []
setattr(self, attr, value)
@staticmethod
def _fixup_history_item(s):
""" strip history item string and add missing ")" if needed """
s = s.strip()
# remove the information about the architecture
s = re.sub(":\w+", "", s)
if "(" in s and not s.endswith(")"):
s += ")"
return s
class AptHistory(PackageHistory):
def __init__(self, use_cache=True):
LOG.debug("AptHistory.__init__()")
self.main_context = GLib.main_context_default()
self.history_file = apt_pkg.config.find_file("Dir::Log::History")
#Copy monitoring of history file changes from historypane.py
self.logfile = Gio.File.new_for_path(self.history_file)
self.monitor = self.logfile.monitor_file(0, None)
self.monitor.connect("changed", self._on_apt_history_changed)
self.update_callback = None
LOG.debug("init history")
# this takes a long time, run it in the idle handler
self._transactions = []
self._history_ready = False
GLib.idle_add(self._rescan, use_cache)
@property
def transactions(self):
return self._transactions
@property
def history_ready(self):
return self._history_ready
def _mtime_cmp(self, a, b):
return cmp(os.path.getmtime(a), os.path.getmtime(b))
def _rescan(self, use_cache=True):
self._history_ready = False
self._transactions = []
p = os.path.join(SOFTWARE_CENTER_CACHE_DIR, "apthistory.p")
cachetime = 0
if os.path.exists(p) and use_cache:
with ExecutionTime("loading pickle cache"):
try:
self._transactions = pickle.load(open(p))
cachetime = os.path.getmtime(p)
except:
LOG.exception("failed to load cache")
for history_gz_file in sorted(glob.glob(self.history_file + ".*.gz"),
cmp=self._mtime_cmp):
if os.path.getmtime(history_gz_file) < cachetime:
LOG.debug("skipping already cached '%s'" % history_gz_file)
continue
self._scan(history_gz_file)
self._scan(self.history_file)
if use_cache:
pickle.dump(self._transactions, open(p, "w"))
self._history_ready = True
def _scan(self, history_file, rescan=False):
LOG.debug("_scan: '%s' (%s)" % (history_file, rescan))
try:
tagfile = apt_pkg.TagFile(open(history_file))
except (IOError, SystemError) as ioe:
LOG.debug(ioe)
return
for stanza in tagfile:
# keep the UI alive
while self.main_context.pending():
self.main_context.iteration()
# ignore records with
try:
trans = AptTransaction(stanza)
except (KeyError, ValueError):
continue
# ignore the ones we have already
if (rescan and
len(self._transactions) > 0 and
trans.start_date <= self._transactions[0].start_date):
continue
# add it
# FIXME: this is a list, so potentially slow, but its sorted
# so we could (and should) do a binary search
if not trans in self._transactions:
self._transactions.insert(0, trans)
def _on_apt_history_changed(self, monitor, afile, other_file, event):
if event == Gio.FileMonitorEvent.CHANGES_DONE_HINT:
self._scan(self.history_file, rescan=True)
if self.update_callback:
self.update_callback()
def set_on_update(self, update_callback):
self.update_callback = update_callback
def get_installed_date(self, pkg_name):
installed_date = None
for trans in self._transactions:
for pkg in trans.install:
if pkg.split(" ")[0] == pkg_name:
installed_date = trans.start_date
return installed_date
return installed_date
def _find_in_terminal_log(self, date, term_file):
found = False
term_lines = []
for line in term_file:
if line.startswith("Log started: %s" % date):
found = True
elif line.endswith("Log ended") or line.startswith("Log started"):
found = False
if found:
term_lines.append(line)
return term_lines
def find_terminal_log(self, date):
"""Find the terminal log part for the given transaction
(this can be rather slow)
"""
# FIXME: try to be more clever here with date/file timestamps
term = apt_pkg.config.find_file("Dir::Log::Terminal")
term_lines = self._find_in_terminal_log(date, open(term))
# now search the older history
if not term_lines:
for f in glob.glob(term + ".*.gz"):
term_lines = self._find_in_terminal_log(date, gzip.open(f))
if term_lines:
return term_lines
return term_lines
|