Page Menu
Home
c4science
Search
Configure Global Search
Log In
Files
F79885386
webinterface.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
Wed, Aug 28, 19:23
Size
22 KB
Mime Type
text/x-python
Expires
Fri, Aug 30, 19:23 (2 d)
Engine
blob
Format
Raw Data
Handle
20248718
Attached To
R3600 invenio-infoscience
webinterface.py
View Options
# -*- coding: utf-8 -*-
## $Id: webmessage_webinterface.py,v 1.13 2008/03/12 16:48:08 tibor Exp $
##
## This file is part of Invenio.
## Copyright (C) 2009, 2010, 2011 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.
"""FieldExporter web interface"""
__revision__
=
"$Id: webmessage_webinterface.py,v 1.13 2008/03/12 16:48:08 tibor Exp $"
__lastupdated__
=
"""$Date: 2008/03/12 16:48:08 $"""
import
re
from
invenio.legacy.webpage
import
page
from
invenio.ext.legacy.handler
import
WebInterfaceDirectory
,
\
wash_urlargd
from
invenio.utils.url
import
redirect_to_url
from
invenio.config
import
CFG_SITE_URL
,
\
CFG_SITE_SECURE_URL
from
invenio.utils.date
import
convert_datestruct_to_datetext
,
\
convert_datetext_to_datestruct
from
invenio.base.i18n
import
gettext_set_language
from
invenio.bibexport_method_fieldexporter
import
get_css
,
\
get_navigation_menu
,
\
perform_request_edit_query
,
\
perform_request_edit_job
,
\
perform_request_jobs
,
\
perform_request_new_job
,
\
perform_request_save_job
,
\
perform_request_delete_jobs
,
\
perform_request_run_jobs
,
\
perform_request_job_queries
,
\
perform_request_new_query
,
\
perform_request_save_query
,
\
perform_request_delete_queries
,
\
perform_request_run_queries
,
\
perform_request_job_history
,
\
perform_request_job_results
,
\
perform_request_display_job_result
,
\
perform_request_download_job_result
,
\
AccessDeniedError
from
invenio.bibexport_method_fieldexporter_dblayer
import
Job
,
\
Query
,
\
JobResult
from
invenio.legacy.webuser
import
collect_user_info
,
\
page_not_authorized
from
invenio.modules.access.engine
import
acc_authorize_action
class
WebInterfaceFieldExporterPages
(
WebInterfaceDirectory
):
"""Defines the set of /fieldexporter pages."""
_exports
=
[
""
,
"jobs"
,
"edit_job"
,
"job_queries"
,
"edit_query"
,
"job_results"
,
"display_job_result"
,
"download_job_result"
,
"history"
,
"not_authorized"
]
# constats containig URL to the pages
_EXPORT_URL
=
"
%s
/exporter"
%
(
CFG_SITE_URL
,
)
_JOBS_URL
=
"
%s
/exporter/jobs"
%
(
CFG_SITE_URL
,
)
_EDIT_JOB_URL
=
"
%s
/exporter/edit_job"
%
(
CFG_SITE_URL
,
)
_EDIT_QUERY_URL
=
"
%s
/exporter/edit_query"
%
(
CFG_SITE_URL
,
)
_JOB_QUERIES_URL
=
"
%s
/exporter/job_queries"
%
(
CFG_SITE_URL
,
)
_JOB_HISTORY_URL
=
"
%s
/exporter/history"
%
(
CFG_SITE_URL
,
)
_NOT_AUTHORIZED_URL
=
"
%s
/exporter/not_authorized"
%
(
CFG_SITE_URL
,
)
_LOGIN_URL
=
"
%s
/youraccount/login"
%
(
CFG_SITE_SECURE_URL
,)
_NAVTRAIL_EXPORT
=
"""<a href="/exporter" class="navtrail">Export</a>"""
def
index
(
self
,
req
,
form
):
""" The function called by default"""
redirect_to_url
(
req
,
self
.
_JOB_HISTORY_URL
)
__call__
=
index
def
jobs
(
self
,
req
,
form
):
"""Displays all the jobs of a given user
and allows creation, deletion and execution of jobs"""
argd
=
wash_urlargd
(
form
,
{
"new_button"
:
(
str
,
""
),
"run_button"
:
(
str
,
""
),
"delete_button"
:
(
str
,
""
),
"selected_jobs"
:
(
list
,
""
)
})
# load the right message language
language
=
argd
[
"ln"
]
_
=
gettext_set_language
(
language
)
self
.
_check_user_credentials
(
req
,
language
)
user_id
=
self
.
_get_user_id
(
req
)
try
:
# if the form is submitted through some of the buttons
# we should perform the appropriate action
if
argd
[
"new_button"
]:
self
.
_redirect_to_page
(
req
,
self
.
_EDIT_JOB_URL
,
language
)
elif
argd
[
"delete_button"
]:
job_ids
=
argd
[
"selected_jobs"
]
perform_request_delete_jobs
(
job_ids
=
job_ids
,
user_id
=
user_id
,
language
=
language
)
elif
argd
[
"run_button"
]:
job_ids
=
argd
[
"selected_jobs"
]
perform_request_run_jobs
(
job_ids
=
job_ids
,
user_id
=
user_id
,
language
=
language
)
self
.
_redirect_to_page
(
req
,
self
.
_JOB_HISTORY_URL
,
language
)
user_id
=
self
.
_get_user_id
(
req
)
body
=
perform_request_jobs
(
user_id
=
user_id
,
language
=
language
)
except
AccessDeniedError
:
self
.
_redirect_to_not_authorised_page
(
req
,
language
)
return
page
(
title
=
_
(
"Export Job Overview"
),
metaheaderadd
=
get_css
(),
body
=
body
,
req
=
req
,
navmenuid
=
"fieldexporter"
,
titleprologue
=
get_navigation_menu
(
language
),
navtrail
=
self
.
_NAVTRAIL_EXPORT
,
language
=
language
)
def
edit_job
(
self
,
req
,
form
):
"""Edits an existing job or creates a new one"""
# Create an empty job and use its default values
# to init missing parameters
job
=
Job
()
argd
=
wash_urlargd
(
form
,
{
"job_name"
:
(
str
,
job
.
get_name
()),
"output_directory"
:
(
str
,
job
.
get_output_directory
()),
"job_frequency"
:
(
int
,
job
.
get_frequency
()),
"output_format"
:
(
int
,
job
.
get_output_format
()),
"last_run"
:
(
str
,
convert_datestruct_to_datetext
(
job
.
get_last_run
())),
"id"
:
(
int
,
job
.
get_id
()),
"save_button"
:
(
str
,
""
),
"cancel_button"
:
(
str
,
""
),
"edit_queries_button"
:
(
str
,
""
)
})
language
=
argd
[
"ln"
]
# load the right message language
_
=
gettext_set_language
(
language
)
self
.
_check_user_credentials
(
req
,
language
)
user_id
=
self
.
_get_user_id
(
req
)
job_id
=
argd
[
"id"
]
job
=
Job
(
job_id
=
job_id
,
name
=
argd
[
"job_name"
],
frequency
=
argd
[
"job_frequency"
],
output_format
=
argd
[
"output_format"
],
last_run
=
convert_datetext_to_datestruct
(
argd
[
"last_run"
]),
output_directory
=
argd
[
"output_directory"
])
try
:
if
argd
[
"cancel_button"
]:
self
.
_redirect_to_page
(
req
,
self
.
_JOBS_URL
,
language
)
elif
argd
[
"save_button"
]:
perform_request_save_job
(
job
=
job
,
user_id
=
user_id
,
language
=
language
)
self
.
_redirect_to_page
(
req
,
self
.
_JOBS_URL
,
language
)
elif
argd
[
"edit_queries_button"
]:
result_job_id
=
perform_request_save_job
(
job
=
job
,
user_id
=
user_id
,
language
=
language
)
edit_queries_url
=
"
%s
?job_id=
%s
"
%
(
self
.
_JOB_QUERIES_URL
,
result_job_id
)
self
.
_redirect_to_page
(
req
,
edit_queries_url
,
language
)
elif
Job
.
ID_MISSING
==
job_id
:
title
=
_
(
"New Export Job"
)
body
=
perform_request_new_job
(
language
=
language
)
else
:
title
=
_
(
"Edit Export Job"
)
body
=
perform_request_edit_job
(
job_id
=
job_id
,
user_id
=
user_id
,
language
=
language
)
except
AccessDeniedError
:
self
.
_redirect_to_not_authorised_page
(
req
,
language
)
return
page
(
title
=
title
,
metaheaderadd
=
get_css
(),
body
=
body
,
req
=
req
,
navmenuid
=
"fieldexporter"
,
titleprologue
=
get_navigation_menu
(
language
),
navtrail
=
self
.
_NAVTRAIL_EXPORT
,
language
=
language
)
def
job_queries
(
self
,
req
,
form
):
"""Allows edition and manipulations of the queries of a job"""
argd
=
wash_urlargd
(
form
,
{
"new_button"
:
(
str
,
""
),
"run_button"
:
(
str
,
""
),
"delete_button"
:
(
str
,
""
),
"selected_queries"
:
(
list
,
""
),
"job_id"
:
(
int
,
-
1
)
})
# load the right message language
language
=
argd
[
"ln"
]
_
=
gettext_set_language
(
language
)
self
.
_check_user_credentials
(
req
,
language
)
user_id
=
self
.
_get_user_id
(
req
)
job_id
=
argd
[
"job_id"
]
try
:
# if the form is submitted through some of the buttons
# we should perform the appropriate action
if
argd
[
"new_button"
]:
new_query_url
=
"
%s
?job_id=
%s
"
%
(
self
.
_EDIT_QUERY_URL
,
job_id
)
self
.
_redirect_to_page
(
req
,
new_query_url
,
language
)
if
argd
[
"delete_button"
]:
query_ids
=
argd
[
"selected_queries"
]
perform_request_delete_queries
(
query_ids
=
query_ids
,
user_id
=
user_id
,
language
=
language
)
if
argd
[
"run_button"
]:
title
=
_
(
"Query Results"
)
query_ids
=
argd
[
"selected_queries"
]
body
=
perform_request_run_queries
(
query_ids
=
query_ids
,
user_id
=
user_id
,
job_id
=
job_id
,
language
=
language
)
else
:
title
=
_
(
"Export Job Queries"
)
body
=
perform_request_job_queries
(
job_id
=
job_id
,
user_id
=
user_id
,
language
=
language
)
except
AccessDeniedError
:
self
.
_redirect_to_not_authorised_page
(
req
,
language
)
return
page
(
title
=
title
,
metaheaderadd
=
get_css
(),
body
=
body
,
req
=
req
,
navmenuid
=
"fieldexporter"
,
titleprologue
=
get_navigation_menu
(
language
),
navtrail
=
self
.
_NAVTRAIL_EXPORT
,
language
=
language
)
def
edit_query
(
self
,
req
,
form
):
"""Edits an existing query or creates a new one"""
# Create an empty job and use its default values
# to init missing parameters
query
=
Query
()
name
=
query
.
get_name
()
output_fields
=
", "
.
join
(
query
.
get_output_fields
())
search_criteria
=
query
.
get_search_criteria
()
comment
=
query
.
get_comment
()
query_id
=
query
.
get_id
()
argd
=
wash_urlargd
(
form
,
{
"name"
:
(
str
,
name
),
"search_criteria"
:
(
str
,
search_criteria
),
"output_fields"
:
(
str
,
output_fields
),
"comment"
:
(
str
,
comment
),
"id"
:
(
int
,
query_id
),
"job_id"
:
(
int
,
Job
.
ID_MISSING
),
"save_button"
:
(
str
,
""
),
"cancel_button"
:
(
str
,
""
)
})
# load the right message language
language
=
argd
[
"ln"
]
_
=
gettext_set_language
(
language
)
self
.
_check_user_credentials
(
req
,
language
)
user_id
=
self
.
_get_user_id
(
req
)
query_id
=
argd
[
"id"
]
job_id
=
argd
[
"job_id"
]
current_job_queries_url
=
"
%s
?job_id=
%s
"
%
(
self
.
_JOB_QUERIES_URL
,
job_id
)
try
:
if
argd
[
"cancel_button"
]:
self
.
_redirect_to_page
(
req
,
current_job_queries_url
,
language
)
elif
argd
[
"save_button"
]:
name
=
argd
[
"name"
]
search_criteria
=
argd
[
"search_criteria"
]
comment
=
argd
[
"comment"
]
# split the text entered by user to different fields
outoutput_fields_text
=
argd
[
"output_fields"
]
re_split_pattern
=
re
.
compile
(
r',\s*'
)
output_fields
=
re_split_pattern
.
split
(
outoutput_fields_text
)
query
=
Query
(
query_id
,
name
,
search_criteria
,
comment
,
output_fields
)
perform_request_save_query
(
query
=
query
,
job_id
=
job_id
,
user_id
=
user_id
,
language
=
language
)
self
.
_redirect_to_page
(
req
,
current_job_queries_url
,
language
)
elif
Query
.
ID_MISSING
==
query_id
:
title
=
_
(
"New Query"
)
body
=
perform_request_new_query
(
job_id
=
job_id
,
user_id
=
user_id
,
language
=
language
)
else
:
title
=
_
(
"Edit Query"
)
body
=
perform_request_edit_query
(
query_id
=
query_id
,
job_id
=
job_id
,
user_id
=
user_id
,
language
=
language
)
except
AccessDeniedError
:
self
.
_redirect_to_not_authorised_page
(
req
,
language
)
return
page
(
title
=
title
,
metaheaderadd
=
get_css
(),
body
=
body
,
req
=
req
,
navmenuid
=
"fieldexporter"
,
titleprologue
=
get_navigation_menu
(
language
),
navtrail
=
self
.
_NAVTRAIL_EXPORT
,
language
=
language
)
def
job_results
(
self
,
req
,
form
):
"""Displays information about the results of a job"""
argd
=
wash_urlargd
(
form
,
{
"result_id"
:
(
int
,
-
1
)
})
# load the right message language
language
=
argd
[
"ln"
]
_
=
gettext_set_language
(
language
)
self
.
_check_user_credentials
(
req
,
language
)
user_id
=
self
.
_get_user_id
(
req
)
job_result_id
=
argd
[
"result_id"
]
title
=
_
(
"Export Job Results"
)
try
:
body
=
perform_request_job_results
(
job_result_id
=
job_result_id
,
user_id
=
user_id
,
language
=
language
)
except
AccessDeniedError
:
self
.
_redirect_to_not_authorised_page
(
req
,
language
)
return
page
(
title
=
title
,
metaheaderadd
=
get_css
(),
body
=
body
,
req
=
req
,
navmenuid
=
"fieldexporter"
,
titleprologue
=
get_navigation_menu
(
language
),
navtrail
=
self
.
_NAVTRAIL_EXPORT
,
language
=
language
)
def
display_job_result
(
self
,
req
,
form
):
"""Displays the results of a job"""
argd
=
wash_urlargd
(
form
,
{
"result_id"
:
(
int
,
JobResult
.
ID_MISSING
),
"output_format"
:
(
int
,
Job
.
OUTPUT_FORMAT_MISSING
)
})
# load the right message language
language
=
argd
[
"ln"
]
_
=
gettext_set_language
(
language
)
self
.
_check_user_credentials
(
req
,
language
)
user_id
=
self
.
_get_user_id
(
req
)
job_result_id
=
argd
[
"result_id"
]
output_format
=
argd
[
"output_format"
]
title
=
_
(
"Export Job Result"
)
try
:
body
=
perform_request_display_job_result
(
job_result_id
=
job_result_id
,
output_format
=
output_format
,
user_id
=
user_id
,
language
=
language
)
except
AccessDeniedError
:
self
.
_redirect_to_not_authorised_page
(
req
,
language
)
return
page
(
title
=
title
,
metaheaderadd
=
get_css
(),
body
=
body
,
req
=
req
,
navmenuid
=
"fieldexporter"
,
titleprologue
=
get_navigation_menu
(
language
),
navtrail
=
self
.
_NAVTRAIL_EXPORT
,
language
=
language
)
def
download_job_result
(
self
,
req
,
form
):
"""Returns to the browser zip file containing the job result"""
argd
=
wash_urlargd
(
form
,
{
"result_id"
:
(
int
,
JobResult
.
ID_MISSING
),
"output_format"
:
(
int
,
Job
.
OUTPUT_FORMAT_MISSING
)
})
# load the right message language
language
=
argd
[
"ln"
]
job_result_id
=
argd
[
"result_id"
]
output_format
=
argd
[
"output_format"
]
user_id
=
self
.
_get_user_id
(
req
)
_
=
gettext_set_language
(
language
)
self
.
_check_user_credentials
(
req
,
language
)
title
=
_
(
"Export Job Result"
)
try
:
perform_request_download_job_result
(
req
=
req
,
job_result_id
=
job_result_id
,
output_format
=
output_format
,
user_id
=
user_id
,
language
=
language
)
except
AccessDeniedError
:
self
.
_redirect_to_not_authorised_page
(
req
,
language
)
def
history
(
self
,
req
,
form
):
"""Displays history of the jobs"""
argd
=
wash_urlargd
(
form
,
{})
# load the right message language
language
=
argd
[
"ln"
]
_
=
gettext_set_language
(
language
)
self
.
_check_user_credentials
(
req
,
language
)
title
=
_
(
"Export Job History"
)
user_id
=
self
.
_get_user_id
(
req
)
body
=
perform_request_job_history
(
user_id
,
language
)
return
page
(
title
=
title
,
metaheaderadd
=
get_css
(),
body
=
body
,
req
=
req
,
navmenuid
=
"fieldexporter"
,
titleprologue
=
get_navigation_menu
(
language
),
navtrail
=
self
.
_NAVTRAIL_EXPORT
,
language
=
language
)
def
not_authorized
(
self
,
req
,
form
):
"""Displays page telling the user that
he is not authorised to access the resource"""
argd
=
wash_urlargd
(
form
,
{})
# load the right message language
language
=
argd
[
"ln"
]
_
=
gettext_set_language
(
language
)
text
=
_
(
"You are not authorised to access this resource."
)
return
page_not_authorized
(
req
=
req
,
ln
=
language
,
text
=
text
)
def
_check_user_credentials
(
self
,
req
,
language
):
"""Check if the user is allowed to use field exporter
@param req: request as received from apache
@param language: the language of the page
"""
user_info
=
collect_user_info
(
req
)
#redirect guests to login page
if
"1"
==
user_info
[
"guest"
]:
referer_url
=
"
%s
?ln=
%s
"
%
(
self
.
_EXPORT_URL
,
language
)
redirect_url
=
"
%s
?ln=
%s
&referer=
%s
"
%
(
self
.
_LOGIN_URL
,
language
,
referer_url
)
redirect_to_url
(
req
,
redirect_url
)
#redirect unauthorized user to not_authorized page
(
auth_code
,
auth_msg
)
=
acc_authorize_action
(
user_info
,
'cfgbibexport'
)
if
0
!=
auth_code
:
self
.
_redirect_to_not_authorised_page
(
req
,
language
)
def
_redirect_to_not_authorised_page
(
self
,
req
,
language
):
"""Redirects user to page telling him that he is not
authorised to do something
@param req: request as received from apache
@param language: the language of the page
"""
self
.
_redirect_to_page
(
req
,
self
.
_NOT_AUTHORIZED_URL
,
language
)
def
_redirect_to_page
(
self
,
req
,
url
,
language
):
"""Redirects user to a page with the given URL
and language.
@param req: request as received from apache
@param language: the language of the page
@param url: url to redirect to
"""
# check which symbol to use for appending the parameters
# if this is the only parameter use ?
if
-
1
==
url
.
find
(
"?"
):
append_symbol
=
"?"
# if there are other parameters already appended, use &
else
:
append_symbol
=
"&"
redirect_url
=
"
%s%s
ln=
%s
"
%
(
url
,
append_symbol
,
language
)
redirect_to_url
(
req
,
redirect_url
)
def
_get_user_id
(
self
,
req
):
"""Return the identifier of the currently loged user.
@param req: request as received from apache
@return: identifier of currently logged user
"""
user_info
=
collect_user_info
(
req
)
user_id
=
user_info
[
'uid'
]
return
user_id
Event Timeline
Log In to Comment