Page Menu
Home
c4science
Search
Configure Global Search
Log In
Files
F78725848
bibknowledge_dblayer.py
No One
Temporary
Actions
Download File
Edit File
Delete File
View Transforms
Subscribe
Mute Notifications
Award Token
Subscribers
None
File Metadata
Details
File Info
Storage
Attached
Created
Thu, Aug 22, 15:29
Size
14 KB
Mime Type
text/x-python
Expires
Sat, Aug 24, 15:29 (2 d)
Engine
blob
Format
Raw Data
Handle
20090548
Attached To
R3600 invenio-infoscience
bibknowledge_dblayer.py
View Options
# -*- coding: utf-8 -*-
##
## This file is part of Invenio.
## Copyright (C) 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010 CERN.
##
## Invenio 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 2 of the
## License, or (at your option) any later version.
##
## Invenio 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 Invenio; if not, write to the Free Software Foundation, Inc.,
## 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
"""
Database access related functions for BibKnowledge.
"""
__revision__
=
"$Id$"
from
invenio.dbquery
import
run_sql
def
get_kbs_info
(
kbtypeparam
=
""
,
searchkbname
=
""
):
"""Returns all kbs as list of dictionaries {id, name, description, kbtype}
If the KB is dynamic, the dynamic kb key are added in the dict.
"""
out
=
[]
query
=
"SELECT id, name, description, kbtype FROM knwKB ORDER BY name"
res
=
run_sql
(
query
)
for
row
in
res
:
doappend
=
1
# by default
kbid
=
row
[
0
]
name
=
row
[
1
]
description
=
row
[
2
]
kbtype
=
row
[
3
]
dynres
=
{}
if
kbtype
==
'd'
:
#get the dynamic config
dynres
=
get_kb_dyn_config
(
kbid
)
if
kbtypeparam
:
doappend
=
0
if
(
kbtype
==
kbtypeparam
):
doappend
=
1
if
searchkbname
:
doappend
=
0
if
(
name
==
searchkbname
):
doappend
=
1
if
doappend
:
mydict
=
{
'id'
:
kbid
,
'name'
:
name
,
'description'
:
description
,
'kbtype'
:
kbtype
}
mydict
.
update
(
dynres
)
out
.
append
(
mydict
)
return
out
def
get_kb_id
(
kb_name
):
"""Returns the id of the kb with given name"""
res
=
run_sql
(
"""SELECT id FROM knwKB WHERE name LIKE %s"""
,
(
kb_name
,))
if
len
(
res
)
>
0
:
return
res
[
0
][
0
]
else
:
return
None
def
get_kb_name
(
kb_id
):
"""Returns the name of the kb with given id
@param kb_id the id
@return string
"""
res
=
run_sql
(
"""SELECT name FROM knwKB WHERE id=%s"""
,
(
kb_id
,))
if
len
(
res
)
>
0
:
return
res
[
0
][
0
]
else
:
return
None
def
get_kb_type
(
kb_id
):
"""Returns the type of the kb with given id
@param kb_id knowledge base id
@return kb_type
"""
res
=
run_sql
(
"""SELECT kbtype FROM knwKB WHERE id=%s"""
,
(
kb_id
,))
if
len
(
res
)
>
0
:
return
res
[
0
][
0
]
else
:
return
None
def
get_kb_mappings
(
kb_name
=
""
,
sortby
=
"to"
,
keylike
=
""
,
valuelike
=
""
,
match_type
=
"s"
):
"""Returns a list of all mappings from the given kb, ordered by key
@param kb_name knowledge base name. if "", return all
@param sortby the sorting criteria ('from' or 'to')
@keylike return only entries where key matches this
@valuelike return only entries where value matches this
"""
out
=
[]
k_id
=
get_kb_id
(
kb_name
)
if
len
(
keylike
)
>
0
:
if
match_type
==
"s"
:
keylike
=
"%"
+
keylike
+
"%"
else
:
keylike
=
'%'
if
len
(
valuelike
)
>
0
:
if
match_type
==
"s"
:
valuelike
=
"%"
+
valuelike
+
"%"
else
:
valuelike
=
'%'
if
not
kb_name
:
res
=
run_sql
(
"""SELECT m.id, m.m_key, m.m_value, m.id_knwKB,
k.name
FROM knwKBRVAL m, knwKB k
where m_key like %s
and m_value like %s
and m.id_knwKB = k.id"""
,
(
keylike
,
valuelike
))
else
:
res
=
run_sql
(
"""SELECT m.id, m.m_key, m.m_value, m.id_knwKB,
k.name
FROM knwKBRVAL m, knwKB k
WHERE id_knwKB=%s
and m.id_knwKB = k.id
and m_key like %s
and m_value like %s"""
,
(
k_id
,
keylike
,
valuelike
))
#sort res
lres
=
list
(
res
)
if
sortby
==
"from"
:
lres
.
sort
(
lambda
x
,
y
:
cmp
(
x
[
1
],
y
[
1
]))
else
:
lres
.
sort
(
lambda
x
,
y
:
cmp
(
x
[
2
],
y
[
2
]))
for
row
in
lres
:
out
.
append
({
'id'
:
row
[
0
],
'key'
:
row
[
1
],
'value'
:
row
[
2
],
'kbid'
:
row
[
3
],
'kbname'
:
row
[
4
]})
return
out
def
get_kb_dyn_config
(
kb_id
):
"""
Returns a dictionary of 'field'=> y, 'expression'=> z
for a knowledge base of type 'd'. The dictionary may have coll_id, collection.
@param kb_id the id
@return dict
"""
res
=
run_sql
(
"""SELECT output_tag, search_expression, id_collection
FROM knwKBDDEF where
id_knwKB = %s"""
,
(
kb_id
,
))
mydict
=
{}
for
row
in
res
:
mydict
[
'field'
]
=
row
[
0
]
mydict
[
'expression'
]
=
row
[
1
]
mydict
[
'coll_id'
]
=
row
[
2
]
#put a collection field if collection exists..
if
mydict
.
has_key
(
'coll_id'
):
c_id
=
mydict
[
'coll_id'
]
res
=
run_sql
(
"""SELECT name from collection where id = %s"""
,
(
c_id
,))
if
res
:
mydict
[
'collection'
]
=
res
[
0
][
0
]
return
mydict
def
save_kb_dyn_config
(
kb_id
,
field
,
expression
,
collection
=
""
):
"""Saves a dynamic knowledge base configuration
@param kb_id the id
@param field the field where values are extracted
@param expression ..using this expression
@param collection ..in a certain collection (default is all)
"""
#check that collection exists
coll_id
=
None
if
collection
:
res
=
run_sql
(
"""SELECT id from collection where name = %s"""
,
(
collection
,))
if
res
:
coll_id
=
res
[
0
][
0
]
run_sql
(
"""DELETE FROM knwKBDDEF where id_knwKB = %s"""
,
(
kb_id
,
))
run_sql
(
"""INSERT INTO knwKBDDEF (id_knwKB, output_tag, search_expression, id_collection)
VALUES (%s,%s,%s,%s)"""
,
(
kb_id
,
field
,
expression
,
coll_id
))
return
""
def
get_kb_description
(
kb_name
):
"""Returns the description of the given kb
@param kb_id the id
@return string
"""
k_id
=
get_kb_id
(
kb_name
)
res
=
run_sql
(
"""SELECT description FROM knwKB WHERE id=%s"""
,
(
k_id
,))
return
res
[
0
][
0
]
def
add_kb
(
kb_name
,
kb_description
,
kb_type
=
None
):
"""
Adds a new kb with given name and description. Returns the id of
the kb.
If name already exists replace old value
@param kb_name the name of the kb to create
@param kb_description a description for the kb
@return the id of the newly created kb
"""
kb_db
=
'w'
#the typical written_as - change_to
if
not
kb_type
:
pass
else
:
if
kb_type
==
'taxonomy'
:
kb_db
=
't'
if
kb_type
==
'dynamic'
:
kb_db
=
'd'
run_sql
(
"""REPLACE INTO knwKB (name, description, kbtype)
VALUES (%s,%s,%s)"""
,
(
kb_name
,
kb_description
,
kb_db
))
return
get_kb_id
(
kb_name
)
def
delete_kb
(
kb_name
):
"""Deletes the given kb"""
k_id
=
get_kb_id
(
kb_name
)
run_sql
(
"""DELETE FROM knwKBRVAL WHERE id_knwKB = %s"""
,
(
k_id
,))
run_sql
(
"""DELETE FROM knwKB WHERE id = %s"""
,
(
k_id
,))
#finally, delete from COLL table
run_sql
(
"""DELETE FROM knwKBDDEF where id_knwKB = %s"""
,
(
k_id
,))
return
True
def
kb_exists
(
kb_name
):
"""Returns True if a kb with the given name exists"""
rows
=
run_sql
(
"""SELECT id FROM knwKB WHERE name = %s"""
,
(
kb_name
,))
if
len
(
rows
)
>
0
:
return
True
else
:
return
False
def
update_kb
(
kb_name
,
new_name
,
new_description
):
"""Updates given kb with new name and new description"""
k_id
=
get_kb_id
(
kb_name
)
run_sql
(
"""UPDATE knwKB
SET name = %s , description = %s
WHERE id = %s"""
,
(
new_name
,
new_description
,
k_id
))
return
True
def
add_kb_mapping
(
kb_name
,
key
,
value
):
"""Adds new mapping key->value in given kb"""
k_id
=
get_kb_id
(
kb_name
)
run_sql
(
"""REPLACE INTO knwKBRVAL (m_key, m_value, id_knwKB)
VALUES (%s, %s, %s)"""
,
(
key
,
value
,
k_id
))
return
True
def
remove_kb_mapping
(
kb_name
,
key
):
"""Removes mapping with given key from given kb"""
k_id
=
get_kb_id
(
kb_name
)
run_sql
(
"""DELETE FROM knwKBRVAL
WHERE m_key = %s AND id_knwKB = %s"""
,
(
key
,
k_id
))
return
True
def
kb_mapping_exists
(
kb_name
,
key
):
"""Returns true if the mapping with given key exists in the given kb"""
if
kb_exists
(
kb_name
):
k_id
=
get_kb_id
(
kb_name
)
rows
=
run_sql
(
"""SELECT id FROM knwKBRVAL
WHERE m_key = %s
AND id_knwKB = %s"""
,
(
key
,
k_id
))
if
len
(
rows
)
>
0
:
return
True
return
False
def
kb_key_rules
(
key
):
"""Returns a list of 4-tuples that have a key->value mapping in some KB
The format of the tuples is [kb_id, kb_name,key,value] """
res
=
run_sql
(
"""SELECT f.id, f.name, m.m_key, m.m_value
from knwKBRVAL as m JOIN
knwKB as f on
m.id_knwKB=f.id WHERE
m.m_key = %s"""
,
(
key
,
))
return
res
def
kb_value_rules
(
value
):
"""Returns a list of 4-tuples that have a key->value mapping in some KB
The format of the tuples is [kb_id, kb_name,key,value] """
res
=
run_sql
(
"""SELECT f.id, f.name, m.m_key, m.m_value from
knwKBRVAL as m JOIN
knwKB as f on
m.id_knwKB=f.id WHERE
m.m_value = %s"""
,
(
value
,
))
return
res
def
get_kb_mapping_value
(
kb_name
,
key
):
"""
Returns a value of the given key from the given kb.
If mapping not found, returns None #'default'
@param kb_name the name of a knowledge base
@param key the key to look for
#@param default a default value to return if mapping is not found
"""
k_id
=
get_kb_id
(
kb_name
)
res
=
run_sql
(
"""SELECT m_value FROM knwKBRVAL
WHERE m_key LIKE %s
AND id_knwKB = %s LIMIT 1"""
,
(
key
,
k_id
))
if
len
(
res
)
>
0
:
return
res
[
0
][
0
]
else
:
return
None
# default
def
update_kb_mapping
(
kb_name
,
key
,
new_key
,
new_value
):
"""Updates the mapping given by key with new key and value"""
k_id
=
get_kb_id
(
kb_name
)
run_sql
(
"""UPDATE knwKBRVAL
SET m_key = %s , m_value = %s
WHERE m_key = %s AND id_knwKB = %s"""
,
(
new_key
,
new_value
,
key
,
k_id
))
return
True
#the following functions should be used by a higher level API
def
get_kba_values
(
kb_name
,
searchname
=
""
,
searchtype
=
"s"
):
"""Returns the "authority file" type of list of values for a
given knowledge base.
@param kb_name the name of the knowledge base
@param searchname search by this..
@param searchtype s=substring, e=exact, sw=startswith
"""
k_id
=
get_kb_id
(
kb_name
)
if
searchtype
==
's'
and
searchname
:
searchname
=
'%'
+
searchname
+
'%'
if
searchtype
==
'sw'
and
searchname
:
#startswith
searchname
=
searchname
+
'%'
if
not
searchname
:
searchname
=
'%'
res
=
run_sql
(
"""SELECT m_value FROM knwKBRVAL
WHERE m_value LIKE %s
AND id_knwKB = %s"""
,
(
searchname
,
k_id
))
return
res
def
get_kbr_keys
(
kb_name
,
searchkey
=
""
,
searchvalue
=
""
,
searchtype
=
's'
):
"""Returns keys from a knowledge base
@param kb_name the name of the knowledge base
@param searchkey search using this key
@param searchvalue search using this value
@param searchtype s=substring, e=exact, sw=startswith
"""
k_id
=
get_kb_id
(
kb_name
)
if
searchtype
==
's'
and
searchkey
:
searchkey
=
'%'
+
searchkey
+
'%'
if
searchtype
==
's'
and
searchvalue
:
searchvalue
=
'%'
+
searchvalue
+
'%'
if
searchtype
==
'sw'
and
searchvalue
:
#startswith
searchvalue
=
searchvalue
+
'%'
if
not
searchvalue
:
searchvalue
=
'%'
if
not
searchkey
:
searchkey
=
'%'
return
run_sql
(
"""SELECT m_key FROM knwKBRVAL
WHERE m_value LIKE %s
AND m_key LIKE %s
AND id_knwKB = %s"""
,
(
searchvalue
,
searchkey
,
k_id
))
def
get_kbr_values
(
kb_name
,
searchkey
=
""
,
searchvalue
=
""
,
searchtype
=
's'
):
"""Returns values from a knowledge base
@param kb_name the name of the knowledge base
@param searchkey search using this key
@param searchvalue search using this value
@param searchtype s=substring, e=exact, sw=startswith
@return a list of values
"""
k_id
=
get_kb_id
(
kb_name
)
if
searchtype
==
's'
and
searchkey
:
searchkey
=
'%'
+
searchkey
+
'%'
if
searchtype
==
's'
and
searchvalue
:
searchvalue
=
'%'
+
searchvalue
+
'%'
if
searchtype
==
'sw'
and
searchvalue
:
#startswith
searchvalue
=
searchvalue
+
'%'
if
not
searchvalue
:
searchvalue
=
'%'
if
not
searchkey
:
searchkey
=
'%'
return
run_sql
(
"""SELECT m_value FROM knwKBRVAL
WHERE m_value LIKE %s
AND m_key LIKE %s
AND id_knwKB = %s"""
,
(
searchvalue
,
searchkey
,
k_id
))
def
get_kbr_items
(
kb_name
,
searchkey
=
""
,
searchvalue
=
""
,
searchtype
=
's'
):
"""Returns dicts of 'key' and 'value' from a knowledge base
@param kb_name the name of the knowledge base
@param searchkey search using this key
@param searchvalue search using this value
@param searchtype s=substring, e=exact, sw=startswith
@return a list of dictionaries [{'key'=>x, 'value'=>y},..]
"""
k_id
=
get_kb_id
(
kb_name
)
if
searchtype
==
's'
and
searchkey
:
searchkey
=
'%'
+
searchkey
+
'%'
if
searchtype
==
's'
and
searchvalue
:
searchvalue
=
'%'
+
searchvalue
+
'%'
if
searchtype
==
'sw'
and
searchvalue
:
#startswith
searchvalue
=
searchvalue
+
'%'
if
not
searchvalue
:
searchvalue
=
'%'
if
not
searchkey
:
searchkey
=
'%'
res
=
[]
rows
=
run_sql
(
"""SELECT m_key, m_value FROM knwKBRVAL
WHERE m_value LIKE %s
AND m_key LIKE %s
AND id_knwKB = %s"""
,
(
searchvalue
,
searchkey
,
k_id
))
for
row
in
rows
:
mdict
=
{}
m_key
=
row
[
0
]
m_value
=
row
[
1
]
mdict
[
'key'
]
=
m_key
mdict
[
'value'
]
=
m_value
res
.
append
(
mdict
)
return
res
Event Timeline
Log In to Comment