Mercurial > feed-push
annotate feed-push @ 14:5c26fc1adbac draft
minor fix in automatic directory creation
Signed-off-by: Changaco <changaco ατ changaco δοτ net>
| author | Changaco <changaco ατ changaco δοτ net> |
|---|---|
| date | Mon, 16 Apr 2012 14:40:51 +0200 |
| parents | ddbbd4ee5d47 |
| children | 68a9b24a182a |
| rev | line source |
|---|---|
| 1 | 1 #!/usr/bin/env python2 |
| 2 # -*- coding: utf-8 -*- | |
| 3 | |
| 4 # This program is free software: you can redistribute it and/or modify | |
| 5 # it under the terms of the GNU General Public License as published by | |
| 6 # the Free Software Foundation, either version 3 of the License, or | |
| 7 # (at your option) any later version. | |
| 8 # | |
| 9 # This program is distributed in the hope that it will be useful, | |
| 10 # but WITHOUT ANY WARRANTY; without even the implied warranty of | |
| 11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | |
| 12 # GNU General Public License for more details. | |
| 13 # | |
| 14 # You should have received a copy of the GNU General Public License | |
| 15 # along with this program. If not, see <http://www.gnu.org/licenses/>. | |
| 16 | |
| 17 import argparse | |
| 18 import calendar | |
| 19 from functools import partial, reduce | |
| 20 from glob import glob | |
| 21 import json | |
| 22 import os | |
| 7 | 23 from os.path import abspath, dirname, isdir |
| 1 | 24 import shlex |
| 25 from subprocess import Popen, PIPE, STDOUT | |
| 26 import sys | |
| 27 from syslog import * | |
| 28 import time | |
| 29 | |
| 30 import feedparser | |
| 31 import gamin | |
| 32 | |
| 33 | |
| 34 # Constants | |
| 35 | |
| 36 log_levels = ['DEBUG', 'INFO', 'NOTICE', 'WARNING', 'ERR', 'CRIT', 'ALERT', 'EMERG'] | |
| 37 | |
| 38 gamin_events = { | |
| 39 1:'GAMChanged', 2:'GAMDeleted', 3:'GAMStartExecuting', 4:'GAMStopExecuting', | |
| 40 5:'GAMCreated', 6:'GAMMoved', 7:'GAMAcknowledge', 8:'GAMExists', 9:'GAMEndExist' | |
| 41 } | |
| 42 | |
| 43 | |
| 44 # Generic utils | |
| 45 | |
| 46 concat = lambda l: reduce(list.__add__, l, []) | |
| 47 | |
| 48 def dict_append(d, k, v): | |
| 49 d.setdefault(k, []) | |
| 50 d[k].append(v) | |
| 51 | |
| 52 | |
| 53 # Logging | |
| 54 | |
| 55 def log(*args): | |
| 56 if len(args) == 1: | |
| 57 priority, msg = LOG_INFO, args[0] | |
| 58 elif len(args) == 2: | |
| 59 priority, msg = args | |
| 60 else: | |
| 61 return | |
| 62 priority = 7 - priority | |
| 63 if priority < global_args.log_level: | |
| 64 return | |
| 65 if global_args.syslog: | |
| 66 syslog(*args) | |
| 67 else: | |
| 68 sys.stderr.write(log_levels[priority]+': '+msg+'\n') | |
| 69 | |
| 70 def ignore_event(path, event): | |
| 71 log(LOG_DEBUG, 'ignoring event '+gamin_events.get(event, str(event))+' on '+path) | |
| 72 | |
| 73 | |
| 74 # Config parsing | |
| 75 | |
| 76 def parse_config_file(config_fd): | |
| 77 feeds_paths = config_to_feed_paths_to_commands[config_fd.name] = {} | |
| 11 | 78 cmd = [] |
|
12
ddbbd4ee5d47
log watched/unwatched feed paths
Changaco <changaco ατ changaco δοτ net>
parents:
11
diff
changeset
|
79 log('parsing config file '+config_fd.name) |
| 1 | 80 config_fd.seek(0) |
| 81 for i, line in enumerate(config_fd): | |
| 82 line = line.strip() | |
| 83 if len(line) == 0 or line[0] == '#': | |
| 84 continue | |
| 85 if line[0] == '%': | |
| 11 | 86 if isinstance(cmd, str): |
| 87 cmd = [] | |
| 88 cmd.append(line[1:].rstrip(';')) | |
| 89 elif not cmd: | |
| 1 | 90 log(LOG_ERR, 'missing command in file '+config_fd.name+' before line '+str(i)) |
| 91 return | |
| 92 else: | |
| 11 | 93 cmd = '; '.join(cmd) |
| 1 | 94 for feed_path in glob(line): |
| 95 feed_path = abspath(feed_path) | |
| 96 dict_append(feeds_paths, feed_path, cmd) | |
| 97 if not feed_path in path_to_feed_fd: | |
| 98 monitor.watch_file(feed_path, handle_feed_change) | |
|
12
ddbbd4ee5d47
log watched/unwatched feed paths
Changaco <changaco ατ changaco δοτ net>
parents:
11
diff
changeset
|
99 log('now watching '+feed_path) |
| 1 | 100 |
| 101 | |
| 102 # Gamin callbacks | |
| 103 | |
| 104 def handle_config_change(path, event): | |
| 105 path = abspath(path) | |
| 106 if os.path.isdir(path): | |
| 107 ignore_event(path, event) | |
| 108 elif not path in path_to_config_fd: | |
| 109 open_config(path, event) | |
| 110 elif event in [gamin.GAMChanged, gamin.GAMDeleted]: | |
| 111 update_config(path, event) | |
| 112 else: | |
| 113 ignore_event(path, event) | |
| 114 | |
| 115 def open_config(path, event): | |
| 116 if event in [gamin.GAMCreated, gamin.GAMExists]: | |
| 117 if (not path.endswith('.conf') or path[0] == '.') and not hasattr(global_args.config, 'read'): | |
| 118 return log('ignoring '+path+' (not a valid config file name)') | |
| 119 try: | |
| 120 config_fd = open(path) | |
| 121 except IOError as e: | |
| 122 return log('failed to open "'+line+'" '+str(e)) | |
| 123 path_to_config_fd[path] = config_fd | |
| 124 parse_config_file(config_fd) | |
| 125 else: | |
| 126 ignore_event(path, event) | |
| 127 | |
| 128 def update_config(path, event): | |
|
10
89f581ebf4b2
fix update_config function
Changaco <changaco ατ changaco δοτ net>
parents:
7
diff
changeset
|
129 feeds_paths = set(concat(d.keys() for d in config_to_feed_paths_to_commands.values())) |
| 1 | 130 if event == gamin.GAMChanged: |
| 131 parse_config_file(path_to_config_fd[path]) | |
| 132 elif event == gamin.GAMDeleted: | |
| 133 log('removing actions from deleted config file '+config_fd.name) | |
| 134 config_to_feed_paths_to_commands.pop(path) | |
| 135 path_to_config_fd.pop(path).close() | |
|
10
89f581ebf4b2
fix update_config function
Changaco <changaco ατ changaco δοτ net>
parents:
7
diff
changeset
|
136 new_feeds_paths = set(concat(d.keys() for d in config_to_feed_paths_to_commands.values())) |
| 1 | 137 for feed_path in feeds_paths.difference(new_feeds_paths): |
| 138 monitor.stop_watch(feed_path) | |
|
12
ddbbd4ee5d47
log watched/unwatched feed paths
Changaco <changaco ατ changaco δοτ net>
parents:
11
diff
changeset
|
139 log('stopped watching '+feed_path) |
| 1 | 140 if feed_path in path_to_feed_fd: |
| 141 path_to_feed_fd.pop(feed_path).close() | |
| 142 | |
| 143 def handle_feed_change(path, event): | |
| 144 if path not in path_to_feed_fd: | |
| 145 if event in [gamin.GAMCreated, gamin.GAMExists, gamin.GAMChanged]: | |
| 146 try: | |
| 147 feed_fd = path_to_feed_fd[path] = open(path) | |
| 148 except IOError as e: | |
| 149 return log('failed to open "'+path+'": '+str(e)) | |
| 150 feed = feedparser.parse(feed_fd.read()) | |
| 151 handle_feed_change(path, gamin.GAMChanged) | |
| 152 else: | |
| 153 ignore_event(path, event) | |
| 154 elif event == gamin.GAMChanged: | |
| 155 feed_fd = path_to_feed_fd[path] | |
| 156 feed_fd.seek(0) | |
| 157 feed = feedparser.parse(feed_fd.read()) | |
| 158 for entry in reversed(feed.entries): | |
| 159 if entry.id in state['id_cache'].get(feed_fd.name, []) or \ | |
| 160 not global_args.flood and calendar.timegm(entry.published_parsed) < time.time() - 86400: | |
| 161 continue | |
| 162 for feed_path_to_commands in config_to_feed_paths_to_commands.values(): | |
| 163 for cmd in feed_path_to_commands.get(path, []): | |
| 164 run_command(format_cmd(cmd, feed=feed.feed, entry=entry), entry.content[0].value) | |
| 165 state['id_cache'][feed_fd.name] = [entry.id for entry in feed.entries] | |
| 166 save_state() | |
| 167 elif event == gamin.GAMDeleted: | |
| 168 path_to_feed_fd.pop(path).close() | |
| 169 else: | |
| 170 ignore_event(path, event) | |
| 171 | |
| 172 def save_state(): | |
| 173 global_args.state_file.truncate(0) | |
| 174 json.dump(state, global_args.state_file) | |
| 175 global_args.state_file.flush() | |
| 176 | |
| 177 | |
| 178 # Commands utils | |
| 179 | |
| 180 def format_cmd(cmd, **kwargs): | |
| 181 """The safe equivalent of str.format() for shell commands, meaning interpolated variables can't do shell injections (I hope).""" | |
| 182 r = u'' | |
| 183 for arg in shlex.split(cmd.encode('utf8')): | |
| 184 a = arg.decode('utf8') | |
| 185 b = a.format(**kwargs) | |
| 186 if a != b: | |
| 187 r += u" '" + b.replace(u"'", u'\'"\'"\'') + u"'" | |
| 188 else: | |
| 189 r += u' ' + arg | |
| 190 return r.lstrip() | |
| 191 | |
| 192 def run_command(cmd, input): | |
| 193 p = Popen(cmd, stdin=PIPE, stdout=PIPE, stderr=STDOUT, shell=True) | |
| 194 output = p.communicate(input.encode('utf8'))[0].decode('utf8') | |
| 195 if p.returncode != 0: | |
| 196 log(LOG_ERR, 'command failed: '+cmd+'\n'+output) | |
| 197 else: | |
| 198 log(LOG_INFO, 'successfully executed '+cmd) | |
| 199 log(LOG_DEBUG, '===== output:\n'+output) | |
| 200 | |
| 201 | |
| 202 # Argparse utils | |
| 203 | |
| 204 def AbsPath(next_type=None): | |
| 205 def f(s): | |
| 206 p = abspath(s) | |
| 207 if next_type is not None: | |
| 208 return next_type(p) | |
| 209 else: | |
| 210 return p | |
| 211 return f | |
| 212 | |
| 213 class Apply(argparse.Action): | |
| 214 def __init__(self, f, *args, **kwargs): | |
| 215 super(self.__class__, self).__init__(**kwargs) | |
| 216 self.f = f | |
| 217 def __call__(self, parser, namespace, values, option_string=None): | |
| 218 setattr(namespace, self.dest, self.f(values[0])) | |
| 219 | |
|
6
455cd8c78862
create state dir if it doesn't exist
Changaco <changaco ατ changaco δοτ net>
parents:
1
diff
changeset
|
220 def MakeDirs(next_type=None): |
|
455cd8c78862
create state dir if it doesn't exist
Changaco <changaco ατ changaco δοτ net>
parents:
1
diff
changeset
|
221 def f(s): |
| 7 | 222 d = dirname(s) |
|
14
5c26fc1adbac
minor fix in automatic directory creation
Changaco <changaco ατ changaco δοτ net>
parents:
12
diff
changeset
|
223 if d and not isdir(d): |
|
6
455cd8c78862
create state dir if it doesn't exist
Changaco <changaco ατ changaco δοτ net>
parents:
1
diff
changeset
|
224 try: |
|
455cd8c78862
create state dir if it doesn't exist
Changaco <changaco ατ changaco δοτ net>
parents:
1
diff
changeset
|
225 os.makedirs(d) |
|
455cd8c78862
create state dir if it doesn't exist
Changaco <changaco ατ changaco δοτ net>
parents:
1
diff
changeset
|
226 except OSError as e: |
|
455cd8c78862
create state dir if it doesn't exist
Changaco <changaco ατ changaco δοτ net>
parents:
1
diff
changeset
|
227 raise argparse.ArgumentTypeError(str(e)) |
|
455cd8c78862
create state dir if it doesn't exist
Changaco <changaco ατ changaco δοτ net>
parents:
1
diff
changeset
|
228 if next_type is not None: |
|
455cd8c78862
create state dir if it doesn't exist
Changaco <changaco ατ changaco δοτ net>
parents:
1
diff
changeset
|
229 return next_type(s) |
|
455cd8c78862
create state dir if it doesn't exist
Changaco <changaco ατ changaco δοτ net>
parents:
1
diff
changeset
|
230 else: |
|
455cd8c78862
create state dir if it doesn't exist
Changaco <changaco ατ changaco δοτ net>
parents:
1
diff
changeset
|
231 return s |
|
455cd8c78862
create state dir if it doesn't exist
Changaco <changaco ατ changaco δοτ net>
parents:
1
diff
changeset
|
232 return f |
|
455cd8c78862
create state dir if it doesn't exist
Changaco <changaco ατ changaco δοτ net>
parents:
1
diff
changeset
|
233 |
| 1 | 234 def Directory(s): |
| 235 try: | |
| 236 os.listdir(s) | |
| 237 return s | |
| 238 except OSError as e: | |
| 239 raise argparse.ArgumentTypeError(str(e)) | |
| 240 | |
| 241 def File(flags): | |
| 242 def f(s): | |
| 243 try: | |
| 244 return os.fdopen(os.open(s, flags), 'w') | |
| 245 except OSError as e: | |
| 246 raise argparse.ArgumentTypeError(str(e)) | |
| 247 return f | |
| 248 | |
| 249 class First(argparse.Action): | |
| 250 def __call__(self, parser, namespace, values, option_string=None): | |
| 251 setattr(namespace, self.dest, values[0]) | |
| 252 | |
| 253 def FirstOf(*types, **kwargs): | |
| 254 kwargs.setdefault('error', 'argument "{}" is not valid') | |
| 255 def f(s): | |
| 256 for t in types: | |
| 257 try: | |
| 258 return t(s) | |
| 259 except: | |
| 260 pass | |
| 261 raise argparse.ArgumentTypeError(error.format(s)) | |
| 262 return f | |
| 263 | |
| 264 | |
| 265 # Main | |
| 266 | |
| 267 if __name__ == '__main__': | |
| 268 | |
| 269 p = argparse.ArgumentParser() | |
| 270 p.add_argument('config', type=FirstOf(AbsPath(argparse.FileType('r')), AbsPath(Directory), error='"{}" is neither a file nor a directory'), help='either a file or a directory') | |
|
6
455cd8c78862
create state dir if it doesn't exist
Changaco <changaco ατ changaco δοτ net>
parents:
1
diff
changeset
|
271 p.add_argument('state_file', type=MakeDirs(argparse.FileType('a+')), help='e.g. /var/lib/feed-push/state') |
| 1 | 272 p.add_argument('--flood', default=False, action='store_true', help='push all articles on startup instead of ignoring the ones older than 24h (useful for debugging)') |
| 273 p.add_argument('--fork', metavar='pid-file', nargs=1, type=File(os.O_WRONLY|os.O_CREAT|os.O_EXCL), action=First, help='useful in init scripts') | |
| 274 p.add_argument('--log-level', nargs=1, default=1, choices=log_levels, action=partial(Apply, log_levels.index), help='default is INFO') | |
| 275 p.add_argument('--syslog', default=False, action='store_true', help='log to syslog instead of stderr') | |
| 276 global_args = p.parse_args() | |
| 277 | |
| 278 if global_args.fork: | |
| 279 pid = os.fork() | |
| 280 if pid != 0: | |
| 281 global_args.fork.write(str(pid)) | |
| 282 exit(0) | |
| 283 | |
| 284 if global_args.syslog: | |
| 285 openlog(facility=LOG_DAEMON) | |
| 286 | |
| 287 state = {'id_cache': {}} | |
| 288 saved_state = global_args.state_file.read().strip() | |
| 289 if len(saved_state) > 0: | |
| 290 state.update(json.loads(saved_state)) | |
| 291 del saved_state | |
| 292 | |
| 293 monitor = gamin.WatchMonitor() | |
| 294 path_to_feed_fd = {} | |
| 295 path_to_config_fd = {} | |
| 296 config_to_feed_paths_to_commands = {} | |
| 297 if hasattr(global_args.config, 'read'): | |
| 298 os.chdir(os.path.dirname(global_args.config.name)) | |
| 299 monitor.watch_file(global_args.config.name, handle_config_change) | |
| 300 else: | |
| 301 os.chdir(global_args.config) | |
| 302 monitor.watch_directory(global_args.config, handle_config_change) | |
| 303 | |
| 304 while True: | |
| 305 monitor.handle_one_event() |