#!/usr/bin/env python
#-*- coding: utf-8 -*-
###########################################################################
## ##
## Copyrights Etienne Chové <chove@crans.org> 2009 ##
## ##
## 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, either version 3 of the License, or ##
## (at your option) any later version. ##
## ##
## 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, see <http://www.gnu.org/licenses/>. ##
## ##
###########################################################################
from __future__ import print_function
from modules import OsmoseLog, download
from cStringIO import StringIO
import sys, os, fcntl, urllib, urllib2, traceback
try:
import poster.encode
import poster.streaminghttp
poster.streaminghttp.register_openers()
has_poster_lib = True
except:
has_poster_lib = False
import psycopg2
import osmose_config as config
import inspect
import fileinput
import shutil
import datetime
import dateutil.parser
import socket
import subprocess
import time
#proxy_support = urllib2.ProxyHandler()
#print proxy_support.proxies
#opener = urllib2.build_opener(proxy_support)
#urllib2.install_opener(opener)
###########################################################################
## fonctions utiles
def get_pstree(pid=os.getpid()):
tree = []
while os.path.isdir("/proc/%d"%pid):
tree.append((pid, open("/proc/%d/cmdline"%pid).read().replace('\x00', ' ').strip()))
pid = int(open("/proc/%d/stat"%pid).read().split(" ")[3])
tree.reverse()
return tree
class lockfile:
def __init__(self, filename):
#return
self.fn = filename
try:
olddata = open(self.fn, "r").read()
except:
olddata = ""
try:
self.fd = open(self.fn, "w")
for l in get_pstree():
self.fd.write("%6d %s\n"%l)
self.fd.flush()
fcntl.flock(self.fd, fcntl.LOCK_NB|fcntl.LOCK_EX)
except:
#restore old data
self.fd.close()
open(self.fn, "w").write(olddata)
raise
self.ok = True
def __del__(self):
#return
if "fd" in dir(self):
try:
fcntl.flock(self.fd, fcntl.LOCK_NB|fcntl.LOCK_UN)
self.fd.close()
except:
pass
if "fn" in dir(self) and "ok" in dir(self):
try:
os.remove(self.fn)
except:
pass
class analyser_config:
pass
def get_version():
cmd = ["git", "describe"]
try:
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE)
version = proc.stdout.readlines()[0].strip()
except:
version = "(unknown)"
return version
def set_pgsql_schema(conf, logger, reset=False):
if reset:
db_schema = '"$user"'
elif conf.db_schema:
db_schema = conf.db_schema
else:
db_schema = conf.country
logger.log("set pgsql schema to %s" % db_schema)
cmd = ["psql"]
cmd += conf.db_psql_args
cmd += ["-c", "ALTER ROLE %s IN DATABASE %s SET search_path = %s,public;" % (conf.db_user, conf.db_base, db_schema)]
logger.execute_out(cmd)
###########################################################################
def lock_osmosis_database(logger):
osmosis_lock = False
for trial in xrange(60):
# acquire lock
try:
lfil = "/tmp/osmose-osmosis_import"
osmosis_lock = lockfile(lfil)
break
except:
logger.log(logger.log_av_r + "can't lock %s" % lfil + logger.log_ap)
logger.log("waiting 2 minutes")
time.sleep(2*60)
if not osmosis_lock:
logger.log(logger.log_av_r + "definitively can't lock" + logger.log_ap)
raise
return osmosis_lock
def check_database(conf, logger):
if "osmosis" in conf.download:
# check if database contains all necessary extensions
logger.sub().log("check database")
gisconn = psycopg2.connect(conf.db_string)
giscurs = gisconn.cursor()
for extension in ["hstore", "fuzzystrmatch", "unaccent"]:
giscurs.execute("""SELECT installed_version FROM pg_available_extensions
WHERE name = %s""",
[extension])
if giscurs.rowcount != 1 or giscurs.fetchone()[0] == None:
logger.log(logger.log_av_r+u"missing extension: "+extension+logger.log_ap)
return False
for table in ["geometry_columns", "spatial_ref_sys"]:
giscurs.execute("""SELECT tablename FROM pg_tables
WHERE tablename = %s""",
[table])
if giscurs.rowcount != 1:
# On PostGIS 2.0, geometry_columns has been moved to a view
giscurs.execute("""SELECT viewname FROM pg_views
WHERE viewname = %s""",
[table])
if giscurs.rowcount != 1:
logger.log(logger.log_av_r+u"missing table: "+table+logger.log_ap)
return False
else:
# No need to check permissions for views
continue
for perm in ["select", "update", "delete"]:
giscurs.execute("SELECT has_table_privilege(%s, %s)",
[table, perm])
if giscurs.fetchone()[0] == False:
logger.log(logger.log_av_r+u"missing permission %s on table: %s" % (perm, table)+logger.log_ap)
return False
giscurs.close()
gisconn.close()
return True
def init_database(conf, logger):
# import osmosis
if "osmosis" in conf.download:
osmosis_lock = lock_osmosis_database(logger)
set_pgsql_schema(conf, logger, reset=True)
# drop schema if present - might be remaining from a previous failing import
logger.sub().log("DROP SCHEMA %s" % conf.download["osmosis"])
gisconn = psycopg2.connect(conf.db_string)
giscurs = gisconn.cursor()
sql = "DROP SCHEMA IF EXISTS %s CASCADE;" % conf.download["osmosis"]
giscurs.execute(sql)
gisconn.commit()
giscurs.close()
gisconn.close()
# schema
logger.log(logger.log_av_r+"import osmosis schema"+logger.log_ap)
for script in conf.osmosis_pre_scripts:
cmd = ["psql"]
cmd += conf.db_psql_args
cmd += ["-f", script]
logger.execute_out(cmd)
# data
logger.log(logger.log_av_r+"import osmosis data"+logger.log_ap)
cmd = [conf.bin_osmosis]
dst_ext = os.path.splitext(conf.download["dst"])[1]
dir_country_tmp = os.path.join(conf.dir_tmp, conf.download["osmosis"])
shutil.rmtree(dir_country_tmp, ignore_errors=True)
os.makedirs(dir_country_tmp)
if dst_ext == ".pbf":
cmd += ["--read-pbf", "file=%s" % conf.download["dst"]]
else:
cmd += ["--read-xml", "file=%s" % conf.download["dst"]]
cmd += ["-quiet"]
cmd += ["--write-pgsql-dump", "directory=%s"%dir_country_tmp, "enableLinestringBuilder=yes"]
logger.execute_err(cmd)
for script in conf.osmosis_import_scripts:
cmd = ["psql"]
cmd += conf.db_psql_args
cmd += ["-f", script]
logger.execute_out(cmd, cwd=dir_country_tmp)
shutil.rmtree(dir_country_tmp, ignore_errors=True)
# post import scripts
logger.log(logger.log_av_r+"import osmosis post scripts"+logger.log_ap)
for script in conf.osmosis_post_scripts:
cmd = ["psql"]
cmd += conf.db_psql_args
cmd += ["-f", script]
logger.execute_out(cmd)
# rename table
logger.log(logger.log_av_r+"rename osmosis tables"+logger.log_ap)
gisconn = psycopg2.connect(conf.db_string)
giscurs = gisconn.cursor()
giscurs.execute("DROP SCHEMA IF EXISTS %s CASCADE" % conf.download["osmosis"])
giscurs.execute("CREATE SCHEMA %s" % conf.download["osmosis"])
for t in ["nodes", "ways", "way_nodes", "relations", "relation_members", "users", "schema_info"]:
sql = "ALTER TABLE %s SET SCHEMA %s;" % (t, conf.download["osmosis"])
giscurs.execute(sql)
gisconn.commit()
giscurs.close()
gisconn.close()
# free lock
del osmosis_lock
def clean_database(conf, logger, no_clean):
if "osmosis" in conf.download:
gisconn = psycopg2.connect(conf.db_string)
giscurs = gisconn.cursor()
if no_clean:
# grant read-only access to everybody
logger.sub().log("GRANT USAGE %s" % conf.download["osmosis"])
sql = "GRANT USAGE ON SCHEMA %s TO public" % conf.download["osmosis"]
logger.sub().log(sql)
giscurs.execute(sql)
for t in ["nodes", "ways", "way_nodes", "relations", "relation_members", "users", "schema_info"]:
sql = "GRANT SELECT ON %s.%s TO public" % (conf.download["osmosis"], t)
logger.sub().log(sql)
giscurs.execute(sql)
else:
# drop all tables
logger.sub().log("DROP SCHEMA %s" % conf.download["osmosis"])
sql = "DROP SCHEMA IF EXISTS %s CASCADE;" % conf.download["osmosis"]
logger.sub().log(sql)
giscurs.execute(sql)
gisconn.commit()
giscurs.close()
gisconn.close()
###########################################################################
def check_osmosis_diff(conf, logger):
logger.log("check osmosis replication")
diff_path = conf.download["diff_path"]
if not os.path.exists(diff_path):
return False
for f_name in ["configuration.txt", "download.lock", "state.txt"]:
f = os.path.join(diff_path, f_name)
if not os.path.exists(f):
return False
return True
def init_osmosis_diff(conf, logger):
logger.log(logger.log_av_r+"init osmosis replication for diff"+logger.log_ap)
diff_path = conf.download["diff_path"]
for f_name in ["configuration.txt", "download.lock", "state.txt"]:
f = os.path.join(diff_path, f_name)
if os.path.exists(f):
os.remove(f)
cmd = [conf.bin_osmosis]
cmd += ["--read-replication-interval-init", "workingDirectory=%s" % diff_path]
cmd += ["-quiet"]
logger.execute_err(cmd)
for line in fileinput.input(os.path.join(diff_path, "configuration.txt"), inplace=1):
if line.startswith("baseUrl"):
sys.stdout.write("baseUrl=" + conf.download["diff"])
elif line.startswith("maxInterval"):
if "geofabrik" in conf.download["diff"]:
# on daily diffs provided by Geofabrik, we should apply only one diff at a time
sys.stdout.write("maxInterval=" + str(60*60*24/2)) # 1/2 day at most
else:
sys.stdout.write("maxInterval=" + str(7*60*60*24)) # 7 day at most
else:
sys.stdout.write(line)
fileinput.close()
if conf.download["diff"].endswith("minute/"):
from modules import OsmTs
OsmTs.run(conf.download["dst"],
os.path.join(diff_path, "state.txt"),
"minute", logger)
else:
download.dl(conf.download["diff"] + "state.txt",
os.path.join(diff_path, "state.txt"),
logger.sub(),
min_file_size=10)
def run_osmosis_diff(conf, logger):
logger.log(logger.log_av_r+"run osmosis replication"+logger.log_ap)
diff_path = conf.download["diff_path"]
xml_change = os.path.join(diff_path, "change.osc.gz")
tmp_pbf_file = conf.download["dst"] + ".tmp"
shutil.copyfile(os.path.join(diff_path, "state.txt"),
os.path.join(diff_path, "state.txt.old"))
try:
prev_state_ts = None
is_uptodate = False
nb_iter = 0
with open(os.path.join(diff_path, "state.txt"), 'r') as f:
state_lines = f.readlines()
for line in state_lines:
print("state: ", line, end=' ')
if line.startswith("timestamp="):
s = line.translate(None, "\\")
state_ts = dateutil.parser.parse(s[len("timestamp="):]).replace(tzinfo=None)
cur_ts = datetime.datetime.today()
if state_ts < (cur_ts - datetime.timedelta(days=10)):
# Skip updates, and directly download .pbf file if extract is too old
logger.log(logger.log_av_r + "stop updates, to download full extract" + logger.log_ap)
return (False, None)
while not is_uptodate and nb_iter < 30:
nb_iter += 1
logger.log("iteration=%d" % nb_iter)
try:
cmd = [conf.bin_osmosis]
cmd += ["--read-replication-interval", "workingDirectory=%s" % diff_path]
cmd += ["--simplify-change", "--write-xml-change", "file=%s" % xml_change]
cmd += ["-quiet"]
logger.execute_err(cmd)
except:
logger.log("waiting 2 minutes")
time.sleep(2*60)
continue
cmd = [conf.bin_osmosis]
cmd += ["--read-xml-change", "file=%s" % xml_change]
cmd += ["--read-pbf", "file=%s" % conf.download["dst"] ]
cmd += ["--apply-change", "--buffer"]
cmd += ["--write-pbf", "file=%s" % tmp_pbf_file]
cmd += ["-quiet"]
logger.execute_err(cmd)
shutil.move(tmp_pbf_file, conf.download["dst"])
# find if state.txt is more recent than one day
with open(os.path.join(diff_path, "state.txt"), 'r') as f:
state_lines = f.readlines()
for line in state_lines:
print("state: ", nb_iter, " - ", line, end=' ')
if line.startswith("timestamp="):
s = line.translate(None, "\\")
state_ts = dateutil.parser.parse(s[len("timestamp="):]).replace(tzinfo=None)
cur_ts = datetime.datetime.today()
if prev_state_ts != None:
print(" ", prev_state_ts - state_ts)
if state_ts > (cur_ts - datetime.timedelta(days=1)):
is_uptodate = True
elif prev_state_ts == state_ts:
is_uptodate = True
else:
prev_state_ts = state_ts
if not is_uptodate:
# we didn't get the latest version of the pbf file
logger.log(logger.log_av_r + "didn't get latest version of osm file" + logger.log_ap)
return (False, None)
elif nb_iter == 1:
return (True, xml_change)
else:
# TODO: we should return a merge of all xml change files
return (True, None)
except:
logger.log(logger.log_av_r+"got error, aborting"+logger.log_ap)
shutil.copyfile(os.path.join(diff_path, "state.txt.old"),
os.path.join(diff_path, "state.txt"))
raise
###########################################################################
def check_osmosis_change(conf, logger):
if not check_osmosis_diff(conf, logger):
return False
logger.log("check osmosis replication for database")
return True
def init_osmosis_change(conf, logger):
init_osmosis_diff(conf, logger)
logger.log(logger.log_av_r+"import osmosis change post scripts"+logger.log_ap)
set_pgsql_schema(conf, logger)
for script in conf.osmosis_change_init_post_scripts:
cmd = ["psql"]
cmd += conf.db_psql_args
cmd += ["-f", script]
logger.execute_out(cmd)
set_pgsql_schema(conf, logger, reset=True)
def run_osmosis_change(conf, logger):
logger.log(logger.log_av_r+"run osmosis replication"+logger.log_ap)
diff_path = conf.download["diff_path"]
xml_change = os.path.join(diff_path, "change.osc.gz")
shutil.copyfile(os.path.join(diff_path, "state.txt"),
os.path.join(diff_path, "state.txt.old"))
try:
osmosis_lock = lock_osmosis_database(logger)
set_pgsql_schema(conf, logger)
cmd = [conf.bin_osmosis]
cmd += ["--read-replication-interval", "workingDirectory=%s" % diff_path]
cmd += ["--simplify-change", "--write-xml-change", "file=%s" % xml_change]
cmd += ["-quiet"]
logger.execute_err(cmd)
cmd = ["psql"]
cmd += conf.db_psql_args
cmd += ["-c", "TRUNCATE TABLE actions"]
logger.execute_out(cmd)
cmd = [conf.bin_osmosis]
cmd += ["--read-xml-change", xml_change]
cmd += ["--write-pgsql-change", "database=%s"%conf.db_base, "user=%s"%conf.db_user, "password=%s"%conf.db_password]
cmd += ["-quiet"]
logger.execute_err(cmd)
logger.log(logger.log_av_r+"import osmosis change post scripts"+logger.log_ap)
for script in conf.osmosis_change_post_scripts:
logger.log(script)
cmd = ["psql"]
cmd += conf.db_psql_args
cmd += ["-f", script]
logger.execute_out(cmd)
set_pgsql_schema(conf, logger, reset=True)
del osmosis_lock
return xml_change
except:
logger.log(logger.log_av_r+"got error, aborting"+logger.log_ap)
shutil.copyfile(os.path.join(diff_path, "state.txt.old"),
os.path.join(diff_path, "state.txt"))
raise
###########################################################################
def run(conf, logger, options):
err_code = 0
country = conf.country
try:
version = get_version()
except:
version = None
if not check_database(conf, logger):
logger.log(logger.log_av_r+u"error in database initialisation"+logger.log_ap)
return 0x10
##########################################################################
## check for working dirs and creates when needed
dirs = [conf.dir_tmp, conf.dir_cache, conf.dir_results, conf.dir_extracts, conf.dir_diffs]
if "diff_path" in conf.download:
dirs.append(conf.download["diff_path"])
for i in dirs:
if not os.path.exists(i):
try:
os.makedirs(i)
except OSError as e:
sys.exit("%s\nCheck 'dir_work' in modules/config.py and its permissions" % str(e))
# variable used by osmosis
if not "JAVACMD_OPTIONS" in os.environ:
os.environ["JAVACMD_OPTIONS"] = ""
os.environ["JAVACMD_OPTIONS"] += " -Djava.io.tmpdir="+conf.dir_tmp
os.environ["JAVACMD_OPTIONS"] += " -Duser.timezone=GMT"
##########################################################################
## download and create database
if options.skip_init:
pass
elif options.change and check_osmosis_change(conf, logger) and not options.change_init:
xml_change = run_osmosis_change(conf, logger)
elif "url" in conf.download:
newer = False
xml_change = None
updated = False # set if extract was updated instead of fully downloaded
if options.diff and check_osmosis_diff(conf, logger) and os.path.exists(conf.download["dst"]):
(status, xml_change) = run_osmosis_diff(conf, logger)
if status:
newer = True
updated = True
if not newer and options.skip_download:
logger.sub().log("skip download")
newer = True
if not newer:
logger.log(logger.log_av_r+u"downloading"+logger.log_ap)
newer = download.dl(conf.download["url"], conf.download["dst"], logger.sub(),
min_file_size=8*1024)
updated = False
if not newer:
return 0
init_database(conf, logger)
if options.change:
init_osmosis_change(conf, logger)
elif options.diff and not updated:
init_osmosis_diff(conf, logger)
if hasattr(conf, "sql_post_scripts"):
logger.log(logger.log_av_r+"import post scripts"+logger.log_ap)
for script in conf.sql_post_scripts:
cmd = ["psql"]
cmd += conf.db_psql_args
cmd += ["-f", script]
logger.execute_out(cmd)
##########################################################################
## analyses
country_timestamp = None
for analyser, password in conf.analyser.iteritems():
logger.log(logger.log_av_r + country + " : " + analyser + logger.log_ap)
if not "analyser_" + analyser in analysers:
logger.sub().log("skipped")
continue
if password == "xxx":
logger.sub().log("code is not correct - won't upload to %s" % conf.updt_url)
elif not conf.results_url and not has_poster_lib:
logger.sub().log("results_url is not correct - won't upload to %s" % conf.updt_url)
try:
analyser_conf = analyser_config()
analyser_conf.dst_dir = conf.dir_results
analyser_conf.db_string = conf.db_string
analyser_conf.db_user = conf.db_user
if conf.db_schema:
analyser_conf.db_schema = conf.db_schema
else:
analyser_conf.db_schema = country
analyser_conf.dir_scripts = conf.dir_scripts
analyser_conf.options = conf.analyser_options
analyser_conf.polygon_id = conf.polygon_id
if options.change and xml_change:
analyser_conf.src = xml_change
elif "dst" in conf.download:
analyser_conf.src = conf.download["dst"]
lunched_analyser = []
lunched_analyser_change = []
for name, obj in inspect.getmembers(analysers["analyser_" + analyser]):
if (inspect.isclass(obj) and obj.__module__ == "analyser_" + analyser and
(name.startswith("Analyser") or name.startswith("analyser"))):
# analyse
analyser_conf.dst_file = name + "-" + country + ".xml"
analyser_conf.dst_file += ".bz2"
analyser_conf.dst = os.path.join(conf.dir_results, analyser_conf.dst_file)
analyser_conf.version = version
analyser_conf.verbose = options.verbose
analyser_conf.timestamp = country_timestamp
with obj(analyser_conf, logger.sub()) as analyser_obj:
if not options.change or not xml_change:
analyser_obj.analyser()
lunched_analyser.append(analyser_obj)
else:
analyser_obj.analyser_change()
lunched_analyser_change.append(analyser_obj)
country_timestamp = analyser_obj.config.timestamp
# update
if (conf.results_url or has_poster_lib) and password != "xxx":
logger.sub().log("update")
if analyser in conf.analyser_updt_url:
list_urls = conf.analyser_updt_url[analyser]
else:
list_urls = [conf.updt_url]
for url in list_urls:
update_finished = False
nb_iter = 0
while not update_finished and nb_iter < 3:
time.sleep(nb_iter * 15)
nb_iter += 1
logger.sub().sub().log("iteration=%d" % nb_iter)
try:
tmp_src = "%s-%s" % (analyser, country)
if has_poster_lib:
(tmp_dat, tmp_headers) = poster.encode.multipart_encode(
{"content": open(analyser_conf.dst, "rb"),
"source": tmp_src,
"code": password})
tmp_req = urllib2.Request(url, tmp_dat, tmp_headers)
fd = urllib2.urlopen(tmp_req, timeout=1800)
else:
tmp_req = urllib2.Request(url)
tmp_url = os.path.join(conf.results_url, analyser_conf.dst_file)
tmp_dat = urllib.urlencode([('url', tmp_url),
('source', tmp_src),
('code', password)])
fd = urllib2.urlopen(tmp_req, tmp_dat, timeout=1800)
dt = fd.read().decode("utf8").strip()
if dt[-2:] != "OK":
sys.stderr.write((u"UPDATE ERROR %s/%s : %s\n"%(country, analyser, dt)).encode("utf8"))
err_code |= 4
else:
logger.sub().sub().log(dt)
update_finished = True
except socket.timeout:
logger.sub().sub().sub().log("got a timeout")
pass
except:
s = StringIO()
traceback.print_exc(file=s)
logger.sub().log("error on update...")
for l in s.getvalue().decode("utf8").split("\n"):
logger.sub().sub().log(l)
if not update_finished:
err_code |= 1
except:
s = StringIO()
traceback.print_exc(file=s)
logger.sub().log("error on analyse...")
for l in s.getvalue().decode("utf8").split("\n"):
logger.sub().sub().log(l)
err_code |= 2
continue
finally:
if not options.no_clean:
for obj in lunched_analyser:
with obj as o:
o.analyser_clean()
for obj in lunched_analyser_change:
with obj as o:
o.analyser_change_clean()
##########################################################################
## vidange
logger.log(logger.log_av_r + u"cleaning : " + country + logger.log_ap)
if options.change:
pass
else:
clean_database(conf, logger, options.no_clean or not conf.clean_at_end)
if options.diff:
# don't erase any file
return err_code
# remove files
if "url" in conf.download and "dst" in conf.download and not options.no_clean:
f = ".osm".join(conf.download["dst"].split(".osm")[:-1])
for ext in ["osm", "osm.bz2", "osm.pbf"]:
try:
os.remove("%s.%s"%(f, ext))
logger.sub().log("DROP FILE %s.%s"%(f, ext))
except:
pass
return err_code
###########################################################################
if __name__ == "__main__":
err_code = 0
#=====================================
# analyse des arguments
from optparse import OptionParser
parser = OptionParser()
parser.add_option("--verbose", dest="verbose", action="store_true",
help="Verbose mode")
parser.add_option("--list-analyser", dest="list_analyser", action="store_true",
help="List all available analysers")
parser.add_option("--list-country", dest="list_country", action="store_true",
help="List all available countries")
parser.add_option("--country", dest="country", action="append",
help="Country to analyse (can be repeated)")
parser.add_option("--analyser", dest="analyser", action="append",
help="Analyser to run (can be repeated)")
parser.add_option("--change", dest="change", action="store_true",
help="Run analyser on change mode when available")
parser.add_option("--change_init", dest="change_init", action="store_true",
help="Initialize database for change mode")
parser.add_option("--skip-download", dest="skip_download", action="store_true",
help="Don't download extract")
parser.add_option("--skip-init", dest="skip_init", action="store_true",
help="Don't initialize database")
parser.add_option("--no-clean", dest="no_clean", action="store_true",
help="Don't remove extract and database after analyses")
parser.add_option("--cron", dest="cron", action="store_true",
help="Record output in a specific log")
parser.add_option("--version", dest="version", action="store_true",
help="Output version information and exit")
(options, args) = parser.parse_args()
analysers_path = os.path.join(os.path.dirname(__file__), "analysers")
if options.list_analyser:
for fn in sorted(os.listdir(analysers_path)):
if fn.startswith("analyser_") and fn.endswith(".py"):
print(fn[9:-3])
sys.exit(0)
if options.list_country:
for k in sorted(config.config.keys()):
print(k)
sys.exit(0)
if options.cron:
output = sys.stdout
logger = OsmoseLog.logger(output, False)
else:
output = sys.stdout
logger = OsmoseLog.logger(output, True)
if options.change_init and not options.change:
logger.log(logger.log_av_b+"--change must be specified "+fn[:-3]+logger.log_ap)
sys.exit(1)
if options.version:
print("osmose backend version: %s" % get_version())
sys.exit(0)
if not options.country:
parser.print_help()
sys.exit(1)
#=====================================
# chargement des analysers
old_path = list(sys.path)
sys.path.insert(0, analysers_path)
logger.log(logger.log_av_v+"loading analyses "+logger.log_ap)
analysers = {}
for fn in os.listdir(analysers_path):
if fn.startswith("analyser_") and fn.endswith(".py"):
if options.analyser and fn[:-3] not in options.analyser:
continue
logger.log(" load "+fn[:-3])
analysers[fn[:-3]] = __import__(fn[:-3])
if options.analyser:
count = 0
for k in options.analyser:
if k not in analysers:
logger.log(logger.log_av_b+"not found "+k+logger.log_ap)
count += 1
# user is passing only non-existent analysers
if len(options.analyser) == count:
sys.exit("No valid analysers specified")
sys.path[:] = old_path # restore previous path
#=====================================
# analyse
for country, country_conf in config.config.iteritems():
# filter
if options.country and country not in options.country:
continue
# acquire lock
try:
lfil = "/tmp/analyse-%s"%country
lock = lockfile(lfil)
except:
logger.log(logger.log_av_r+"can't lock %s"%country+logger.log_ap)
if options.cron:
sys.stderr.write("can't lock %s\n"%country)
for l in open(lfil).read().rstrip().split("\n"):
logger.log(" "+l)
if options.cron:
sys.stderr.write(" "+l+"\n")
if options.cron:
sys.stderr.flush()
err_code |= 0x80
continue
country_conf.init()
options.diff = not options.change and "diff" in country_conf.download
# analyse
err_code |= run(country_conf, logger, options)
# free lock
del lock
logger.log(logger.log_av_v+u"end of analyses"+logger.log_ap)
sys.exit(err_code)