-
Notifications
You must be signed in to change notification settings - Fork 0
/
mrp.py
55 lines (37 loc) · 1.27 KB
/
mrp.py
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
import os
import luigi
import attr
from luigi.configuration.core import add_config_path
from luigi.util import requires
from tasks.base import ParseJavaScript, GzipToFtp, BaseConfig
from tcomapi.common.javascript import parse_json_from_js
from tcomapi.common.utils import save_csvrows
from settings import CONFIG_DIR
@attr.s
class Row:
id = attr.ib(converter=lambda x: str(x))
value = attr.ib(converter=lambda x: str(x))
date = attr.ib(default='')
config_path = os.path.join(CONFIG_DIR, 'mrp.conf')
add_config_path(config_path)
class kgd_mrp(BaseConfig):
url = luigi.Parameter(default='')
pattern = luigi.Parameter(default='')
class MrpParse(ParseJavaScript):
pattern = luigi.Parameter(default='')
def run(self):
d = parse_json_from_js(self.url, self.pattern)
# wrap each row and get tuple
rows = [attr.astuple(Row(**_d)) for _d in d]
save_csvrows(self.output().path, rows)
@requires(MrpParse)
class GzipMrpToFtp(GzipToFtp):
pass
class Mrp(luigi.WrapperTask):
def requires(self):
return GzipMrpToFtp(url=kgd_mrp().url,
name=kgd_mrp().name(),
monthly=True,
pattern=kgd_mrp().pattern)
if __name__ == '__main__':
luigi.run()