-
Notifications
You must be signed in to change notification settings - Fork 0
/
commands.py
58 lines (46 loc) · 1.62 KB
/
commands.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
56
57
58
import json
import click
import requests
from core.infrastructure import PostgresThreadRepository
from core.usecases import create_message
@click.group()
def cli():
"""CLI entrypoint"""
@cli.group("data-gouv")
def data_gouv():
"""Handles Data Gouv usecases"""
@data_gouv.command("messages")
@click.option("-f", "--file", is_flag=True)
@click.option("-d", "--dump", is_flag=True)
def data_gouv_messages(file, dump):
"""Data Gouv threads and messages"""
repository = PostgresThreadRepository()
if not file:
url = (
"https://www.data.gouv.fr/api/1/organizations/ministere-de-leconomie-des-finances-et-de-la-"
"souverainete-industrielle-et-numerique/discussions"
)
response = requests.get(url)
source = response.json()
if dump:
source = json.dumps(response.json(), indent=2, ensure_ascii=False)
with open("data/discussions.json", "w") as file:
json.dump(response.json(), file, indent=2, ensure_ascii=False)
else:
with open("data/discussions.json", "r") as file:
source = json.load(file)
for discussion in source:
for message in discussion["discussion"]:
create_message(
repository=repository,
thread_id=discussion["id"],
author=message["posted_by"]["slug"],
content=message["content"],
posted_on=message["posted_on"],
)
@cli.group("data-eco")
def data_eco():
"""Handles Data Economie usecases"""
click.echo("Handle Data Economie usecases")
if __name__ == "__main__":
cli()