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
|
#!/usr/bin/python3
# SPDX-License-Identifier: GPL-2.0+
import logging
import sqlalchemy
logger = logging.getLogger()
def strip_title(title, package):
for prefix in (package + ": ", package + " "):
if title.startswith(prefix):
return title[len(prefix):]
return title
def get_affects(result, kind):
for bugnum, package, affected, title, patched in result:
row = dict(bugnum=bugnum, package=package, patched=patched, kind=kind)
somesource = False
for pkg in package.split(","):
if not pkg.startswith("src:"):
continue
arow = row.copy()
arow.update(affects=pkg[4:], title=strip_title(title, pkg[4:]))
yield arow
if somesource:
logger.warning("%s #%d assigned to multiple source packages",
kind, bugnum)
somesource = True
if not somesource and affected:
for aff in affected.split(","):
if not aff.startswith("src:"):
continue
arow = row.copy()
arow.update(affects=aff[4:], title=strip_title(title, aff[4:]))
yield arow
somesource = True
if not somesource:
logger.warning("%s #%d affects no packages", kind, bugnum)
def make_table(name, columns):
return sqlalchemy.table(name, *map(sqlalchemy.column, columns.split()))
udd_all_bugs = make_table(
"all_bugs", "id package affected_packages title severity affects_unstable")
udd_bugs_tags = make_table("bugs_tags", "id tag")
udd_bugs_usertags = make_table("bugs_usertags", "id email tag")
def tagged_clause(id_, tag):
return sqlalchemy.exists().where(
sqlalchemy.and_(udd_bugs_tags.c.id == id_, udd_bugs_tags.c.tag == tag))
def get_bugs_where(conn, whereclause, kind):
query = sqlalchemy.select(
[udd_all_bugs.c.id, udd_all_bugs.c.package,
udd_all_bugs.c.affected_packages, udd_all_bugs.c.title,
tagged_clause(udd_all_bugs.c.id, "patch")],
sqlalchemy.and_(whereclause, udd_all_bugs.c.affects_unstable == 't'))
return get_affects(conn.execute(query), kind)
def get_ftbfs(conn):
clause = sqlalchemy.and_(
tagged_clause(udd_all_bugs.c.id, "ftbfs"),
udd_all_bugs.c.severity.in_(["serious", "critical", "grave"]))
return get_bugs_where(conn, clause, 'ftbfs')
def get_usertagged(conn, email, tag, kind):
clause = sqlalchemy.exists().where(
sqlalchemy.and_(udd_bugs_usertags.c.id == udd_all_bugs.c.id,
udd_bugs_usertags.c.email == email,
udd_bugs_usertags.c.tag == tag))
return get_bugs_where(conn, clause, kind)
def get_bugs(conn):
yield from get_ftbfs(conn)
email = 'debian-cross@lists.debian.org'
yield from get_usertagged(conn, email, 'cross-satisfiability', 'bdsat')
yield from get_usertagged(conn, email, 'ftcbfs', 'ftcbfs')
def main():
udde = sqlalchemy.create_engine(
'postgresql://udd-mirror:udd-mirror@udd-mirror.debian.net/'
'udd?client_encoding=utf8')
with udde.connect() as conn:
bugs = list(get_bugs(conn))
query = sqlalchemy.text("""
INSERT INTO bugs (kind, bugnum, package, affects, title, patched)
VALUES (:kind, :bugnum, :package, :affects, :title, :patched);""")
crosse = sqlalchemy.create_engine('sqlite:///db')
with crosse.connect() as conn:
with conn.begin():
conn.execute("DELETE FROM bugs;")
conn.execute(query, bugs)
if __name__ == "__main__":
main()
|
